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

feat(invariants): scoped-dispatch carrier/subject checks and the setup-drives tripwire

Browse Source 
Three dev-mode invariants close the leak-by-default regression class at
runtime: (1) every scope-filtered event family must dispatch with a scope
carrier — a bare dispatch throws at the call site naming the carrier rule;
(2) where the subject is recoverable from the arguments (agent/*, the tool
pipeline, prompt assembly) the carrier's key must BE that subject, and an
assembly context must never carry agent without scope (use
assembleContextFor); (3) a turn/start logged before the owning agent's
agent/session-start is the setup-drives teaching error (setup registers the
scoped world, it never drives the agent).
This commit is contained in:
Tianyi Cui
2026-07-09 02:11:21 +08:00
parent 15f4d1cd03
commit 1ac7857349
5 changed files with 196 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
packages/support/invariants/package.json
View File

@@ -24,13 +24,19 @@
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^0.0.1",
"@deepseek-ai/dsh-scope": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
"@deepseek-ai/dsh-tools": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

90
packages/support/invariants/src/index.ts
View File

@@ -20,6 +20,9 @@
*/
import type { Context } from 'cordis'
import { carrierKeyOf, isScopeCarrier } from '@deepseek-ai/dsh-scope'
import type { AssembleContext } from '@deepseek-ai/dsh-system-prompt'
import type { ToolExecution } from '@deepseek-ai/dsh-tools'
import { HarnessError } from '@deepseek-ai/dsh-llm'
import type { CallId, GenerateOptions } from '@deepseek-ai/dsh-llm'
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
@@ -362,6 +365,93 @@ export function apply(ctx: Context, config: Config = {}): void {
lastStatus.set(agent, status)
})
// --- Scoped-dispatch invariants (the agent-scoping seam) ---------------
//
// Every scope-filtered event family must dispatch with a scope carrier
// (scopeTarget) whose key IS the subject the event's arguments name —
// a dispatch without one silently reverts that event to global delivery
// (agent-scoped listeners over-hear foreign agents), and a mis-keyed one
// delivers to the wrong agent's listeners. `internal/dispatch` fires
// synchronously before listener delivery, so a violation throws at the
// dispatching call site. The table maps each family to how its subject is
// read from the event arguments; `null` = the subject is not recoverable
// from the arguments (session events key by the OWNING agent; subagent
// lifecycle events key by the delegating parent), so only carrier
// PRESENCE is asserted there.
const scopedSubject: Record<string, ((args: unknown[]) => unknown) | null> = {
'agent/created': args => args[0],
'agent/disposed': args => args[0],
'agent/status': args => args[0],
'agent/queued': args => args[0],
'agent/session-start': args => args[0],
'agent/pre-step': args => args[0],
'agent/prompt-submit': args => args[0],
'agent/request': args => args[0],
'agent/step-result': args => args[0],
'agent/turn-continuation': args => args[0],
'agent/error': args => args[0],
'tools/pre-execute': args => (args[0] as ToolExecution).agent,
'tools/post-execute': args => (args[0] as ToolExecution).agent,
'system-prompt/assemble': args => (args[1] as AssembleContext).scope,
'session/created': null,
'session/event': null,
'session/flush': null,
'subagent/start': null,
'subagent/end': null,
}
ctx.on('internal/dispatch', (_mode, name, args, thisArg) => {
const subjectOf = scopedSubject[name]
if (subjectOf === undefined) return
if (!isScopeCarrier(thisArg)) {
throw new InvariantError(
`"${name}" is a scope-filtered event but was dispatched without a scope carrier — `
+ 'pass scopeTarget(base, subject) as the dispatch thisArg (agent events: use agentEvents(ctx, agent))')
}
if (subjectOf !== null && carrierKeyOf(thisArg) !== subjectOf(args)) {
throw new InvariantError(
`"${name}" was dispatched with a scope carrier keyed to a DIFFERENT subject than its arguments name — `
+ 'the carrier key and the event\'s subject must be the same object (use agentEvents(ctx, agent))')
}
// The assembly context must never carry the agent DX field without the
// scope layer selector: the assembly would silently miss the agent's
// scoped sections/tools (use assembleContextFor(agent)).
if (name === 'system-prompt/assemble') {
const context = args[1] as AssembleContext
if (context.agent !== undefined && context.scope !== context.agent) {
throw new InvariantError(
'an assembly context carries `agent` without `scope` (or with a mismatched scope) — '
+ 'use assembleContextFor(agent) so the assembly resolves the agent\'s scoped layer')
}
}
}, { global: true })
// --- Setup-drives invariant ---------------------------------------------
//
// CreateAgentOptions.setup REGISTERS the agent's scoped world; it must not
// DRIVE the agent — an inject() there opens a turn before
// `agent/session-start`, inverting the "session-start fires before the
// first turn" contract every bridge keys on. A turn/start appended to a
// live agent's session before its agent/session-start fired is therefore a
// creation-time misuse, reported at the appending call site. Sessions of
// agents that exist BEFORE this plugin applies are marked started (their
// ordering is unknowable after the fact — never a false positive on HMR).
// `agents` is read via ctx.get (a strict, optional store lookup) rather
// than injected: the invariants plugin must load in harnesses that carry
// no agent registry at all (bare session tests), where this check simply
// never trips.
const sessionStarted = new WeakSet<Session>()
for (const agent of ctx.get('agents')?.list() ?? []) sessionStarted.add(agent.session)
ctx.on('agent/session-start', (agent) => { sessionStarted.add(agent.session) })
ctx.on('session/event', (session, event) => {
if (event.type !== 'turn/start' || sessionStarted.has(session)) return
const owner = ctx.get('agents')?.list().find(agent => agent.session === session)
if (owner === undefined) return
throw new InvariantError(
`agent "${owner.id}": a turn opened before agent/session-start fired — `
+ 'CreateAgentOptions.setup registers the scoped world, it must not drive the agent '
+ '(send/steer/inject belong after creation returns)')
})
// 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

100
packages/support/invariants/tests/invariants.spec.ts
View File

@@ -1,5 +1,6 @@
import { describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
import { CallId } from '@deepseek-ai/dsh-llm'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
@@ -41,8 +42,8 @@ describe('session-log invariants', () => {
const session = ctx.sessions.create()
// Session.append enforces seq-contiguity at the source, so drive the
// invariants seq check directly via session/event with a regressing seq.
ctx.emit('session/event', session, { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } as never)
expect(() => { ctx.emit('session/event', session, { type: 'turn/end', seq: 0, time: 2, data: { turn: 1, reason: { kind: 'completed' } } } as never) })
ctx.emit(scopeTarget(session, undefined), 'session/event', session, { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } as never)
expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, { type: 'turn/end', seq: 0, time: 2, data: { turn: 1, reason: { kind: 'completed' } } } as never) })
.toThrow(/seq must strictly increase/)
})
@@ -340,11 +341,11 @@ describe('dev-freeze', () => {
// handler directly via hand-built session/events — exactly the shape the
// invariants listener receives. Open a turn first (seq 0) so the cyclic
// user/message (seq 1) satisfies the turn-enclosure invariant.
ctx.emit('session/event', session, { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } as never)
ctx.emit(scopeTarget(session, undefined), 'session/event', session, { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } as never)
const cyclic: Record<string, unknown> = { type: 'text', text: 'x' }
cyclic['self'] = cyclic
const event = { type: 'user/message', seq: 1, time: 1, data: { content: [cyclic], source: { kind: 'user' } } }
expect(() => { ctx.emit('session/event', session, event as never) }).not.toThrow()
expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, event as never) }).not.toThrow()
expect(Object.isFrozen(cyclic)).toBe(true)
})
})
@@ -354,41 +355,41 @@ describe('agent status invariants', () => {
const { ctx } = await setup({ freeze: false })
const agent = mockAgent('a1')
expect(() => {
ctx.emit('agent/status', agent, 'idle')
ctx.emit('agent/status', agent, 'running')
ctx.emit('agent/status', agent, 'idle')
ctx.emit('agent/status', agent, 'disposed')
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'idle')
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running')
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'idle')
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'disposed')
}).not.toThrow()
})
it('accepts running→disposed', async () => {
const { ctx } = await setup({ freeze: false })
const agent = mockAgent('a2')
ctx.emit('agent/status', agent, 'running')
expect(() => { ctx.emit('agent/status', agent, 'disposed') }).not.toThrow()
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'disposed') }).not.toThrow()
})
it('rejects a no-op transition', async () => {
const { ctx } = await setup({ freeze: false })
const agent = mockAgent('a3')
ctx.emit('agent/status', agent, 'running')
expect(() => { ctx.emit('agent/status', agent, 'running') }).toThrow(/no-op transition/)
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running') }).toThrow(/no-op transition/)
})
it('rejects leaving the terminal disposed state', async () => {
const { ctx } = await setup({ freeze: false })
const agent = mockAgent('a4')
ctx.emit('agent/status', agent, 'disposed')
expect(() => { ctx.emit('agent/status', agent, 'idle') }).toThrow(/left terminal state disposed/)
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'disposed')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'idle') }).toThrow(/left terminal state disposed/)
})
it('tracks status per agent independently', async () => {
const { ctx } = await setup({ freeze: false })
const a = mockAgent('a5')
const b = mockAgent('b5')
ctx.emit('agent/status', a, 'running')
ctx.emit(scopeTarget(a, a), 'agent/status', a, 'running')
// b's first observation is independent of a.
expect(() => { ctx.emit('agent/status', b, 'running') }).not.toThrow()
expect(() => { ctx.emit(scopeTarget(b, b), 'agent/status', b, 'running') }).not.toThrow()
})
})
@@ -406,8 +407,8 @@ describe('HMR safety', () => {
expect(Object.isFrozen(event)).toBe(false)
// A no-op status transition no longer throws either.
const agent = mockAgent('hmr')
ctx.emit('agent/status', agent, 'idle')
expect(() => { ctx.emit('agent/status', agent, 'idle') }).not.toThrow()
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'idle')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'idle') }).not.toThrow()
})
it('InvariantError carries a stable code', () => {
@@ -780,3 +781,66 @@ describe('request cross-check ordering (prepend)', () => {
}).toThrow(/diverges from the boundary derivation/)
})
})
describe('scoped-dispatch invariants', () => {
async function scopedCtx() {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(Invariants)
return ctx
}
it('rejects a scoped-family dispatch without a carrier (teaching error)', async () => {
const ctx = await scopedCtx()
const agent = { id: 'a1' } as unknown as Agent
expect(() => { ctx.emit('agent/error', agent, 1, 0, new Error('x')) })
.toThrow(/dispatched without a scope carrier/)
})
it('rejects a carrier keyed to a different subject than the arguments name', async () => {
const ctx = await scopedCtx()
const agent = { id: 'a1' } as unknown as Agent
const other = { id: 'a2' } as unknown as Agent
expect(() => { ctx.emit(scopeTarget(agent, other), 'agent/error', agent, 1, 0, new Error('x')) })
.toThrow(/keyed to a DIFFERENT subject/)
// The correct spelling passes.
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/error', agent, 1, 0, new Error('x')) })
.not.toThrow()
})
it('rejects an assembly context carrying agent without scope', async () => {
const ctx = await scopedCtx()
const agent = { id: 'a1' } as unknown as Agent
const base = { name: 'systemPrompt' }
const assembly = { sections: [], tools: [], variables: {} }
const bad = { agent }
expect(() => {
// The carrier base stands in for the SystemPrompt service (the declared `this`); the invariant only reads the carrier marks.
void ctx.waterfall(scopeTarget(base, undefined) as never, 'system-prompt/assemble', assembly as never, bad as never, () => Promise.resolve(assembly as never))
}).toThrow(/agent.*without.*scope|assembleContextFor/)
const good = { agent, scope: agent }
expect(() => {
void ctx.waterfall(scopeTarget(base, agent) as never, 'system-prompt/assemble', assembly as never, good as never, () => Promise.resolve(assembly as never))
}).not.toThrow()
})
it('rejects a turn opened before agent/session-start (setup drives the agent)', async () => {
const ctx = await scopedCtx()
// A live agent whose session is in the store but whose session-start has
// not fired: appending turn/start must throw the teaching error.
const session = ctx.sessions.create(SessionId('drive-s'))
const agent = { id: 'driver', session } as unknown as Agent
// Provide a minimal agents lookup: the invariant reads ctx.get('agents').
const registryStub = { list: () => [agent] }
ctx.root.provide('agents', registryStub as never)
expect(() => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
}).toThrow(/turn opened before agent\/session-start/)
// After session-start fires, turns open freely.
ctx.emit(scopeTarget(agent, agent), 'agent/session-start', agent, 'startup')
expect(() => {
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
}).not.toThrow()
})
})

9
packages/support/invariants/tsconfig.json
View File

@@ -22,6 +22,15 @@
},
{
"path": "../../core/agent"
},
{
"path": "../../core/scope"
},
{
"path": "../../core/system-prompt"
},
{
"path": "../../core/tools"
}
]
}