Merge remote-tracking branch 'origin/master' into codex/truncated-design
# Conflicts: # docs/config-catalog.md # docs/event-producer-consumer.md # docs/rfc/INDEX.md # packages/cordis/tool-cordis/src/api-catalog.ts # pnpm-lock.yaml
This commit is contained in:
Dudu-0223
@@ -5,7 +5,7 @@ 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) |
|
||||
| `invariants/` | Dev-mode event-contract invariants + session-log freeze | (listens on `session/*`, `agent/*`) |
|
||||
| `invariants/` | Dev-mode event-contract assertions | (listens on `session/*`, `agent/*`) |
|
||||
| `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`) |
|
||||
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
# dsh-invariants
|
||||
|
||||
Dev-mode event-contract invariants and session-log freeze. A pure-listener plugin (everything is a plugin) that asserts the harness event contract at runtime and, optionally, freezes logged session-event data so any code that mutates history throws instead of corrupting silently.
|
||||
Dev-mode event-contract assertions. This pure-listener plugin checks relationships among session events, agent states, scoped dispatches, and model requests at runtime; it does not own or change product behavior.
|
||||
|
||||
**Off in production.** Enable it in tests and the demos, where a contract violation should fail loudly. It costs nothing when not registered, and 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-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.
|
||||
|
||||
## Plugin
|
||||
|
||||
A functional plugin — register the module namespace (this is what loading by name in `cordis.yml` does):
|
||||
@@ -14,17 +18,10 @@ import * as Invariants from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
declare const ctx: Context
|
||||
|
||||
await ctx.plugin(Invariants) // freeze on (default)
|
||||
await ctx.plugin(Invariants, { freeze: false }) // assert contract, don't freeze
|
||||
await ctx.plugin(Invariants)
|
||||
```
|
||||
|
||||
`inject`: `['sessions']` — it reads `ctx.sessions.list()` at apply time to rebuild trace state for sessions that already exist (so a hot reload mid-turn doesn't falsely reject the next event). It listens on `session/created`, `session/event`, and `agent/status`.
|
||||
|
||||
### Config
|
||||
|
||||
| Key | Default | Meaning |
|
||||
|---|---|---|
|
||||
| `freeze` | `true` | Deep-freeze each logged event's data so mutating a logged event throws. Set `false` to assert the contract without freezing. |
|
||||
`inject`: `['sessions']` — it reads `ctx.sessions.list()` at apply time to rebuild trace state for sessions that already exist, so a hot reload mid-turn does not falsely reject the next event. The oracle listeners are explicitly global so pre-commit staging and post-commit application keep the same audience even if the plugin is mounted under a scoped context; their cleanup still belongs to that mounting fiber. The plugin has no configuration.
|
||||
|
||||
## Invariants asserted
|
||||
|
||||
@@ -46,10 +43,10 @@ Model requests (on `llm/stream`):
|
||||
|
||||
On any violation it throws `InvariantError` (`code: 'INVARIANT'`).
|
||||
|
||||
## Why runtime, not deep-readonly types
|
||||
## Why runtime assertions remain useful
|
||||
|
||||
A `DeepReadonly<SessionEvent>` is high type-noise across every log consumer, and a plugin can cast straight through it. A dev-mode freeze plus these assertions catch real corruption at zero production cost and zero type noise. The always-on half of that defense — cloning derived messages so request/adapter mutation can't reach back into the log — lives in `dsh-session`'s `deriveMessages`. This package is the dev-mode tripwire. See [dev-mode invariants](../../../docs/rfc/implemented/architecture/2026-06-11-dev-invariants-over-deep-readonly.md).
|
||||
Session enforces the per-record storage boundary at runtime, where a cast cannot bypass it. Pervasive `DeepReadonly<SessionEvent>` types would add noise across consumers without expressing relationships such as turn/step nesting, subject-correct scoped dispatch, or equality between a request and its log reconstruction. This plugin checks those relationships in development while `dsh-session` keeps history immutable in every composition. See [source-owned session immutability and dev-mode invariants](../../../docs/rfc/implemented/architecture/2026-06-11-dev-invariants-over-deep-readonly.md).
|
||||
|
||||
## Seeded sessions
|
||||
|
||||
A seeded/forked session arrives with events already in its log (the `Session` constructor copies the seed without emitting `session/event`). On `session/created` the plugin replays the existing log through the checker and freezes those entries, so seeded history is held to the same contract.
|
||||
A seeded or forked session arrives with events already in its log because construction does not emit `session/event` for each seed record. `Session` validates, snapshots, and freezes every seed record before accepting it; on `session/created`, this plugin replays the accepted log only to rebuild and check its relational trace state.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-invariants",
|
||||
"description": "Dev-mode event-contract invariants + session-log freeze for the DeepSeek Harness",
|
||||
"description": "Dev-mode event-contract assertions for the DeepSeek Harness",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
@@ -24,12 +24,14 @@
|
||||
"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",
|
||||
"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:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
|
||||
@@ -1,26 +1,25 @@
|
||||
/**
|
||||
* Dev-mode invariants: a pure-listener plugin that asserts the harness event
|
||||
* contract at runtime, and (optionally) freezes logged session-event data so
|
||||
* any code that mutates history throws instead of corrupting silently.
|
||||
* Dev-mode invariants: a pure-listener plugin that asserts relationships in
|
||||
* the harness event contract at runtime.
|
||||
*
|
||||
* Everything is a plugin — this is just listeners on `session/created`,
|
||||
* `session/event`, and `agent/status`. It is **off in production**: enable it
|
||||
* in tests and the demos, where a contract violation should be a loud failure,
|
||||
* not a subtle one. It doubles as executable documentation of the event
|
||||
* taxonomy: the assertions below ARE the contract.
|
||||
* `session/event`, `agent/status`, and the scoped dispatch and request seams.
|
||||
* It is **off in production**: enable it in tests and demos, where a contract
|
||||
* violation should be a loud failure rather than a subtle one. It doubles as
|
||||
* executable documentation of the event taxonomy: the assertions below are
|
||||
* the contract.
|
||||
*
|
||||
* Why runtime assertions instead of compile-time deep-readonly types? See
|
||||
* the dev-invariants RFC. Briefly: a `DeepReadonly<SessionEvent>` is high type-noise across
|
||||
* every log consumer and a plugin casts straight through it; a dev-mode freeze
|
||||
* + assertions catch real corruption at zero production cost and zero type
|
||||
* noise. The always-on half of that defense (cloning derived messages) lives
|
||||
* in dsh-session; this package is the dev-mode tripwire.
|
||||
* Session owns immutable log storage: it snapshots and deep-freezes every
|
||||
* accepted event at the source. This plugin checks relationships that one
|
||||
* event's types and immutability cannot express, including turn/step nesting,
|
||||
* scoped dispatch, status transitions, and request reconstructability.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-invariants
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { HarnessError } from '@deepseek-ai/dsh-llm'
|
||||
import { carrierKeyOf, isScopeCarrier } from '@deepseek-ai/dsh-scope'
|
||||
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'
|
||||
@@ -41,16 +40,6 @@ export class InvariantError extends HarnessError {
|
||||
}
|
||||
}
|
||||
|
||||
/** Plugin config. */
|
||||
export interface Config {
|
||||
/**
|
||||
* Deep-freeze logged session-event data so mutating a logged event throws.
|
||||
* Default true — this plugin only runs in dev/test, where freezing is the
|
||||
* point. Set false to assert the event contract without freezing.
|
||||
*/
|
||||
freeze?: boolean
|
||||
}
|
||||
|
||||
/** Per-session bookkeeping for the session-log invariants. */
|
||||
interface SessionTrace {
|
||||
/** Highest `seq` seen so far (must strictly increase). */
|
||||
@@ -79,27 +68,32 @@ interface SessionTrace {
|
||||
surface: number[]
|
||||
}
|
||||
|
||||
/**
|
||||
* Deep-freeze a value and everything reachable from it.
|
||||
*
|
||||
* Walks every object's own properties even when the object itself is already
|
||||
* frozen: `Session.append()` accepts event data from arbitrary plugins/tools,
|
||||
* so a caller can hand us a SHALLOW-frozen object whose descendants are still
|
||||
* mutable. Skipping an already-frozen node (the obvious idempotence shortcut)
|
||||
* would leave exactly the kind of mutable history the dev-invariants RFC means to catch. A
|
||||
* `WeakSet` of visited objects keeps it terminating on cycles and avoids
|
||||
* re-walking shared subtrees / already-processed seed events.
|
||||
*/
|
||||
function deepFreeze(value: unknown, seen: WeakSet<object> = new WeakSet()): void {
|
||||
if (value === null || typeof value !== 'object') return
|
||||
if (seen.has(value)) return
|
||||
seen.add(value)
|
||||
// Freeze the node (no-op if a caller pre-froze it), then ALWAYS descend —
|
||||
// a frozen container can still hold mutable children.
|
||||
Object.freeze(value)
|
||||
for (const key of Object.keys(value)) {
|
||||
deepFreeze((value as Record<string, unknown>)[key], seen)
|
||||
}
|
||||
/** One accepted event's deferred mutation of a live session trace. */
|
||||
interface SessionTraceTransition {
|
||||
/** Scalar state after the event commits. */
|
||||
scalars: Pick<SessionTrace, 'lastSeq' | 'openTurn' | 'openStep' | 'nextTurn' | 'nextStep'>
|
||||
/** The event's mutation of the open step's pending call set. */
|
||||
pendingCalls:
|
||||
| { 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
|
||||
}
|
||||
|
||||
/** Event payload prefix for scoped seams whose first argument names its agent. */
|
||||
interface AgentSubject {
|
||||
agent: Agent
|
||||
}
|
||||
|
||||
/** Structural subject fields used without coupling this dev plugin to owning services. */
|
||||
interface ScopedSubjectFields {
|
||||
agent?: Agent
|
||||
scope?: object
|
||||
}
|
||||
|
||||
/** Assert that a step-scoped event names the currently open turn and step. */
|
||||
@@ -111,14 +105,19 @@ function requireOpenStep(trace: SessionTrace, kind: string, turn: number, step:
|
||||
}
|
||||
}
|
||||
|
||||
/** Assert one appended event against the per-session invariants. */
|
||||
function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
/** Validate one candidate event without mutating the committed session trace. */
|
||||
function validateEvent(trace: SessionTrace, event: SessionEvent): SessionTraceTransition {
|
||||
// seq is strictly monotonic — the spine of replay equivalence. lastSeq
|
||||
// starts at -1, so the first event (seq 0) passes.
|
||||
if (event.seq <= trace.lastSeq) {
|
||||
throw new InvariantError(`seq must strictly increase: saw ${event.seq} after ${trace.lastSeq}`)
|
||||
}
|
||||
trace.lastSeq = event.seq
|
||||
let openTurn = trace.openTurn
|
||||
let openStep = trace.openStep
|
||||
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
|
||||
@@ -160,7 +159,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
// positional range — every shadowed node must appear in sourceEventSeqs.
|
||||
if (se.surfaceOp !== undefined) {
|
||||
if (se.surfaceOp === 'append') {
|
||||
trace.surface.push(event.seq)
|
||||
surface = { kind: 'append' }
|
||||
} else {
|
||||
const { start, end } = se.surfaceOp
|
||||
const startIdx = trace.surface.indexOf(start)
|
||||
@@ -182,9 +181,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
if (missing.length > 0) {
|
||||
throw new InvariantError(`surface replace: sourceEventSeqs must include every shadowed surface node; missing ${missing.join(', ')}`)
|
||||
}
|
||||
// Apply the replace to the tracked surface: the new node takes the
|
||||
// range's position so order stays in sync for later replaces.
|
||||
trace.surface.splice(startIdx, shadowed.length, event.seq)
|
||||
surface = { kind: 'replace', start: startIdx, count: shadowed.length }
|
||||
}
|
||||
}
|
||||
|
||||
@@ -203,8 +200,8 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
if (event.data.turn !== trace.nextTurn) {
|
||||
throw new InvariantError(`turn/start expected turn ${trace.nextTurn}, got ${event.data.turn}`)
|
||||
}
|
||||
trace.openTurn = event.data.turn
|
||||
trace.nextStep = 1
|
||||
openTurn = event.data.turn
|
||||
nextStep = 1
|
||||
break
|
||||
}
|
||||
case 'turn/end': {
|
||||
@@ -214,8 +211,8 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
if (trace.openStep !== null) {
|
||||
throw new InvariantError(`turn/end ${event.data.turn} while step ${trace.openStep} is still open`)
|
||||
}
|
||||
trace.openTurn = null
|
||||
trace.nextTurn += 1
|
||||
openTurn = null
|
||||
nextTurn += 1
|
||||
break
|
||||
}
|
||||
case 'step/start': {
|
||||
@@ -229,16 +226,16 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
if (event.data.step !== trace.nextStep) {
|
||||
throw new InvariantError(`step/start expected step ${trace.nextStep} in turn ${event.data.turn}, got ${event.data.step}`)
|
||||
}
|
||||
trace.openStep = event.data.step
|
||||
openStep = event.data.step
|
||||
break
|
||||
}
|
||||
case 'step/end': {
|
||||
requireOpenStep(trace, 'step/end', event.data.turn, event.data.step)
|
||||
// A result must arrive in the step that issued the call; orphan calls
|
||||
// (a step that errored before its result) do not carry to the next step.
|
||||
trace.pendingCalls.clear()
|
||||
trace.openStep = null
|
||||
trace.nextStep += 1
|
||||
pendingCalls = { kind: 'clear' }
|
||||
openStep = null
|
||||
nextStep += 1
|
||||
break
|
||||
}
|
||||
case 'assistant/chunk': {
|
||||
@@ -251,7 +248,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
}
|
||||
case 'tool/call': {
|
||||
requireOpenStep(trace, 'tool/call', event.data.turn, event.data.step)
|
||||
trace.pendingCalls.add(event.data.callId)
|
||||
pendingCalls = { kind: 'add', callId: event.data.callId }
|
||||
break
|
||||
}
|
||||
case 'tool/result': {
|
||||
@@ -260,9 +257,10 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
// does NOT hold: a call may have no result — a throwing tool-execution
|
||||
// pipeline step ends the turn with no tool/result, which is legal.)
|
||||
const syntheticInterrupted = event.data.isError && event.data.error?.code === 'interrupted'
|
||||
if (!trace.pendingCalls.delete(event.data.callId) && !syntheticInterrupted) {
|
||||
if (!trace.pendingCalls.has(event.data.callId) && !syntheticInterrupted) {
|
||||
throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call in this step`)
|
||||
}
|
||||
pendingCalls = { kind: 'delete', callId: event.data.callId }
|
||||
break
|
||||
}
|
||||
// Turn-enclosure (the turn-enclosure RFC): EVERY session event not handled by a boundary
|
||||
@@ -282,8 +280,52 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
break
|
||||
}
|
||||
}
|
||||
// Track every seq seen — used above to validate sourceEventSeqs references.
|
||||
trace.knownSeqs.add(event.seq)
|
||||
return {
|
||||
scalars: { lastSeq: event.seq, openTurn, openStep, nextTurn, nextStep },
|
||||
pendingCalls,
|
||||
surface,
|
||||
seq: event.seq,
|
||||
}
|
||||
}
|
||||
|
||||
/** Apply one already-validated transition after its event commits. */
|
||||
function applyTransition(trace: SessionTrace, transition: SessionTraceTransition): void {
|
||||
Object.assign(trace, transition.scalars)
|
||||
switch (transition.pendingCalls.kind) {
|
||||
case 'none':
|
||||
break
|
||||
case 'add':
|
||||
trace.pendingCalls.add(transition.pendingCalls.callId)
|
||||
break
|
||||
case 'delete':
|
||||
trace.pendingCalls.delete(transition.pendingCalls.callId)
|
||||
break
|
||||
case 'clear':
|
||||
trace.pendingCalls.clear()
|
||||
break
|
||||
/* v8 ignore next -- validateEvent produces this closed transition union */
|
||||
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. */
|
||||
function replayEvent(trace: SessionTrace, event: SessionEvent): void {
|
||||
applyTransition(trace, validateEvent(trace, event))
|
||||
}
|
||||
|
||||
/** Legal agent status transitions (the only state machine the loop guarantees). */
|
||||
@@ -303,14 +345,19 @@ function checkTransition(from: AgentStatus | undefined, to: AgentStatus): void {
|
||||
|
||||
/**
|
||||
* Register the dev-mode invariants. Contributions are effect-scoped, so
|
||||
* disposing the plugin fiber removes all listeners and stops freezing
|
||||
* (HMR-safe). On (re-)apply the trace state is rebuilt by replaying each
|
||||
* existing session's log, so a hot reload mid-turn does not falsely reject the
|
||||
* next event.
|
||||
* disposing the plugin fiber removes all listeners (HMR-safe). On (re-)apply
|
||||
* the trace state is rebuilt by replaying each existing session's log, so a
|
||||
* hot reload mid-turn does not falsely reject the next event.
|
||||
*
|
||||
* @param ctx - Cordis context that receives the invariant listeners.
|
||||
*/
|
||||
export function apply(ctx: Context, config: Config = {}): void {
|
||||
const freeze = config.freeze ?? true
|
||||
export function apply(ctx: Context): void {
|
||||
const traces = new WeakMap<Session, SessionTrace>()
|
||||
const stagedTransitions = new WeakMap<SessionEvent, {
|
||||
session: Session
|
||||
trace: SessionTrace
|
||||
transition: SessionTraceTransition
|
||||
}>()
|
||||
// Agent status has no stored history to replay; the first observation after
|
||||
// (re-)apply seeds the baseline, so a reload never produces a false positive.
|
||||
const lastStatus = new WeakMap<Agent, AgentStatus>()
|
||||
@@ -326,20 +373,19 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
surface: [],
|
||||
})
|
||||
|
||||
/** Build (or rebuild) a session's trace by replaying its whole log; freeze it. */
|
||||
/** Build (or rebuild) a session's trace by replaying its whole log. */
|
||||
const seedSession = (session: Session): SessionTrace => {
|
||||
const trace = freshTrace()
|
||||
traces.set(session, trace)
|
||||
for (const event of session.events) {
|
||||
checkEvent(trace, event)
|
||||
if (freeze) deepFreeze(event)
|
||||
replayEvent(trace, event)
|
||||
}
|
||||
return trace
|
||||
}
|
||||
|
||||
// Every store-created session (the only kind that emits session/event) is
|
||||
// seeded first — via ctx.sessions.list() at apply or session/created — so
|
||||
// the fallback is a defensive guard, never hit in practice.
|
||||
// seeded first — via ctx.sessions.list() at apply or session/created — so the
|
||||
// fallback is a defensive guard, never hit in practice.
|
||||
/* v8 ignore next -- traceFor's fallback: session/event always follows a seed */
|
||||
const traceFor = (session: Session): SessionTrace => traces.get(session) ?? seedSession(session)
|
||||
|
||||
@@ -350,17 +396,90 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
|
||||
// A newly created session may arrive seeded/forked (the constructor copies
|
||||
// the seed WITHOUT emitting session/event), so replay its log here too.
|
||||
ctx.on('session/created', (session) => { seedSession(session) })
|
||||
ctx.on('session/created', (session) => { seedSession(session) }, { global: true })
|
||||
|
||||
ctx.on('session/event', (session, event) => {
|
||||
checkEvent(traceFor(session), event)
|
||||
if (freeze) deepFreeze(event)
|
||||
})
|
||||
// Session resolves dispatch before committing, so internal/dispatch has
|
||||
// already staged this exact event. A later dispatch veto skips every
|
||||
// session/event callback and therefore leaves the live trace unchanged.
|
||||
const staged = stagedTransitions.get(event)
|
||||
/* v8 ignore next 2 -- internal/dispatch stages the exact callback arguments */
|
||||
if (staged === undefined || staged.session !== session) {
|
||||
throw new InvariantError('session/event reached publication without matching pre-commit validation')
|
||||
}
|
||||
stagedTransitions.delete(event)
|
||||
applyTransition(staged.trace, staged.transition)
|
||||
}, { global: true })
|
||||
|
||||
ctx.on('agent/status', (agent, status) => {
|
||||
checkTransition(lastStatus.get(agent), status)
|
||||
lastStatus.set(agent, status)
|
||||
})
|
||||
}, { global: true })
|
||||
|
||||
// --- 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/session-prefix': args => args[0],
|
||||
'agent/step-result': args => args[0],
|
||||
'agent/turn-continuation': args => args[0],
|
||||
'agent/turn-stop': args => args[0],
|
||||
'agent/error': args => args[0],
|
||||
'approval/request': args => (args[0] as AgentSubject).agent,
|
||||
'tools/pre-execute': args => (args[0] as ScopedSubjectFields).agent,
|
||||
'tools/execute': args => (args[0] as ScopedSubjectFields).agent,
|
||||
'tools/post-execute': args => (args[0] as ScopedSubjectFields).agent,
|
||||
'tools/result': args => (args[0] as ScopedSubjectFields).agent,
|
||||
'system-prompt/assemble': args => (args[1] as ScopedSubjectFields).scope,
|
||||
'session/created': null,
|
||||
'session/disposed': 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))')
|
||||
}
|
||||
if (name === 'session/event') {
|
||||
const [session, event] = args as [Session, SessionEvent]
|
||||
const trace = traceFor(session)
|
||||
const transition = validateEvent(trace, event)
|
||||
// The exact event identity reaches the contained post-commit listener.
|
||||
// A later internal/dispatch listener may still veto; because validation
|
||||
// is pure, abandoning this weakly keyed transition does not advance the
|
||||
// committed trace or retain the session.
|
||||
stagedTransitions.set(event, { session, trace, transition })
|
||||
}
|
||||
}, { global: true })
|
||||
|
||||
// Request-reconstruction cross-check (the reconstructability RFC): a
|
||||
// loop-built request — frozen envelope + live sessionId is the marker; a
|
||||
@@ -437,5 +556,5 @@ export function apply(ctx: Context, config: Config = {}): void {
|
||||
throw new InvariantError(`llm request for session "${String(session.id)}" diverges from the folded request header`)
|
||||
}
|
||||
return next()
|
||||
}, { prepend: true })
|
||||
}, { global: true, prepend: true })
|
||||
}
|
||||
|
||||
@@ -1,16 +1,17 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { createScope, 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'
|
||||
import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import * as Invariants from '@deepseek-ai/dsh-invariants'
|
||||
import { InvariantError } from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
/** A Context with the session store and the invariants plugin registered. */
|
||||
async function setup(config?: { freeze?: boolean }) {
|
||||
async function setup() {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const fiber = await ctx.plugin(Invariants, config ?? {})
|
||||
const fiber = await ctx.plugin(Invariants)
|
||||
return { ctx, fiber }
|
||||
}
|
||||
|
||||
@@ -20,8 +21,27 @@ function mockAgent(id: string): Agent {
|
||||
}
|
||||
|
||||
describe('session-log invariants', () => {
|
||||
it('keeps pre-commit staging and post-commit application global when mounted under a scope', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
let scopedCtx!: Context
|
||||
await ctx.plugin(Object.assign((inner: Context) => {
|
||||
scopedCtx = createScope(inner, {}).ctx
|
||||
}, { inject: ['sessions'] }))
|
||||
await scopedCtx.plugin(Invariants)
|
||||
const globalSession = ctx.sessions.create(SessionId('global-under-scoped-invariants'))
|
||||
|
||||
expect(() => {
|
||||
globalSession.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'message', source: { kind: 'user' } },
|
||||
})
|
||||
globalSession.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}).not.toThrow()
|
||||
})
|
||||
|
||||
it('accepts a well-formed turn/step/tool sequence', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -36,18 +56,59 @@ describe('session-log invariants', () => {
|
||||
}).not.toThrow()
|
||||
})
|
||||
|
||||
it('does not advance the trace when a later internal-dispatch listener vetoes', async () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create(SessionId('dispatch-veto-rollback'))
|
||||
let veto = true
|
||||
ctx.on('internal/dispatch', (_mode, name) => {
|
||||
if (name !== 'session/event' || !veto) return
|
||||
veto = false
|
||||
throw new Error('later dispatch veto')
|
||||
})
|
||||
|
||||
expect(() => session.append('turn/start', {
|
||||
turn: 1,
|
||||
trigger: { kind: 'message', source: { kind: 'user' } },
|
||||
})).toThrow('later dispatch veto')
|
||||
expect(session.events).toEqual([])
|
||||
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}).not.toThrow()
|
||||
expect(session.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
|
||||
})
|
||||
|
||||
it('applies the committed transition after a prepended observer throws', async () => {
|
||||
const { ctx } = await setup()
|
||||
const warnings: string[] = []
|
||||
ctx.logger.warn = ((message: unknown) => { warnings.push(String(message)) }) as typeof ctx.logger.warn
|
||||
const session = ctx.sessions.create(SessionId('postcommit-peer'))
|
||||
ctx.on('session/event', () => { throw new Error('hostile observer') }, { prepend: true })
|
||||
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
}).not.toThrow()
|
||||
expect(session.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
|
||||
expect(warnings).toEqual([
|
||||
'session "postcommit-peer": session/event listener threw: Error: hostile observer',
|
||||
'session "postcommit-peer": session/event listener threw: Error: hostile observer',
|
||||
])
|
||||
})
|
||||
|
||||
it('rejects a non-monotonic seq (replay spine)', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
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/)
|
||||
})
|
||||
|
||||
it('rejects a turn/start while another turn is open', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }))
|
||||
@@ -55,7 +116,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a turn/end that does not match the open turn', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => session.append('turn/end', { turn: 2, reason: { kind: 'completed' } }))
|
||||
@@ -63,14 +124,14 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a step/start outside its declared turn', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => session.append('step/start', { turn: 2, step: 1 })).toThrow(/open turn is 1/)
|
||||
})
|
||||
|
||||
it('rejects a step/end that does not match the open step', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -78,7 +139,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects an assistant/chunk outside an open step', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'x' } }))
|
||||
@@ -86,7 +147,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a message event appended outside any open turn (turn-enclosure)', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
// No turn open: every message-bearing event must be turn-enclosed (the turn-enclosure RFC).
|
||||
expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }))
|
||||
@@ -96,7 +157,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects steering and plugin-added events appended outside any open turn', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
// steering/message is turn-scoped: outside a turn it would land past the
|
||||
// commit boundary and be dropped on resume (the turn-enclosure RFC).
|
||||
@@ -112,7 +173,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('accepts message events once a turn is open', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' }))
|
||||
@@ -120,7 +181,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a tool/result with no prior tool/call', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -129,7 +190,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('allows a synthetic interrupted tool/result from crash repair without a prior tool/call event', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -151,7 +212,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('allows a tool/call with no matching tool/result (thrown waterfall ends the step)', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -163,7 +224,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('holds seeded sessions to the contract on session/created', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
// A seq-contiguous, serializable seed (so it passes Session's constructor
|
||||
// validation) that nonetheless violates turn nesting — a second turn/start
|
||||
// while the first turn is still open — must be rejected by the invariants
|
||||
@@ -176,7 +237,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('tracks turns per session independently', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const a = ctx.sessions.create(SessionId('a'))
|
||||
const b = ctx.sessions.create(SessionId('b'))
|
||||
a.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -185,7 +246,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('accepts multiple steps in a turn and consecutive turns', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
expect(() => {
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -202,7 +263,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a skipped turn number', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
|
||||
@@ -211,7 +272,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a skipped step number within a turn', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -221,7 +282,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a turn/end while a step is still open', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -230,7 +291,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a step/start while a step is still open', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -238,7 +299,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects a tool/result satisfying a call from a previous step', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -251,7 +312,7 @@ describe('session-log invariants', () => {
|
||||
})
|
||||
|
||||
it('rejects an assistant/message naming the wrong step', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
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 })
|
||||
@@ -265,7 +326,7 @@ describe('HMR state rebuild', () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
// First registration, mid-turn: a turn is open when the plugin reloads.
|
||||
const first = await ctx.plugin(Invariants, { freeze: false })
|
||||
const first = await ctx.plugin(Invariants)
|
||||
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 })
|
||||
@@ -273,7 +334,7 @@ describe('HMR state rebuild', () => {
|
||||
|
||||
// Re-apply (HMR): the fresh fiber must replay the existing log so the open
|
||||
// step is known — the next chunk must NOT be a false positive.
|
||||
await ctx.plugin(Invariants, { freeze: false })
|
||||
await ctx.plugin(Invariants)
|
||||
expect(() => session.append('assistant/chunk', { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'h' } }))
|
||||
.not.toThrow()
|
||||
// And a genuine violation is still caught after the rebuild.
|
||||
@@ -282,113 +343,86 @@ describe('HMR state rebuild', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('dev-freeze', () => {
|
||||
it('freezes appended event data so mutating a logged event throws', async () => {
|
||||
const { ctx } = await setup() // freeze defaults true
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
describe('session immutability', () => {
|
||||
it('always freezes appended event data without the invariants plugin', () => {
|
||||
const session = new Session(SessionId('appended'))
|
||||
const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(Object.isFrozen(event)).toBe(true)
|
||||
expect(Object.isFrozen(event.data)).toBe(true)
|
||||
expect(Object.isFrozen(event.data.content)).toBe(true)
|
||||
expect(Object.isFrozen(event.data.content[0])).toBe(true)
|
||||
expect(Object.isFrozen(session.events)).toBe(true)
|
||||
expect(() => { (event.data.content[0] as { text: string }).text = 'HACKED' }).toThrow()
|
||||
})
|
||||
|
||||
it('does not freeze when freeze:false', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
expect(Object.isFrozen(event)).toBe(false)
|
||||
})
|
||||
|
||||
it('freezes seeded events on session/created', async () => {
|
||||
const { ctx } = await setup()
|
||||
it('always freezes seeded events without the invariants plugin', () => {
|
||||
const seed = [
|
||||
{ type: 'turn/start' as const, seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } },
|
||||
{ type: 'user/message' as const, seq: 1, time: 0, data: { content: [{ type: 'text' as const, text: 'seeded' }], source: { kind: 'user' as const } }, surfaceOp: 'append' as const },
|
||||
]
|
||||
const session = ctx.sessions.create(undefined, { seed })
|
||||
const session = new Session(SessionId('seeded'), seed)
|
||||
expect(Object.isFrozen(seed[0])).toBe(false)
|
||||
expect(Object.isFrozen(session.events)).toBe(true)
|
||||
expect(Object.isFrozen(session.events[0])).toBe(true)
|
||||
expect(Object.isFrozen(session.events[0]?.data)).toBe(true)
|
||||
expect(Object.isFrozen(session.events[1]?.data)).toBe(true)
|
||||
})
|
||||
|
||||
it('freezes mutable descendants of a shallow-frozen event datum', async () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
// A caller hands in a SHALLOW-frozen block whose nested array is still
|
||||
// mutable. deepFreeze must descend into the already-frozen object and
|
||||
// freeze the descendant, not short-circuit on the frozen container —
|
||||
// otherwise dev-mode misses exactly the history mutation the dev-invariants RFC catches.
|
||||
// `append` snapshots `data`, so the freeze applies to the LOGGED clone, not
|
||||
// the caller's input — read the event back and assert on its data.
|
||||
it('snapshots and freezes descendants of a shallow-frozen caller value', () => {
|
||||
const session = new Session(SessionId('shallow-frozen'))
|
||||
const innerContent: { type: 'text'; text: string }[] = [{ type: 'text', text: 'inner' }]
|
||||
const block = Object.freeze({ type: 'tool-result' as const, toolCallId: CallId('c1'), content: innerContent, isError: false })
|
||||
const event = session.append('user/message', { content: [block], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
const logged = event.data.content[0] as { content: { type: 'text'; text: string }[] }
|
||||
expect(Object.isFrozen(innerContent)).toBe(false)
|
||||
expect(Object.isFrozen(logged.content)).toBe(true)
|
||||
expect(Object.isFrozen(logged.content[0])).toBe(true)
|
||||
innerContent[0]!.text = 'caller mutation'
|
||||
expect(logged.content[0]!.text).toBe('inner')
|
||||
expect(() => { logged.content.push({ type: 'text', text: 'mutation' }) }).toThrow()
|
||||
})
|
||||
|
||||
it('terminates on a cyclic event datum (WeakSet guard)', async () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
// The deep-freeze WeakSet guard must terminate on a self-referential
|
||||
// structure rather than recursing forever. Session.append now rejects
|
||||
// non-serializable (incl. cyclic) data at the source, so drive the 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)
|
||||
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(Object.isFrozen(cyclic)).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('agent status invariants', () => {
|
||||
it('accepts legal transitions: idle→running→idle and →disposed', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
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 { ctx } = await setup()
|
||||
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 { ctx } = await setup()
|
||||
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 { ctx } = await setup()
|
||||
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 { ctx } = await setup()
|
||||
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()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -400,14 +434,14 @@ describe('HMR safety', () => {
|
||||
|
||||
await fiber.dispose()
|
||||
|
||||
// After disposal: no freezing, no assertions. An event that WOULD have
|
||||
// violated the open-turn rule now passes silently, and is not frozen.
|
||||
// After disposal the plugin's assertions are gone, so an event that would
|
||||
// violate the open-turn rule passes. Session still owns immutability.
|
||||
const event = session.append('turn/start', { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
expect(Object.isFrozen(event)).toBe(false)
|
||||
expect(Object.isFrozen(event)).toBe(true)
|
||||
// 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', () => {
|
||||
@@ -418,16 +452,17 @@ describe('HMR safety', () => {
|
||||
expect(err.message).toBe('invariant violated: seq must strictly increase')
|
||||
})
|
||||
|
||||
it('does not leak listeners across dispose (no stale freezing)', async () => {
|
||||
it('does not leak listeners across dispose', async () => {
|
||||
const { ctx, fiber } = await setup()
|
||||
await fiber.dispose()
|
||||
const spy = vi.fn()
|
||||
ctx.on('session/event', spy)
|
||||
const session = ctx.sessions.create()
|
||||
session.append('user/message', { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
|
||||
// our own spy fires, proving events still flow — but the plugin's frozen.
|
||||
// The spy proves events still flow after plugin disposal. Session, not the
|
||||
// disposed listener, freezes the accepted record.
|
||||
expect(spy).toHaveBeenCalledOnce()
|
||||
expect(Object.isFrozen(session.events[0])).toBe(false)
|
||||
expect(Object.isFrozen(session.events[0])).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -639,7 +674,7 @@ describe('surface invariants', () => {
|
||||
})
|
||||
|
||||
it('catches an incomplete-provenance replace on the load/seed path', async () => {
|
||||
const { ctx } = await setup({ freeze: false })
|
||||
const { ctx } = await setup()
|
||||
const badSeed = [
|
||||
{ type: 'turn/start' as const, seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } },
|
||||
{ type: 'step/start' as const, seq: 1, time: 0, data: { turn: 1, step: 1 } },
|
||||
@@ -654,10 +689,10 @@ describe('surface invariants', () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
// Type system prevents surface metadata on non-surface events; this test
|
||||
// exercises the runtime guard against casts or persisted-data bypass.
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-return
|
||||
expect(() => (session.append as any)('turn/end', { turn: 1, reason: { kind: 'completed' } }, { sourceEventSeqs: [0] }))
|
||||
// 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/)
|
||||
})
|
||||
|
||||
@@ -665,8 +700,8 @@ describe('surface invariants', () => {
|
||||
const { ctx } = await setup()
|
||||
const session = ctx.sessions.create()
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-return
|
||||
expect(() => (session.append as any)('turn/end', { turn: 1, reason: { kind: 'completed' } }, { surfaceOp: 'append' }))
|
||||
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/)
|
||||
})
|
||||
})
|
||||
@@ -674,7 +709,7 @@ describe('surface invariants', () => {
|
||||
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 { ctx } = await setup()
|
||||
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' })
|
||||
@@ -736,7 +771,7 @@ describe('request-reconstruction cross-check (llm/stream)', () => {
|
||||
})
|
||||
|
||||
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 { 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 })
|
||||
@@ -777,7 +812,7 @@ describe('request cross-check ordering (prepend)', () => {
|
||||
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 })
|
||||
await ctx.plugin(Invariants)
|
||||
|
||||
const session = ctx.sessions.create(SessionId('prepend-check'))
|
||||
session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
|
||||
@@ -795,3 +830,67 @@ 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('accepts a matching carrier and rejects a mismatched one for EVERY agent-subject event', async () => {
|
||||
const ctx = await scopedCtx()
|
||||
// Real Session objects: the session-start tracker WeakSet-keys them.
|
||||
const agent = { id: 'a1', session: new Session(SessionId('a1-s')) } as unknown as Agent
|
||||
const other = { id: 'a2', session: new Session(SessionId('a2-s')) } as unknown as Agent
|
||||
// One dispatch per table row keeps every subject extractor covered: the
|
||||
// matching carrier passes, the foreign-keyed one throws.
|
||||
const rows: [string, unknown[]][] = [
|
||||
['agent/created', [agent]],
|
||||
['agent/disposed', [agent]],
|
||||
['agent/status', [agent, 'idle']],
|
||||
['agent/queued', [agent, [], { source: { kind: 'user' }, steering: false }]],
|
||||
['agent/session-start', [agent, 'startup']],
|
||||
['agent/pre-step', [agent, 1, 1, '', new AbortController().signal]],
|
||||
['agent/prompt-submit', [agent, [], { kind: 'user' }, () => Promise.resolve({ kind: 'allow' })]],
|
||||
['agent/request', [agent, 1, 1, { model: 'm' }, () => Promise.resolve({ model: 'm' })]],
|
||||
['agent/session-prefix', [agent, [], new AbortController().signal, () => Promise.resolve([])]],
|
||||
['agent/step-result', [agent, 1, 1, { role: 'assistant', content: [] }, () => Promise.resolve({ role: 'assistant', content: [] })]],
|
||||
['agent/turn-continuation', [agent, 1, { action: 'stop' }, () => Promise.resolve({ action: 'stop' })]],
|
||||
['agent/turn-stop', [agent, 1]],
|
||||
['agent/error', [agent, 1, 0, new Error('x')]],
|
||||
['approval/request', [{ agent, toolName: 'echo' }, () => Promise.resolve('unavailable')]],
|
||||
['tools/pre-execute', [{ callId: 'c', name: 't', arguments: {}, agent }, () => Promise.resolve({ kind: 'allow' })]],
|
||||
['tools/execute', [{ callId: 'c', name: 't', arguments: {}, agent }, () => Promise.resolve({ callId: 'c', content: [], isError: false })]],
|
||||
['tools/post-execute', [{ callId: 'c', name: 't', arguments: {}, agent }, { callId: 'c', content: [], isError: false }, () => Promise.resolve({ kind: 'accept' })]],
|
||||
['tools/result', [{ callId: 'c', name: 't', arguments: {}, agent }, { callId: 'c', content: [], isError: false }]],
|
||||
]
|
||||
for (const [event, args] of rows) {
|
||||
const subject = event.startsWith('tools/') ? agent : agent
|
||||
expect(() => { (ctx.emit as (...a: unknown[]) => void)(scopeTarget(agent, subject), event, ...args) },
|
||||
`${event} with matching carrier`).not.toThrow()
|
||||
expect(() => { (ctx.emit as (...a: unknown[]) => void)(scopeTarget(agent, other), event, ...args) },
|
||||
`${event} with foreign carrier`).toThrow(/DIFFERENT subject/)
|
||||
}
|
||||
})
|
||||
|
||||
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()
|
||||
})
|
||||
|
||||
})
|
||||
|
||||
@@ -22,6 +22,9 @@
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../../core/scope"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
A scripted `SubagentProvider` for testing the [subagent seam](../../subagent/subagent/README.md) without a model or a real child agent — the subagent analog of [`dsh-llm-replay`](../llm-replay/README.md).
|
||||
|
||||
It lets a test drive `ctx.subagents` and the model-facing `dsh-tool-subagent` through the **real cordis Loader / export path**, exercising provider registration, start-time capability validation, the run lifecycle (`result` / `cancel` / `dispose`), and the structured-output branch — all deterministically and keylessly.
|
||||
It lets a test drive `ctx.subagents` and the model-facing `dsh-tool-subagent` through the real Cordis loader/export path, exercising provider registration, async start, start-time capability validation, required-signal cancellation, `result`, `dispose`, and structured output deterministically and keylessly.
|
||||
|
||||
## Usage
|
||||
|
||||
@@ -13,8 +13,8 @@ Load it as a plugin (functional shape: `name`/`inject`/`Config`/`apply`, no defa
|
||||
| `name` | `mock` | Registry name to register the provider under. |
|
||||
| `reply` | `mock subagent reply` | The scripted child's final answer text. |
|
||||
| `stopReason` | `completed` | The stop reason `result` settles with. |
|
||||
| `capabilities` | all `true` | Which start-time capabilities (`outputSchema`/`depthLimit`/`toolFilter`) the provider advertises. |
|
||||
| `inheritsParentContext` | `false` | The context contract to declare; `true` exercises the fork-shaped tool wording in consumer tests. |
|
||||
| `capabilities` | all `true` | Which start-time capabilities (`outputSchema`, `depthLimit`, `toolFilter`, and `persona`) the provider advertises. |
|
||||
| `inheritsParentContext` | `false` | Conversation-history descriptor: `false` means fresh, while `true` exercises seeded/fork wording. It says nothing about tool, service, scope, or authority inheritance. |
|
||||
| `structured` | `{ reply }` | Structured value surfaced when a request carries an `outputSchema` and the capability is on. |
|
||||
|
||||
A `cancel()` issued before `result` settles flips the stop reason to `aborted`, so the cancellation path is observable.
|
||||
Aborting the required request signal or disposing before `result` settles flips the stop reason to `aborted`, so both holder-facing cancellation paths are observable.
|
||||
|
||||
@@ -26,14 +26,13 @@ import type {
|
||||
|
||||
const STOP_REASONS = ['completed', 'aborted', 'error', 'max-tokens', 'refusal'] as const
|
||||
|
||||
const DEFAULT_CAPS: SubagentCapabilities = { outputSchema: true, depthLimit: true, toolFilter: true }
|
||||
const DEFAULT_CAPS: SubagentCapabilities = { outputSchema: true, depthLimit: true, toolFilter: true, persona: true }
|
||||
|
||||
/**
|
||||
* A scripted provider: every {@link start} returns a run whose `result`
|
||||
* resolves on a microtask with the configured reply (and a structured value
|
||||
* when the request asked for one and the capability is on). `dispose` is a
|
||||
* no-op; a `cancel()` before the result settles flips the stop reason to
|
||||
* `aborted`, so the cancellation path is observable in a test.
|
||||
* A scripted provider: every {@link start} returns a ready run whose `result`
|
||||
* resolves on the next task with the configured reply (and a structured value
|
||||
* when the request asked for one and the capability is on). The required
|
||||
* signal and `dispose()` both flip an unsettled result to `aborted`.
|
||||
*/
|
||||
class MockSubagentProvider implements SubagentProvider {
|
||||
readonly capabilities: SubagentCapabilities
|
||||
@@ -47,12 +46,22 @@ class MockSubagentProvider implements SubagentProvider {
|
||||
this.inheritsParentContext = config.inheritsParentContext ?? false
|
||||
}
|
||||
|
||||
start(request: SubagentStartRequest): SubagentRun {
|
||||
async start(request: SubagentStartRequest): Promise<SubagentRun> {
|
||||
if (request.signal.aborted) throw new Error('mock subagent start aborted before publication')
|
||||
const reply = this.config.reply ?? 'mock subagent reply'
|
||||
const output: ContentBlock[] = [{ type: 'text', text: reply }]
|
||||
const wantsStructured = request.outputSchema !== undefined && this.capabilities.outputSchema
|
||||
const baseStop: SubagentStopReason = this.config.stopReason ?? 'completed'
|
||||
let cancelled = false
|
||||
const flags = { cancelled: false }
|
||||
const onAbort = (): void => { flags.cancelled = true }
|
||||
request.signal.addEventListener('abort', onAbort, { once: true })
|
||||
// Make publication genuinely asynchronous so a same-turn abort is still
|
||||
// a provider-owned startup failure rather than a returned live run.
|
||||
await Promise.resolve()
|
||||
if (flags.cancelled) {
|
||||
request.signal.removeEventListener('abort', onAbort)
|
||||
throw new Error('mock subagent start aborted before publication')
|
||||
}
|
||||
|
||||
// A deterministic child id derived from the parent — no clock/random (both
|
||||
// banned in deterministic paths here, and unnecessary for a scripted run).
|
||||
@@ -60,18 +69,22 @@ class MockSubagentProvider implements SubagentProvider {
|
||||
|
||||
const resultFor = (): SubagentResult => ({
|
||||
output,
|
||||
structured: wantsStructured ? (this.config.structured ?? { reply }) : undefined,
|
||||
stopReason: cancelled ? 'aborted' : baseStop,
|
||||
...wantsStructured ? { structured: this.config.structured ?? { reply } } : {},
|
||||
stopReason: flags.cancelled ? 'aborted' : baseStop,
|
||||
})
|
||||
|
||||
const result = new Promise<SubagentResult>((resolve) => {
|
||||
setTimeout(() => { resolve(resultFor()) }, 0)
|
||||
}).finally(() => {
|
||||
request.signal.removeEventListener('abort', onAbort)
|
||||
})
|
||||
return {
|
||||
id,
|
||||
result: Promise.resolve().then(resultFor),
|
||||
cancel() {
|
||||
cancelled = true
|
||||
},
|
||||
async dispose() {
|
||||
// Scripted run holds no resources — nothing to await.
|
||||
result,
|
||||
dispose(): Promise<void> {
|
||||
flags.cancelled = true
|
||||
request.signal.removeEventListener('abort', onAbort)
|
||||
return Promise.resolve()
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -91,9 +104,11 @@ export interface Config {
|
||||
/** Which start-time capabilities to advertise (default: all `true`). */
|
||||
capabilities?: Partial<SubagentCapabilities>
|
||||
/**
|
||||
* The context contract to declare ({@link SubagentProvider.inheritsParentContext});
|
||||
* default `false` (spawn-like). Set `true` to exercise the fork-shaped tool
|
||||
* wording in consumer tests.
|
||||
* The conversation-history descriptor to declare
|
||||
* ({@link SubagentProvider.inheritsParentContext}); default `false` (fresh
|
||||
* conversation). Set `true` to exercise seeded/fork wording in consumer
|
||||
* tests. This flag says nothing about tool, service, scope, or authority
|
||||
* inheritance.
|
||||
*/
|
||||
inheritsParentContext?: boolean
|
||||
/**
|
||||
@@ -111,6 +126,7 @@ export const Config: z<Config> = z.object({
|
||||
outputSchema: z.boolean(),
|
||||
depthLimit: z.boolean(),
|
||||
toolFilter: z.boolean(),
|
||||
persona: z.boolean(),
|
||||
}),
|
||||
inheritsParentContext: z.boolean(),
|
||||
structured: z.any(),
|
||||
|
||||
@@ -11,7 +11,7 @@ function fakeParent(id = 'parent-1'): Agent {
|
||||
}
|
||||
|
||||
function baseRequest(over: Partial<SubagentStartRequest> = {}): SubagentStartRequest {
|
||||
return { prompt: [{ type: 'text', text: 'task' }], parent: fakeParent(), ...over }
|
||||
return { prompt: [{ type: 'text', text: 'task' }], parent: fakeParent(), signal: new AbortController().signal, ...over }
|
||||
}
|
||||
|
||||
async function mount(config: Partial<mock.Config> = {}): Promise<Context> {
|
||||
@@ -26,12 +26,13 @@ describe('dsh-subagent-mock', () => {
|
||||
const ctx = await mount({ reply: 'hello from mock' })
|
||||
expect(ctx.subagents.list()).toEqual(['mock'])
|
||||
|
||||
const run = ctx.subagents.start('mock', baseRequest())
|
||||
const run = await ctx.subagents.start('mock', baseRequest())
|
||||
await expect(run.result).resolves.toEqual({
|
||||
output: [{ type: 'text', text: 'hello from mock' }],
|
||||
structured: undefined,
|
||||
stopReason: 'completed',
|
||||
})
|
||||
await run.dispose()
|
||||
})
|
||||
|
||||
it('registers under a configurable name', async () => {
|
||||
@@ -41,13 +42,13 @@ describe('dsh-subagent-mock', () => {
|
||||
|
||||
it('surfaces a structured result when the request carries an outputSchema', async () => {
|
||||
const ctx = await mount({ reply: 'r', structured: { answer: 42 } })
|
||||
const run = ctx.subagents.start('mock', baseRequest({ outputSchema: { type: 'object', properties: { answer: { type: 'number' } } } }))
|
||||
const run = await ctx.subagents.start('mock', baseRequest({ outputSchema: { type: 'object', properties: { answer: { type: 'number' } } } }))
|
||||
await expect(run.result).resolves.toMatchObject({ structured: { answer: 42 } })
|
||||
})
|
||||
|
||||
it('defaults structured output to { reply } when outputSchema is requested but no structured value is configured', async () => {
|
||||
const ctx = await mount({ reply: 'fallback reply' })
|
||||
const run = ctx.subagents.start('mock', baseRequest({ outputSchema: { type: 'object', properties: { answer: { type: 'number' } } } }))
|
||||
const run = await ctx.subagents.start('mock', baseRequest({ outputSchema: { type: 'object', properties: { answer: { type: 'number' } } } }))
|
||||
await expect(run.result).resolves.toMatchObject({ structured: { reply: 'fallback reply' } })
|
||||
})
|
||||
|
||||
@@ -56,23 +57,44 @@ describe('dsh-subagent-mock', () => {
|
||||
// The service rejects an outputSchema request against a no-cap provider, so
|
||||
// the structured path is only reachable when the cap is on; with it off and
|
||||
// no schema requested, the result has no structured field.
|
||||
const run = ctx.subagents.start('mock', baseRequest())
|
||||
await expect(run.result).resolves.toMatchObject({ structured: undefined })
|
||||
const run = await ctx.subagents.start('mock', baseRequest())
|
||||
const result = await run.result
|
||||
expect(result).not.toHaveProperty('structured')
|
||||
})
|
||||
|
||||
it('honors a configured stop reason', async () => {
|
||||
const ctx = await mount({ stopReason: 'refusal' })
|
||||
const run = ctx.subagents.start('mock', baseRequest())
|
||||
const run = await ctx.subagents.start('mock', baseRequest())
|
||||
await expect(run.result).resolves.toMatchObject({ stopReason: 'refusal' })
|
||||
})
|
||||
|
||||
it('flips the stop reason to aborted when cancelled before the result settles', async () => {
|
||||
it('flips the stop reason to aborted when the signal fires before the result settles', async () => {
|
||||
const ctx = await mount()
|
||||
const run = ctx.subagents.start('mock', baseRequest())
|
||||
run.cancel()
|
||||
const controller = new AbortController()
|
||||
const run = await ctx.subagents.start('mock', baseRequest({ signal: controller.signal }))
|
||||
controller.abort()
|
||||
await expect(run.result).resolves.toMatchObject({ stopReason: 'aborted' })
|
||||
})
|
||||
|
||||
it('rejects an already-aborted request before starting publication', async () => {
|
||||
const ctx = await mount()
|
||||
const controller = new AbortController()
|
||||
controller.abort()
|
||||
|
||||
await expect(ctx.subagents.start('mock', baseRequest({ signal: controller.signal })))
|
||||
.rejects.toThrow('mock subagent start aborted before publication')
|
||||
})
|
||||
|
||||
it('rejects when cancellation wins the asynchronous publication handoff', async () => {
|
||||
const ctx = await mount()
|
||||
const controller = new AbortController()
|
||||
const pending = ctx.subagents.start('mock', baseRequest({ signal: controller.signal }))
|
||||
|
||||
controller.abort()
|
||||
|
||||
await expect(pending).rejects.toThrow('mock subagent start aborted before publication')
|
||||
})
|
||||
|
||||
it('unregisters the provider when the owning fiber is disposed (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SubagentService)
|
||||
|
||||
Reference in New Issue
Block a user