feat(agent): the agent is a registration scope — Agent.ctx, setup slot, fused scoped dispatch
Every live agent owns a dsh-scope context (Agent.ctx, key = the agent), minted inside the loop's composite lifecycle effect: registrations through it are agent-visible and agent-lifetime, and agent.ctx listeners hear only that agent's dispatches. The composite yields the scope's raw disposer first (identity-nested, no un-nested window), then session entry (scoped enter captures the session carrier), then registration; teardown runs stop/drain -> unregister -> detach session -> unwind scope, keeping store/registry rollback synchronous on every failure path. CreateAgentOptions.setup(agentCtx) runs after the scope is minted and the agent registered, before agent/session-start and the loop start — the slot where a creator composes the agent's scoped world (persona sections, restrict(), scoped tools); a throwing setup unwinds inside the rollback boundary. Setup registers, it never drives. agentEvents(ctx, agent) fuses the scope carrier with the injected subject argument for every agent/* dispatch (the correct dispatch is the shortest spelling); assembleContextFor(agent) pairs the agent DX field with the scope layer selector. All loop/agent/registry dispatch sites converted; agent/* event declarations carry this: Scoped<Agent>; ctx.agent is a safe root accessor defaulting undefined, shadowed by each agent context.
This commit is contained in:
@@ -24,6 +24,7 @@
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-llm": "^0.0.1",
|
||||
"@deepseek-ai/dsh-scope": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
@@ -37,6 +38,7 @@
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-invariants": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-scope": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence": "workspace:^",
|
||||
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
|
||||
|
||||
@@ -7,6 +7,8 @@
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import { scopeTarget } from '@deepseek-ai/dsh-scope'
|
||||
import type { Scoped } from '@deepseek-ai/dsh-scope'
|
||||
import type { AgentId, AgentOptions, AgentStatus, SendOptions } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
|
||||
@@ -28,6 +30,27 @@ export class ReactLoopAgent implements Agent {
|
||||
*/
|
||||
readonly inbox = new Inbox()
|
||||
|
||||
/**
|
||||
* The agent's scope context ({@link Agent.ctx}), wired by the factory right
|
||||
* after the scope is minted — before the agent is registered, announced, or
|
||||
* driven, so no consumer can observe it unset. Definite-assignment (`!`)
|
||||
* expresses that two-phase construction: the agent object and its scope
|
||||
* context are mutually referential (the scope is keyed BY this agent), so
|
||||
* neither can exist strictly before the other.
|
||||
*/
|
||||
ctx!: Context
|
||||
|
||||
/**
|
||||
* The dispatch carrier for this agent's own emits (`agent/status`,
|
||||
* `agent/queued`, `agent/error`): keyed by the agent, base = the agent
|
||||
* (listener `this` is the agent). Built lazily because it is self-referential.
|
||||
*/
|
||||
private get carrier(): Scoped<Agent> {
|
||||
return (this.#carrier ??= scopeTarget(this, this))
|
||||
}
|
||||
|
||||
#carrier: Scoped<Agent> | undefined
|
||||
|
||||
private _status: AgentStatus = 'idle'
|
||||
private currentAbort: AbortController | undefined
|
||||
/**
|
||||
@@ -63,7 +86,7 @@ export class ReactLoopAgent implements Agent {
|
||||
private idleWaiters: (() => void)[] = []
|
||||
|
||||
constructor(
|
||||
private ctx: Context,
|
||||
private loopCtx: Context,
|
||||
public readonly id: AgentId,
|
||||
public readonly options: AgentOptions,
|
||||
public readonly session: Session,
|
||||
@@ -87,9 +110,9 @@ export class ReactLoopAgent implements Agent {
|
||||
// not hang on one bad listener).
|
||||
if (status !== 'running') this.settleIdleWaiters()
|
||||
try {
|
||||
this.ctx.emit('agent/status', this, status)
|
||||
this.loopCtx.emit(this.carrier, 'agent/status', this, status)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`agent "${this.id}": agent/status listener threw on ${status}: ${String(error)}`)
|
||||
this.loopCtx.logger.warn(`agent "${this.id}": agent/status listener threw on ${status}: ${String(error)}`)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -112,7 +135,7 @@ export class ReactLoopAgent implements Agent {
|
||||
if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
|
||||
const source = this.resolveSource(options)
|
||||
this.inbox.enqueue({ content, source })
|
||||
this.ctx.emit('agent/queued', this, content, { source, steering: false })
|
||||
this.loopCtx.emit(this.carrier, 'agent/queued', this, content, { source, steering: false })
|
||||
}
|
||||
|
||||
steer(content: ContentBlock[], options?: SendOptions): void {
|
||||
@@ -120,7 +143,7 @@ export class ReactLoopAgent implements Agent {
|
||||
if (this._status !== 'running') { this.send(content, options); return }
|
||||
const source = this.resolveSource(options)
|
||||
this.inbox.steer({ content, source })
|
||||
this.ctx.emit('agent/queued', this, content, { source, steering: true })
|
||||
this.loopCtx.emit(this.carrier, 'agent/queued', this, content, { source, steering: true })
|
||||
}
|
||||
|
||||
inject(content: ContentBlock[], options?: SendOptions): void {
|
||||
@@ -181,11 +204,12 @@ export class ReactLoopAgent implements Agent {
|
||||
// plugins monitoring agent/error see idle-injection persistence failures
|
||||
// too. A throwing agent/error listener is contained.
|
||||
if (turnRecorded) {
|
||||
void Promise.resolve(this.ctx.parallel('session/flush', this.session)).catch((error: unknown) => {
|
||||
// Through the store's flush (the carrier owner), never a raw parallel.
|
||||
void this.loopCtx.sessions.flush(this.session).catch((error: unknown) => {
|
||||
const err = error instanceof Error ? error : new Error(String(error))
|
||||
this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
|
||||
this.loopCtx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
|
||||
try {
|
||||
this.ctx.emit('agent/error', this, turn, 0, err)
|
||||
this.loopCtx.emit(this.carrier, 'agent/error', this, turn, 0, err)
|
||||
} catch {
|
||||
// contained: the failure is already logged; a throwing agent/error
|
||||
// listener must not escape this fire-and-forget catch.
|
||||
@@ -264,7 +288,7 @@ export class ReactLoopAgent implements Agent {
|
||||
* fiber's LIFO disposal chain, where a throw would skip later disposers).
|
||||
*/
|
||||
start(): () => void {
|
||||
this.done = runLoop(this.ctx, this, {
|
||||
this.done = runLoop(this.loopCtx, this, {
|
||||
setStatus: (status) => { this.setStatus(status) },
|
||||
setAbort: controller => void (this.currentAbort = controller),
|
||||
disposed: this.disposed,
|
||||
@@ -296,7 +320,7 @@ export class ReactLoopAgent implements Agent {
|
||||
// 'disposed' is part of the agent/status contract. Guarded: a throwing
|
||||
// listener must not break the disposal chain.
|
||||
try {
|
||||
this.ctx.emit('agent/status', this, 'disposed')
|
||||
this.loopCtx.emit(this.carrier, 'agent/status', this, 'disposed')
|
||||
} catch {
|
||||
// listener error during disposal — nothing safe left to do with it
|
||||
}
|
||||
|
||||
@@ -10,6 +10,8 @@
|
||||
import { Context, Service } from 'cordis'
|
||||
import { randomUUID } from 'node:crypto'
|
||||
import z from 'schemastery'
|
||||
import { createScope } from '@deepseek-ai/dsh-scope'
|
||||
import { agentEvents } from '@deepseek-ai/dsh-agent'
|
||||
import type { AgentFactory, AgentHandle, AgentId, AgentOptions, CreateAgentOptions, ResumeAgentOptions, SessionStartSource } from '@deepseek-ai/dsh-agent'
|
||||
import type {} from '@deepseek-ai/dsh-llm'
|
||||
import { SessionId } from '@deepseek-ai/dsh-session'
|
||||
@@ -174,7 +176,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
})
|
||||
// A seeded (forked) create is still a fresh start, NOT a resume — `resume`
|
||||
// is reserved for reloading a PERSISTED session via resume()/resumeWith().
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'startup')
|
||||
return this.startOwned(options.agentId, options.agentOptions ?? {}, session, 'startup', options.setup)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -298,17 +300,46 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
*/
|
||||
private start(
|
||||
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
|
||||
setup?: (agentCtx: Context) => void,
|
||||
): { agent: ReactLoopAgent; disposeAgent: () => Promise<void> } {
|
||||
const agent = new ReactLoopAgent(this.ctx, id, options, session)
|
||||
const dispose = this.ctx.effect(function* (this: AgentLoop) {
|
||||
yield this.ctx.sessions.enter(session)
|
||||
// Mint the agent's scope (key = the agent) and wire the two-phase
|
||||
// reference: the scope context tags registrations + filters dispatch;
|
||||
// the extend adds the `ctx.agent` DX own-property on top. The raw
|
||||
// disposer is yielded IMMEDIATELY (exact function identity nests the
|
||||
// scope fiber out of the loop fiber's concurrent sibling list), so
|
||||
// there is no window in which a throw leaves the scope un-nested.
|
||||
//
|
||||
// Yield order is the REVERSE of teardown (LIFO). Teardown runs:
|
||||
// stop/drain → unregister → detach session → unwind scope
|
||||
// Detach BEFORE the scope unwind is deliberate: the scope fiber's
|
||||
// unload is asynchronous (fiber inertia), and every disposer chained
|
||||
// after an async one waits for it — detaching first keeps the
|
||||
// store/registry rollback SYNCHRONOUS on every failure path (a caller
|
||||
// that catches a throwing create() observes no half-created agent or
|
||||
// session, and the ids are immediately reusable), at the cost that a
|
||||
// scoped listener's own disposer runs after the session left the store
|
||||
// (it heard the final stop/drain flush while still attached, so
|
||||
// nothing durable is lost).
|
||||
const scope = createScope(this.ctx, agent)
|
||||
agent.ctx = scope.ctx.extend({ agent })
|
||||
yield scope.rawDispose
|
||||
// Enter the session THROUGH agent.ctx so the store captures the agent's
|
||||
// scope as the session's dispatch carrier.
|
||||
yield agent.ctx.sessions.enter(session)
|
||||
this.ctx.sessions.announce(session)
|
||||
yield this.ctx.agents.register(agent)
|
||||
// The creator's scoped composition, inside the rollback boundary: a
|
||||
// throwing setup unwinds LIFO through register → scope → detach, so a
|
||||
// half-created agent never leaks. Setup REGISTERS (through agent.ctx),
|
||||
// it never drives — see CreateAgentOptions.setup.
|
||||
setup?.(agent.ctx)
|
||||
// Fire AFTER register (a listener can ctx.agents.get(id) + inject()) and
|
||||
// BEFORE the loop's first turn. Contained: a throwing listener is logged,
|
||||
// never aborts construction (no open turn to balance here).
|
||||
try {
|
||||
this.ctx.emit('agent/session-start', agent, source)
|
||||
agentEvents(this.ctx, agent).emit('agent/session-start', source)
|
||||
} catch (error: unknown) {
|
||||
this.ctx.logger.warn(`agent "${id}": agent/session-start listener threw: ${String(error)}`)
|
||||
}
|
||||
@@ -338,8 +369,11 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
* `AgentHandle.dispose(): Promise<void>` contract (mirrors the ACP `quiesce()`
|
||||
* helper).
|
||||
*/
|
||||
private startOwned(id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource): AgentHandle {
|
||||
const { agent, disposeAgent } = this.start(id, options, session, source)
|
||||
private startOwned(
|
||||
id: AgentId, options: AgentOptions, session: Session, source: SessionStartSource,
|
||||
setup?: (agentCtx: Context) => void,
|
||||
): AgentHandle {
|
||||
const { agent, disposeAgent } = this.start(id, options, session, source, setup)
|
||||
let disposing: Promise<void> | undefined
|
||||
return { agent, dispose: () => (disposing ??= disposeAgent()) }
|
||||
}
|
||||
|
||||
@@ -10,7 +10,8 @@
|
||||
import type { Context } from 'cordis'
|
||||
import type { FinishReason, GenerateOptions, LlmCallConfig, Message } from '@deepseek-ai/dsh-llm'
|
||||
import { BlockAssembler, HarnessError, deepFreeze } from '@deepseek-ai/dsh-llm'
|
||||
import type { ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import { agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
|
||||
import type { AgentEventDispatch, ContinuationDecision, HookContext, PromptDecision } from '@deepseek-ai/dsh-agent'
|
||||
import { canonicalHeader } from '@deepseek-ai/dsh-session'
|
||||
import type { Session, TurnEndReason, TurnTrigger } from '@deepseek-ai/dsh-session'
|
||||
import { createTransmissionLog, recordRequestHeader } from './request-log.ts'
|
||||
@@ -155,9 +156,10 @@ export interface LoopHandle {
|
||||
* every prompt blocked → 'turn/end'(rejected), 0 steps
|
||||
* STEP loop:
|
||||
* drain steering → session('steering/message') ⟵ catches late steering
|
||||
* assembly = ctx.systemPrompt.assemble({agent}) ⟵ waterfall system-prompt/assemble; renderPrompt
|
||||
* assembly = ctx.systemPrompt.assemble(assembleContextFor(agent)) ⟵ waterfall system-prompt/assemble
|
||||
* (scope-filtered; scoped sections/tools join); renderPrompt
|
||||
* (persona section + {{variables}}) IS the full prompt
|
||||
* await ctx.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
|
||||
* await events.serial('agent/pre-step') ⟵ surface mutation (compaction) OUTSIDE the step
|
||||
* boundary = session.deriveMessages() ⟵ the reconstruction boundary: snapshot in the
|
||||
* session('step/start') same sync frame, strictly before step/start
|
||||
* config = waterfall agent/request(config) ⟵ frozen seed; a returned replacement switches
|
||||
@@ -181,7 +183,7 @@ export interface LoopHandle {
|
||||
* if action==stop && steering arrived (step/end/continuation listeners): continue anyway
|
||||
* if action==stop: break
|
||||
* session('turn/end') ⟵ durable turn boundary (no agent/* mirror)
|
||||
* await ctx.parallel('session/flush', session) ⟵ durability checkpoint
|
||||
* await ctx.sessions.flush(session) ⟵ durability checkpoint (store-owned carrier)
|
||||
* re-enqueue leftover steering as queued ⟵ steering is never stranded
|
||||
* idle (emit agent/status) unless more queued
|
||||
* ```
|
||||
@@ -198,6 +200,10 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
const transmission = createTransmissionLog()
|
||||
|
||||
const { session } = agent
|
||||
// The fused agent-subject dispatcher: every agent/* dispatch below carries
|
||||
// the agent's scope (an `agent.ctx` listener hears only this agent) with
|
||||
// the subject injected — one spelling, checked by the dev invariants.
|
||||
const events = agentEvents(ctx, agent)
|
||||
|
||||
while (!handle.isDisposed()) {
|
||||
await agent.inbox.waitForQueued(handle.disposed)
|
||||
@@ -252,7 +258,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
// turn number is actually last in the log — a stale counter would collide.
|
||||
const turn = lastTurnNumber(session) + 1
|
||||
try {
|
||||
await runTurn(ctx, agent, handle, turn, transmission)
|
||||
await runTurn(ctx, events, agent, handle, turn, transmission)
|
||||
} catch (error: unknown) {
|
||||
// Backstop: runTurn rethrows only a PRE-turn throw (the invariant guard
|
||||
// before turn/start) — no turn/start was appended, so no turn is open and
|
||||
@@ -263,7 +269,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
const err = toError(error)
|
||||
ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, 0, err)
|
||||
events.emit('agent/error', turn, 0, err)
|
||||
} catch { /* contained: a throwing agent/error listener must not kill the driver */ }
|
||||
}
|
||||
|
||||
@@ -288,7 +294,7 @@ export async function runLoop(ctx: Context, agent: ReactLoopAgent, handle: LoopH
|
||||
}
|
||||
|
||||
async function runTurn(
|
||||
ctx: Context, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
|
||||
ctx: Context, events: AgentEventDispatch, agent: ReactLoopAgent, handle: LoopHandle, turn: number, transmission: TransmissionLog,
|
||||
): Promise<void> {
|
||||
const { session } = agent
|
||||
|
||||
@@ -354,7 +360,7 @@ async function runTurn(
|
||||
// boundary is durable). So set the error reason for closeTurn to append.
|
||||
reason = { kind: 'error', step, ...errorData(err) }
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, step, err)
|
||||
events.emit('agent/error', turn, step, err)
|
||||
} catch {
|
||||
// contained: the error is already captured on `reason`; a throwing
|
||||
// agent/error listener must not prevent the turn from closing.
|
||||
@@ -398,8 +404,8 @@ async function runTurn(
|
||||
// batch always reports the last vetoing reason.
|
||||
let lastBlockReason = 'prompt blocked by hook'
|
||||
for (const message of queued) {
|
||||
const decision = await ctx.waterfall(
|
||||
'agent/prompt-submit', agent, message.content, message.source,
|
||||
const decision = await events.waterfall(
|
||||
'agent/prompt-submit', message.content, message.source,
|
||||
() => Promise.resolve<PromptDecision>({ kind: 'allow' }),
|
||||
)
|
||||
if (decision.kind === 'block') {
|
||||
@@ -456,7 +462,7 @@ async function runTurn(
|
||||
// step. renderPrompt IS the full prompt — the persona is the order-0
|
||||
// section (registered by the AgentLoop plugin) and `{{variable}}`
|
||||
// interpolation happens in the render, so there is no separate join.
|
||||
const assembly = await ctx.systemPrompt.assemble({ agent })
|
||||
const assembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
|
||||
const fullSystemPrompt = renderPrompt(assembly)
|
||||
|
||||
// Interruption landing after assembly: dispose() or cancel() in a
|
||||
@@ -483,7 +489,7 @@ async function runTurn(
|
||||
// throwing listener escapes to the outer catch, which closes the (not-yet-
|
||||
// open) step as a no-op and ends the turn via failTurn — a broken
|
||||
// pre-step plugin ends the turn, not the loop.
|
||||
await ctx.serial('agent/pre-step', agent, turn, step, fullSystemPrompt, abort.signal)
|
||||
await events.serial('agent/pre-step', turn, step, fullSystemPrompt, abort.signal)
|
||||
|
||||
// Interruption landing during the pre-step seam: do not open an empty step.
|
||||
if (handle.isCancelled() || handle.isDisposed()) {
|
||||
@@ -524,7 +530,8 @@ async function runTurn(
|
||||
|
||||
let stepOutcome: { hadToolCalls: boolean; finish: FinishReason } | { error: Error }
|
||||
try {
|
||||
stepOutcome = await runStep(ctx, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
|
||||
stepOutcome = await runStep(
|
||||
ctx, events, agent, turn, step, assembly, fullSystemPrompt, boundaryMessages, transmission, abort.signal)
|
||||
} catch (error: unknown) {
|
||||
stepOutcome = { error: toError(error) }
|
||||
} finally {
|
||||
@@ -566,8 +573,8 @@ async function runTurn(
|
||||
const defaultDecision: ContinuationDecision = { action: stepOutcome.hadToolCalls || steered ? 'continue' : 'stop' }
|
||||
let decision: ContinuationDecision
|
||||
try {
|
||||
decision = await ctx.waterfall(
|
||||
'agent/turn-continuation', agent, turn, defaultDecision,
|
||||
decision = await events.waterfall(
|
||||
'agent/turn-continuation', turn, defaultDecision,
|
||||
() => Promise.resolve(defaultDecision),
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
@@ -644,8 +651,9 @@ async function runTurn(
|
||||
|
||||
// Durability checkpoint: persistence plugins drain write-behind buffers.
|
||||
// A failing persistence plugin is reported but doesn't kill the agent.
|
||||
// Through the store's flush (the carrier owner), never a raw parallel.
|
||||
try {
|
||||
await ctx.parallel('session/flush', session)
|
||||
await ctx.sessions.flush(session)
|
||||
} catch (error: unknown) {
|
||||
// The turn is already closed (turn/end appended above) and flush must run
|
||||
// AFTER turn/end to be a checkpoint — so there is no in-turn position left
|
||||
@@ -657,7 +665,7 @@ async function runTurn(
|
||||
const err = toError(error)
|
||||
ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
|
||||
try {
|
||||
ctx.emit('agent/error', agent, turn, step, err)
|
||||
events.emit('agent/error', turn, step, err)
|
||||
} catch {
|
||||
// contained: a throwing agent/error listener must not escape the loop.
|
||||
}
|
||||
@@ -681,6 +689,7 @@ function drainSteering(agent: ReactLoopAgent, turn: number): boolean {
|
||||
* step/start and already reflects any compaction. */
|
||||
async function runStep(
|
||||
ctx: Context,
|
||||
events: AgentEventDispatch,
|
||||
agent: ReactLoopAgent,
|
||||
turn: number,
|
||||
step: number,
|
||||
@@ -713,7 +722,7 @@ async function runStep(
|
||||
// model-visible content flows through the log channels). The header event
|
||||
// below records whatever the request ACTUALLY uses, so a listener's switch
|
||||
// is a logged, reconstructable fact, never silent drift.
|
||||
const config = await ctx.waterfall('agent/request', agent, turn, step, seedConfig, () => Promise.resolve(seedConfig))
|
||||
const config = await events.waterfall('agent/request', turn, step, seedConfig, () => Promise.resolve(seedConfig))
|
||||
if (!config.model) {
|
||||
throw new Error(`agent "${agent.id}" has no model: set AgentOptions.model or supply one via the agent/request waterfall`)
|
||||
}
|
||||
@@ -764,7 +773,7 @@ async function runStep(
|
||||
|
||||
if (assembler.finish.kind === 'max-tokens') {
|
||||
let message: Message = withoutToolCalls(assembler.message())
|
||||
message = withoutToolCalls(await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message)))
|
||||
message = withoutToolCalls(await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message)))
|
||||
// Fire the assistant/message when there is content OR usage: a max-tokens
|
||||
// step can be cut off with empty content but still carry token accounting,
|
||||
// and assistant/message is the only host for usage (there is no standalone
|
||||
@@ -787,7 +796,7 @@ async function runStep(
|
||||
// source of truth for derived history and replay) records the message that
|
||||
// tool dispatch actually uses.
|
||||
let message: Message = assembler.message()
|
||||
message = await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message))
|
||||
message = await events.waterfall('agent/step-result', turn, step, message, () => Promise.resolve(message))
|
||||
|
||||
// Same content-or-usage guard as the max-tokens branch: a step that finishes
|
||||
// with neither assembled content nor usage (e.g. a bare `stop` finish that
|
||||
|
||||
172
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
Normal file
172
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
Normal file
@@ -0,0 +1,172 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import LlmService from '@deepseek-ai/dsh-llm'
|
||||
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
import ToolRegistry from '@deepseek-ai/dsh-tools'
|
||||
import AgentRegistry, { AgentId, agentEvents, assembleContextFor } from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { scopeOf } from '@deepseek-ai/dsh-scope'
|
||||
import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
|
||||
import type { ContentBlock } from '@deepseek-ai/dsh-llm'
|
||||
import { MockAdapter, textResponse } from './mock-adapter.ts'
|
||||
|
||||
async function harness(adapter: MockAdapter = new MockAdapter([textResponse('ok')])) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(LlmService)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SystemPrompt, { persona: 'You are the deployment.' })
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(AgentLoop, { agents: [] })
|
||||
ctx.llm.registerAdapter(['mock'], adapter)
|
||||
return ctx
|
||||
}
|
||||
|
||||
function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const dispose = ctx.on('agent/status', (subject, status) => {
|
||||
if (subject === agent && status === 'idle') {
|
||||
dispose()
|
||||
resolve()
|
||||
}
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }]
|
||||
|
||||
describe('agent scope lifecycle', () => {
|
||||
it('wires agent.ctx: tagged with the agent, DX field set, ctx.agent safe elsewhere', async () => {
|
||||
const ctx = await harness()
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
expect(scopeOf(agent.ctx)).toBe(agent)
|
||||
expect(agent.ctx.agent).toBe(agent)
|
||||
// The root accessor default: a plain context answers undefined, not a throw.
|
||||
expect(ctx.agent).toBeUndefined()
|
||||
await ctx.agents.get(AgentId('a1'))?.whenIdle()
|
||||
})
|
||||
|
||||
it('scoped registrations live in the agent world and die with the agent', async () => {
|
||||
const ctx = await harness()
|
||||
const handle = ctx.agents.create({ agentId: AgentId('a1'), sessionId: SessionId('s1'), agentOptions: { model: 'mock' } })
|
||||
const { agent } = handle
|
||||
agent.ctx.systemPrompt.section({ name: 'deployment:persona', order: 0, text: 'You run tests.' })
|
||||
agent.ctx.tools.register({
|
||||
name: 'mine', description: 'scoped', parameters: {},
|
||||
execute: () => Promise.resolve(text('ran')),
|
||||
})
|
||||
|
||||
const scopedAssembly = await ctx.systemPrompt.assemble(assembleContextFor(agent))
|
||||
expect(scopedAssembly.sections.find(s => s.name === 'deployment:persona')?.text).toBe('You run tests.')
|
||||
expect(scopedAssembly.tools.map(t => t.name)).toContain('mine')
|
||||
// Other assemblies are untouched.
|
||||
const globalAssembly = await ctx.systemPrompt.assemble()
|
||||
expect(globalAssembly.sections.find(s => s.name === 'deployment:persona')?.text).toBe('You are the deployment.')
|
||||
expect(globalAssembly.tools.map(t => t.name)).not.toContain('mine')
|
||||
|
||||
await handle.dispose()
|
||||
// The scoped world unwound with the agent: nothing leaked into the registries.
|
||||
expect(ctx.tools.get('mine', agent)).toBeUndefined()
|
||||
const after = await ctx.systemPrompt.assemble(assembleContextFor(agent))
|
||||
expect(after.sections.find(s => s.name === 'deployment:persona')?.text).toBe('You are the deployment.')
|
||||
})
|
||||
|
||||
it('agent.ctx listeners hear only their own agent (scoped dispatch end to end)', async () => {
|
||||
const ctx = await harness(new MockAdapter([textResponse('one'), textResponse('two')]))
|
||||
const a = ctx.agentLoop.create(AgentId('a'), { model: 'mock' })
|
||||
const b = ctx.agentLoop.create(AgentId('b'), { model: 'mock' })
|
||||
|
||||
const heard: string[] = []
|
||||
a.ctx.on('agent/status', (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')
|
||||
})
|
||||
|
||||
b.send(text('for b'))
|
||||
await waitForIdle(ctx, b)
|
||||
expect(heard).toEqual([]) // nothing of b's leaked into a's scope
|
||||
|
||||
a.send(text('for a'))
|
||||
await waitForIdle(ctx, a)
|
||||
expect(heard).toContain('a-sees:a:running')
|
||||
expect(heard).toContain('a-sees:user-message')
|
||||
})
|
||||
|
||||
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) => {
|
||||
order.push('session-start')
|
||||
// The scoped section is already registered by the time session-start fires.
|
||||
void ctx.systemPrompt.assemble(assembleContextFor(agent)).then((assembly) => {
|
||||
order.push(`persona:${assembly.sections.find(s => s.name === 'deployment:persona')?.text}`)
|
||||
})
|
||||
})
|
||||
|
||||
const handle = ctx.agents.create({
|
||||
agentId: AgentId('child'),
|
||||
sessionId: SessionId('child-s'),
|
||||
agentOptions: { model: 'mock' },
|
||||
setup: (agentCtx) => {
|
||||
order.push('setup')
|
||||
agentCtx.systemPrompt.section({ name: 'deployment:persona', order: 0, text: 'You are the child.' })
|
||||
},
|
||||
})
|
||||
await new Promise(resolve => setTimeout(resolve, 0))
|
||||
expect(order).toEqual(['setup', 'session-start', 'persona:You are the child.'])
|
||||
await handle.dispose()
|
||||
})
|
||||
|
||||
it('a throwing setup unwinds the half-created agent completely', async () => {
|
||||
const ctx = await harness()
|
||||
expect(() => ctx.agents.create({
|
||||
agentId: AgentId('bad'),
|
||||
sessionId: SessionId('bad-s'),
|
||||
agentOptions: { model: 'mock' },
|
||||
setup: () => { throw new Error('boom setup') },
|
||||
})).toThrow('boom setup')
|
||||
|
||||
// Nothing leaked: no agent, no session, and the ids are reusable.
|
||||
expect(ctx.agents.get(AgentId('bad'))).toBeUndefined()
|
||||
expect(ctx.sessions.get(SessionId('bad-s'))).toBeUndefined()
|
||||
const retry = ctx.agents.create({ agentId: AgentId('bad'), sessionId: SessionId('bad-s'), agentOptions: { model: 'mock' } })
|
||||
await retry.dispose()
|
||||
})
|
||||
|
||||
it('a throwing session/created listener disposes the scope (pre-nesting rollback window)', async () => {
|
||||
const ctx = await harness()
|
||||
let boom = true
|
||||
ctx.on('session/created', () => {
|
||||
if (boom) { boom = false; throw new Error('boom created') }
|
||||
})
|
||||
expect(() => ctx.agents.create({
|
||||
agentId: AgentId('bad'), sessionId: SessionId('bad-s'), agentOptions: { model: 'mock' },
|
||||
})).toThrow('boom created')
|
||||
expect(ctx.agents.get(AgentId('bad'))).toBeUndefined()
|
||||
expect(ctx.sessions.get(SessionId('bad-s'))).toBeUndefined()
|
||||
// The rollback also disposed the scope fiber: re-creating works cleanly.
|
||||
const retry = ctx.agents.create({ agentId: AgentId('bad'), sessionId: SessionId('bad-s'), agentOptions: { model: 'mock' } })
|
||||
expect(scopeOf(retry.agent.ctx)).toBe(retry.agent)
|
||||
await retry.dispose()
|
||||
})
|
||||
|
||||
it('registrations through a disposed agent ctx throw INACTIVE_EFFECT', async () => {
|
||||
const ctx = await harness()
|
||||
const handle = ctx.agents.create({ agentId: AgentId('a1'), sessionId: SessionId('s1'), agentOptions: { model: 'mock' } })
|
||||
await handle.dispose()
|
||||
expect(() => handle.agent.ctx.on('agent/status', () => {})).toThrow(/inactive context/)
|
||||
})
|
||||
|
||||
it('agentEvents fuses carrier and subject for custom drivers', async () => {
|
||||
const ctx = await harness()
|
||||
const agent = ctx.agentLoop.create(AgentId('a1'), { model: 'mock' })
|
||||
const other = ctx.agentLoop.create(AgentId('a2'), { model: 'mock' })
|
||||
const heard: string[] = []
|
||||
agent.ctx.on('agent/error', (subject: Agent, turn: number) => 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'))
|
||||
expect(heard).toEqual(['a1:2'])
|
||||
})
|
||||
})
|
||||
@@ -34,6 +34,9 @@
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../../core/scope"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user