fix(invariants): assert runtime relationships, not API shapes

This commit is contained in:
Tianyi Cui
2026-07-20 19:34:19 +08:00
parent 1254c07025
commit 1145ee5fc3
124 changed files with 2923 additions and 2334 deletions

View File

@@ -1,29 +1,24 @@
/** Package-owned runtime contract checks for `@deepseek-ai/dsh-tool-workflow`. @module @deepseek-ai/dsh-tool-workflow/invariant */
/**
* Package-owned invariant companion for `@deepseek-ai/dsh-tool-workflow`.
* @module @deepseek-ai/dsh-tool-workflow/invariant
*/
/* jscpd:ignore-start */
import type { Context } from 'cordis'
import { observePluginInvariant, type InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
const PACKAGE_NAME = '@deepseek-ai/dsh-tool-workflow'
/** Cordis companion plugin name. */
export const name = 'tool-workflow-invariant'
/** Services required before the companion can register. */
/** Service required before the companion can reserve package ownership. */
export const inject = ['invariants']
/** Install checks for this package's active plugin fibers. */
const install: InvariantInstaller = (ctx, fail) => {
observePluginInvariant(ctx, fail, {
name: 'tool-workflow',
inject: [
'tools',
'workflows',
'systemPrompt',
],
effects: [
'tools.register()',
],
})
}
/**
* No runtime invariant: this model-facing adapter has no independent lifecycle stream; execution
* relations are owned by the capability seam it calls.
*/
const install: InvariantInstaller = () => {}
/**
* Register this package's invariant companion.
@@ -32,3 +27,4 @@ const install: InvariantInstaller = (ctx, fail) => {
*/
export const apply = (ctx: Context): Promise<() => void> =>
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
/* jscpd:ignore-end */

View File

@@ -1,33 +1,24 @@
/**
* Package-owned runtime contract checks for `@deepseek-ai/dsh-workflow-workerthread`.
* Package-owned invariant companion for `@deepseek-ai/dsh-workflow-workerthread`.
* @module @deepseek-ai/dsh-workflow-workerthread/invariant
*/
/* jscpd:ignore-start */
import type { Context } from 'cordis'
import { observePluginInvariant, type InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
const PACKAGE_NAME = '@deepseek-ai/dsh-workflow-workerthread'
/** Cordis companion plugin name. */
export const name = 'workflow-workerthread-invariant'
/** Services required before the companion can register. */
/** Service required before the companion can reserve package ownership. */
export const inject = ['invariants']
/** Install checks for this package's active plugin fibers. */
const install: InvariantInstaller = (ctx, fail) => {
observePluginInvariant(ctx, fail, {
name: 'WorkerWorkflowEngine',
inject: [
'subagents',
],
effects: [
'ctx.provide("workflows")',
],
services: [
'workflows',
],
})
}
/**
* No runtime invariant: this process-boundary implementation exposes no same-process event relation;
* worker protocol and built-worker tests cover it.
*/
const install: InvariantInstaller = () => {}
/**
* Register this package's invariant companion.
@@ -36,3 +27,4 @@ const install: InvariantInstaller = (ctx, fail) => {
*/
export const apply = (ctx: Context): Promise<() => void> =>
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
/* jscpd:ignore-end */

View File

@@ -1,24 +1,134 @@
/** Package-owned runtime contract checks for `@deepseek-ai/dsh-workflow`. @module @deepseek-ai/dsh-workflow/invariant */
/** Package-owned workflow lifecycle invariants. @module @deepseek-ai/dsh-workflow/invariant */
import type { Context } from 'cordis'
import { observeServiceInvariant, serviceShapeViolation, type InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
import type {
WorkflowAgentEndInfo,
WorkflowAgentInfo,
WorkflowResultInfo,
WorkflowRunInfo,
} from './types.ts'
const PACKAGE_NAME = '@deepseek-ai/dsh-workflow'
/** Cordis companion plugin name. */
export const name = 'workflow-invariant'
/** Services required before the companion can register. */
/** Service required before the companion can reserve package ownership. */
export const inject = ['invariants']
/** Validate every implementation bound to this package's service seam. */
interface WorkflowTrace {
meta: string
agents: Map<number, WorkflowAgentInfo>
starts: number
}
/** Require every event for a run to retain its validated identity snapshot. */
function traceFor(
traces: ReadonlyMap<string, WorkflowTrace>,
info: WorkflowRunInfo,
fail: InvariantFailure,
): WorkflowTrace {
const trace = traces.get(info.id)
if (trace === undefined) fail(`workflow event has no matching workflow/start for run ${JSON.stringify(info.id)}`)
if (trace.meta !== JSON.stringify(info.meta)) {
fail(`workflow event meta diverges from workflow/start for run ${JSON.stringify(info.id)}`)
}
return trace
}
/** Assert the immutable identity fields shared by an agent pair. */
function validateAgentEnd(start: WorkflowAgentInfo, end: WorkflowAgentEndInfo, fail: InvariantFailure): void {
if (start.label !== end.label || start.phase !== end.phase || start.childId !== end.childId) {
fail(`workflow/agent-end identity diverges from workflow/agent-start for seq ${end.seq}`)
}
const outcome: string = end.outcome
if (outcome !== 'completed' && outcome !== 'failed' && outcome !== 'cancelled') {
fail(`workflow/agent-end carries unknown outcome ${JSON.stringify(outcome)}`)
}
}
/** Validate a terminal result against the accumulated run trace. */
function validateWorkflowEnd(trace: WorkflowTrace, result: WorkflowResultInfo, fail: InvariantFailure): void {
if (trace.agents.size > 0) fail(`workflow/end has ${trace.agents.size} agent call(s) without workflow/agent-end`)
if (!Number.isSafeInteger(result.agentsStarted) || result.agentsStarted < trace.starts) {
fail('workflow/end agentsStarted must be a safe integer covering every observed agent start')
}
if (result.stopReason === 'completed' ? result.error !== undefined : typeof result.error !== 'string') {
fail('workflow/end error must be absent exactly for completed runs')
}
}
/** Install workflow start/end and child-call pairing checks. */
const install: InvariantInstaller = (ctx, fail) => {
observeServiceInvariant(ctx, fail, 'workflows', value => serviceShapeViolation(value, {
methods: ['start'],
}))
const traces = new Map<string, WorkflowTrace>()
const stagedStarts = new WeakSet<WorkflowRunInfo>()
const stagedAgentStarts = new WeakSet<WorkflowAgentInfo>()
const stagedAgentEnds = new WeakSet<WorkflowAgentEndInfo>()
const stagedEnds = new WeakSet<WorkflowResultInfo>()
ctx.on('internal/dispatch', (_mode, eventName, args) => {
if (eventName === 'workflow/start') {
const info = args[0] as WorkflowRunInfo
if (String(info.id).length === 0 || info.meta.name.length === 0 || info.meta.description.length === 0) {
fail('workflow/start id, meta.name, and meta.description must be non-empty')
}
if (traces.has(info.id)) fail(`workflow/start repeated run id ${JSON.stringify(info.id)}`)
stagedStarts.add(info)
return
}
if (!eventName.startsWith('workflow/')) return
const info = args[0] as WorkflowRunInfo
const trace = traceFor(traces, info, fail)
if (eventName === 'workflow/agent-start') {
const agent = args[1] as WorkflowAgentInfo
if (!Number.isSafeInteger(agent.seq) || agent.seq < 1 || String(agent.childId).length === 0) {
fail('workflow/agent-start seq must be positive and childId must be non-empty')
}
if (trace.agents.has(agent.seq)) fail(`workflow/agent-start repeated seq ${agent.seq}`)
stagedAgentStarts.add(agent)
return
}
if (eventName === 'workflow/agent-end') {
const agent = args[1] as WorkflowAgentEndInfo
const start = trace.agents.get(agent.seq)
if (start === undefined) return fail(`workflow/agent-end has no matching start for seq ${agent.seq}`)
validateAgentEnd(start, agent, fail)
stagedAgentEnds.add(agent)
return
}
if (eventName === 'workflow/end') {
const result = args[1] as WorkflowResultInfo
validateWorkflowEnd(trace, result, fail)
stagedEnds.add(result)
}
}, { global: true })
ctx.on('workflow/start', (info) => {
/* v8 ignore next -- internal/dispatch stages the same run-info object */
if (!stagedStarts.delete(info)) return
traces.set(info.id, { meta: JSON.stringify(info.meta), agents: new Map(), starts: 0 })
}, { global: true })
ctx.on('workflow/agent-start', (info, agent) => {
/* v8 ignore next -- internal/dispatch stages the same agent object */
if (!stagedAgentStarts.delete(agent)) return
const trace = traceFor(traces, info, fail)
trace.agents.set(agent.seq, agent)
trace.starts += 1
}, { global: true })
ctx.on('workflow/agent-end', (info, agent) => {
/* v8 ignore next -- internal/dispatch stages the same agent object */
if (!stagedAgentEnds.delete(agent)) return
traceFor(traces, info, fail).agents.delete(agent.seq)
}, { global: true })
ctx.on('workflow/end', (info, result) => {
/* v8 ignore next -- internal/dispatch stages the same result object */
if (!stagedEnds.delete(result)) return
traces.delete(info.id)
}, { global: true })
}
/**
* Register this package's invariant companion.
* Register the workflow invariant companion.
* @param ctx - Cordis context carrying the invariant service.
* @returns the installed registration's disposer after setup succeeds.
*/

View File

@@ -0,0 +1,108 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { SessionId } from '@deepseek-ai/dsh-session'
import { WorkflowRunId } from '@deepseek-ai/dsh-workflow'
import type {
WorkflowAgentEndInfo,
WorkflowAgentInfo,
WorkflowResultInfo,
WorkflowRunInfo,
} from '@deepseek-ai/dsh-workflow'
import * as WorkflowInvariant from '@deepseek-ai/dsh-workflow/invariant'
import InvariantService from '@deepseek-ai/dsh-invariants'
async function setup(): Promise<Context> {
const ctx = new Context()
await ctx.plugin(InvariantService)
await ctx.plugin(WorkflowInvariant)
return ctx
}
const info = (overrides: Partial<WorkflowRunInfo> = {}): WorkflowRunInfo => ({
id: WorkflowRunId('workflow-1'),
meta: { name: 'review', description: 'Review a change' },
...overrides,
})
const agent = (overrides: Partial<WorkflowAgentInfo> = {}): WorkflowAgentInfo => ({
seq: 1,
label: 'reviewer',
childId: SessionId('child-1'),
...overrides,
})
const agentEnd = (overrides: Partial<WorkflowAgentEndInfo> = {}): WorkflowAgentEndInfo => ({
...agent(),
outcome: 'completed',
...overrides,
})
const result = (overrides: Partial<WorkflowResultInfo> = {}): WorkflowResultInfo => ({
stopReason: 'completed',
agentsStarted: 1,
...overrides,
})
describe('workflow invariants', () => {
it('accepts a complete workflow and child lifecycle', async () => {
const ctx = await setup()
const run = info()
ctx.emit('workflow/start', run)
ctx.emit('workflow/phase', run, 'inspect')
ctx.emit('workflow/log', run, 'working')
ctx.emit('workflow/agent-start', run, agent())
ctx.emit('workflow/agent-end', run, agentEnd())
ctx.emit('workflow/end', run, result())
ctx.emit('tools/change')
})
it('rejects invalid run identity and enclosure', async () => {
const ctx = await setup()
expect(() => { ctx.emit('workflow/start', info({ id: WorkflowRunId('') })) }).toThrow(/must be non-empty/)
const run = info()
ctx.emit('workflow/start', run)
expect(() => { ctx.emit('workflow/start', run) }).toThrow(/repeated run id/)
expect(() => { ctx.emit('workflow/log', info({ meta: { name: 'other', description: 'x' } }), 'x') })
.toThrow(/meta diverges/)
const fresh = await setup()
expect(() => { fresh.emit('workflow/log', info(), 'x') }).toThrow(/no matching workflow\/start/)
})
it('rejects malformed and unpaired child lifecycles', async () => {
const ctx = await setup()
const run = info()
ctx.emit('workflow/start', run)
expect(() => { ctx.emit('workflow/agent-start', run, agent({ seq: 0 })) }).toThrow(/seq must be positive/)
ctx.emit('workflow/agent-start', run, agent())
expect(() => { ctx.emit('workflow/agent-start', run, agent()) }).toThrow(/repeated seq/)
expect(() => { ctx.emit('workflow/agent-end', run, agentEnd({ seq: 2 })) }).toThrow(/no matching start/)
expect(() => { ctx.emit('workflow/agent-end', run, agentEnd({ childId: SessionId('other') })) })
.toThrow(/identity diverges/)
expect(() => { ctx.emit('workflow/agent-end', run, agentEnd({ outcome: 'unknown' as never })) })
.toThrow(/unknown outcome/)
})
it('rejects inconsistent terminal results', async () => {
const active = await setup()
active.emit('workflow/start', info())
active.emit('workflow/agent-start', info(), agent())
expect(() => { active.emit('workflow/end', info(), result()) }).toThrow(/without workflow\/agent-end/)
const count = await setup()
count.emit('workflow/start', info())
count.emit('workflow/agent-start', info(), agent())
count.emit('workflow/agent-end', info(), agentEnd())
expect(() => { count.emit('workflow/end', info(), result({ agentsStarted: 0 })) })
.toThrow(/covering every observed agent start/)
const completed = await setup()
completed.emit('workflow/start', info())
expect(() => { completed.emit('workflow/end', info(), result({ error: 'unexpected' })) })
.toThrow(/absent exactly for completed/)
const failed = await setup()
failed.emit('workflow/start', info())
expect(() => { failed.emit('workflow/end', info(), result({ stopReason: 'error' })) })
.toThrow(/absent exactly for completed/)
})
})

View File

@@ -58,8 +58,11 @@ describe('dsh-workflow (interface)', () => {
ctx.on('workflow/log', (info, message) => { seen.push([info, message]) })
ctx.on('workflow/agent-start', (info, agent) => { seen.push([info, agent]) })
const engine = ctx.workflows as StubEngine
engine.emit('workflow/start', INFO)
engine.emit('workflow/log', INFO, 'hello')
engine.emit('workflow/agent-start', INFO, { seq: 1, label: 'l', childId: 'c' })
engine.emit('workflow/agent-end', INFO, { seq: 1, label: 'l', childId: 'c', outcome: 'completed' })
engine.emit('workflow/end', INFO, { stopReason: 'completed', agentsStarted: 1 })
expect(seen).toEqual([
[INFO, 'hello'],
[INFO, { seq: 1, label: 'l', childId: 'c' }],
@@ -77,8 +80,11 @@ describe('dsh-workflow (interface)', () => {
ctx.on('workflow/agent-start', (_info, agent) => { seen.push(agent.label) })
const engine = ctx.workflows as StubEngine
const payload = { seq: 1, label: 'original', childId: 'c' }
engine.emit('workflow/start', INFO)
engine.emit('workflow/agent-start', INFO, payload)
await Promise.resolve()
engine.emit('workflow/agent-end', INFO, { ...payload, outcome: 'completed' })
engine.emit('workflow/end', INFO, { stopReason: 'completed', agentsStarted: 1 })
expect(seen).toEqual(['original'])
expect(String(warn.mock.calls[0]![0])).toContain('listener rejected')
})
@@ -91,7 +97,9 @@ describe('dsh-workflow (interface)', () => {
ctx.on('workflow/phase', () => { throw new Error('bad listener') })
ctx.on('workflow/phase', (_info, title) => { reached.push(title) })
const engine = ctx.workflows as StubEngine
engine.emit('workflow/start', INFO)
expect(() => { engine.emit('workflow/phase', INFO, 'Scan') }).not.toThrow()
engine.emit('workflow/end', INFO, { stopReason: 'completed', agentsStarted: 0 })
expect(reached).toEqual(['Scan'])
expect(warn).toHaveBeenCalledOnce()
expect(String(warn.mock.calls[0]![0])).toContain('workflow/phase listener threw')
@@ -107,7 +115,9 @@ describe('dsh-workflow (interface)', () => {
})
ctx.on('workflow/phase', (_info, title) => { reached.push(title) })
const engine = ctx.workflows as StubEngine
engine.emit('workflow/start', INFO)
expect(() => { engine.emit('workflow/phase', INFO, 'Scan') }).not.toThrow()
engine.emit('workflow/end', INFO, { stopReason: 'completed', agentsStarted: 0 })
expect(reached).toEqual(['Scan'])
expect(warn).toHaveBeenCalledOnce()
expect(String(warn.mock.calls[0]![0])).toContain('[unrenderable thrown value]')