Merge remote-tracking branch 'origin/master' into worktree/provider-routed-llm-adapters

# Conflicts:
#	docs/cordis-catalog/events.md
#	docs/cordis-catalog/services.md
#	docs/event-producer-consumer.md
#	docs/module-graph.md
#	packages/context/time-context/tests/time-context.spec.ts
#	packages/cordis/tool-cordis/src/api-catalog.ts
#	packages/core/session/README.md
#	packages/core/session/src/types.ts
#	packages/core/session/tests/surface.spec.ts
#	packages/examples/acp-demo/tests/acp-agent.spec.ts
#	packages/examples/stdio-demo/tests/stdio-agent.spec.ts
#	packages/hooks/hooks-claude/tests/coverage.spec.ts
#	packages/hooks/hooks-codex/tests/coverage.spec.ts
#	packages/session-query/session-query/tests/session-query.spec.ts
#	packages/support/invariants/tests/invariants.spec.ts
#	packages/ui/acp/tests/harness.ts
This commit is contained in:
Tianyi Cui
2026-07-17 21:56:10 +08:00
358 changed files with 18578 additions and 2877 deletions

View File

@@ -5,9 +5,10 @@ Packages that exist to serve development, testing, and the examples rather than
| Package | Role | ctx key |
|---|---|---|
| `acp-snapshot/` | ACP snapshot suite kit: subprocess scenario harness + golden normalizers + the `defineAcpSnapshotSuite` factory | (library — imported by example `*.snapshot.ts` suites) |
| `agent-loop-testkit/` | Shared prerequisite mounting for tests that exercise the concrete agent loop | (library — imported by AgentLoop integration tests) |
| `invariants/` | Runtime event-contract assertions for development diagnostics | (listens on `session/*`, `agent/*`) |
| `loader-smoke/` | Shared real-Loader subprocess harness for keyless example smokes | (library — imported by example e2e suites) |
| `llm-replay/` | Record/replay adapter: short-circuits `llm/stream` from a recorded session JSONL (keyless snapshot tests) | (listens on `llm/stream`) |
| `subagent-mock/` | Scripted `SubagentProvider` for deterministic seam/tool tests | (registers on `ctx.subagents`) |
`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the snapshot tier's harness/normalizer/suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.
`invariants` is development support but has no environment guard: it runs wherever registered, and the default `dsh-agent-spine-demo` bundle mounts it unconditionally. `agent-loop-testkit` centralizes the mandatory service spine for hand-built AgentLoop tests without owning their loop or scenario. `llm-replay` backs the demos and the snapshot test tier under the per-file coverage gate. `acp-snapshot` carries the snapshot tier's harness/normalizer/suite machinery, while `loader-smoke` owns the parallel stdio/Loader process boundary used by keyless example e2e suites. `subagent-mock` exercises the real `ctx.subagents` load path without a model or child agent. A package graduates OUT of `support/` into a product group only when it gains documented product consumers.

View File

@@ -4,7 +4,7 @@ The ACP snapshot suite kit: the shared machinery behind the keyless snapshot tie
Three layers, importable separately:
- **`runScenario` (harness)** — boots the real agent bin as a subprocess via tsx (unbuilt, Loader path), drives it over ACP JSON-RPC stdio from a deterministic `input.json` script, tees raw stdout for the golden + purity check, and harvests every persisted session JSONL (parent + subagent children, primary-first) after a graceful stdin-EOF shutdown. Parameterized by `AgentUnderTest` (`binScript`, `configPath`, `tsconfigPath` — absolute paths; the subprocess cwd is a temp dir outside the repo).
- **`runScenario` (harness)** — boots the real agent bin as a subprocess via tsx (unbuilt, Loader path), drives it over ACP JSON-RPC stdio from a deterministic `input.json` script, tees raw stdout for the golden + purity check, and harvests every persisted session JSONL (parent + subagent children, primary-first) after a graceful stdin-EOF shutdown. Parameterized by `AgentUnderTest` (`binScript`, `configPath`, `tsconfigPath` — absolute paths; the subprocess cwd is a temp dir outside the repo). Startup failures preserve captured agent stderr in the rejected diagnostic.
- **Normalizers** — pure functions turning the two captured surfaces into stable text: `normalizeStdout` (JSON-RPC ids → first-seen sequence; UUIDs/cwd → tokens; doubles as the stdout-purity check), `normalizeSessionLog` (times zeroed, `seq` kept), `scrubSystemPrompts` (prompt text → `{{system}}`), `scrubToolSchemas` (schema bulk → `{{tools}}`), and `scrubRequestHeaders` (all header bulk → `{{system}}`/`{{tools}}`/`{{messagePrefix}}` outside each pin, structure kept — [pinned-header RFC](../../../docs/rfc/implemented/testing/2026-07-06-pin-request-header-content-in-one-scenario.md)).
- **`defineAcpSnapshotSuite` (factory)** — registers the whole describe/it tree for a scenario table: per-scenario golden + re-persisted-log compares, record/refresh fixture write-back, rejection of structured `UNKNOWN_TOOL` results, the per-header-class pin (`system-prompt.golden.md` plus `tool-schemas.golden.json`) with its live uniformity guard, and the fixture guard block (no orphan scenario dirs, required files present, exactly one pin per class, every JSONL prompt/schema-scrubbed, non-pinning fixtures fully header-scrubbed). Must be called at vitest collection time.

View File

@@ -168,6 +168,9 @@ export interface RunOptions {
export async function runScenario(input: InputScript, opts: RunOptions): Promise<RunResult> {
const cwd = await mkdtemp(join(tmpdir(), 'acp-snap-cwd-'))
const sessionsRoot = await mkdtemp(join(tmpdir(), 'acp-snap-sessions-'))
// Fixed path length: spill-policy budgets the preview against the REAL path
// before stdout normalization, so tmpdir() length differences churn goldens.
const spillRoot = '/tmp/dsh-acp-snapshot-spill'
// Everything past the temp-dir creation runs under a try/finally that always
// removes both dirs — so a failure in workspace seeding, spawn, or any step
// never leaks them (the "e2e tests own their resources" rule).
@@ -187,6 +190,7 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
DSH_SNAPSHOT: opts.mode,
DSH_SNAPSHOT_FILE: opts.fixtureFile,
DSH_SNAPSHOT_SESSIONS_ROOT: sessionsRoot,
DSH_SNAPSHOT_SPILL_ROOT: spillRoot,
DSH_HOME: join(cwd, '.dsh'),
DSH_AGENTS_HOME: join(cwd, '.agents'),
...opts.overrideFile !== undefined ? { DSH_SNAPSHOT_OVERRIDE: opts.overrideFile } : {},
@@ -281,6 +285,10 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
// Harvest EVERY persisted log (parent + any subagent children) while the
// temp dirs still exist, ordered primary-first.
sessionLogs = await harvestSessionLogs(sessionsRoot)
} catch (error: unknown) {
const stderr = stderrChunks.join('')
if (stderr === '') throw error
throw new Error(`snapshot-harness: scenario failed: ${String(error)}\nagent stderr:\n${stderr}`, { cause: error })
} finally {
// Failure-safe teardown: kill a still-running child and drop the temp dirs
// even if seeding/spawn/a step/harvest threw, so a flaky run never leaks a
@@ -291,6 +299,7 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
}
await rm(cwd, { recursive: true, force: true })
await rm(sessionsRoot, { recursive: true, force: true })
await rm(spillRoot, { recursive: true, force: true })
}
return {

View File

@@ -14,6 +14,16 @@ const MESSAGE_PREFIX = '{{messagePrefix}}'
/** A UUID v4 string, the shape `randomUUID()` produces for session ids. */
const UUID_RE = /[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}/gi
const LOCAL_SPILL_PATH_RE = new RegExp(
String.raw`\{\{cwd\}\}/\.spill/session-[0-9a-f]{12}/[0-9a-f]{12}-([A-Za-z0-9._~-]+?)`
+ String.raw`(?=\. Use read with offset/limit|[\s)]|$)`,
'g',
)
const SNAPSHOT_SPILL_PATH_RE = new RegExp(
String.raw`/tmp/dsh-acp-snapshot-spill/session-[0-9a-f]{12}/[0-9a-f]{12}-([A-Za-z0-9._~-]+?)`
+ String.raw`(?=\. Use read with offset/limit|[\s)]|$)`,
'g',
)
/** Inputs the normalizers need to recognize a run's volatile values. */
export interface NormalizeContext {
@@ -29,6 +39,9 @@ function scrubString(value: string, ctx: NormalizeContext): string {
// cwd first (longest, most specific), then explicit session ids, then any
// residual UUID (covers ids that appear in places we didn't enumerate).
out = out.split(ctx.cwd).join(CWD)
out = out.split(`/private${CWD}`).join(CWD)
out = out.replace(LOCAL_SPILL_PATH_RE, (_match, name: string) => `{{spillLocator:${name}}}`)
out = out.replace(SNAPSHOT_SPILL_PATH_RE, (_match, name: string) => `{{spillLocator:${name}}}`)
for (const id of ctx.sessionIds) out = out.split(id).join(SESSION_ID)
out = out.replace(UUID_RE, SESSION_ID)
return out

View File

@@ -3,6 +3,8 @@
* compares normalized stdout; comparable session fixtures are both replay input and expected
* output. Record mode refreshes reproducible model scenarios from the live API, while refresh
* mode replays committed scripts and rewrites derived artifacts without a key.
* Replay scenarios run concurrently because each subprocess owns unique temp cwd and persistence
* roots and only reads committed fixtures. Record and refresh scenarios stay serial while writing.
*
* Exactly one scenario per header-composition class pins the system prompt and tool schemas in
* dedicated sidecars. Every live header is checked against that pin, so session-dependent
@@ -461,7 +463,7 @@ export function stabilizeRefreshLog(fresh: string, existing: string, replacement
}
/**
* Register the suite: one `describe` per scenario (the golden/log compares and
* Register the suite: one test per scenario (the golden/log compares and
* the header-uniformity guard) plus the fixture guard block (no orphan
* scenario dirs, required files present, exactly one pin per header class,
* pinning fixtures well-formed, every JSONL prompt-scrubbed, non-pinning
@@ -477,6 +479,7 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
const RECORDING = mode === 'record'
const REFRESHING = mode === 'refresh'
const childMode: 'replay' | 'record' = RECORDING ? 'record' : 'replay'
const scenarioSuite = mode === 'replay' ? describe.concurrent : describe
/** The class a scenario's header composition belongs to (see {@link Scenario.headerClass}). */
const classOf = (scenario: Scenario): string => scenario.headerClass ?? 'default'
@@ -496,11 +499,11 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
}
}
for (const scenario of scenarios) {
describe(`snapshot: ${scenario.name}`, () => {
scenarioSuite('snapshot scenarios', () => {
for (const scenario of scenarios) {
// In RECORD mode, only re-run the `recorded` (live-API) scenarios; the `authored` ones
// (sidecar-driven errors/cancel) are never re-recorded.
it.skipIf(RECORDING && !scenario.recorded)('matches the goldens', async () => {
it.skipIf(RECORDING && !scenario.recorded)(`snapshot: ${scenario.name} matches the goldens`, async ({ expect }) => {
const dir = join(snapshotsDir, scenario.name)
const input = JSON.parse(await readFile(join(dir, 'input.json'), 'utf8')) as InputScript
const overrideFile = join(dir, 'replay.override.json')
@@ -658,8 +661,8 @@ export function defineAcpSnapshotSuite(options: SnapshotSuiteOptions): void {
}
}
})
})
}
}
})
describe('snapshot fixtures', () => {
it('every scenario directory is registered (no orphans)', async () => {

View File

@@ -24,6 +24,8 @@ interface ScriptedLog {
/** The whole scripted behavior for one run. Every field defaults to the least surprising choice. */
interface Behavior {
/** Exit during startup after writing any configured stderr note. */
failOnBoot?: boolean
/** Reject every `session/new` (exercises the expect-error step without extra dirs). */
rejectNewSession?: boolean
/** Reject `session/new` only when `additionalDirectories` is non-empty (the real bridge's rule). */
@@ -62,6 +64,7 @@ const behavior: Behavior = fixtureFile === ''
: JSON.parse(readFileSync(join(dirname(fixtureFile), 'behavior.json'), 'utf8')) as Behavior
if (behavior.stderrNote !== undefined) process.stderr.write(`${behavior.stderrNote}\n`)
if (behavior.failOnBoot === true) process.exit(7)
let nextOutboundId = 1000
let sessionId = ''

View File

@@ -38,6 +38,14 @@ async function scenario(behavior: object): Promise<{ dir: string; fixtureFile: s
const boot: InputStep[] = [{ op: 'initialize' }, { op: 'newSession' }]
describe('runScenario', () => {
it('includes agent stderr when the ACP connection closes during startup', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({ failOnBoot: true, stderrNote: 'fake agent requested startup failure' })
await expect(runScenario(
{ steps: [{ op: 'initialize' }] },
{ agent: AGENT, mode: 'replay', fixtureFile },
)).rejects.toThrow(/agent stderr:\nfake agent requested startup failure/)
})
it('drives a full turn: initialize (terminal caps), session, prompt, permission stub, harvest', { timeout: 20_000 }, async () => {
const { fixtureFile } = await scenario({
permissionProbe: true,

View File

@@ -93,6 +93,52 @@ describe('normalizeSessionLog', () => {
expect(out).not.toContain(ctx.cwd)
})
it('scrubs random local spill paths under the snapshot cwd', () => {
const ev = JSON.stringify({
type: 'tool/result', seq: 2, time: 5,
data: {
content: [{
type: 'text',
text: `Full formatted result stored at: ${ctx.cwd}/.spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt. Use read with offset/limit, or grep this path to search within it.`,
}],
},
})
const out = normalizeSessionLog(`${header({ cwd: ctx.cwd })}\n${ev}\n`, ctx)
expect(out).toContain('{{spillLocator:bash.txt}}')
expect(out).not.toContain('session-c22bc3f1d2af')
expect(out).not.toContain('8a7b6c5d4e3f')
})
it('scrubs macOS /private aliases for local spill paths', () => {
const ev = JSON.stringify({
type: 'tool/result', seq: 2, time: 5,
data: {
content: [{
type: 'text',
text: `Full formatted result stored at: /private${ctx.cwd}/.spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt. Use read with offset/limit, or grep this path to search within it.`,
}],
},
})
const out = normalizeSessionLog(`${header({ cwd: ctx.cwd })}\n${ev}\n`, ctx)
expect(out).toContain('{{spillLocator:bash.txt}}')
expect(out).not.toContain('/private{{spillLocator')
})
it('scrubs fixed snapshot spill paths', () => {
const ev = JSON.stringify({
type: 'tool/result', seq: 2, time: 5,
data: {
content: [{
type: 'text',
text: 'Full formatted result stored at: /tmp/dsh-acp-snapshot-spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt. Use read with offset/limit, or grep this path to search within it.',
}],
},
})
const out = normalizeSessionLog(`${header({ cwd: ctx.cwd })}\n${ev}\n`, ctx)
expect(out).toContain('{{spillLocator:bash.txt}}')
expect(out).not.toContain('/tmp/dsh-acp-snapshot-spill')
})
it('scrubs the session id in the header', () => {
const out = normalizeSessionLog(`${header({ id: ctx.sessionIds[0] })}\n`, ctx)
expect(out).toContain('{{sessionId}}')

View File

@@ -0,0 +1,27 @@
# `@deepseek-ai/dsh-agent-loop-testkit`
Shared prerequisite mounting for tests that exercise the concrete `AgentLoop`. `mountAgentLoopTestDependencies(ctx, options?)` installs the LLM, session, system-prompt, tool, and agent services in dependency order, then returns before the loop is mounted.
The caller registers adapters and optional plugins, mounts `AgentLoop` with the configuration under test, and disposes its own Context. System-prompt and tool-registry configuration can be forwarded through `options`; the helper does not provide test defaults beyond those owned by the services. A plugin-load failure rejects the helper call, while services activated earlier in the sequence remain owned by the caller's Context.
```ts
import { Context } from 'cordis'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx)
// Register the test adapter and any optional plugins here.
await ctx.plugin(AgentLoop, { agents: [] })
```
Tests of injection failures, partial topology, service load order, or service teardown mount their dependencies directly instead of using this helper.
## Model Experience
None, as this test-only composition helper neither drives nor modifies model requests.
## Known Limitations and Deferred Work
- **Only the mandatory prerequisite spine is shared** — adapters, optional plugins, `AgentLoop`, agents, and Context teardown remain caller-owned so scenario-specific ordering stays visible.

View File

@@ -0,0 +1,41 @@
{
"name": "@deepseek-ai/dsh-agent-loop-testkit",
"description": "Shared prerequisite mounting for tests that exercise the concrete agent loop",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-llm": "^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.7"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-agent-loop": "workspace:^",
"@deepseek-ai/dsh-llm": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.7"
}
}

View File

@@ -0,0 +1,46 @@
/**
* Shared mounting for the services required before tests load the concrete
* agent loop. The caller retains ownership of the context, loop, adapters,
* optional plugins, and teardown.
* @module @deepseek-ai/dsh-agent-loop-testkit
*/
import type { Context } from 'cordis'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import type { Config as SystemPromptConfig } from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import type { Config as ToolRegistryConfig } from '@deepseek-ai/dsh-tools'
/** Configuration forwarded to the prerequisite service plugins. */
export interface AgentLoopTestDependenciesOptions {
/** Configuration for the system-prompt registry. */
readonly systemPrompt?: SystemPromptConfig
/** Configuration for the tool registry. */
readonly tools?: ToolRegistryConfig
}
/**
* Mount the standard prerequisite services for an AgentLoop test.
*
* The function deliberately does not mount AgentLoop or register an adapter,
* so tests retain control of load order and the topology under test. The
* context owns every mounted service and remains responsible for disposal. A
* plugin-load failure rejects the promise; services activated earlier in the
* sequence remain context-owned and unwind with that context.
* @param ctx - test context that owns the mounted services.
* @param options - optional service configuration forwarded without mutation.
* @returns after every prerequisite service has activated.
*/
export async function mountAgentLoopTestDependencies(
ctx: Context,
options: AgentLoopTestDependenciesOptions = {},
): Promise<void> {
await ctx.plugin(LlmService)
await ctx.plugin(SessionStore)
await ctx.plugin(SystemPrompt, options.systemPrompt ?? {})
await ctx.plugin(ToolRegistry, options.tools ?? {})
await ctx.plugin(AgentRegistry)
}

View File

@@ -0,0 +1,20 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import { mountAgentLoopTestDependencies } from '../src/index.ts'
describe('dsh-agent-loop-testkit', () => {
it('mounts a configurable prerequisite spine that can activate AgentLoop', async () => {
const ctx = new Context()
await mountAgentLoopTestDependencies(ctx, {
systemPrompt: { persona: 'Test persona.' },
tools: { mode: 'native' },
})
expect(renderPrompt(await ctx.systemPrompt.assemble())).toContain('Test persona.')
await expect(ctx.plugin(AgentLoop, { agents: [] })).resolves.toBeDefined()
await ctx.fiber.dispose()
})
})

View File

@@ -0,0 +1,33 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../core/agent"
},
{
"path": "../../llm/llm"
},
{
"path": "../../core/session"
},
{
"path": "../../core/system-prompt"
},
{
"path": "../../core/tools"
}
]
}

View File

@@ -4,7 +4,7 @@ Runtime event-contract assertions intended for development diagnostics. This pur
The plugin has no environment guard: it is active wherever it is registered. The default [`dsh-agent-spine-demo`](../../examples/agent-spine-demo/README.md) bundle mounts it unconditionally; a custom composition can omit it when the runtime cost is undesirable. It doubles as executable documentation of the event taxonomy — the assertions *are* the contract.
Session itself owns immutable log storage in every composition: it takes one lossless JSON snapshot of each accepted event, deep-freezes that record, and exposes the log through immutable array snapshots. The invariants plugin checks the cross-record and cross-seam rules that storage immutability cannot express.
Session itself owns immutable, surface-valid log storage in every composition: it takes one lossless JSON snapshot of each candidate, validates the complete surface transition, deep-freezes the accepted record, and exposes the log through immutable array snapshots. The invariants plugin checks the remaining cross-record and cross-seam rules that Session does not own.
Session-log assertions run during Cordis `internal/dispatch`, while `Session.append()` is resolving the `session/event` callback snapshot but before it pushes the candidate into the log. A valid transition is staged by exact event identity and applied to the live trace only when that same committed event reaches the plugin's contained post-commit listener. A later internal dispatch check can therefore veto without advancing either the log or the invariant trace, while ordinary `session/event` observer failures remain observe-only.

View File

@@ -3,7 +3,8 @@
* turn and step nesting, scoped dispatch, status transitions, and request
* reconstruction. The plugin has no environment guard and is active wherever
* mounted, including the default `dsh-agent-spine-demo` bundle; custom compositions
* may omit it. Sessions still own event snapshots and freezing.
* may omit it. Sessions own immutable, surface-valid event storage; this plugin
* checks only relationships that event acceptance cannot express.
* @module @deepseek-ai/dsh-invariants
*/
@@ -13,7 +14,7 @@ import { assertNever, HarnessError } from '@deepseek-ai/dsh-llm'
import type { CallId, GenerateOptions } from '@deepseek-ai/dsh-llm'
import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent'
import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import { scopedSubjectResolverFor } from './scoped-events.generated.ts'
export const name = 'invariants'
@@ -48,15 +49,6 @@ interface SessionTrace {
* `step/end` — a result must arrive in the same step as its call.
*/
pendingCalls: Set<CallId>
/** Every seq seen so far — validates `sourceEventSeqs` references. */
knownSeqs: Set<number>
/**
* The seqs currently on the surface linked list, in linked-list order
* (head to tail). A replace reorders this relative to seq order (the new
* node takes the replaced range's position), so range validation is
* positional, not by seq comparison.
*/
surface: number[]
}
/** One accepted event's deferred mutation of a live session trace. */
@@ -68,12 +60,6 @@ interface SessionTraceTransition {
| { kind: 'none' }
| { kind: 'add' | 'delete'; callId: CallId }
| { kind: 'clear' }
/** The event's mutation of the derived surface order. */
surface:
| { kind: 'none' | 'append' }
| { kind: 'replace'; start: number; count: number }
/** The committed event sequence to add to the known-sequence set. */
seq: number
}
/** Assert that a step-scoped event names the currently open turn and step. */
@@ -97,73 +83,6 @@ function validateEvent(trace: SessionTrace, event: SessionEvent): SessionTraceTr
let nextTurn = trace.nextTurn
let nextStep = trace.nextStep
let pendingCalls: SessionTraceTransition['pendingCalls'] = { kind: 'none' }
let surface: SessionTraceTransition['surface'] = { kind: 'none' }
// --- Surface invariants ---
// Surface metadata (sourceEventSeqs, surfaceOp) is only valid on
// surface-eligible event types. The compiler enforces this at append()
// call sites; this runtime check catches casts and persisted data.
const SURFACE_TYPES = new Set<string>(['user/message', 'assistant/message', 'tool/result', 'context/message', 'steering/message'])
// Cast to surface-eligible event type so we can access surfaceOp and
// sourceEventSeqs (optional on SessionEvent, mandatory on SurfaceEvent).
// SurfaceEvent's mandatory surfaceOp is too strict here — we need to
// CHECK whether surface metadata is present, not assume it.
const se = event as SessionEvent<SurfaceEventType>
if (!SURFACE_TYPES.has(event.type)) {
if (se.sourceEventSeqs !== undefined) {
throw new InvariantError(`${event.type} cannot carry sourceEventSeqs (non-surface event)`)
}
if (se.surfaceOp !== undefined) {
throw new InvariantError(`${event.type} cannot carry surfaceOp (non-surface event)`)
}
}
if (se.sourceEventSeqs !== undefined) {
if (se.sourceEventSeqs.length === 0) {
throw new InvariantError('sourceEventSeqs must not be empty when present')
}
const unique = new Set(se.sourceEventSeqs)
if (unique.size !== se.sourceEventSeqs.length) {
throw new InvariantError('sourceEventSeqs must not contain duplicates')
}
for (const ref of se.sourceEventSeqs) {
if (ref >= event.seq) {
throw new InvariantError(`sourceEventSeqs must reference earlier events: ${ref} >= current seq ${event.seq}`)
}
if (!trace.knownSeqs.has(ref)) {
throw new InvariantError(`sourceEventSeqs references unknown seq ${ref}`)
}
}
}
// Fold this event into the tracked surface linked list, validating the
// replace contract as we go. `append` adds a tail node; `replace` shadows a
// positional range — every shadowed node must appear in sourceEventSeqs.
if (se.surfaceOp !== undefined) {
if (se.surfaceOp === 'append') {
surface = { kind: 'append' }
} else {
const { start, end } = se.surfaceOp
const startIdx = trace.surface.indexOf(start)
if (startIdx === -1) {
throw new InvariantError(`surface replace: start seq ${start} is not on the surface`)
}
const endIdx = trace.surface.indexOf(end)
if (endIdx === -1) {
throw new InvariantError(`surface replace: end seq ${end} is not on the surface`)
}
if (startIdx > endIdx) {
throw new InvariantError(`surface replace: start seq ${start} (pos ${startIdx}) is after end seq ${end} (pos ${endIdx}) on the surface`)
}
// Every node the replace shadows (surface positions [startIdx, endIdx]
// inclusive) must appear in sourceEventSeqs — the provenance contract.
const shadowed = trace.surface.slice(startIdx, endIdx + 1)
const recorded = new Set(se.sourceEventSeqs ?? [])
const missing = shadowed.filter(seq => !recorded.has(seq))
if (missing.length > 0) {
throw new InvariantError(`surface replace: sourceEventSeqs must include every shadowed surface node; missing ${missing.join(', ')}`)
}
surface = { kind: 'replace', start: startIdx, count: shadowed.length }
}
}
// Boundary/step-scoped events have explicit cases; every OTHER event type —
// including plugin-added (merge-extensible) SessionEventMap keys — is caught
@@ -263,8 +182,6 @@ function validateEvent(trace: SessionTrace, event: SessionEvent): SessionTraceTr
return {
scalars: { lastSeq: event.seq, openTurn, openStep, nextTurn, nextStep },
pendingCalls,
surface,
seq: event.seq,
}
}
@@ -287,20 +204,6 @@ function applyTransition(trace: SessionTrace, transition: SessionTraceTransition
default:
assertNever(transition.pendingCalls, 'session trace pending-call transition')
}
switch (transition.surface.kind) {
case 'none':
break
case 'append':
trace.surface.push(transition.seq)
break
case 'replace':
trace.surface.splice(transition.surface.start, transition.surface.count, transition.seq)
break
/* v8 ignore next -- validateEvent produces this closed transition union */
default:
assertNever(transition.surface, 'session trace surface transition')
}
trace.knownSeqs.add(transition.seq)
}
/** Validate and apply one event while rebuilding an already-committed log. */
@@ -345,8 +248,6 @@ export function apply(ctx: Context): void {
nextTurn: 1,
nextStep: 1,
pendingCalls: new Set(),
knownSeqs: new Set(),
surface: [],
})
/** Build (or rebuild) a session's trace by replaying its whole log. */

View File

@@ -464,7 +464,7 @@ describe('HMR safety', () => {
})
})
describe('surface invariants', () => {
describe('surface contract under the invariants composition', () => {
it('accepts well-formed surface metadata', async () => {
const { ctx } = await setup()
const session = ctx.sessions.create()
@@ -493,7 +493,7 @@ describe('surface invariants', () => {
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [] })
}).toThrow(InvariantError)
}).toThrow(/must not be empty/)
})
it('rejects duplicate sourceEventSeqs', async () => {
@@ -518,7 +518,8 @@ describe('surface invariants', () => {
})
it('accepts sourceEventSeqs referencing a valid earlier event', async () => {
// Positive test: ref < current seq and ref is in knownSeqs → passes.
// Session seqs are contiguous, so every non-negative ref below the current
// seq necessarily names an existing earlier event.
const { ctx } = await setup()
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
@@ -538,23 +539,6 @@ describe('surface invariants', () => {
}).toThrow(/must reference earlier/)
})
it('rejects sourceEventSeqs referencing unknown seq (gap in event log)', async () => {
// Create an impossible-through-public-API gap so seq 2 is earlier but unknown.
const { ctx } = await setup()
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
session.append('step/start', { turn: 1, step: 1 })
;(session as unknown as { log: unknown[] }).log.push({
type: 'assistant/chunk',
seq: 3,
time: Date.now(),
data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'x' } },
})
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [2] })
}).toThrow(/unknown seq 2/)
})
it('rejects a replace whose start is positioned after its end on the surface', async () => {
const { ctx } = await setup()
const session = ctx.sessions.create()
@@ -565,7 +549,7 @@ describe('surface invariants', () => {
// Reversed range: start seq 3 is at a later surface position than end seq 2.
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 3, end: 2 }, sourceEventSeqs: [2, 3] })
}).toThrow(/is after end seq 2 .* on the surface/)
}).toThrow(/is after end seq 2/)
})
it('rejects a replace whose sourceEventSeqs omits a shadowed surface node', async () => {
@@ -602,7 +586,7 @@ describe('surface invariants', () => {
// seq 1 (step/start) is a real earlier event but never entered the surface.
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 1, end: 2 }, sourceEventSeqs: [1, 2] })
}).toThrow(/start seq 1 is not on the surface/)
}).toThrow(/start seq 1 not found in surface/)
})
it('rejects a replace naming an end seq that is not on the surface', async () => {
@@ -614,7 +598,7 @@ describe('surface invariants', () => {
// start (2) is on the surface but end (99) never entered it.
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 2, end: 99 }, sourceEventSeqs: [2] })
}).toThrow(/end seq 99 is not on the surface/)
}).toThrow(/end seq 99 not found in surface/)
})
it('rejects a replace whose range is reversed in surface position after a prior replace reordered it', async () => {
@@ -631,7 +615,7 @@ describe('surface invariants', () => {
// reversed positionally (3 is at pos 1, 4 is at pos 0).
expect(() => {
session.append('assistant/message', { provenance: { provider: 'mock', model: 'mock' }, turn: 1, step: 1, content: [] }, { surfaceOp: { op: 'replace', start: 3, end: 4 }, sourceEventSeqs: [3, 4] }) // seq 5
}).toThrow(/is after end seq 4 .* on the surface/)
}).toThrow(/is after end seq 4/)
})
it('accepts a replace whose start seq exceeds its end seq when the surface position order is valid', async () => {
@@ -675,25 +659,6 @@ describe('surface invariants', () => {
expect(() => ctx.sessions.create(undefined, { seed: badSeed })).toThrow(/must include every shadowed surface node; missing 3/)
})
it('rejects sourceEventSeqs on a non-surface event', async () => {
const { ctx } = await setup()
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
// Session rejects this at its own acceptance boundary. Emit a hand-built
// record to cover the listener's defensive check for alternate producers.
const event = { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } }, sourceEventSeqs: [0] }
expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, event as never) })
.toThrow(/cannot carry sourceEventSeqs/)
})
it('rejects surfaceOp on a non-surface event', async () => {
const { ctx } = await setup()
const session = ctx.sessions.create()
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
const event = { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } }, surfaceOp: 'append' }
expect(() => { ctx.emit(scopeTarget(session, undefined), 'session/event', session, event as never) })
.toThrow(/cannot carry surfaceOp/)
})
})
describe('request-reconstruction cross-check (llm/stream)', () => {