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

Merge branch 'master' into feat/py-types-code-mode

Browse Source
This commit is contained in:
Chinesezjc
2026-08-07 11:23:41 +08:00
committed by GitHub
parent 0de132b927 70937db8a0
commit 20651af6a6
1051 changed files with 38456 additions and 7801 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -7,13 +7,15 @@
import type {
Agent,
AgentCancelCause,
AgentEventDispatch,
AgentOptions,
AgentStatus,
CancelOptions,
InboxTarget,
PreStepDecision,
RequestErrorAction,
} from '@deepseek-ai/dsh-agent'
import { Inbox, agentCarrier, agentEvents, assembleContextFor, emitAgentEvent } from '@deepseek-ai/dsh-agent'
import { Inbox, agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
import type { GenerateOptions, LlmCallConfig, Message, PreparedLlmCall } from '@deepseek-ai/dsh-llm'
import {
BlockAssembler,
@@ -27,7 +29,7 @@ import type { Scope } from '@deepseek-ai/dsh-scope'
import { createScope } from '@deepseek-ai/dsh-scope'
import type { EpochHeader, RequestContext, Session, SessionId, TurnEndReason, UserMessage } from '@deepseek-ai/dsh-session'
import { canonicalHeader, headerEquals } from '@deepseek-ai/dsh-session'
import { renderContextSnapshot, renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import { joinContextSections, renderContextSections, renderPrompt } from '@deepseek-ai/dsh-system-prompt'
import type { PromptAssembly } from '@deepseek-ai/dsh-system-prompt'
import type { Context } from 'cordis'
import { RuntimeContextProjection } from './runtime-context.ts'
@@ -68,6 +70,9 @@ export class ReactLoopAgent implements Agent {
readonly scope: Scope
readonly ctx: Context
/** Fused dispatcher, built once in the constructor so hot-path dispatches never allocate. */
private readonly dispatch: AgentEventDispatch
/** Whether this loop instance has appended its initial/resume request anchor. */
private requestHeaderLogged = false
private readonly runtimeContext: RuntimeContextProjection
@@ -78,10 +83,11 @@ export class ReactLoopAgent implements Agent {
public readonly options: AgentOptions,
public readonly session: Session,
) {
this.dispatch = agentEvents(loopCtx, this)
this.inbox = new Inbox(session, {
inserted: (message) => { emitAgentEvent(loopCtx, this, 'agent/inbox/inserted', { message }) },
discarded: (message) => { emitAgentEvent(loopCtx, this, 'agent/inbox/discarded', { message }) },
claimed: (message, turn) => { emitAgentEvent(loopCtx, this, 'agent/inbox/claimed', { message, turn }) },
inserted: (message) => { this.dispatch.emit('agent/inbox/inserted', { message }) },
discarded: (message) => { this.dispatch.emit('agent/inbox/discarded', { message }) },
claimed: (message, turn) => { this.dispatch.emit('agent/inbox/claimed', { message, turn }) },
})
const lastTurn = session.events.findLast(event => event.type === 'turn/start')?.data.turn ?? 0
this.phase = { kind: 'idle', lastTurn }
@@ -100,7 +106,7 @@ export class ReactLoopAgent implements Agent {
this.phase = next
const status = this.status
if (status !== previousStatus) {
emitAgentEvent(this.loopCtx, this, 'agent/status', status)
this.dispatch.emit('agent/status', { status })
}
}
@@ -178,7 +184,7 @@ export class ReactLoopAgent implements Agent {
private throwError(error: unknown): never {
const turn = this.phase.kind === 'running' ? this.phase.turn : this.phase.lastTurn
const step = this.phase.kind === 'running' ? this.phase.step : 0
emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error)
this.dispatch.emit('agent/error', { turn, step, error })
throw error
}
@@ -202,10 +208,11 @@ export class ReactLoopAgent implements Agent {
const claimed = this.inbox.claim(target, position.turn)
const assembly = await this.loopCtx.systemPrompt.assemble(assembleContextFor(this, signal))
signal.throwIfAborted()
const context = this.runtimeContext.project(renderContextSnapshot(assembly))
const decision = await agentEvents(this.loopCtx, this).waterfall(
'agent/pre-step', claimed, { ...position, signal },
() => Promise.resolve({
const sections = renderContextSections(assembly)
const context = this.runtimeContext.project(joinContextSections(sections), sections)
const decision = await this.dispatch.waterfall(
'agent/pre-step', { messages: claimed, ...position, signal },
(): Promise<PreStepDecision> => Promise.resolve<PreStepDecision>({
kind: 'enter',
messages: context === undefined ? claimed : [...claimed, context],
}),
@@ -265,7 +272,7 @@ export class ReactLoopAgent implements Agent {
}
signal.throwIfAborted()
if (turnEnds && this.inbox.nextStep.length === 0) {
await this.loopCtx.serial(agentCarrier(this), 'agent/turn-stopping', this, turn, signal)
await this.dispatch.serial('agent/turn-stopping', { turn, signal })
signal.throwIfAborted()
}
if (turnEnds && this.inbox.nextStep.length === 0) break
@@ -322,14 +329,15 @@ export class ReactLoopAgent implements Agent {
signal.throwIfAborted()
const finish = assembler.finish
if (finish.kind === 'error' || finish.kind === 'aborted') {
const action = await this.loopCtx.waterfall(
agentCarrier(this), 'agent/request-error', this, {
const action = await this.dispatch.waterfall(
'agent/request-error', {
turn,
step,
provider: request.provider,
failure: finish.failure,
retryPolicy: preparedCall?.retryPolicy,
}, signal,
signal,
},
() => Promise.resolve<RequestErrorAction>(undefined),
)
signal.throwIfAborted()
@@ -404,8 +412,8 @@ export class ReactLoopAgent implements Agent {
...maxTokens === undefined ? {} : { maxTokens },
},
))
const proposedConfig = await this.loopCtx.waterfall(
agentCarrier(this), 'agent/request', this, turn, step, signal,
const proposedConfig = await this.dispatch.waterfall(
'agent/request', { turn, step, signal },
() => Promise.resolve(seedConfig),
)
signal.throwIfAborted()

93
packages/core/agent-loop/src/index.ts
View File

@@ -20,7 +20,7 @@ import type {
SessionStartSource,
} from '@deepseek-ai/dsh-agent'
import { errorChain } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import { SessionId, SessionPreparation } from '@deepseek-ai/dsh-session'
import type { Session, SessionHeader } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
import type {} from '@deepseek-ai/dsh-tools'
@@ -104,6 +104,30 @@ async function raceAbort<T>(operation: PromiseLike<T> | T, signal: AbortSignal,
}
}
/** Start an abortable operation and release a value that arrives after cancellation. */
async function raceAbortCall<T>(
operation: () => PromiseLike<T> | T,
signal: AbortSignal,
id: SessionId,
releaseAbandoned?: (value: T) => void,
): Promise<T> {
if (signal.aborted) {
throw signal.reason instanceof Error
? signal.reason
: new Error(`agent "${id}" creation aborted`, { cause: signal.reason })
}
const pending = Promise.resolve().then(operation)
try {
return await raceAbort(pending, signal, id)
} catch (error: unknown) {
// oxlint-disable-next-line typescript/no-unnecessary-condition -- the signal can abort while the operation is awaited.
if (signal.aborted && releaseAbandoned !== undefined) {
void pending.then(releaseAbandoned, () => undefined)
}
throw error
}
}
/** Resolve the deployment-wide scheduler cap at the owning config boundary. */
function resolveMaxParallelToolCalls(value: number | undefined): number {
const maxParallelToolCalls = value ?? DEFAULT_MAX_PARALLEL_TOOL_CALLS
@@ -151,11 +175,11 @@ declare module 'cordis' {
* Consumers that buffer work for the configured identity use this
* transient signal to reject that work instead of waiting forever. Normal
* factory teardown suppresses failures from the cancelled startup attempt.
* @param sessionId - exact shared agent/session identity that failed startup.
* @param error - persistence, setup, or publication failure.
* @param payload.sessionId - exact shared agent/session identity that failed startup.
* @param payload.error - persistence, setup, or publication failure.
* @mode emit
*/
'agent-loop/config-start-failed'(sessionId: SessionId, error: unknown): void
'agent-loop/config-start-failed'(payload: { sessionId: SessionId; error: unknown }): void
}
}
@@ -327,7 +351,7 @@ export class AgentLoop extends Service implements AgentFactory {
): void {
if (!this.ownership.isActive()) return
this.ctx.logger.warn(`agent "${configId}": config-driven ${action} of "${sessionId}" failed: ${errorChain(error)}`)
const args: unknown[] = ['agent-loop/config-start-failed', sessionId, error]
const args: unknown[] = ['agent-loop/config-start-failed', { sessionId, error }]
for (const callback of this.ctx.events.dispatch('emit', args)) {
try {
const returned: unknown = callback(...args)
@@ -376,7 +400,7 @@ export class AgentLoop extends Service implements AgentFactory {
released.resolve()
}
}
const disposeAgentListener = ownerCtx.on('agent/disposed', checkReleased)
const disposeAgentListener = ownerCtx.on('agent/disposed', () => { checkReleased() })
const disposeSessionListener = ownerCtx.on('session/disposed', checkReleased)
try {
checkReleased()
@@ -501,7 +525,7 @@ export class AgentLoop extends Service implements AgentFactory {
// A synchronous announce/session-start listener may have started
// teardown; the machine is already live (delivery works from the
// session-start seam), so only the liveness recheck is owed.
emitAgentEvent(loopCtx, agent, 'agent/session-start', source)
emitAgentEvent(loopCtx, agent, 'agent/session-start', { source })
assertLive()
return { agent, dispose }
},
@@ -524,8 +548,8 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published running agent.
*/
create(id: SessionId, options: AgentOptions = {}, meta: Pick<SessionHeader, 'cwd'> = {}): Agent {
const session = this.runtime.ctx.sessions.prepare(id, { meta })
const prepared = this.prepare(this.ctx, id, options, session)
using preparation = SessionPreparation.create(this.runtime.ctx.sessions.prepare(id, { meta }))
const prepared = this.prepare(this.ctx, id, options, preparation.session)
try {
return prepared.publish('startup').agent
} catch (error: unknown) {
@@ -541,14 +565,14 @@ export class AgentLoop extends Service implements AgentFactory {
* @returns the published handle.
*/
async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle> {
const session = this.runtime.ctx.sessions.prepare(options.sessionId, {
const preparation = SessionPreparation.create(this.runtime.ctx.sessions.prepare(options.sessionId, {
...options.seed === undefined ? {} : { seed: options.seed },
...options.meta === undefined ? {} : { meta: options.meta },
})
}))
const published = this.setupAndPublish(
ownerCtx,
options.sessionId,
session,
preparation,
options.agentOptions ?? {},
options.setup,
options.signal,
@@ -562,12 +586,14 @@ export class AgentLoop extends Service implements AgentFactory {
private async setupAndPublish(
ownerCtx: Context,
id: SessionId,
session: Session,
preparation: SessionPreparation,
agentOptions: AgentOptions,
setup: AgentSetup | undefined,
signal: AbortSignal | undefined,
source: SessionStartSource,
): Promise<AgentHandle> {
using ownedPreparation = preparation
const session = ownedPreparation.session
const prepared = this.prepare(ownerCtx, id, agentOptions, session, signal)
try {
const setupCommit = await raceAbort(setup?.(prepared.agent.ctx), prepared.signal, id)
@@ -613,26 +639,31 @@ export class AgentLoop extends Service implements AgentFactory {
ownerAbort.signal,
this.ownership.signal,
])
let loaded: Awaited<ReturnType<SessionPersistence['load']>>
let preparation: SessionPreparation | undefined
try {
loaded = await raceAbort(persistence.load(id), fused, id)
try {
preparation = await raceAbortCall(
() => persistence.prepare(id, fused),
fused,
id,
(abandoned) => { abandoned[Symbol.dispose]() },
)
} finally {
await unfollowOwner()
}
ownerCtx.fiber.assertActive()
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
return await this.setupAndPublish(
ownerCtx,
id,
preparation,
options.agentOptions ?? {},
options.setup,
options.signal,
'resume',
)
} finally {
await unfollowOwner()
}
ownerCtx.fiber.assertActive()
if (!this.ownership.isActive()) throw new Error('agent loop is not active')
const session = this.runtime.ctx.sessions.prepare(id, {
seed: loaded.events,
meta: loaded.meta,
})
const prepared = this.prepare(ownerCtx, id, options.agentOptions ?? {}, session, options.signal)
try {
const setupCommit = await raceAbort(options.setup?.(prepared.agent.ctx), prepared.signal, id)
setupCommit?.commit()
return prepared.publish('resume')
} catch (error: unknown) {
await prepared.dispose()
throw error
preparation?.[Symbol.dispose]()
}
})()
this.ownership.trackWrapper(published)

9
packages/core/agent-loop/src/runtime-context.ts
View File

@@ -4,6 +4,7 @@
*/
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { ContextSnapshotSection } from '@deepseek-ai/dsh-llm'
import type { Session, UserMessage } from '@deepseek-ai/dsh-session'
import { isReplacementSurfaceEvent } from '@deepseek-ai/dsh-session'
import type { Context } from 'cordis'
@@ -57,15 +58,19 @@ export class RuntimeContextProjection {
/**
* Create an uncommitted snapshot only when the retained value differs.
* @param current - fully rendered dynamic context.
* @param sections - named contributions that formed the current snapshot.
* @returns a candidate user message, or `undefined` when no update is needed.
*/
project(current: string): UserMessage | undefined {
project(current: string, sections: readonly ContextSnapshotSection[]): UserMessage | undefined {
if (this.retained === undefined && current.length === 0) return
const snapshot = current.length === 0 ? CLEARED : current
if (this.retained?.text === snapshot) return
return createUserMessage({
content: [{ type: 'text', text: snapshot }],
source: { kind: 'plugin', plugin: SOURCE },
// The cleared marker has no contributions left to attribute.
source: sections.length === 0
? { kind: 'plugin', plugin: SOURCE }
: { kind: 'plugin', plugin: SOURCE, form: 'snapshot', sections },
})
}
}

8
packages/core/agent-loop/tests/agent-initiator.spec.ts
View File

@@ -31,7 +31,7 @@ async function harness(adapter: LlmAdapter): Promise<Harness> {
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -164,18 +164,18 @@ describe('AgentLoop initiator scope', () => {
if (context.agent === agent) capture(context.signal)
return next()
})
ctx.on('agent/pre-step', async (subject, _message, { signal }, next) => {
ctx.on('agent/pre-step', async ({ agent: subject, signal }, next) => {
if (subject === agent) {
expect(ctx.agents.requireInitiator()).toBe(agent)
preStepSignals.push(signal)
}
return next()
})
ctx.on('agent/request', async (subject, _turn, _step, signal, next) => {
ctx.on('agent/request', async ({ agent: subject, signal }, next) => {
if (subject === agent) capture(signal)
return next()
})
ctx.on('agent/turn-stopping', (subject, _turn, signal) => {
ctx.on('agent/turn-stopping', ({ agent: subject, signal }) => {
if (subject === agent) capture(signal)
})
ctx.tools.register(defineContentToolFixture({

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

@@ -60,17 +60,17 @@ describe('Agent', () => {
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/start') lifecycle.push('turn/start')
})
ctx.on('agent/inbox/inserted', (subject, event) => {
if (subject === agent) inserted.push(event)
ctx.on('agent/inbox/inserted', ({ agent: subject, message }) => {
if (subject === agent) inserted.push({ message })
})
ctx.on('agent/inbox/claimed', (subject, event) => {
ctx.on('agent/inbox/claimed', ({ agent: subject, message, turn }) => {
if (subject === agent) {
lifecycle.push('agent/inbox/claimed')
claimed.push(event)
claimed.push({ message, turn })
}
})
ctx.on('agent/inbox/discarded', (subject, event) => {
if (subject === agent) discarded.push(event)
ctx.on('agent/inbox/discarded', ({ agent: subject, message }) => {
if (subject === agent) discarded.push({ message })
})
const context = createUserMessage({
content: [{ type: 'text', text: 'discard me' }],
@@ -114,7 +114,7 @@ describe('Agent', () => {
const ctx = await harness(new MockAdapter([textResponse('ok')]))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const statuses: string[] = []
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent) statuses.push(status)
})
@@ -152,7 +152,7 @@ describe('Agent', () => {
const ctx = await harness(new MockAdapter([textResponse('ok')]))
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/status', (_subject, status) => {
ctx.on('agent/status', ({ status }) => {
throw new Error(`bad ${status} listener`)
})

20
packages/core/agent-loop/tests/cancel.spec.ts
View File

@@ -40,7 +40,7 @@ function send(agent: Agent, text: string) {
/** Resolve on the agent's next idle transition (event-based, not status poll). */
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') { dispose(); resolve() }
})
})
@@ -156,7 +156,7 @@ describe('Agent.cancel()', () => {
const running = Promise.withResolvers<undefined>()
let disposalDone: Promise<void> | undefined
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject !== agent || status !== 'running') return
disposalDone = handle.dispose()
running.resolve(undefined)
@@ -200,7 +200,7 @@ describe('Agent.cancel()', () => {
const replacementRegistered = Promise.withResolvers<undefined>()
let replacementObservation: Promise<{ status: string; requests: number; turns: number }> | undefined
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject !== agent || status !== 'idle' || replacementObservation !== undefined) return
send(agent, 'cancelled replacement')
replacementObservation = agent.whenIdle().then(() => ({
@@ -239,7 +239,7 @@ describe('Agent.cancel()', () => {
const replacementRegistered = Promise.withResolvers<undefined>()
let replacementIdle: Promise<void> | undefined
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject !== agent || status !== 'idle' || replacementIdle !== undefined) return
send(agent, 'cancelled replacement')
agent.cancel({ kind: 'user' })
@@ -440,7 +440,7 @@ describe('Agent.cancel()', () => {
})
let cancelled = false
ctx.on('agent/turn-stopping', (subject) => {
ctx.on('agent/turn-stopping', ({ agent: subject }) => {
if (subject === agent && !cancelled) {
cancelled = true
agent.cancel({ kind: 'user' })
@@ -465,7 +465,7 @@ describe('Agent.cancel()', () => {
// durable turn-start commit and must drop the reserved work.
let streamed = false
ctx.on('session/event', (_s, event) => { if (event.type === 'assistant/chunk') streamed = true })
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'running') agent.cancel({ kind: 'user' })
})
@@ -485,7 +485,7 @@ describe('Agent.cancel()', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let replaced = false
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject !== agent || status !== 'running' || replaced) return
replaced = true
agent.cancel({ kind: 'user' })
@@ -664,7 +664,7 @@ describe('Agent.cancel()', () => {
switch (stage) {
case 'pre-step':
ctx.on('agent/pre-step', async (subject, _message, { signal }, next) => {
ctx.on('agent/pre-step', async ({ agent: subject, signal }, next) => {
if (subject === agent) await blockUntilAbort(signal)
return next()
})
@@ -679,13 +679,13 @@ describe('Agent.cancel()', () => {
})
break
case 'request':
ctx.on('agent/request', async (subject, _turn, _step, signal, next) => {
ctx.on('agent/request', async ({ agent: subject, signal }, next) => {
if (subject === agent) await blockUntilAbort(signal)
return next()
})
break
case 'stopping':
ctx.on('agent/turn-stopping', async (subject, _turn, signal) => {
ctx.on('agent/turn-stopping', async ({ agent: subject, signal }) => {
if (subject === agent) await blockUntilAbort(signal)
})
break

36
packages/core/agent-loop/tests/config-session-id.spec.ts
View File

@@ -5,7 +5,7 @@ import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SessionStore, { SessionId, SessionPreparation } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
@@ -19,7 +19,7 @@ afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive:
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') { dispose(); resolve() }
})
})
@@ -170,7 +170,7 @@ describe('config-driven session id', () => {
await cleanupStarted.promise
expect(first.status).toBe('idle')
const failures: unknown[] = []
ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
ctx.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
const secondLoop = await ctx.plugin(AgentLoop, config)
await new Promise(resolve => setTimeout(resolve, 0))
expect(ctx.agents.get(sessionId)).toBe(first)
@@ -234,7 +234,7 @@ describe('config-driven session id', () => {
const failures: { sessionId: SessionId; error: unknown }[] = []
ctx.on('agent-loop/config-start-failed', () => { throw listenerFailure })
ctx.on('agent-loop/config-start-failed', () => Promise.reject(asyncListenerFailure) as never)
ctx.on('agent-loop/config-start-failed', (sessionId, error) => {
ctx.on('agent-loop/config-start-failed', ({ sessionId, error }) => {
failures.push({ sessionId, error })
})
vi.spyOn(ctx.sessionPersistence, 'list').mockRejectedValue(failure)
@@ -274,7 +274,7 @@ describe('config-driven session id', () => {
// Deliberately violate the normal Error-only rejection rule to exercise the unknown boundary.
// oxlint-disable-next-line typescript/prefer-promise-reject-errors
ctx.on('agent-loop/config-start-failed', () => Promise.reject(unrenderable) as never)
ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) })
ctx.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
vi.spyOn(ctx.sessionPersistence, 'list').mockRejectedValue(unrenderable)
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
@@ -296,35 +296,33 @@ describe('config-driven session id', () => {
})
it.each(['resolve', 'reject'] as const)(
'abandons an exact-id persistence lookup that later %s when AgentLoop disposal starts',
'abandons an exact-id preparation that later %s when AgentLoop disposal starts',
async (outcome) => {
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-exact-dispose-'))
dirs.push(root)
const ctx = await makeCoreContext()
await ctx.plugin(SessionPersistenceJsonl, { root })
const loading = Promise.withResolvers<Awaited<ReturnType<typeof ctx.sessionPersistence.load>>>()
vi.spyOn(ctx.sessionPersistence, 'load').mockReturnValue(loading.promise)
const preparing = Promise.withResolvers<SessionPreparation>()
vi.spyOn(ctx.sessionPersistence, 'prepare').mockReturnValue(preparing.promise)
const released = vi.fn()
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const failures: unknown[] = []
ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) })
ctx.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
const loop = await ctx.plugin(AgentLoop, {
agents: [{ id: 'main', sessionId: SessionId('config-exact-dispose'), model: 'mock' }],
})
await loop.dispose()
if (outcome === 'resolve') {
loading.resolve({
meta: {
id: SessionId('config-exact-dispose'),
version: 0,
createdAt: Date.now(),
},
events: [],
})
preparing.resolve(SessionPreparation.create(
ctx.sessions.prepare(SessionId('config-exact-dispose')),
{ release: released },
))
} else {
loading.reject(new Error('startup cancelled by teardown'))
preparing.reject(new Error('startup cancelled by teardown'))
}
await Promise.resolve()
if (outcome === 'resolve') await expect.poll(() => released).toHaveBeenCalledOnce()
expect(ctx.agents.get(SessionId('config-exact-dispose'))).toBeUndefined()
expect(failures).toEqual([])
expect(warn).not.toHaveBeenCalled()
@@ -481,7 +479,7 @@ describe('startup reporting after factory teardown', () => {
gate.promise.catch(() => undefined)
vi.spyOn(ctx.sessionPersistence, 'list').mockReturnValue(gate.promise)
const failures: unknown[] = []
ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
ctx.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
const loop = await ctx.plugin(AgentLoop, {

36
packages/core/agent-loop/tests/contract-regressions.spec.ts
View File

@@ -40,7 +40,7 @@ async function harness(adapter: MockAdapter) {
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -191,7 +191,7 @@ describe('abort during tool execution ends the turn', () => {
const adapter = new MockAdapter([textResponse('must not run')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a-empty-batch'), { provider: 'mock', model: 'mock' })
ctx.on('agent/pre-step', (subject, _messages, _context, next) => {
ctx.on('agent/pre-step', ({ agent: subject }, next) => {
if (subject !== agent) return next()
return Promise.resolve({ kind: 'enter', messages: [] })
})
@@ -288,7 +288,7 @@ describe('abort during tool execution ends the turn', () => {
send(agent, 'leave an unmatched historical call')
await waitForIdle(ctx, agent)
const disposeInjection = ctx.on('agent/pre-step', async (subject, _messages, { turn }, next) => {
const disposeInjection = ctx.on('agent/pre-step', async ({ agent: subject, turn }, next) => {
const decision = await next()
if (subject === agent && turn === 2 && decision.kind === 'enter') {
disposeInjection()
@@ -382,7 +382,7 @@ describe('disposal leaves the two-state status contract balanced', () => {
const statuses: string[] = []
const reasons: TurnEndReason[] = []
ctx.on('agent/status', (_agent, status) => void statuses.push(status))
ctx.on('agent/status', ({ status }) => void statuses.push(status))
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
send(agent, 'go')
@@ -411,7 +411,7 @@ describe('disposal leaves the two-state status contract balanced', () => {
agent = inner.agentLoop.create(SessionId('scoped'), { provider: 'mock', model: 'mock' })
}, { inject: ['agentLoop'] }))
ctx.on('agent/status', (_agent, status) => {
ctx.on('agent/status', ({ status }) => {
if (status === 'idle') throw new Error('broken status listener')
})
@@ -457,7 +457,7 @@ describe('adapter registration, routing, and accepted-input ownership', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), {}) // no model — router plugin decides
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
return { ...await next(), provider: 'mock', model: 'mock' }
})
@@ -540,7 +540,7 @@ describe('turn numbering continues across seeded sessions', () => {
ctx2.on('session/event', (_s, event) => { if (event.type === 'turn/start') turns.push(event.data.turn) })
forked.followup(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }))
await new Promise<void>((resolve) => {
ctx2.on('agent/status', (subject, status) => {
ctx2.on('agent/status', ({ agent: subject, status }) => {
if (subject === forked && status === 'idle') resolve()
})
})
@@ -586,7 +586,7 @@ describe('a finish-error stream chunk ends the turn as error, not completed', ()
const reasons: TurnEndReason[] = []
const errors: unknown[] = []
ctx.on('agent/error', (_agent, turn, step, error) => {
ctx.on('agent/error', ({ turn, step, error }) => {
expect({ turn, step }).toEqual({ turn: 1, step: 1 })
errors.push(error)
})
@@ -710,7 +710,7 @@ describe('turn and step boundary recovery', () => {
if (event.type === 'step/start' && !threw) { threw = true; throw new Error('boom step-start') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -743,7 +743,7 @@ describe('turn and step boundary recovery', () => {
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -800,7 +800,7 @@ describe('turn and step boundary recovery', () => {
}
})
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -893,14 +893,14 @@ describe('turn and step boundary recovery', () => {
}, { inject: ['agentLoop'] }))
let threw = false
ctx.on('agent/pre-step', (_subject, _messages, _context, next) => {
ctx.on('agent/pre-step', (_payload, next) => {
if (threw) return next()
threw = true
void fiber.dispose()
throw new Error('boom pre-step during disposal')
})
const errorEmits: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errorEmits.push(error)
})
@@ -926,7 +926,7 @@ describe('turn and step boundary recovery', () => {
if (!threw && event.type === 'turn/start') { threw = true; throw new Error('boom turn/start append') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -959,7 +959,7 @@ describe('turn and step boundary recovery', () => {
if (event.type === 'step/end' && !threw) { threw = true; throw new Error('boom step-end') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -1000,7 +1000,7 @@ describe('turn and step boundary recovery', () => {
if (!threw && event.type === 'step/end') { threw = true; throw new Error('boom step/end listener') }
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -1215,7 +1215,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
await mountInvariants(ctx)
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async (_subject, _messages, _context, next) => {
ctx.on('agent/pre-step', async (_payload, next) => {
await blocker
return next()
})
@@ -1261,7 +1261,7 @@ describe('disposal and cancellation during pre-step assembly', () => {
await mountInvariants(ctx)
ctx.llm.registerAdapter(['mock'], adapter)
ctx.on('agent/pre-step', async (_subject, _messages, _context, next) => {
ctx.on('agent/pre-step', async (_payload, next) => {
await blocker
return next()
})

20
packages/core/agent-loop/tests/coverage-edges.spec.ts
View File

@@ -28,7 +28,7 @@ async function harness(adapter: MockAdapter) {
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -120,7 +120,7 @@ describe('thrown-value propagation', () => {
})
const errors: unknown[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
ctx.on('agent/error', ({ error }) => void errors.push(error))
send(agent, 'fails before turn start')
send(agent, 'survives as the next item')
@@ -143,7 +143,7 @@ describe('thrown-value propagation', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threwOnce = false
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
if (!threwOnce) {
threwOnce = true
throw { code: 500 }
@@ -167,7 +167,7 @@ describe('durable error rendering', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threwOnce = false
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
if (!threwOnce) {
threwOnce = true
throw new LlmError('server overloaded', 'RATE_LIMIT')
@@ -250,7 +250,7 @@ describe('request-error action edges', () => {
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('retry-after-cancel'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request-error', async (subject) => {
ctx.on('agent/request-error', async ({ agent: subject }) => {
subject.cancel({ kind: 'user' })
return { kind: 'retry' }
})
@@ -271,7 +271,7 @@ describe('request-error action edges', () => {
])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('retry-raced'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request-error', async (subject, _context, signal, next) => {
ctx.on('agent/request-error', async ({ agent: subject, signal }, next) => {
await next()
subject.cancel({ kind: 'user' })
expect(signal.aborted).toBe(true)
@@ -350,7 +350,7 @@ describe('persistent step-close rejection', () => {
if (event.type === 'step/end') throw new Error('step close permanently rejected')
})
const statuses: string[] = []
ctx.on('agent/status', (subject, status) => { if (subject === agent) statuses.push(status) })
ctx.on('agent/status', ({ agent: subject, status }) => { if (subject === agent) statuses.push(status) })
send(agent, 'go')
await agent.whenIdle()
@@ -406,7 +406,7 @@ describe('turn close failure containment', () => {
}
})
const errors: unknown[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
ctx.on('agent/error', ({ error }) => { errors.push(error) })
send(agent, 'go')
await agent.whenIdle()
@@ -484,11 +484,11 @@ describe('driver bookkeeping edges', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('reject-next-step'), { provider: 'mock', model: 'mock' })
let proposals = 0
ctx.on('agent/pre-step', async (_subject, _messages, _context, next) => {
ctx.on('agent/pre-step', async (_payload, next) => {
proposals += 1
return proposals === 2 ? { kind: 'reject' } : next()
})
ctx.on('agent/turn-stopping', (subject) => {
ctx.on('agent/turn-stopping', ({ agent: subject }) => {
subject.inject(createUserMessage({
content: [{ type: 'text', text: 'do not enter the next step' }],
source: { kind: 'plugin', plugin: 'test' },

42
packages/core/agent-loop/tests/interception.spec.ts
View File

@@ -41,7 +41,7 @@ async function harness(adapter: MockAdapter) {
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -65,7 +65,7 @@ describe('agent/pre-step', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const seen: string[] = []
ctx.on('agent/pre-step', async (_agent, messages, _signal, next) => {
ctx.on('agent/pre-step', async ({ messages }, next) => {
seen.push(messages[0]!.content.map(b => (b.type === 'text' ? b.text : '')).join(''))
return next()
})
@@ -92,8 +92,8 @@ describe('agent/pre-step', () => {
}))
const agent = ctx.agentLoop.create(SessionId('prompt-coordinates'), { provider: 'mock', model: 'mock' })
const seen: Array<{ turn: number; step: number; messages: number }> = []
ctx.on('agent/pre-step', async (_agent, messages, context, next) => {
seen.push({ turn: context.turn, step: context.step, messages: messages.length })
ctx.on('agent/pre-step', async ({ messages, turn, step }, next) => {
seen.push({ turn, step, messages: messages.length })
return next()
})
@@ -113,7 +113,7 @@ describe('agent/pre-step', () => {
const entered = Promise.withResolvers<undefined>()
const decision = Promise.withResolvers<PreStepDecision>()
const observed: UserMessage[] = []
ctx.on('agent/pre-step', async (subject, messages) => {
ctx.on('agent/pre-step', async ({ agent: subject, messages }) => {
if (subject !== agent) return { kind: 'enter', messages }
const message = messages[0]!
expect(Object.isFrozen(message)).toBe(true)
@@ -161,7 +161,7 @@ describe('agent/pre-step', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/pre-step', async (_agent, messages): Promise<PreStepDecision> =>
ctx.on('agent/pre-step', async ({ messages }): Promise<PreStepDecision> =>
({
kind: 'enter',
messages: [{ ...messages[0]!, content: [{ type: 'text', text: 'REWRITTEN' }] }],
@@ -182,7 +182,7 @@ describe('agent/pre-step', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/pre-step', async (_agent, messages): Promise<PreStepDecision> =>
ctx.on('agent/pre-step', async ({ messages }): Promise<PreStepDecision> =>
({
kind: 'enter',
messages: [...messages, createUserMessage({
@@ -211,15 +211,15 @@ describe('agent/pre-step', () => {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/turn-stopping', (subject) => {
ctx.on('agent/turn-stopping', ({ agent: subject }) => {
subject.inject(createUserMessage({
content: [{ type: 'text', text: 'pending context' }],
source: { kind: 'plugin', plugin: 'test' },
}))
})
ctx.on('agent/pre-step', async (_subject, _messages, context, next) => {
ctx.on('agent/pre-step', async ({ step }, next) => {
const decision = await next()
return context.step === 1 || decision.kind === 'reject'
return step === 1 || decision.kind === 'reject'
? decision
: { kind: 'enter', messages: [] }
})
@@ -262,7 +262,7 @@ describe('agent/pre-step', () => {
const decision = Promise.withResolvers<PreStepDecision>()
let claimed: UserMessage[] = []
let firstProposal = true
ctx.on('agent/pre-step', async (_agent, messages) => {
ctx.on('agent/pre-step', async ({ messages }) => {
if (!firstProposal) return { kind: 'enter', messages }
firstProposal = false
claimed = messages
@@ -372,14 +372,14 @@ describe('agent/pre-step', () => {
provider: 'mock',
model: 'mock',
})
ctx.on('agent/pre-step', async (_agent, messages, _signal, next) => {
ctx.on('agent/pre-step', async ({ messages }, next) => {
const decision = await next()
return messages.some(message =>
message.content.some(block => block.type === 'text' && block.text === 'blocked prompt'))
? { kind: 'reject' as const }
: decision
})
ctx.on('agent/pre-step', async (subject, messages, _signal, next) => {
ctx.on('agent/pre-step', async ({ agent: subject, messages }, next) => {
if (messages.some(message =>
message.content.some(block => block.type === 'text' && block.text === 'blocked prompt'))) {
subject.inject(createUserMessage({
@@ -482,7 +482,7 @@ describe('agent/pre-step', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/pre-step', async (_agent, messages, _signal, next): Promise<PreStepDecision> => {
ctx.on('agent/pre-step', async ({ messages }, next): Promise<PreStepDecision> => {
const text = messages.flatMap(message => message.content)
.map(b => (b.type === 'text' ? b.text : '')).join('')
return text === 'secret'
@@ -519,17 +519,17 @@ describe('agent/pre-step', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let threw = false
ctx.on('agent/pre-step', async (_agent, messages) => {
ctx.on('agent/pre-step', async ({ messages }) => {
if (!threw) { threw = true; throw new Error('prompt hook broke') }
return { kind: 'enter' as const, messages }
})
const errors: Error[] = []
const reasons: TurnEndReason[] = []
const statuses: string[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
ctx.on('agent/status', (subject, status) => { if (subject === agent) statuses.push(status) })
ctx.on('agent/status', ({ agent: subject, status }) => { if (subject === agent) statuses.push(status) })
ctx.on('session/event', (session, event) => {
if (session === agent.session && event.type === 'turn/end') reasons.push(event.data.reason)
})
@@ -559,7 +559,7 @@ describe('agent/session-start', () => {
const ctx = await harness(adapter)
const sources: SessionStartSource[] = []
ctx.on('agent/session-start', (_agent, source) => void sources.push(source))
ctx.on('agent/session-start', ({ source }) => void sources.push(source))
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
// fires synchronously at create, before any turn
@@ -576,7 +576,7 @@ describe('agent/session-start', () => {
const adapter = new MockAdapter([textResponse('ok')])
const ctx = await harness(adapter)
ctx.on('agent/session-start', (agent) => {
ctx.on('agent/session-start', ({ agent }) => {
agent.inject(createUserMessage({ content: [{ type: 'text', text: 'session preamble' }], source: { kind: 'plugin', plugin: 'test' } }))
})
@@ -724,11 +724,11 @@ describe('worked example: a native hook plugin is just a cordis plugin on the se
name: 'native-guard',
apply(ctx: Context) {
// 1. SessionStart: seed a standing instruction.
ctx.on('agent/session-start', (agent, source) => {
ctx.on('agent/session-start', ({ agent, source }) => {
agent.inject(createUserMessage({ content: [{ type: 'text', text: `policy active (started: ${source})` }], source: { kind: 'plugin', plugin: 'native-guard' } }))
})
// 2. PreStep: reject a forbidden prompt, annotate the rest.
ctx.on('agent/pre-step', async (_agent, messages, _signal, next): Promise<PreStepDecision> => {
ctx.on('agent/pre-step', async ({ messages }, next): Promise<PreStepDecision> => {
const text = messages.flatMap(message => message.content)
.map(b => (b.type === 'text' ? b.text : '')).join('')
if (text.includes('rm -rf')) {

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

@@ -28,7 +28,7 @@ async function harness(adapter: MockAdapter, persona = '') {
/** Wait for the agent's next transition to idle after a waking send. */
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -216,7 +216,7 @@ describe('agent loop', () => {
const adapter = new MockAdapter([textResponse('ok after rescue')])
const ctx = await harness(adapter, 'In {{cwd}}.')
const errors: Error[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -263,7 +263,7 @@ describe('agent loop', () => {
assembly.variables['model'] = 'mock'
return next()
})
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
const config = await next()
return { ...config, provider: 'mock', model: 'mock' }
})
@@ -553,7 +553,7 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('failed-steering'), { provider: 'mock', model: 'mock' })
let fail = true
ctx.on('agent/pre-step', (subject, _messages, _context, next) => {
ctx.on('agent/pre-step', ({ agent: subject }, next) => {
if (subject !== agent || !fail) return next()
fail = false
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'pending steering' }], source: { kind: 'user' } }))
@@ -713,7 +713,7 @@ describe('agent loop', () => {
let steps = 0
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
ctx.on('agent/turn-stopping', (subject) => {
ctx.on('agent/turn-stopping', ({ agent: subject }) => {
if (steps < 3) {
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'plugin', plugin: 'loop-test' } }))
}
@@ -785,7 +785,7 @@ describe('agent loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
const config = await next()
// The seed is frozen — config is not a mutable per-call knob; a switch
// is proposed by returning a replacement, and the loop logs it.
@@ -816,7 +816,7 @@ describe('agent loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const fires: { turn: number; step: number; signal: AbortSignal }[] = []
ctx.on('agent/pre-step', (subject, _messages, { turn, step, signal }, next) => {
ctx.on('agent/pre-step', ({ agent: subject, turn, step, signal }, next) => {
if (subject === agent) fires.push({ turn, step, signal })
return next()
})
@@ -837,7 +837,7 @@ describe('agent loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let boundaryOpen = true
ctx.on('agent/pre-step', (subject, _messages, _context, next) => {
ctx.on('agent/pre-step', ({ agent: subject }, next) => {
if (subject === agent) boundaryOpen = subject.session.events.at(-1)?.type === 'step/start'
return next()
})
@@ -855,13 +855,13 @@ describe('agent loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let throwOnce = true
ctx.on('agent/pre-step', (_agent, _messages, _context, next) => {
ctx.on('agent/pre-step', (_payload, next) => {
if (throwOnce) { throwOnce = false; throw new Error('boom in pre-step') }
return next()
})
const errors: Error[] = []
ctx.on('agent/error', (_a, _t, _s, error) => {
ctx.on('agent/error', ({ error }) => {
if (error instanceof Error) errors.push(error)
})
@@ -933,7 +933,7 @@ describe('agent loop', () => {
ctx.on('session/event', (_session, event) => { if (event.type === 'step/end') steps++ })
// Force exactly one continuation (step 1 → step 2), then defer to default
// (step 2 is a plain stop with no tool calls → stops).
ctx.on('agent/turn-stopping', (subject) => {
ctx.on('agent/turn-stopping', ({ agent: subject }) => {
if (steps < 2) {
subject.steer(createUserMessage({ content: [{ type: 'text', text: 'continue after truncation' }], source: { kind: 'plugin', plugin: 'max-tokens-test' } }))
}
@@ -1296,7 +1296,7 @@ describe('agent loop', () => {
const errors: unknown[] = []
const reasons: TurnEndReason[] = []
ctx.on('agent/error', (_agent, _turn, _step, error) => {
ctx.on('agent/error', ({ error }) => {
errors.push(error)
})
ctx.on('session/event', (_s, event) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })

4
packages/core/agent-loop/tests/properties.spec.ts
View File

@@ -50,7 +50,7 @@ async function harness() {
/** Resolve on the agent's next transition to idle (event-based, not polled). */
function nextIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -63,7 +63,7 @@ function nextIdle(ctx: Context, agent: Agent): Promise<void> {
* the seen list plus a disposer for the listener (per the registry convention). */
function recordStatus(ctx: Context, agent: Agent): { seen: string[]; dispose: () => void } {
const seen: string[] = []
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent) seen.push(status)
})
return { seen, dispose }

2
packages/core/agent-loop/tests/request-cache.e2e.ts
View File

@@ -59,7 +59,7 @@ async function loopHarness(): Promise<Context> {
function waitForIdle(context: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = context.on('agent/status', (subject, status) => {
const dispose = context.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()

8
packages/core/agent-loop/tests/request-error.spec.ts
View File

@@ -62,12 +62,12 @@ describe('agent/request-error', () => {
retryPolicy: ResolvedRetryPolicy | undefined
}[] = []
const statuses: string[] = []
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent) statuses.push(status)
})
ctx.on('agent/request-error', async (subject, context) => {
ctx.on('agent/request-error', async ({ agent: subject, turn, step, failure, retryPolicy }) => {
expect(subject).toBe(agent)
seen.push(context)
seen.push({ turn, step, failure, retryPolicy })
return { kind: 'retry' }
})
@@ -102,7 +102,7 @@ describe('agent/request-error', () => {
const adapter = new MockAdapter([fail('busy', 'RATE_LIMIT'), textResponse('unused')])
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('request-error-cancel'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request-error', async (subject) => {
ctx.on('agent/request-error', async ({ agent: subject }) => {
subject.cancel({ kind: 'user' })
return { kind: 'retry' }
})

18
packages/core/agent-loop/tests/request-reconstruction.spec.ts
View File

@@ -38,7 +38,7 @@ async function harnessRoutes(
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -122,7 +122,7 @@ describe('request stability across the loop', () => {
const adapter = new MockAdapter([textResponse('one'), textResponse('two')], reasoning)
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('effort'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request', async (_agent, turn, _step, _signal, next) => {
ctx.on('agent/request', async ({ turn }, next) => {
const config = await next()
return turn === 2 ? { ...config, reasoningEffort: ReasoningEffortId('max') } : config
})
@@ -198,7 +198,7 @@ describe('request stability across the loop', () => {
provider: 'deepseek',
model: 'deepseek-model',
})
ctx.on('agent/request', async (_agent, turn, _step, _signal, next) => {
ctx.on('agent/request', async ({ turn }, next) => {
const config = await next()
return turn === 2
? { ...config, provider: 'other', model: 'other-model' }
@@ -232,7 +232,7 @@ describe('request stability across the loop', () => {
model: 'deepseek-model',
maxTokens: 4_096,
})
ctx.on('agent/request', async (_agent, turn, _step, _signal, next) => {
ctx.on('agent/request', async ({ turn }, next) => {
const config = await next()
return turn === 2
? { ...config, provider: 'other', model: 'other-model' }
@@ -460,7 +460,7 @@ describe('request stability across the loop', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
let injected = false
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
if (!injected) {
injected = true
agent.inject(createUserMessage({ content: [{ type: 'text', text: '[late context]' }], source: { kind: 'plugin', plugin: 'test' } }))
@@ -539,7 +539,7 @@ describe('request stability across the loop', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => {
ctx.on('agent/request', async (_payload, next) => {
const config = await next()
// next() resolves the SAME frozen seed — in-place shaping after
// delegation is unrepresentable, so a "mutate what next() returned"
@@ -576,7 +576,7 @@ describe('request stability across the loop', () => {
send(agent, 'go')
await waitForIdle(ctx, agent)
ctx.systemPrompt.section({ name: 'extra', order: 2, text: 'now with guidance' })
ctx.on('agent/request', async (_agent, _turn, _step, _signal, next) => ({
ctx.on('agent/request', async (_payload, next) => ({
...await next(), temperature: 0.5, maxTokens: 99, stop: ['<END>'],
}))
send(agent, 'again')
@@ -658,7 +658,7 @@ describe('request/context capacity records', () => {
send(agent, 'first')
await waitForIdle(ctx, agent)
ctx.on('agent/request', (subject, _turn, _step, _signal, next) => subject === agent
ctx.on('agent/request', ({ agent: subject }, next) => subject === agent
? Promise.resolve({ provider: 'mock', model: 'large' })
: next())
send(agent, 'second')
@@ -686,7 +686,7 @@ describe('request/context capacity records', () => {
const ctx = await harness(adapter)
const agent = ctx.agentLoop.create(SessionId('capacity-clear'), { provider: 'mock', model: 'known' })
let model = 'known'
ctx.on('agent/request', (subject, _turn, _step, _signal, next) => subject === agent
ctx.on('agent/request', ({ agent: subject }, next) => subject === agent
? Promise.resolve({ provider: 'mock', model })
: next())

168
packages/core/agent-loop/tests/resume.spec.ts
View File

@@ -1,12 +1,12 @@
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { afterEach, describe, expect, it } from 'vitest'
import { afterEach, describe, expect, it, vi } from 'vitest'
import { Context } from 'cordis'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import LlmService from '@deepseek-ai/dsh-llm'
import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import SessionStore, { SESSION_FORMAT_VERSION, Session, SessionId, SessionPreparation } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
@@ -52,9 +52,21 @@ async function persistSession(sessionId: SessionId): Promise<string> {
return root
}
/** Build a detached preparation for lifecycle-race test doubles. */
function preparationFromSnapshot(
ctx: Context,
snapshot: { meta: SessionHeader; events: readonly SessionEvent[] },
): SessionPreparation {
return SessionPreparation.create(ctx.sessions.prepare(snapshot.meta.id, {
seed: structuredClone(snapshot.events) as SessionEvent[],
meta: structuredClone(snapshot.meta),
seedSource: 'persistence',
}))
}
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') { dispose(); resolve() }
})
})
@@ -196,7 +208,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
await ctx.sessions.flush(first.session)
await expect(ctx.agents.resume({ resumeSessionId: sessionId }))
.rejects.toThrow(/live turn is open/)
.rejects.toThrow(/while it is live/)
first.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
await ctx.sessions.flush(first.session)
@@ -248,7 +260,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
const adapter1 = new MockAdapter([textResponse('a')])
const { ctx: ctx1, root } = await persistentHarness(adapter1)
const sources1: string[] = []
ctx1.on('agent/session-start', (_agent, source) => void sources1.push(source))
ctx1.on('agent/session-start', ({ source }) => void sources1.push(source))
const a1 = (await ctx1.agents.create({ sessionId: SessionId('start-sess') })).agent
expect(sources1).toEqual(['startup'])
a1.followup(createUserMessage({ content: [{ type: 'text', text: 'q' }], source: { kind: 'user' } }))
@@ -267,7 +279,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
await ctx2.plugin(SessionPersistenceJsonl, { root })
ctx2.llm.registerAdapter(['mock'], adapter2)
const sources2: string[] = []
ctx2.on('agent/session-start', (_agent, source) => void sources2.push(source))
ctx2.on('agent/session-start', ({ source }) => void sources2.push(source))
await ctx2.agents.resume({ resumeSessionId: SessionId('start-sess') })
expect(sources2).toEqual(['resume'])
await ctx2.fiber.dispose()
@@ -286,11 +298,11 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
expect(ctx.agents.get(sessionId)?.session).toBe(session)
order.push('session/created')
})
ctx.on('agent/created', (agent) => {
ctx.on('agent/created', ({ agent }) => {
expect(agent.status).toBe('idle')
order.push('agent/created')
})
ctx.on('agent/session-start', (agent) => {
ctx.on('agent/session-start', ({ agent }) => {
expect(() => { agent.cancel({ kind: 'user' }) }).not.toThrow()
order.push('agent/session-start')
})
@@ -446,22 +458,24 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
await ctx.fiber.dispose()
})
it('owner unload aborts a never-settling persistence load, releases the identity, and blocks late publication', async () => {
it('owner unload aborts a never-settling persistence preparation, releases the identity, and blocks late publication', async () => {
const sessionId = SessionId('resume-load-owner-unload')
const root = await persistSession(sessionId)
const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
const snapshot = await ctx.sessionPersistence.load(sessionId)
const lateLoad = Promise.withResolvers<typeof snapshot>()
const loadStarted = Promise.withResolvers<undefined>()
let loads = 0
ctx.sessionPersistence.load = (id) => {
const abandoned = preparationFromSnapshot(ctx, snapshot)
const latePreparation = Promise.withResolvers<SessionPreparation>()
const preparationStarted = Promise.withResolvers<undefined>()
const originalPrepare = ctx.sessionPersistence.prepare.bind(ctx.sessionPersistence)
let preparations = 0
ctx.sessionPersistence.prepare = (id, signal) => {
expect(id).toBe(sessionId)
loads += 1
if (loads === 1) {
loadStarted.resolve(undefined)
return lateLoad.promise
preparations += 1
if (preparations === 1) {
preparationStarted.resolve(undefined)
return latePreparation.promise
}
return Promise.resolve(structuredClone(snapshot))
return originalPrepare(id, signal)
}
const published: string[] = []
@@ -473,7 +487,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
const owner = await ctx.plugin(Object.assign((inner: Context) => {
resuming = inner.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } })
}, { inject: ['agents'] }))
await loadStarted.promise
await preparationStarted.promise
const rejection = expect(promptly(resuming)).rejects.toThrow(/owner disposed during setup/)
await promptly(owner.dispose())
@@ -485,23 +499,24 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
// can be reused before awaiting the public rejection.
const retry = await promptly(ctx.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } }))
await rejection
expect(loads).toBe(2)
expect(preparations).toBe(2)
expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start'])
// Settlement of the abandoned backend promise cannot resume the old
// transaction or emit a second publication after the retry owns the ids.
lateLoad.resolve(structuredClone(snapshot))
latePreparation.resolve(abandoned)
await Promise.resolve()
await Promise.resolve()
expect(ctx.agents.get(sessionId)).toBe(retry.agent)
expect(ctx.sessions.get(sessionId)).toBe(retry.agent.session)
expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start'])
abandoned[Symbol.dispose]()
await retry.dispose()
await ctx.fiber.dispose()
})
it('AgentLoop unload aborts persistence load and awaits wrapper settlement', async () => {
it('AgentLoop unload aborts persistence preparation and awaits wrapper settlement', async () => {
const sessionId = SessionId('resume-load-factory-unload')
const root = await persistSession(sessionId)
const ctx = new Context()
@@ -515,19 +530,20 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('next')]))
const snapshot = await ctx.sessionPersistence.load(sessionId)
const lateLoad = Promise.withResolvers<typeof snapshot>()
const loadStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.load = (id) => {
const abandoned = preparationFromSnapshot(ctx, snapshot)
const latePreparation = Promise.withResolvers<SessionPreparation>()
const preparationStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.prepare = (id) => {
expect(id).toBe(sessionId)
loadStarted.resolve(undefined)
return lateLoad.promise
preparationStarted.resolve(undefined)
return latePreparation.promise
}
const published: string[] = []
ctx.on('session/created', () => void published.push('session/created'))
ctx.on('agent/created', () => void published.push('agent/created'))
const resuming = ctx.agents.resume({ resumeSessionId: sessionId, agentOptions: { provider: 'mock', model: 'mock' } })
await loadStarted.promise
await preparationStarted.promise
const rejection = expect(promptly(resuming)).rejects.toThrow(/agent loop is not active/)
await promptly(loopFiber.dispose())
await rejection
@@ -535,10 +551,11 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
expect(published).toEqual([])
expect(ctx.agents.get(sessionId)).toBeUndefined()
expect(ctx.sessions.get(sessionId)).toBeUndefined()
lateLoad.resolve(structuredClone(snapshot))
latePreparation.resolve(abandoned)
await Promise.resolve()
await Promise.resolve()
expect(published).toEqual([])
abandoned[Symbol.dispose]()
await ctx.fiber.dispose()
})
@@ -730,6 +747,24 @@ describe('creation and resume cancellation edges', () => {
await ctx.fiber.dispose()
})
it('rejects when setup synchronously aborts its caller signal', async () => {
const { ctx } = await persistentHarness(new MockAdapter([]))
const controller = new AbortController()
const creating = ctx.agents.create({
sessionId: SessionId('setup-synchronous-abort'),
agentOptions: { provider: 'mock', model: 'mock' },
signal: controller.signal,
setup() {
controller.abort(new Error('setup synchronously cancelled'))
},
})
await expect(promptly(creating)).rejects.toThrow('setup synchronously cancelled')
expect(ctx.agents.get(SessionId('setup-synchronous-abort'))).toBeUndefined()
await ctx.fiber.dispose()
})
it('resume with a pre-aborted caller signal rejects out of the load race', async () => {
const sessionId = SessionId('resume-pre-aborted')
const root = await persistSession(sessionId)
@@ -743,19 +778,45 @@ describe('creation and resume cancellation edges', () => {
signal: controller.signal,
}))).rejects.toThrow('resume abandoned')
const stringReason = new AbortController()
stringReason.abort('resume string reason')
await expect(promptly(ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: { provider: 'mock', model: 'mock' },
signal: stringReason.signal,
}))).rejects.toThrow(/creation aborted/)
expect(ctx.agents.get(sessionId)).toBeUndefined()
await ctx.fiber.dispose()
})
it('factory teardown during a hung resume load rejects with loop-inactive', async () => {
it('releases a restored preparation if the loop becomes inactive before setup', async () => {
const sessionId = SessionId('resume-loop-inactive-after-prepare')
const root = await persistSession(sessionId)
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
const loop = ctx.agentLoop as unknown as {
ownership: { isActive: () => boolean }
}
vi.spyOn(loop.ownership, 'isActive').mockReturnValueOnce(false)
await expect(ctx.agents.resume({
resumeSessionId: sessionId,
agentOptions: { provider: 'mock', model: 'mock' },
})).rejects.toThrow('agent loop is not active')
expect(ctx.agents.get(sessionId)).toBeUndefined()
await ctx.fiber.dispose()
})
it('factory teardown during a hung resume preparation rejects with loop-inactive', async () => {
const sessionId = SessionId('resume-loop-teardown')
const root = await persistSession(sessionId)
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
const snapshot = await ctx.sessionPersistence.load(sessionId)
const gate = Promise.withResolvers<typeof snapshot>()
const loadStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.load = () => {
loadStarted.resolve(undefined)
const abandoned = preparationFromSnapshot(ctx, snapshot)
const gate = Promise.withResolvers<SessionPreparation>()
const preparationStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.prepare = () => {
preparationStarted.resolve(undefined)
return gate.promise
}
@@ -763,27 +824,28 @@ describe('creation and resume cancellation edges', () => {
resumeSessionId: sessionId,
agentOptions: { provider: 'mock', model: 'mock' },
})
await loadStarted.promise
// Resolve the load only after teardown began: the post-load ownership
await preparationStarted.promise
// Resolve the preparation only after teardown began: the post-prepare ownership
// check, not the abort race, must reject the wrapper.
const rejection = expect(promptly(resuming)).rejects.toThrow()
const disposal = ctx.fiber.dispose()
gate.resolve(structuredClone(snapshot))
gate.resolve(abandoned)
await rejection
await disposal
abandoned[Symbol.dispose]()
})
})
describe('configured-start failure edges', () => {
it('a non-Error mid-load abort reason is wrapped for the resume caller', async () => {
it('a non-Error mid-prepare abort reason is wrapped for the resume caller', async () => {
const sessionId = SessionId('resume-string-mid-abort')
const root = await persistSession(sessionId)
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
const gate = Promise.withResolvers<never>()
gate.promise.catch(() => undefined)
const loadStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.load = () => {
loadStarted.resolve(undefined)
const preparationStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.prepare = () => {
preparationStarted.resolve(undefined)
return gate.promise
}
const controller = new AbortController()
@@ -793,7 +855,7 @@ describe('configured-start failure edges', () => {
agentOptions: { provider: 'mock', model: 'mock' },
signal: controller.signal,
})
await loadStarted.promise
await preparationStarted.promise
controller.abort('operator string reason')
await expect(promptly(resuming)).rejects.toThrow(/creation aborted/)
@@ -808,7 +870,7 @@ describe('configured-start failure edges', () => {
// The artifact exists (list reports it) but its load fails: this is
// corruption, not first creation — the failure must be reported, and no
// fresh same-id session may shadow the broken one.
ctx.sessionPersistence.load = () => Promise.reject(new Error('artifact corrupt'))
ctx.sessionPersistence.prepare = () => Promise.reject(new Error('artifact corrupt'))
const configured = new Context()
await configured.plugin(LlmService)
@@ -818,9 +880,9 @@ describe('configured-start failure edges', () => {
await configured.plugin(AgentRegistry)
await configured.plugin(SessionPersistenceJsonl, { root })
configured.llm.registerAdapter(['mock'], new MockAdapter([]))
configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id)
configured.sessionPersistence.prepare = (id, signal) => ctx.sessionPersistence.prepare(id, signal)
const configFailures: unknown[] = []
configured.on('agent-loop/config-start-failed', (_id, error) => { configFailures.push(error) })
configured.on('agent-loop/config-start-failed', ({ error }) => { configFailures.push(error) })
const configWarnings: string[] = []
const configWarn = configured.logger.warn.bind(configured.logger)
configured.logger.warn = ((...args: unknown[]) => {
@@ -847,13 +909,13 @@ describe('configured-start failure edges', () => {
const ctx = await mountPersistentHarness(root, new MockAdapter([]))
const gate = Promise.withResolvers<never>()
gate.promise.catch(() => undefined)
const loadStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.load = () => {
loadStarted.resolve(undefined)
const preparationStarted = Promise.withResolvers<undefined>()
ctx.sessionPersistence.prepare = () => {
preparationStarted.resolve(undefined)
return gate.promise
}
const failures: unknown[] = []
ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
ctx.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
const configured = new Context()
await configured.plugin(LlmService)
@@ -863,12 +925,12 @@ describe('configured-start failure edges', () => {
await configured.plugin(AgentRegistry)
await configured.plugin(SessionPersistenceJsonl, { root })
configured.llm.registerAdapter(['mock'], new MockAdapter([]))
configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id)
configured.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
configured.sessionPersistence.prepare = (id, signal) => ctx.sessionPersistence.prepare(id, signal)
configured.on('agent-loop/config-start-failed', ({ error }) => { failures.push(error) })
const loop = await configured.plugin(AgentLoop, {
agents: [{ id: 'main', resumeSessionId: sessionId, provider: 'mock', model: 'mock' }],
})
await loadStarted.promise
await preparationStarted.promise
const disposal = loop.dispose()
gate.reject(new Error('late backend failure'))
await disposal

10
packages/core/agent-loop/tests/runtime-context.spec.ts
View File

@@ -30,10 +30,16 @@ describe('RuntimeContextProjection', () => {
const projection = new RuntimeContextProjection(ctx, session)
expect(session.surface.nodes).toContain(retained.seq)
expect(projection.project('retained')).toBeUndefined()
expect(projection.project('retained', [])).toBeUndefined()
expect(projection.project('next', [{ name: 'sandbox:policy', text: 'policy' }])?.source).toEqual({
kind: 'plugin',
plugin: SOURCE,
form: 'snapshot',
sections: [{ name: 'sandbox:policy', text: 'policy' }],
})
const other = ctx.sessions.create(SessionId('runtime-context-other'))
other.append('user/message', contextMessage('other'), { surfaceOp: 'append' })
expect(projection.project('retained')).toBeUndefined()
expect(projection.project('retained', [])).toBeUndefined()
})
})

36
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
View File

@@ -31,7 +31,7 @@ async function harness(adapter: MockAdapter = new MockAdapter([textResponse('ok'
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()
@@ -199,7 +199,7 @@ describe('agent scope lifecycle', () => {
const b = ctx.agentLoop.create(SessionId('b'), { provider: 'mock', model: 'mock' })
const heard: string[] = []
a.ctx.on('agent/status', (subject, status) => void heard.push(`a-sees:${subject.id}:${status}`))
a.ctx.on('agent/status', ({ agent: subject, status }) => void heard.push(`a-sees:${subject.id}:${status}`))
a.ctx.on('session/event', (_s, event) => {
if (event.type === 'user/message') heard.push('a-sees:user-message')
})
@@ -217,7 +217,7 @@ describe('agent scope lifecycle', () => {
it('runs setup in the guaranteed slot: scoped world complete before session-start and the first assembly', async () => {
const ctx = await harness()
const order: string[] = []
ctx.on('agent/session-start', (agent) => {
ctx.on('agent/session-start', ({ agent }) => {
order.push('session-start')
// The scoped section is already registered by the time session-start fires.
void ctx.systemPrompt.assemble(assembleContextFor(agent)).then((assembly) => {
@@ -673,19 +673,19 @@ describe('agent scope lifecycle', () => {
ctx.on('session/created', (session) => {
if (session.id === SessionId('agent-created-barrier-s')) lifecycle.push('session-created')
})
ctx.on('agent/created', (agent) => {
ctx.on('agent/created', ({ agent }) => {
if (agent.id !== SessionId('agent-created-barrier-s')) return
lifecycle.push('agent-created:dispose')
disposeCurrentLifecycle(ownerCtx)
})
ctx.on('agent/created', (agent) => {
ctx.on('agent/created', ({ agent }) => {
if (agent.id !== SessionId('agent-created-barrier-s')) return
expect(ctx.agents.get(agent.id)).toBe(agent)
expect(ctx.sessions.get(agent.session.id)).toBe(agent.session)
agent.ctx.effect(() => () => { lifecycle.push('scope-disposed') })
lifecycle.push('agent-created:observer')
})
ctx.on('agent/disposed', (agent) => {
ctx.on('agent/disposed', ({ agent }) => {
if (agent.id === SessionId('agent-created-barrier-s')) lifecycle.push('agent-disposed')
})
ctx.on('session/disposed', (session) => {
@@ -720,8 +720,8 @@ describe('agent scope lifecycle', () => {
const starts: string[] = []
let ownerCtx!: Context
let creating!: ReturnType<typeof ctx.agents.create>
ctx.on('agent/session-start', agent => void starts.push(agent.id))
ctx.on('agent/created', (agent) => {
ctx.on('agent/session-start', ({ agent }) => void starts.push(agent.id))
ctx.on('agent/created', ({ agent }) => {
if (agent.id === SessionId('listener-dispose-s')) disposeCurrentLifecycle(ownerCtx)
})
@@ -749,15 +749,15 @@ describe('agent scope lifecycle', () => {
const statuses: string[] = []
let scopeDisposed = false
let observerSawLive = false
ctx.on('agent/status', (agent, status) => {
ctx.on('agent/status', ({ agent, status }) => {
if (agent.id === SessionId('session-start-dispose-s')) statuses.push(status)
})
ctx.on('agent/session-start', (agent) => {
ctx.on('agent/session-start', ({ agent }) => {
if (agent.id !== SessionId('session-start-dispose-s')) return
announced = agent
disposeCurrentLifecycle(ownerCtx)
})
ctx.on('agent/session-start', (agent) => {
ctx.on('agent/session-start', ({ agent }) => {
if (agent.id !== SessionId('session-start-dispose-s')) return
expect(ctx.agents.get(agent.id)).toBe(agent)
expect(ctx.sessions.get(agent.session.id)).toBe(agent.session)
@@ -840,7 +840,7 @@ describe('agent scope lifecycle', () => {
const ctx = await harness()
let boom = true
const disposed: string[] = []
ctx.on('agent/disposed', agent => void disposed.push(agent.id))
ctx.on('agent/disposed', ({ agent }) => void disposed.push(agent.id))
ctx.on('session/created', () => {
if (boom) { boom = false; throw new Error('boom created') }
})
@@ -861,11 +861,11 @@ describe('agent scope lifecycle', () => {
const lifecycle: string[] = []
ctx.on('session/created', (session) => { lifecycle.push(`session-created:${session.id}`) })
ctx.on('session/disposed', (session) => { lifecycle.push(`session-disposed:${session.id}`) })
ctx.on('agent/created', (agent) => {
ctx.on('agent/created', ({ agent }) => {
lifecycle.push(`agent-created:${agent.id}`)
throw new Error('agent observer failed')
})
ctx.on('agent/disposed', (agent) => { lifecycle.push(`agent-disposed:${agent.id}`) })
ctx.on('agent/disposed', ({ agent }) => { lifecycle.push(`agent-disposed:${agent.id}`) })
await expect(ctx.agents.create({
sessionId: SessionId('partial-session'),
@@ -911,10 +911,10 @@ describe('agent scope lifecycle', () => {
const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
const other = ctx.agentLoop.create(SessionId('a2'), { provider: 'mock', model: 'mock' })
const heard: string[] = []
agent.ctx.on('agent/error', (subject: Agent, turn: number) => void heard.push(`${subject.id}:${turn}`))
agent.ctx.on('agent/error', ({ agent: subject, turn }) => void heard.push(`${subject.id}:${turn}`))
agentEvents(ctx, other).emit('agent/error', 1, 0, new Error('not for a1'))
agentEvents(ctx, agent).emit('agent/error', 2, 0, new Error('for a1'))
agentEvents(ctx, other).emit('agent/error', { turn: 1, step: 0, error: new Error('not for a1') })
agentEvents(ctx, agent).emit('agent/error', { turn: 2, step: 0, error: new Error('for a1') })
expect(heard).toEqual(['a1:2'])
})
@@ -1064,7 +1064,7 @@ describe('agent scope lifecycle', () => {
})
const agent = handle.agent
let reentered = false
ctx.on('agent/status', (subject, status) => {
ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject !== agent || status !== 'idle' || reentered) return
reentered = true
agent.followup(createUserMessage({ content: [{ type: 'text', text: 'reentrant' }], source: { kind: 'user' } }))

2
packages/core/agent-loop/tests/tool-calls.spec.ts
View File

@@ -31,7 +31,7 @@ async function harness(adapter: MockAdapter, maxParallelToolCalls?: number) {
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') { dispose(); resolve() }
})
})

2
packages/core/agent-loop/tests/tool-order.spec.ts
View File

@@ -33,7 +33,7 @@ async function harness(adapter: MockAdapter, toolOrder?: SystemPromptConfig['too
function waitForIdle(ctx: Context, agent: Agent): Promise<void> {
return new Promise((resolve) => {
const dispose = ctx.on('agent/status', (subject, status) => {
const dispose = ctx.on('agent/status', ({ agent: subject, status }) => {
if (subject === agent && status === 'idle') {
dispose()
resolve()

4
packages/core/agent/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write packages/core/agent/README.md
README.md: c3d6e6c24480894b6059417c1ab89db7aa0d7fa2
README.zh.md: 16ee8f5e6c483555839b0c3ab174e2e2356b1359
README.md: 2a69ab380eaad3929e27039582807037969eba64
README.zh.md: 176f3f75cf0f6e3309b2f5d34afb4d562105608e

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

@@ -50,7 +50,7 @@ Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-age
The lifecycle edges have two important local caveats. `agent/created` runs after scoped setup and after both session and agent registry entries exist. Setup is trusted composition-only code; the immediately following non-vetoing `agent/session-start` notification is the first supported startup injection point. `agent/disposed` always means the exact agent has left the registry. AgentLoop emits it after its driver is quiescent, while ordered teardown may still be detaching the session and unwinding the scope; custom agents registered directly own any stronger driver-ordering contract themselves.
Most interception points are cooperative waterfalls. `agent/pre-step` receives the exclusive claimed `UserMessage[]` plus a `PreStepContext` containing the proposed `turn`, `step`, and cancellation `signal`; its batch may be empty when tools already require another request. Other turn-scoped asynchronous seams receive their explicit `AbortSignal` positionally. Listeners may cooperate with a signal but must not retain it as authority over another turn. `agent/request-error` is the failed-model-request recovery waterfall: it receives request coordinates, normalized failure facts, the serving registration's retry policy when available, and the signal. A listener returns `{ kind: 'retry' }` without calling `next()` when it owns recovery. `agent/turn-stopping` runs before an otherwise completed turn closes. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns signal lifetime; the [agent-scope runtime-design Agent Note](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way) owns scoped dispatch and terminal settlement.
Most interception points are cooperative waterfalls. `agent/pre-step` receives a payload carrying the subject `agent`, the exclusive claimed `UserMessage[]`, and the proposed `turn`, `step`, and cancellation `signal`; its batch may be empty when tools already require another request. Agent-scoped turn seams carry their explicit `AbortSignal` in the payload; the remaining turn-scoped seams receive it through their request value. Listeners may cooperate with a signal but must not retain it as authority over another turn. `agent/request-error` is the failed-model-request recovery waterfall: it receives request coordinates, normalized failure facts, the serving registration's retry policy when available, and the signal. A listener returns `{ kind: 'retry' }` without calling `next()` when it owns recovery. `agent/turn-stopping` runs before an otherwise completed turn closes. The [explicit-cancellation decision](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md) owns signal lifetime; the [agent-scope runtime-design Agent Note](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way) owns scoped dispatch and terminal settlement.
`PreStepDecision` is either `{ kind: 'reject' }` or `{ kind: 'enter', messages }`. The enter branch is the complete identified, frozen batch for the proposed step. A listener that wraps downstream entry preserves that batch unless it intentionally replaces it; additions follow the waterfall's natural return order. Claiming already removed the offered messages from the inbox, so rejection does not retain them. Messages inserted after the claim remain pending for a later boundary.

2
packages/core/agent/README.zh.md
View File

@@ -50,7 +50,7 @@ Agent *创建* 由实现 `AgentFactory` 的插件（`dsh-agent-loop`）提供，
生命周期边有两个重要的本地注意事项。`agent/created` 在作用域 setup 之后、会话与 agent 注册表条目都存在之后运行。Setup 是受信任、仅用于组合的代码；紧随其后且不可 veto 的 `agent/session-start` 通知是第一个受支持的启动注入点。`agent/disposed` 始终表示确切 agent 已离开注册表。AgentLoop 在其驱动器完全停稳后发出该事件，而有序 teardown 此时可能仍在分离会话并撤销作用域；直接注册的自定义 agent 自行拥有任何更强的驱动器顺序契约。
大多数拦截点都是协作式 waterfall（瀑布式事件）。`agent/pre-step` 接收独占的已领取 `UserMessage[]`，以及包含拟进入 `turn`、`step` 与取消 `signal` 的 `PreStepContext`；当工具已经要求继续请求时，该批次可以为空。其他轮次作用域异步 seam 仍按位置接收显式 `AbortSignal`。监听器可以配合信号，但不得将它保留为控制另一轮次的权限。`agent/request-error` 是失败模型请求的恢复 waterfall：它接收请求坐标、规范化失败事实、可用时提供服务的注册项重试策略以及信号。拥有恢复权的监听器返回 `{ kind: 'retry' }` 且不调用 `next()`。`agent/turn-stopping` 在本可完成的轮次关闭前运行。信号生命周期由[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)拥有；作用域分发与终止结算由 [agent 作用域 runtime 设计 Agent Note（agent 决策记录）](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way)拥有。
大多数拦截点都是协作式 waterfall（瀑布式事件）。`agent/pre-step` 接收一个 payload，携带主体 `agent`、独占的已领取 `UserMessage[]` 以及拟进入的 `turn`、`step` 与取消 `signal`；当工具已经要求继续请求时，该批次可以为空。agent 作用域轮次 seam 在 payload 中携带显式 `AbortSignal`；其余轮次作用域 seam 通过其请求值接收它。监听器可以配合信号，但不得将它保留为控制另一轮次的权限。`agent/request-error` 是失败模型请求的恢复 waterfall：它接收请求坐标、规范化失败事实、可用时提供服务的注册项重试策略以及信号。拥有恢复权的监听器返回 `{ kind: 'retry' }` 且不调用 `next()`。`agent/turn-stopping` 在本可完成的轮次关闭前运行。信号生命周期由[显式取消决策](../../../.agents/notes/implemented/architecture/2026-07-16-explicit-turn-cancellation.md)拥有；作用域分发与终止结算由 [agent 作用域 runtime 设计 Agent Note（agent 决策记录）](../../../.agents/notes/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way)拥有。
`PreStepDecision` 要么是 `{ kind: 'reject' }`，要么是 `{ kind: 'enter', messages }`。enter 分支是拟进入步骤的完整、带标识且冻结的批次。包装下游 enter 的监听器会保留该批次，除非有意替换它；新增消息遵循 waterfall 的自然返回顺序。领取操作已经把候选消息从 inbox 删除，因此 reject 不会保留它们；领取后插入的消息仍等待后续边界。

94
packages/core/agent/src/dispatch.ts
View File

@@ -1,7 +1,8 @@
/**
* Agent-scoped dispatch and prompt assembly helpers. Ordinary events use the
* fused dispatcher so subject and scope key cannot diverge; registry lifecycle
* code instead captures one stable carrier for both edges.
* Agent-scoped dispatch and prompt assembly helpers. The fused dispatcher
* {@link agentEvents} couples the agent subject to its scope carrier, so the
* scope key and the payload's `agent` cannot diverge; repeat dispatchers (the
* loop driver) build it once in the agent's constructor and reuse it.
* @module @deepseek-ai/dsh-agent/dispatch
*/
@@ -17,25 +18,38 @@ type Params<F> = F extends (...args: infer P) => unknown ? P : never
type Return<F> = F extends (...args: never[]) => infer R ? R : never
/**
* The event names whose subject is an agent: handler parameters start with an
* `Agent` AND the handler declares a `Scoped<Agent>` `this` (the scope-carrier
* contract). The `this` check keeps accidental first-parameter-happens-to-be-
* an-Agent events (or zero-arg events, whose parameter tuple would satisfy a
* bare rest-tuple check via callability) out of the fused-dispatch surface.
* The event names whose subject is an agent: the handler's first parameter is
* a payload object carrying the `agent` subject AND the handler declares a
* `Scoped<Agent>` `this` (the scope-carrier contract). The `this` check keeps
* accidental payload-happens-to-carry-an-Agent events (or zero-arg events,
* whose parameter tuple would satisfy a bare rest-tuple check via callability)
* out of the fused-dispatch surface.
*/
export type AgentSubjectEvent = {
[K in keyof Events]: Events[K] extends (this: Scoped<Agent>, ...args: infer P) => unknown
? P extends [Agent, ...unknown[]] ? K : never
? P extends [infer Payload, ...unknown[]]
? Payload extends { agent: Agent } ? K : never
: never
: never
}[keyof Events]
/** The event arguments AFTER the injected agent subject. */
type Tail<K extends AgentSubjectEvent> = Params<Events[K]> extends [Agent, ...infer R] ? R : never
/** The full payload object of one agent-subject event. */
type PayloadOf<K extends AgentSubjectEvent> = Params<Events[K]> extends [infer Payload, ...unknown[]] ? Payload : never
/** The event arguments AFTER the payload: the waterfall `next` when present. */
type Tail<K extends AgentSubjectEvent> = Params<Events[K]> extends [unknown, ...infer R] ? R : never
/**
* The payload as emit-side callers pass it: the full payload minus the agent
* field, which the fused dispatcher injects so subject and scope key cannot
* diverge.
*/
type PayloadRest<K extends AgentSubjectEvent> = Omit<PayloadOf<K> & object, 'agent'>
/**
* The fused dispatcher {@link agentEvents} returns: each method dispatches the
* named agent-subject event with the agent's scope carrier as `thisArg` and
* the agent itself injected as the first event argument.
* the agent itself injected into the payload.
*/
export interface AgentEventDispatch {
/**
@@ -44,30 +58,36 @@ export interface AgentEventDispatch {
* contained per listener, so a notification cannot veto lifecycle progress
* or starve a later observer.
* @param name - the agent-subject event to emit.
* @param rest - the event's arguments after the injected agent.
* @param payload - the event's payload fields; `agent` is injected.
*/
emit<K extends AgentSubjectEvent>(name: K, ...rest: Tail<K>): void
emit<K extends AgentSubjectEvent>(name: K, payload: PayloadRest<K>): void
/**
* Awaited in-order dispatch (Cordis `serial`) in the agent's scope.
* @param name - the agent-subject event to dispatch.
* @param rest - the event's arguments after the injected agent.
* @param payload - the event's payload fields; `agent` is injected.
* @returns the serial chain's result (the first bail value, if any).
*/
serial<K extends AgentSubjectEvent>(name: K, ...rest: Tail<K>): Promise<Awaited<Return<Events[K]>>>
serial<K extends AgentSubjectEvent>(name: K, payload: PayloadRest<K>): Promise<Awaited<Return<Events[K]>>>
/**
* Around-middleware dispatch (Cordis `waterfall`) in the agent's scope. The
* declared event parameters already end with the `next` callback, so `rest`
* is exactly the event's arguments after the injected agent — the final
* element being the innermost `next` (the default the listener chain wraps).
* is exactly the event's arguments after the payload — the final element
* being the innermost `next` (the default the listener chain wraps).
* @param name - the agent-subject event to dispatch.
* @param rest - the event's arguments after the injected agent.
* @param payload - the event's payload fields; `agent` is injected.
* @param rest - the event's arguments after the payload (the `next` callback).
* @returns the waterfall's composed result.
*/
waterfall<K extends AgentSubjectEvent>(name: K, ...rest: Tail<K>): Return<Events[K]>
waterfall<K extends AgentSubjectEvent>(name: K, payload: PayloadRest<K>, ...rest: Tail<K>): Return<Events[K]>
}
/**
* Return the fused scope carrier for one agent subject.
* Build the fused scope carrier for one agent subject.
*
* The carrier is a stateless routing object. {@link agentEvents} accepts an
* existing carrier, so callers that dispatch repeatedly for the same agent
* (the loop driver) build it once in the agent's constructor and reuse it,
* keeping hot-path dispatches allocation-free.
* @param agent - the subject agent and scope key.
* @returns the carrier passed as the event dispatcher `this` value.
*/
@@ -79,22 +99,30 @@ export function agentCarrier(agent: Agent): Scoped<Agent> {
* Build a dispatcher that couples the agent subject to its scope carrier.
* @param ctx - the context to dispatch through (any context of the app).
* @param agent - the subject agent; also the scope-carrier key.
* @param carrier - the scope carrier to dispatch through; defaults to
* {@link agentCarrier} for the agent. Pass a constructor-built carrier to
* avoid rebuilding it for every dispatch.
* @returns the fused dispatcher.
*/
export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch {
const carrier = agentCarrier(agent)
export function agentEvents(ctx: Context, agent: Agent, carrier: Scoped<Agent> = agentCarrier(agent)): AgentEventDispatch {
// The ordinary dispatch methods forward through Cordis' variadic mixins. The
// fused (carrier, name, agent, ...rest) tuple is provably a valid argument
// fused (carrier, name, payload, ...rest) tuple is provably a valid argument
// list for the matching thisArg overload, but TypeScript cannot relate the
// generic Tail<K> spread back to that overload's conditional parameter
// tuple — hence one contained, shape-preserving cast per method.
const fused = <K extends AgentSubjectEvent>(payload: PayloadRest<K>): PayloadOf<K> =>
// The dispatcher owns the subject injection; callers pass PayloadRest, so
// the fused record is exactly the declared payload. The spread comes
// first, so a structurally acceptable payload that happens to carry an
// `agent` field can never override the injected subject.
({ ...payload, agent } as PayloadOf<K>)
return {
emit(name, ...rest) {
emit(name, payload) {
// Cordis emit invokes callbacks through Array.map: one synchronous throw
// starves later listeners, and returned promises are discarded. Agent
// notifications are non-vetoing, so resolve the same filtered callback
// set ourselves and contain both failure modes independently.
const args: unknown[] = [carrier, name, agent, ...rest]
const args: unknown[] = [carrier, name, fused(payload)]
const callbacks = ctx.events.dispatch('emit', args)
for (const callback of callbacks) {
try {
@@ -107,15 +135,15 @@ export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch {
}
}
},
async serial(name, ...rest) {
async serial(name, payload) {
// oxlint-disable-next-line typescript/unbound-method -- the events mixin accessor returns a pre-bound function
const serial = ctx.serial as (thisArg: Scoped<Agent>, name: string, ...args: unknown[]) => Promise<never>
return await serial(carrier, name, agent, ...rest)
return await serial(carrier, name, fused(payload))
},
waterfall(name, ...rest) {
waterfall(name, payload, ...rest) {
// oxlint-disable-next-line typescript/unbound-method -- the events mixin accessor returns a pre-bound function
const waterfall = ctx.waterfall as (thisArg: Scoped<Agent>, name: string, ...args: unknown[]) => never
return waterfall(carrier, name, agent, ...rest)
return waterfall(carrier, name, fused(payload), ...rest)
},
}
}
@@ -125,15 +153,15 @@ export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch {
* @param ctx - the context to dispatch through.
* @param agent - the subject agent and scope key.
* @param name - the agent-subject event to emit.
* @param rest - the event arguments after the injected agent.
* @param payload - the event's payload fields; `agent` is injected.
*/
export function emitAgentEvent<K extends AgentSubjectEvent>(
ctx: Context,
agent: Agent,
name: K,
...rest: Tail<K>
payload: PayloadRest<K>,
): void {
agentEvents(ctx, agent).emit(name, ...rest)
agentEvents(ctx, agent).emit(name, payload)
}
/**

8
packages/core/agent/src/index.ts
View File

@@ -187,8 +187,8 @@ export interface AgentFactory {
*/
createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle>
/**
* Load a persisted session and resume an agent on it. Async because it awaits
* both `ctx.sessionPersistence.load` and the optional unpublished setup
* Prepare a persisted session and resume an agent on it. Async because it awaits
* both `ctx.sessionPersistence.prepare` and the optional unpublished setup
* transaction; must be called after that service exists (consumers inject
* `sessionPersistence`). Publication follows the same setup-commit and
* ordered boundary as {@link createAgent}.
@@ -498,7 +498,7 @@ export class AgentRegistry extends Service {
/** Emit the paired disposal edge through the entry's stable carrier. */
private emitDisposed(entry: AgentEntry): void {
const args: unknown[] = [entry.carrier, 'agent/disposed', entry.agent]
const args: unknown[] = [entry.carrier, 'agent/disposed', { agent: entry.agent }]
for (const callback of this.ctx.events.dispatch('emit', args)) {
try {
const returned: unknown = callback(...args)
@@ -530,7 +530,7 @@ export class AgentRegistry extends Service {
// lifecycle edge; detach still pairs a partially delivered first edge.
entry.announcing = true
entry.announced = true
const args: unknown[] = [entry.carrier, 'agent/created', entry.agent]
const args: unknown[] = [entry.carrier, 'agent/created', { agent: entry.agent }]
try {
for (const callback of this.ctx.events.dispatch('emit', args)) {
// A synchronous creation failure vetoes publication and rolls back.

2
packages/core/agent/src/invariant.ts
View File

@@ -14,7 +14,7 @@ export const inject = ['invariants']
/** Install the agent contribution into its child registration fiber. */
const install: InvariantInstaller = (ctx, fail) => {
const lastStatus = new WeakMap<Agent, AgentStatus>()
ctx.on('agent/status', (agent, status) => {
ctx.on('agent/status', ({ agent, status }) => {
const previous = lastStatus.get(agent)
if (previous === status) {
fail(`agent/status repeated ${status} (no-op transition)`)

2
packages/core/agent/src/llm-target.ts
View File

@@ -53,7 +53,7 @@ export function installAgentLlmTarget(agentCtx: Context, target: AgentLlmTargetR
})
const disposeRequest = agentCtx.on(
'agent/request',
async (_agent, _turn, _step, _signal, next): Promise<LlmCallConfig> => {
async (_payload, next): Promise<LlmCallConfig> => {
const resolved = await next()
const selected = target.assembled
if (selected === undefined) return resolved

113
packages/core/agent/src/types.ts
View File

@@ -48,35 +48,11 @@ export interface CancelOptions {
*/
export type AgentStatus = 'idle' | 'running'
/** Coordinates and cancellation for a proposed step. */
export interface PreStepContext {
/** Turn that will own the step. */
readonly turn: number
/** Step proposed by the loop. */
readonly step: number
/** Current turn cancellation signal. */
readonly signal: AbortSignal
}
/** Whether and with which messages the loop enters a proposed step. */
export type PreStepDecision =
| { kind: 'reject' }
| { kind: 'enter'; messages: UserMessage[] }
/** One failed model-request attempt presented to recovery listeners. */
export interface RequestFailureContext {
/** Turn containing the failed request. */
readonly turn: number
/** Step containing the failed request attempt. */
readonly step: number
/** Provider selected for the failed request. */
readonly provider: string
/** Serializable facts normalized at the final adapter boundary. */
readonly failure: LlmFailure
/** Policy of the adapter registration that served the failed request. */
readonly retryPolicy: ResolvedRetryPolicy | undefined
}
/** Action returned by a listener that owns model-request recovery. */
export type RequestErrorAction = { kind: 'retry' } | undefined
@@ -171,105 +147,112 @@ declare module 'cordis' {
* Synchronous listener failure vetoes publication, while returned-promise
* rejection is reported. Detach requested during dispatch waits until every
* creation listener has observed the stable entry.
* @param agent - the newly registered agent with its live session and completed setup.
* @param payload.agent - the newly registered agent with its live session and completed setup.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/created'(this: Scoped<Agent>, agent: Agent): void
'agent/created'(this: Scoped<Agent>, payload: { agent: Agent }): void
/**
* An agent left the registry; AgentLoop emits this after driver quiescence
* and scoped-registration unwind, but before session detachment. Custom
* registry users own their driver-ordering contract.
* @param agent - the exact agent removed from the registry.
* @param payload.agent - the exact agent removed from the registry.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/disposed'(this: Scoped<Agent>, agent: Agent): void
'agent/disposed'(this: Scoped<Agent>, payload: { agent: Agent }): void
/**
* Agent status changed (`idle` ⇄ `running`). A waking delivery enters
* `running` synchronously after reserving cancellation; `idle` means no
* driver remains scheduled or active.
* @param agent - the agent whose status flipped.
* @param status - the status just entered (the transition's destination).
* @param payload.agent - the agent whose status flipped.
* @param payload.status - the status just entered (the transition's destination).
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/status'(this: Scoped<Agent>, agent: Agent, status: AgentStatus): void
'agent/status'(this: Scoped<Agent>, payload: { agent: Agent; status: AgentStatus }): void
/**
* One message entered the live inbox.
* @param agent - the agent whose inbox changed.
* @param event - the inserted message.
* @param payload.agent - the agent whose inbox changed.
* @param payload.message - the inserted message.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/inbox/inserted'(this: Scoped<Agent>, agent: Agent, event: { message: UserMessage }): void
'agent/inbox/inserted'(this: Scoped<Agent>, payload: { agent: Agent; message: UserMessage }): void
/**
* One message left the inbox inside its open turn. If the proposed step
* is rejected, the claimed message ends here: it is neither discarded nor
* re-emitted as a user/message, and the turn closes without a step.
* @param agent - the agent whose inbox changed.
* @param event - the claimed message and owning turn.
* @param payload.agent - the agent whose inbox changed.
* @param payload.message - the claimed message.
* @param payload.turn - the owning turn.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/inbox/claimed'(this: Scoped<Agent>, agent: Agent, event: { message: UserMessage; turn: number }): void
'agent/inbox/claimed'(this: Scoped<Agent>, payload: { agent: Agent; message: UserMessage; turn: number }): void
/**
* One message was discarded from the live inbox.
* @param agent - the agent whose inbox changed.
* @param event - the discarded message.
* @param payload.agent - the agent whose inbox changed.
* @param payload.message - the discarded message.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/inbox/discarded'(this: Scoped<Agent>, agent: Agent, event: { message: UserMessage }): void
'agent/inbox/discarded'(this: Scoped<Agent>, payload: { agent: Agent; message: UserMessage }): void
// ---- session lifecycle (emit) ----
/**
* The session lifecycle began, once before the first turn. Use
* `agent.inject()` to seed model-facing context. This is a notification, not
* a veto; disposal requested by a lifecycle owner is rechecked before the
* driver starts.
* @param agent - the agent whose session lifecycle began.
* @param source - why the session started (fresh startup, resume, …).
* @param payload.agent - the agent whose session lifecycle began.
* @param payload.source - why the session started (fresh startup, resume, …).
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/session-start'(this: Scoped<Agent>, agent: Agent, source: SessionStartSource): void
'agent/session-start'(this: Scoped<Agent>, payload: { agent: Agent; source: SessionStartSource }): void
// ---- the machine's extension seams ----
/**
* Reject a proposed step or replace the messages that enter it. Calling
* `next()` preserves the current messages.
* @param agent - the agent proposing the step.
* @param messages - messages removed from the inbox for this step.
* @param context - proposed turn and step coordinates plus cancellation.
* @param payload.agent - the agent proposing the step.
* @param payload.messages - messages removed from the inbox for this step.
* @param payload.turn - the turn that will own the step.
* @param payload.step - the step proposed by the loop.
* @param payload.signal - the current turn's cancellation signal.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode waterfall
*/
'agent/pre-step'(this: Scoped<Agent>, agent: Agent, messages: UserMessage[], context: PreStepContext, next: () => Promise<PreStepDecision>): Promise<PreStepDecision>
'agent/pre-step'(this: Scoped<Agent>, payload: { agent: Agent; messages: UserMessage[]; turn: number; step: number; signal: AbortSignal }, next: () => Promise<PreStepDecision>): Promise<PreStepDecision>
/**
* Replace the frozen call configuration. `await next()` yields the config
* the machine would use (agent options on the first request, the logged
* header afterwards); return a replacement to switch. Model-visible
* content must use logged channels; this seam cannot mutate messages.
* @param agent - the agent making the model call.
* @param turn - the open turn number.
* @param step - the step whose request this is.
* @param signal - the current turn's explicit abort signal.
* @param payload.agent - the agent making the model call.
* @param payload.turn - the open turn number.
* @param payload.step - the step whose request this is.
* @param payload.signal - the current turn's explicit abort signal.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode waterfall
*/
'agent/request'(this: Scoped<Agent>, agent: Agent, turn: number, step: number, signal: AbortSignal, next: () => Promise<LlmCallConfig>): Promise<LlmCallConfig>
'agent/request'(this: Scoped<Agent>, payload: { agent: Agent; turn: number; step: number; signal: AbortSignal }, next: () => Promise<LlmCallConfig>): Promise<LlmCallConfig>
/**
* Handle one failed model-request attempt before the loop retries or closes
* its step. A listener returns `{ kind: 'retry' }` without calling `next()`
* when it owns recovery, or calls `next()` to delegate. The default
* `undefined` leaves the failure terminal.
* @param agent - the agent whose request failed.
* @param context - request coordinates, provider, normalized failure, and serving policy.
* @param signal - the turn abort signal.
* @param payload.agent - the agent whose request failed.
* @param payload.turn - the turn containing the failed request.
* @param payload.step - the step containing the failed request attempt.
* @param payload.provider - the provider selected for the failed request.
* @param payload.failure - serializable facts normalized at the final adapter boundary.
* @param payload.retryPolicy - the policy of the adapter registration that served the failed request.
* @param payload.signal - the turn abort signal.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode waterfall
*/
'agent/request-error'(this: Scoped<Agent>, agent: Agent, context: RequestFailureContext, signal: AbortSignal, next: () => Promise<RequestErrorAction>): Promise<RequestErrorAction>
'agent/request-error'(this: Scoped<Agent>, payload: { agent: Agent; turn: number; step: number; provider: string; failure: LlmFailure; retryPolicy: ResolvedRetryPolicy | undefined; signal: AbortSignal }, next: () => Promise<RequestErrorAction>): Promise<RequestErrorAction>
/**
* The turn is about to close: the model owes no response (no live tool
* calls, no fresh steering). Awaited before the boundary commits — a
@@ -281,25 +264,25 @@ declare module 'cordis' {
* never short-circuits already-submitted next-step work: same-step
* `additionalContexts` or racing steering still runs, and the turn
* closes only when that inbox drains.
* @param agent - the agent whose turn is at its stop boundary.
* @param turn - the turn about to close.
* @param signal - the current turn's explicit abort signal.
* @param payload.agent - the agent whose turn is at its stop boundary.
* @param payload.turn - the turn about to close.
* @param payload.signal - the current turn's explicit abort signal.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode serial
*/
'agent/turn-stopping'(this: Scoped<Agent>, agent: Agent, turn: number, signal: AbortSignal): Promise<void> | void
'agent/turn-stopping'(this: Scoped<Agent>, payload: { agent: Agent; turn: number; signal: AbortSignal }): Promise<void> | void
// ---- error notifications (emit) ----
/**
* A step or turn errored. The machine reports a failure here even when
* the error has no in-turn position for a durable record.
* @param agent - the agent whose turn errored.
* @param turn - the turn in which the failure surfaced.
* @param step - the step at which the failure surfaced.
* @param error - the failure, verbatim.
* @param payload.agent - the agent whose turn errored.
* @param payload.turn - the turn in which the failure surfaced.
* @param payload.step - the step at which the failure surfaced.
* @param payload.error - the failure, verbatim.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.
* @mode emit
*/
'agent/error'(this: Scoped<Agent>, agent: Agent, turn: number, step: number, error: unknown): void
'agent/error'(this: Scoped<Agent>, payload: { agent: Agent; turn: number; step: number; error: unknown }): void
}
}

38
packages/core/agent/tests/agent.spec.ts
View File

@@ -11,6 +11,7 @@ import type {
Agent,
AgentCancelCause,
AgentFactory,
AgentStatus,
CreateAgentOptions,
ResumeAgentOptions,
} from '@deepseek-ai/dsh-agent'
@@ -145,8 +146,8 @@ describe('AgentRegistry', () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
const lifecycle: string[] = []
ctx.on('agent/created', agent => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/disposed', agent => void lifecycle.push(`disposed:${agent.id}`))
ctx.on('agent/created', ({ agent }) => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/disposed', ({ agent }) => void lifecycle.push(`disposed:${agent.id}`))
const agent = stubAgent('a1')
const dispose = ctx.agents.register(agent)
@@ -195,9 +196,9 @@ describe('AgentRegistry', () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
const lifecycle: string[] = []
ctx.on('agent/created', agent => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/created', ({ agent }) => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/created', () => { throw new Error('creation veto') })
ctx.on('agent/disposed', agent => void lifecycle.push(`disposed:${agent.id}`))
ctx.on('agent/disposed', ({ agent }) => void lifecycle.push(`disposed:${agent.id}`))
expect(() => ctx.agents.register(stubAgent('vetoed'))).toThrow('creation veto')
expect(ctx.agents.get(SessionId('vetoed'))).toBeUndefined()
@@ -213,7 +214,7 @@ describe('AgentRegistry', () => {
ctx.on('agent/created', () => Promise.reject(new Error('created async')) as never)
ctx.on('agent/disposed', () => { throw new Error('disposed sync') })
ctx.on('agent/disposed', () => Promise.reject(new Error('disposed async')) as never)
ctx.on('agent/disposed', agent => void heard.push(agent.id))
ctx.on('agent/disposed', ({ agent }) => void heard.push(agent.id))
const dispose = ctx.agents.register(stubAgent('contained'))
await Promise.resolve()
@@ -232,8 +233,8 @@ describe('AgentRegistry', () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
const lifecycle: string[] = []
ctx.on('agent/created', agent => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/disposed', agent => void lifecycle.push(`disposed:${agent.id}`))
ctx.on('agent/created', ({ agent }) => void lifecycle.push(`created:${agent.id}`))
ctx.on('agent/disposed', ({ agent }) => void lifecycle.push(`disposed:${agent.id}`))
const first = stubAgent('split')
const detachFirst = ctx.agents.enter(first, undefined)
@@ -280,9 +281,9 @@ describe('agentEvents()', () => {
const agent = stubAgent('event')
ctx.on('agent/status', () => { throw new Error('sync listener') })
ctx.on('agent/status', () => Promise.reject(new Error('async listener')) as never)
ctx.on('agent/status', (_agent, status) => void heard.push(status))
ctx.on('agent/status', ({ status }) => void heard.push(status))
agentEvents(ctx, agent).emit('agent/status', 'running')
agentEvents(ctx, agent).emit('agent/status', { status: 'running' })
await Promise.resolve()
expect(heard).toEqual(['running'])
expect(warnings).toEqual([
@@ -296,15 +297,30 @@ describe('agentEvents()', () => {
const agent = stubAgent('serial-event')
const signal = new AbortController().signal
const heard: Array<{ agent: Agent; turn: number; signal: AbortSignal }> = []
ctx.on('agent/turn-stopping', async (subject, turn, receivedSignal) => {
ctx.on('agent/turn-stopping', async ({ agent: subject, turn, signal: receivedSignal }) => {
await Promise.resolve()
heard.push({ agent: subject, turn, signal: receivedSignal })
})
await agentEvents(ctx, agent).serial('agent/turn-stopping', 3, signal)
await agentEvents(ctx, agent).serial('agent/turn-stopping', { turn: 3, signal })
expect(heard).toEqual([{ agent, turn: 3, signal }])
})
it('injects the fused subject even when the payload carries a conflicting agent field', async () => {
const ctx = new Context()
const agent = stubAgent('fused-subject')
const other = stubAgent('payload-agent')
const heard: Agent[] = []
ctx.on('agent/status', ({ agent: subject }) => void heard.push(subject))
// A structurally acceptable payload may carry an extra `agent` field; the
// dispatcher's injected subject must win over it.
const payload: { status: AgentStatus; agent: Agent } = { status: 'running', agent: other }
agentEvents(ctx, agent).emit('agent/status', payload)
expect(heard).toEqual([agent])
})
})
describe('explicit cancellation contract', () => {

14
packages/core/agent/tests/invariant.spec.ts
View File

@@ -21,17 +21,17 @@ describe('agent status invariants', () => {
const ctx = await setup()
const agent = mockAgent('a1')
expect(() => {
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, status: 'idle' })
ctx.emit(scopeTarget(agent, agent), 'agent/status', { agent, status: 'running' })
ctx.emit(scopeTarget(agent, agent), 'agent/status', { agent, status: 'idle' })
}).not.toThrow()
})
it('rejects a no-op transition', async () => {
const ctx = await setup()
const agent = mockAgent('a3')
ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running')
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', agent, 'running') })
ctx.emit(scopeTarget(agent, agent), 'agent/status', { agent, status: 'running' })
expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/status', { agent, status: 'running' }) })
.toThrow(/no-op transition/)
})
@@ -39,7 +39,7 @@ describe('agent status invariants', () => {
const ctx = await setup()
const a = mockAgent('a5')
const b = mockAgent('b5')
ctx.emit(scopeTarget(a, a), 'agent/status', a, 'running')
expect(() => { ctx.emit(scopeTarget(b, b), 'agent/status', b, 'running') }).not.toThrow()
ctx.emit(scopeTarget(a, a), 'agent/status', { agent: a, status: 'running' })
expect(() => { ctx.emit(scopeTarget(b, b), 'agent/status', { agent: b, status: 'running' }) }).not.toThrow()
})
})

8
packages/core/agent/tests/llm-target.spec.ts
View File

@@ -21,7 +21,7 @@ describe('installAgentLlmTarget()', () => {
expect((await ctx.systemPrompt.assemble()).variables).toEqual({})
await expect(agentEvents(ctx, agent).waterfall(
'agent/request', 1, 0, signal, () => Promise.resolve(seed),
'agent/request', { turn: 1, step: 0, signal }, () => Promise.resolve(seed),
)).resolves.toBe(seed)
target.current = {
@@ -32,7 +32,7 @@ describe('installAgentLlmTarget()', () => {
expect((await ctx.systemPrompt.assemble()).variables).toMatchObject({ provider: 'alpha', model: 'a1' })
target.current = { provider: 'beta', model: 'b1' }
await expect(agentEvents(ctx, agent).waterfall(
'agent/request', 1, 0, signal, () => Promise.resolve(seed),
'agent/request', { turn: 1, step: 0, signal }, () => Promise.resolve(seed),
)).resolves.toEqual({
provider: 'alpha',
model: 'a1',
@@ -48,13 +48,13 @@ describe('installAgentLlmTarget()', () => {
temperature: 0.2,
}
await expect(agentEvents(ctx, agent).waterfall(
'agent/request', 1, 1, signal, () => Promise.resolve(inherited),
'agent/request', { turn: 1, step: 1, signal }, () => Promise.resolve(inherited),
)).resolves.toEqual({ provider: 'beta', model: 'b1', temperature: 0.2 })
dispose()
expect((await ctx.systemPrompt.assemble()).variables).toEqual({})
await expect(agentEvents(ctx, agent).waterfall(
'agent/request', 2, 0, signal, () => Promise.resolve(seed),
'agent/request', { turn: 2, step: 0, signal }, () => Promise.resolve(seed),
)).resolves.toBe(seed)
await ctx.fiber.dispose()
})

26
packages/core/scope/src/scoped-events.generated.ts
View File

@@ -8,20 +8,20 @@
type ScopedSubjectResolver = (args: readonly unknown[]) => unknown
const scopedSubjectResolvers: Readonly<Record<string, ScopedSubjectResolver | null>> = Object.freeze({
'agent/created': args => args[0],
'agent/disposed': args => args[0],
'agent/error': args => args[0],
'agent/inbox/claimed': args => args[0],
'agent/inbox/discarded': args => args[0],
'agent/inbox/inserted': args => args[0],
'agent/pre-step': args => args[0],
'agent/request': args => args[0],
'agent/request-error': args => args[0],
'agent/session-start': args => args[0],
'agent/status': args => args[0],
'agent/turn-stopping': args => args[0],
'agent/created': args => (args[0] as Record<string, unknown>)['agent'],
'agent/disposed': args => (args[0] as Record<string, unknown>)['agent'],
'agent/error': args => (args[0] as Record<string, unknown>)['agent'],
'agent/inbox/claimed': args => (args[0] as Record<string, unknown>)['agent'],
'agent/inbox/discarded': args => (args[0] as Record<string, unknown>)['agent'],
'agent/inbox/inserted': args => (args[0] as Record<string, unknown>)['agent'],
'agent/pre-step': args => (args[0] as Record<string, unknown>)['agent'],
'agent/request': args => (args[0] as Record<string, unknown>)['agent'],
'agent/request-error': args => (args[0] as Record<string, unknown>)['agent'],
'agent/session-start': args => (args[0] as Record<string, unknown>)['agent'],
'agent/status': args => (args[0] as Record<string, unknown>)['agent'],
'agent/turn-stopping': args => (args[0] as Record<string, unknown>)['agent'],
'approval/request': args => (args[0] as Record<string, unknown>)['agent'],
'goal/changed': args => args[0],
'goal/changed': args => (args[0] as Record<string, unknown>)['agent'],
'session/created': null,
'session/disposed': null,
'session/event': null,

30
packages/core/scope/tests/invariant.spec.ts
View File

@@ -28,7 +28,7 @@ describe('scoped-dispatch invariants', () => {
const ctx = await setup()
expect(() => { emit(ctx, undefined, 'ordinary/event', []) }).not.toThrow()
const agent = { id: 'a1' }
expect(() => { emit(ctx, undefined, 'agent/error', [agent, 1, 0, new Error('x')]) })
expect(() => { emit(ctx, undefined, 'agent/error', [{ agent, turn: 1, step: 0, error: new Error('x') }]) })
.toThrow(/dispatched without a scope carrier/)
})
@@ -45,34 +45,34 @@ describe('scoped-dispatch invariants', () => {
source: { kind: 'user' },
})
const agentRows = {
'agent/created': [agent],
'agent/disposed': [agent],
'agent/status': [agent, 'idle'],
'agent/inbox/inserted': [agent, { message }],
'agent/inbox/claimed': [agent, { message, turn: 1 }],
'agent/inbox/discarded': [agent, { message }],
'agent/session-start': [agent, 'startup'],
'agent/pre-step': [agent, [message], { turn: 1, step: 1, signal }, () => Promise.resolve({ kind: 'enter', messages: [message] })],
'agent/request': [agent, 1, 1, signal, () => Promise.resolve(config)],
'agent/created': [{ agent }],
'agent/disposed': [{ agent }],
'agent/status': [{ agent, status: 'idle' }],
'agent/inbox/inserted': [{ agent, message }],
'agent/inbox/claimed': [{ agent, message, turn: 1 }],
'agent/inbox/discarded': [{ agent, message }],
'agent/session-start': [{ agent, source: 'startup' }],
'agent/pre-step': [{ agent, messages: [message], turn: 1, step: 1, signal }, () => Promise.resolve({ kind: 'enter', messages: [message] })],
'agent/request': [{ agent, turn: 1, step: 1, signal }, () => Promise.resolve(config)],
'agent/request-error': [
agent,
{
agent,
turn: 1,
step: 1,
provider: 'p',
failure: { message: 'request', code: 'UNKNOWN' },
retryPolicy: undefined,
signal,
},
signal,
() => Promise.resolve(undefined),
],
'agent/turn-stopping': [agent, 1, signal],
'agent/error': [agent, 1, 0, new Error('x')],
'agent/turn-stopping': [{ agent, turn: 1, signal }],
'agent/error': [{ agent, turn: 1, step: 0, error: new Error('x') }],
} satisfies { [K in AgentEventName]: EventArgs<K> }
const rows: Array<[string, unknown[]]> = [
...Object.entries(agentRows),
['approval/request', [{ agent, toolName: 'echo' }, () => Promise.resolve('unavailable')]],
['goal/changed', [agent, { operation: 'create', ref: { id: 'goal-a', revision: 1 } }]],
['goal/changed', [{ agent, change: { operation: 'create', ref: { id: 'goal-a', revision: 1 } } }]],
['system-prompt/assemble', [[], { scope: agent }]],
['tools/code-dispatch-log', [{ exec: { callId: 'c', name: 't', arguments: {} }, agent, subCallId: 'c:code:1', name: 't', isError: false, content: [] }, () => Promise.resolve([])]],
['tools/execute', [{ callId: 'c', name: 't', arguments: {}, agent }, () => Promise.resolve({ content: [], isError: false })]],

160
packages/core/session/src/index.ts
View File

@@ -13,13 +13,15 @@ import { scopeOf, scopeTarget } from '@deepseek-ai/dsh-scope'
import type { Scoped } from '@deepseek-ai/dsh-scope'
import type { Message } from '@deepseek-ai/dsh-llm'
import { SESSION_FORMAT_VERSION, SessionId } from './types.ts'
import type { CreateSessionOptions, EpochHeader, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
import type { CreateSessionOptions, EpochHeader, PrepareSessionOptions, RequestContext, SessionEvent, SessionEventMap, SessionEventType, SessionHeader, SurfaceIntent, SurfaceEventType } from './types.ts'
import { snapshotJsonValue } from './json.ts'
import { SurfaceManager } from './surface.ts'
import { deriveEventMessage, SurfaceManager } from './surface.ts'
import type { SessionSurface } from './surface.ts'
import { foldRequestHeader } from './request-header.ts'
export * from './types.ts'
export { SessionPreparation } from './preparation.ts'
export type { SessionPreparationOptions } from './preparation.ts'
export type { AssistantMessage, ToolResultMessage, UserMessage } from '@deepseek-ai/dsh-llm'
export { isJsonValue, snapshotJsonValue } from './json.ts'
export type { JsonValue } from './json.ts'
@@ -27,7 +29,7 @@ export { interruptedTurnClosers, lastActivityTime, TOOL_NOT_STARTED, TOOL_OUTCOM
export { decodeStorageRecord, packChunkRuns } from './chunk-rows.ts'
export type { ChunkRow, StorageRecord } from './chunk-rows.ts'
export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from './surface.ts'
export { foldSurface, isAppendSurfaceEvent, isReplacementSurfaceEvent, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export { deriveEventMessage, foldSurface, isAppendSurfaceEvent, isReplacementSurfaceEvent, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
/**
@@ -143,6 +145,17 @@ function validateSessionHeader(id: SessionId, input: unknown): SessionHeader {
return deepFreeze(record as unknown as SessionHeader)
}
/** Validate and freeze one exclusively owned persistence header in place. */
function validateRestoredSessionHeader(id: SessionId, input: unknown): SessionHeader {
if (input !== null && typeof input === 'object' && !Array.isArray(input)) {
const prototype = Reflect.getPrototypeOf(input)
if (prototype !== Object.prototype && prototype !== null) {
throw new Error('session header is not a plain JSON record')
}
}
return validateSessionHeader(id, input)
}
/** Detach, validate, and freeze the creation metadata published by a session. */
function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHeader {
const input: unknown = source === undefined
@@ -190,23 +203,58 @@ export function snapshotSessionEvent<T extends SessionEvent>(event: T): T {
return adoptSessionEvent(structuredClone(event))
}
/** Deep-freeze one acyclic JSON tree without consuming the JavaScript call stack. */
function freezeRestoredObject<T extends object>(value: T): T {
const pending: object[] = [value]
while (pending.length > 0) {
// The non-empty check proves an object remains to visit.
// oxlint-disable-next-line typescript/no-non-null-assertion
const current = pending.pop()!
Object.freeze(current)
for (const key in current) {
const child = (current as Record<string, unknown>)[key]
if (child !== null && typeof child === 'object') pending.push(child)
}
}
return value
}
/** Validate the fixed event envelope after one-pass JSON materialization. */
function assertSessionEventEnvelope(value: Record<string, unknown>, index: number): asserts value is SessionEvent {
const event = value
if (event['type'] === 'request/header-delta') {
throw new Error(`seed event at index ${index} uses unsupported legacy request/header-delta format`)
}
const allowed = new Set(['type', 'seq', 'time', 'data', 'surfaceOp', 'sourceEventSeqs'])
if (Object.keys(event).some(key => !allowed.has(key))
|| !Object.hasOwn(event, 'type') || typeof event['type'] !== 'string'
|| !Object.hasOwn(event, 'seq') || typeof event['seq'] !== 'number'
|| !Number.isSafeInteger(event['seq']) || event['seq'] < 0
|| !Object.hasOwn(event, 'time') || typeof event['time'] !== 'number'
|| !Number.isSafeInteger(event['time']) || event['time'] < 0
|| !Object.hasOwn(event, 'data')) {
for (const key in event) {
switch (key) {
case 'type':
case 'seq':
case 'time':
case 'data':
case 'surfaceOp':
case 'sourceEventSeqs':
break
default:
throw new Error(`seed event at index ${index} has an invalid event envelope`)
}
}
const type = event['type']
const seq = event['seq']
const time = event['time']
if (typeof type !== 'string'
|| typeof seq !== 'number' || !Number.isSafeInteger(seq) || seq < 0
|| typeof time !== 'number' || !Number.isSafeInteger(time)
|| event['data'] === undefined) {
throw new Error(`seed event at index ${index} has an invalid event envelope`)
}
assertCurrentLlmShape(event, index)
switch (type) {
case 'request/header':
case 'user/message':
case 'assistant/message':
case 'tool/result':
assertCurrentLlmShape(event, index)
break
}
}
/** Reject obsolete request headers and malformed messages at the seed/load boundary. */
@@ -236,6 +284,8 @@ function assertCurrentLlmShape(event: Record<string, unknown>, index: number): v
assertMessageEventShape(event, `seed ${type} at index ${index}`)
}
const allowedAdapterKeys = new Set(['reasoningEffort', 'maxTokens'])
/** Validate adapter-default provenance imported from a durable request header. */
function assertAdapterDefaults(
value: unknown,
@@ -247,8 +297,7 @@ function assertAdapterDefaults(
throw new Error(`seed request/header at index ${index} has invalid adapterDefaults`)
}
const defaults = value as Record<string, unknown>
const allowed = new Set(['reasoningEffort', 'maxTokens'])
if (Object.keys(defaults).some(key => !allowed.has(key))
if (Object.keys(defaults).some(key => !allowedAdapterKeys.has(key))
|| Object.values(defaults).some(marker => marker !== true)
|| defaults['reasoningEffort'] === true && config['reasoningEffort'] === undefined
|| defaults['maxTokens'] === true && config['maxTokens'] === undefined) {
@@ -442,7 +491,28 @@ export class Session {
return new Session(id, seed, header)
}
private constructor(id: SessionId, seed?: readonly SessionEvent[], header?: SessionHeader) {
/**
* Restore a detached session by taking ownership of fresh persistence values.
* Storage shape, event envelopes, sequence continuity, surface transitions,
* and header fields are validated before the graphs are frozen in place.
* @param id - restored session identity.
* @param seed - fresh detached events whose ownership is transferred.
* @param header - fresh detached metadata whose ownership is transferred.
* @returns a restored detached session.
*/
static fromRestore(id: SessionId, seed: readonly SessionEvent[], header: SessionHeader): Session {
return new Session(id, seed, header, 'restore')
}
private constructor(
id: SessionId,
seed?: readonly SessionEvent[],
header?: SessionHeader,
mode: 'snapshot' | 'restore' = 'snapshot',
) {
const restoredHeader = mode === 'restore'
? validateRestoredSessionHeader(id, header)
: undefined
if (seed !== undefined) {
// Validate the seed to the SAME invariants `append` enforces, so a
// replay/fork (`ctx.sessions.create(id, { seed })`) cannot construct a
@@ -454,7 +524,7 @@ export class Session {
for (const [index, source] of seed.entries()) {
// The seed is a persistence/replay boundary: validate and detach the
// complete event in one lossless-JSON pass.
const snapshot = snapshotJsonValue(source)
const snapshot = mode === 'restore' ? source : snapshotJsonValue(source)
if (snapshot === undefined) {
throw new Error(`seed event at index ${index} is not losslessly JSON-serializable`)
}
@@ -471,11 +541,11 @@ export class Session {
} catch (error: unknown) {
throw new Error(`invalid seed event at index ${index}: ${error instanceof Error ? error.message : 'invalid surface metadata'}`)
}
this.log.push(deepFreeze(snapshot))
this.log.push(mode === 'restore' ? freezeRestoredObject(snapshot) : deepFreeze(snapshot))
}
}
this.firstLiveSeq = this.log.length
this.header = snapshotSessionHeader(id, header)
this.header = restoredHeader ?? snapshotSessionHeader(id, header)
// Appended here so the marker is already in `events` when a backend
// captures the creation seed: no load-time write. Re-marking is skipped
// because a cold session is resumed on first touch, so repeatedly opening
@@ -685,50 +755,13 @@ export class Session {
}
/**
* Project a single event into the LLM message it derives to, or null when
* it produces none — a non-surface event (chunk, boundary, log-only record)
* or an empty-content assistant/message (which exists only to host usage).
* The per-node pure function {@link deriveMessages} folds over the surface;
* an external reconstructor (or the dev invariant) folds the same function
* over a log prefix's surface to rebuild the exact messages any request was
* built from (the reconstructability Agent Note). The returned message is
* the already frozen message nested in the event wrapper and shared by
* delivery, durable history, and model requests.
* Instance face of the pure per-node `deriveEventMessage` export from
* `surface.ts`.
* @param event - the event to project.
* @returns the derived message, or null when the event produces none.
*/
deriveEventMessage(event: SessionEvent): Message | null {
// Intentionally non-exhaustive: only message-producing events derive
// history; turn/step boundaries, chunks, usage, and errors are
// trace/replay data.
switch (event.type) {
// Ordinary prompts and injected context project in user role: the
// event's model-facing content stays verbatim. Do NOT
// re-add per-type framing (e.g. `<context>`) here: framing is
// caller-owned — a producer bakes it into `content`, as workspace-context
// does with `<system-reminder>` — or, if reintroduced, must be driven by
// the event `meta` map and a dedicated renderer, keeping this projection a
// verbatim pass-through. See the deferred design note in
// ../../../../.agents/notes/implemented/simplification/2026-07-20-unwrap-injected-content-envelopes.md
case 'user/message': {
return event.data
}
case 'assistant/message': {
// Skip an empty-content assistant/message: it exists only to host a
// max-tokens step's usage and must not inject a content-less assistant
// turn into the provider transcript.
if (event.data.message.content.length === 0) return null
return event.data.message
}
case 'tool/result': {
return event.data.message
}
default:
// A non-surface event (boundary, chunk, log-only record) projects to
// no message. Merge-extensible union: no assertNever here.
return null
}
return deriveEventMessage(event)
}
}
@@ -816,13 +849,17 @@ export class SessionStore extends Service {
* before the driver's closing events commit, dropping them.
*
* @param id - the session id; omitted, the store mints `session-<n>`.
* @param options - seed events and/or creation metadata for the header.
* @param options - seed events and/or creation metadata for the header. With
* `seedSource: 'persistence'`, metadata and events must be fresh detached
* graphs whose ownership transfers to this call: they are validated and
* frozen in place through {@link Session.fromRestore}, so the caller must
* retain no mutable aliases.
* @returns the constructed session, NOT yet in the store.
* @throws if a session with `id` already exists, metadata is not a plain
* lossless-JSON record with valid scalar fields, or `meta.cwd` is a
* non-absolute path.
*/
prepare(id?: SessionId, options?: CreateSessionOptions): Session {
prepare(id?: SessionId, options?: PrepareSessionOptions): Session {
let sessionId: SessionId
if (id === undefined) {
do sessionId = SessionId(`session-${++this.counter}`)
@@ -831,6 +868,9 @@ export class SessionStore extends Service {
sessionId = SessionId(id)
}
if (this.store.has(sessionId)) throw new Error(`session "${sessionId}" already exists`)
if (options?.seedSource === 'persistence') {
return Session.fromRestore(sessionId, options.seed, options.meta)
}
const seed = options?.seed
const meta = options?.meta
const header: SessionHeader = {

49
packages/core/session/src/preparation.ts Normal file
View File

@@ -0,0 +1,49 @@
/**
* Ownership of one unpublished Session before registry publication.
* @module @deepseek-ai/dsh-session/preparation
*/
import type { Session } from './index.ts'
/** Options for a preparation whose provider retains unpublished state. */
export interface SessionPreparationOptions {
/** Release provider-owned state when the Session was not published. */
readonly release?: () => void
}
/**
* One exact unpublished Session and the provider state that keeps it usable.
* Disposal is synchronous and idempotent. Providers decide whether release
* returns the Session to a cache or discards it; publication may consume that
* state before disposal, making the callback a no-op.
*/
export class SessionPreparation implements Disposable {
private released = false
/** The exact Session to use for setup and publication. */
readonly session: Session
private constructor(
session: Session,
private readonly options: SessionPreparationOptions,
) {
this.session = session
}
/**
* Wrap an unpublished Session in one preparation lifetime.
* @param session - exact unpublished Session.
* @param options - optional provider release behavior.
* @returns a preparation disposed after publication or rollback.
*/
static create(session: Session, options?: SessionPreparationOptions): SessionPreparation {
return new SessionPreparation(session, options ?? {})
}
/** Release provider state once when this preparation leaves its caller. */
[Symbol.dispose](): void {
if (this.released) return
this.released = true
this.options.release?.()
}
}

77
packages/core/session/src/surface.ts
View File

@@ -8,6 +8,7 @@
* @module @deepseek-ai/dsh-session/surface
*/
import type { Message } from '@deepseek-ai/dsh-llm'
import type { SessionEvent, SurfaceEvent, SurfaceEventType, SurfaceOp } from './types.ts'
/** Runtime counterpart of the message-producing event union. */
@@ -66,6 +67,52 @@ export function isReplacementSurfaceEvent(
return isSurfaceEvent(event) && event.surfaceOp !== 'append'
}
/**
* Project a single event into the LLM message it derives to, or null when it
* produces none — a non-surface event (chunk, boundary, log-only record) or an
* empty-content assistant/message (which exists only to host usage). This is
* THE per-node projection rule: `Session.deriveMessages` folds it over the
* live surface, external reconstructors and pure projections fold the same
* function over a log prefix's surface to rebuild the exact messages any
* request was built from. The returned message is the already frozen message
* nested in the event wrapper and shared by delivery, durable history, and
* model requests.
* @param event - the event to project.
* @returns the derived message, or null when the event produces none.
*/
export function deriveEventMessage(event: SessionEvent): Message | null {
// Intentionally non-exhaustive: only message-producing events derive
// history; turn/step boundaries, chunks, usage, and errors are trace/replay
// data.
switch (event.type) {
// Ordinary prompts and injected context project in user role: the event's
// model-facing content stays verbatim. Do NOT re-add per-type framing
// (e.g. `<context>`) here: framing is caller-owned — a producer bakes it
// into `content`, as workspace-context does with `<system-reminder>` — or,
// if reintroduced, must be driven by the event `meta` map and a dedicated
// renderer, keeping this projection a verbatim pass-through. See the
// deferred design note in
// ../../../../.agents/notes/implemented/simplification/2026-07-20-unwrap-injected-content-envelopes.md
case 'user/message': {
return event.data
}
case 'assistant/message': {
// Skip an empty-content assistant/message: it exists only to host a
// max-tokens step's usage and must not inject a content-less assistant
// turn into the provider transcript.
if (event.data.message.content.length === 0) return null
return event.data.message
}
case 'tool/result': {
return event.data.message
}
default:
// A non-surface event (boundary, chunk, log-only record) projects to
// no message. Merge-extensible union: no assertNever here.
return null
}
}
/** One replacement operation observed while folding a session surface. */
export interface SurfaceFoldReplacement {
/** Seq of the event that replaced the prior surface range. */
@@ -308,6 +355,14 @@ function applySurfaceEvent(
baseSeq: number,
): SurfaceFoldReplacement | undefined {
const plan = planSurfaceEvent(state, event, expectedSeq, events, baseSeq)
return applySurfacePlan(state, plan)
}
/** Commit one previously validated surface transition. */
function applySurfacePlan(
state: SurfaceFoldState,
plan: SurfacePlan | undefined,
): SurfaceFoldReplacement | undefined {
if (plan?.kind === 'append') {
state.nodes.push(plan.seq)
} else if (plan?.kind === 'replace') {
@@ -345,6 +400,8 @@ export class SurfaceManager implements SessionSurface {
private _state = createFoldState()
/** Last processed absolute seq. */
private _lastProcessedSeq: number
/** Candidate already validated by `validateNext`, pending exact log admission. */
private _pendingPlan: { event: SessionEvent; expectedSeq: number; plan: SurfacePlan | undefined } | undefined
/**
* @param log - Contiguous complete log or loaded event window.
@@ -363,13 +420,12 @@ export class SurfaceManager implements SessionSurface {
*/
validateNext(event: SessionEvent): void {
if (this._lastProcessedSeq < this.baseSeq + this.log.length - 1) this._processDelta()
planSurfaceEvent(
this._state,
const expectedSeq = this.baseSeq + this.log.length
this._pendingPlan = {
event,
this.baseSeq + this.log.length,
this.log,
this.baseSeq,
)
expectedSeq,
plan: planSurfaceEvent(this._state, event, expectedSeq, this.log, this.baseSeq),
}
}
/** Monotonic count of folded positional replacements. */
@@ -390,7 +446,14 @@ export class SurfaceManager implements SessionSurface {
for (let seq = this._lastProcessedSeq + 1; seq <= tailSeq; seq++) {
const index = seq - this.baseSeq
// oxlint-disable-next-line typescript/no-non-null-assertion -- bounded by the loop condition
applySurfaceEvent(this._state, this.log[index]!, seq, this.log, this.baseSeq)
const event = this.log[index]!
const pending = this._pendingPlan
if (pending?.event === event && pending.expectedSeq === seq) {
applySurfacePlan(this._state, pending.plan)
} else {
applySurfaceEvent(this._state, event, seq, this.log, this.baseSeq)
}
if (pending !== undefined && pending.expectedSeq <= seq) this._pendingPlan = undefined
this._lastProcessedSeq = seq
}
}

18
packages/core/session/src/types.ts
View File

@@ -93,6 +93,24 @@ export interface CreateSessionOptions {
}
}
/**
* Fresh storage values transferred to {@link SessionStore.prepare} without a
* second serialization copy. Callers retain no mutable aliases.
*/
export interface RestoredSessionOptions {
/** Fresh detached storage events to validate and freeze in place. */
readonly seed: SessionEvent[]
/** Fresh detached storage metadata to validate and freeze in place. */
readonly meta: SessionHeader
/** Select the persistence ownership-transfer path. */
readonly seedSource: 'persistence'
}
/** Inputs accepted while constructing an unpublished Session. */
export type PrepareSessionOptions =
| (CreateSessionOptions & { readonly seedSource?: undefined })
| RestoredSessionOptions
/** Why an active agent driver was cancelled. */
export type AgentCancelCause =
| { readonly kind: 'user' }

40
packages/core/session/tests/session.spec.ts
View File

@@ -941,6 +941,36 @@ describe('Session', () => {
expect(() => { appendedEvent.data.todos[0]!.content = 'mutated' }).toThrow(TypeError)
})
it('iteratively freezes deeply nested restored event data', () => {
const depth = 20_000
const data: Record<string, unknown> = {}
let tail = data
for (let index = 0; index < depth; index += 1) {
const child: Record<string, unknown> = {}
tail['child'] = child
tail = child
}
const event = {
type: 'test/deep-restore', seq: 0, time: 1, data,
} as unknown as SessionEvent
expect(() => Session.fromRestore(SessionId('deep-restore'), [event], {
version: SESSION_FORMAT_VERSION,
id: SessionId('deep-restore'),
createdAt: 1,
})).not.toThrow()
let current: unknown = event
let frozenNodes = 0
for (let index = 0; index <= depth + 1; index += 1) {
if (!Object.isFrozen(current)) break
frozenNodes += 1
current = (current as Record<string, unknown>)['data']
?? (current as Record<string, unknown>)['child']
}
expect(frozenNodes).toBe(depth + 2)
})
it('returns cached frozen event-array snapshots that do not grow after append', () => {
const session = Session.create(SessionId('events-snapshot'))
session.append('turn/start', { turn: 1 })
@@ -998,6 +1028,15 @@ describe('Session', () => {
expect(() => Session.create(SessionId('header-invalid'), undefined, new ExoticHeader()))
.toThrow(/not losslessly JSON-serializable/)
expect(() => Session.fromRestore(SessionId('header-invalid'), [], new ExoticHeader()))
.toThrow(/not a plain JSON record/)
for (const header of [null, 1, []]) {
expect(() => Session.fromRestore(
SessionId('header-invalid'),
[],
header as unknown as SessionHeader,
)).toThrow(/not a plain JSON record/)
}
expect(() => Session.create(SessionId('header-invalid'), undefined, {
version: SESSION_FORMAT_VERSION,
id: SessionId('header-invalid'),
@@ -1050,7 +1089,6 @@ describe('Session', () => {
{ ...base, seq: -1 },
{ ...base, time: '1' },
{ ...base, time: 0.5 },
{ ...base, time: -1 },
{ type: base.type, seq: base.seq, time: base.time },
]

36
packages/core/system-prompt/src/index.ts
View File

@@ -8,7 +8,7 @@ import { Context, Service } from 'cordis'
import z from 'schemastery'
import { AnonymousEntries, NamedEntries, ScopedLayers, scopeTarget } from '@deepseek-ai/dsh-scope'
import type { ScopeKey, ScopeLayer, Scoped } from '@deepseek-ai/dsh-scope'
import type { ToolSchema } from '@deepseek-ai/dsh-llm'
import type { ContextSnapshotSection, ToolSchema } from '@deepseek-ai/dsh-llm'
declare module 'cordis' {
interface Context {
@@ -200,15 +200,39 @@ export function renderPrompt(assembly: PromptAssembly): string {
* @returns the current full snapshot, or `''` when no context is active.
*/
export function renderContextSnapshot(assembly: PromptAssembly): string {
const body = assembly.contexts
.map(context => interpolate(context, assembly.variables, 'context'))
.filter(text => text.length > 0)
.join('\n\n')
return joinContextSections(renderContextSections(assembly))
}
/**
* The model-facing snapshot text for an already-rendered section list.
*
* A caller that also needs the sections renders them once and joins here, so a
* request does not interpolate every context twice.
* @param sections - sections from {@link renderContextSections}.
* @returns the current full snapshot, or `''` when no context is active.
*/
export function joinContextSections(sections: readonly ContextSnapshotSection[]): string {
const body = sections.map(section => section.text).join('\n\n')
if (body.length === 0) return ''
return `Current runtime context. This snapshot supersedes earlier runtime-context snapshots.\n\n${body}`
}
/** Interpolate one section or context and attribute diagnostics to its owner. */
/**
* The same snapshot, kept as the named contributions it was assembled from.
*
* {@link renderContextSnapshot} joins these for the model; a consumer that
* presents the snapshot uses them to attribute each part to the subsystem that
* contributed it, without re-splitting the joined prose.
* @param assembly - the assembly whose contexts and variables to render.
* @returns one entry per contributing context that rendered to non-empty text.
*/
export function renderContextSections(assembly: PromptAssembly): ContextSnapshotSection[] {
return assembly.contexts
.map(context => ({ name: context.name, text: interpolate(context, assembly.variables, 'context') }))
.filter(section => section.text.length > 0)
}
/** Interpolate one section or context and attribute diagnostics to its owning input. */
function interpolate(
input: AssembledSection | AssembledContext,
variables: Record<string, string | undefined>,