From f387b774a9a84d992bb73db6b1db984f971343c3 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Thu, 9 Jul 2026 01:17:47 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20the=20agent=20is=20a=20registrat?= =?UTF-8?q?ion=20scope=20=E2=80=94=20Agent.ctx,=20setup=20slot,=20fused=20?= =?UTF-8?q?scoped=20dispatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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; ctx.agent is a safe root accessor defaulting undefined, shadowed by each agent context. --- packages/core/agent-loop/package.json | 2 + packages/core/agent-loop/src/agent.ts | 44 ++++- packages/core/agent-loop/src/index.ts | 44 ++++- packages/core/agent-loop/src/loop.ts | 49 +++-- .../agent-loop/tests/scope-lifecycle.spec.ts | 172 ++++++++++++++++++ packages/core/agent-loop/tsconfig.json | 3 + packages/core/agent/package.json | 2 + packages/core/agent/src/dispatch.ts | 117 ++++++++++++ packages/core/agent/src/index.ts | 45 ++++- packages/core/agent/src/types.ts | 100 ++++++++-- packages/core/agent/tests/agent.spec.ts | 3 + packages/core/agent/tsconfig.json | 3 + 12 files changed, 532 insertions(+), 52 deletions(-) create mode 100644 packages/core/agent-loop/tests/scope-lifecycle.spec.ts create mode 100644 packages/core/agent/src/dispatch.ts diff --git a/packages/core/agent-loop/package.json b/packages/core/agent-loop/package.json index 6e92adb6ab..03acc7e17f 100644 --- a/packages/core/agent-loop/package.json +++ b/packages/core/agent-loop/package.json @@ -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:^", diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 49de0f77c4..5f56cdc8a7 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -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 { + return (this.#carrier ??= scopeTarget(this, this)) + } + + #carrier: Scoped | 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 } diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index e7e0e562d9..306f8707e2 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -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 } { 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` 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 | undefined return { agent, dispose: () => (disposing ??= disposeAgent()) } } diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index eb5b07ba23..dbb66ad7b0 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/packages/core/agent-loop/src/loop.ts @@ -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 { 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({ 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 diff --git a/packages/core/agent-loop/tests/scope-lifecycle.spec.ts b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts new file mode 100644 index 0000000000..28d1da192d --- /dev/null +++ b/packages/core/agent-loop/tests/scope-lifecycle.spec.ts @@ -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 { + 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']) + }) +}) diff --git a/packages/core/agent-loop/tsconfig.json b/packages/core/agent-loop/tsconfig.json index 03df67cfc4..5d7cf98bb7 100644 --- a/packages/core/agent-loop/tsconfig.json +++ b/packages/core/agent-loop/tsconfig.json @@ -34,6 +34,9 @@ }, { "path": "../../core/agent" + }, + { + "path": "../../core/scope" } ] } diff --git a/packages/core/agent/package.json b/packages/core/agent/package.json index e38a6c8d61..b8e6108904 100644 --- a/packages/core/agent/package.json +++ b/packages/core/agent/package.json @@ -24,6 +24,7 @@ "peerDependencies": { "@deepseek-ai/dsh-brand": "^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-system-prompt": "^0.0.1", "cordis": "^4.0.0-rc.6" @@ -31,6 +32,7 @@ "devDependencies": { "@deepseek-ai/dsh-brand": "workspace:^", "@deepseek-ai/dsh-llm": "workspace:^", + "@deepseek-ai/dsh-scope": "workspace:^", "@deepseek-ai/dsh-session": "workspace:^", "@deepseek-ai/dsh-system-prompt": "workspace:^", "cordis": "^4.0.0-rc.6" diff --git a/packages/core/agent/src/dispatch.ts b/packages/core/agent/src/dispatch.ts new file mode 100644 index 0000000000..bed4981135 --- /dev/null +++ b/packages/core/agent/src/dispatch.ts @@ -0,0 +1,117 @@ +/** + * Fused scope-carrier dispatch for agent-subject events, plus the assembly + * context builder. The ONE sanctioned spelling for dispatching `agent/*` + * events: `agentEvents(ctx, agent).waterfall('agent/request', …)` builds the + * scope carrier ({@link scopeTarget} keyed by the agent) AND injects the + * subject as the first event argument in one move, so the correct dispatch is + * also the shortest — a dispatch site cannot pass a carrier keyed to one + * agent while naming another as the subject, which is the invariant the + * dev-mode scoped-dispatch check asserts at runtime. + * + * @module @deepseek-ai/dsh-agent/dispatch + */ + +import type { Context, Events } from 'cordis' +import { scopeTarget } from '@deepseek-ai/dsh-scope' +import type { Scoped } from '@deepseek-ai/dsh-scope' +import type { AssembleContext } from '@deepseek-ai/dsh-system-prompt' +import type { Agent } from './types.ts' + +/** Extract the parameter tuple from an event handler type (its `this` is not part of the tuple). */ +type Params = F extends (...args: infer P) => unknown ? P : never +/** Extract the return type from an event handler type. */ +type Return = 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` `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. + */ +export type AgentSubjectEvent = { + [K in keyof Events]: Events[K] extends (this: Scoped, ...args: infer P) => unknown + ? P extends [Agent, ...unknown[]] ? K : never + : never +}[keyof Events] + +/** The event arguments AFTER the injected agent subject. */ +type Tail = Params extends [Agent, ...infer R] ? R : never + +/** + * 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. + */ +export interface AgentEventDispatch { + /** + * Fire-and-forget notification (Cordis `emit`) in the agent's scope. + * @param name - the agent-subject event to emit. + * @param rest - the event's arguments after the injected agent. + */ + emit(name: K, ...rest: Tail): 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. + * @returns the serial chain's result (the first bail value, if any). + */ + serial(name: K, ...rest: Tail): Promise>> + /** + * 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). + * @param name - the agent-subject event to dispatch. + * @param rest - the event's arguments after the injected agent. + * @returns the waterfall's composed result. + */ + waterfall(name: K, ...rest: Tail): Return +} + +/** + * Build the fused dispatcher for `agent`'s events (see the module doc). Cheap + * (one carrier + one small object) — dispatch sites create it per run/turn + * rather than caching it on the agent. + * @param ctx - the context to dispatch through (any context of the app). + * @param agent - the subject agent; also the scope-carrier key. + * @returns the fused dispatcher. + */ +export function agentEvents(ctx: Context, agent: Agent): AgentEventDispatch { + const carrier: Scoped = scopeTarget(agent, agent) + // The three dispatch methods forward through cordis' variadic mixins. The + // fused (carrier, name, agent, ...rest) tuple is provably a valid argument + // list for the matching thisArg overload, but TypeScript cannot relate the + // generic Tail spread back to that overload's conditional parameter + // tuple — hence one contained, shape-preserving cast per method. + return { + emit(name, ...rest) { + // eslint-disable-next-line @typescript-eslint/unbound-method -- the events mixin accessor returns a pre-bound function + const emit = ctx.emit as (thisArg: Scoped, name: string, ...args: unknown[]) => void + emit(carrier, name, agent, ...rest) + }, + async serial(name, ...rest) { + // eslint-disable-next-line @typescript-eslint/unbound-method -- the events mixin accessor returns a pre-bound function + const serial = ctx.serial as (thisArg: Scoped, name: string, ...args: unknown[]) => Promise + return await serial(carrier, name, agent, ...rest) + }, + waterfall(name, ...rest) { + // eslint-disable-next-line @typescript-eslint/unbound-method -- the events mixin accessor returns a pre-bound function + const waterfall = ctx.waterfall as (thisArg: Scoped, name: string, ...args: unknown[]) => never + return waterfall(carrier, name, agent, ...rest) + }, + } +} + +/** + * The assembly context for one agent's prompt: the typed `agent` DX field and + * the `scope` layer selector, set together (setting `agent` without `scope` + * silently drops the agent's scoped sections/tools from the assembly — the + * dev invariants flag it). THE way the loop (and any custom driver) builds + * its per-step `ctx.systemPrompt.assemble(…)` input. + * @param agent - the agent the assembly is for. + * @returns the context to pass to `assemble()`. + */ +export function assembleContextFor(agent: Agent): AssembleContext { + return { agent, scope: agent } +} diff --git a/packages/core/agent/src/index.ts b/packages/core/agent/src/index.ts index 096555f925..b9c9e6f0df 100644 --- a/packages/core/agent/src/index.ts +++ b/packages/core/agent/src/index.ts @@ -6,14 +6,28 @@ */ import { Context, Service } from 'cordis' +import { scopeTarget } from '@deepseek-ai/dsh-scope' import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session' import type { Agent, AgentId, AgentOptions } from './types.ts' export * from './types.ts' +export { agentEvents, assembleContextFor } from './dispatch.ts' +export type { AgentEventDispatch, AgentSubjectEvent } from './dispatch.ts' declare module 'cordis' { interface Context { agents: AgentRegistry + /** + * The agent whose scope this context belongs to, or `undefined` on any + * context not derived from an agent scope. Pure DX sugar over the + * `dsh-scope` tag: the agent loop sets it as an own property on each + * `Agent.ctx`, and {@link AgentRegistry} registers a root accessor + * defaulting to `undefined` so the read is safe on every context (a plain + * plugin context answers `undefined` instead of throwing the Cordis + * unknown-property error). Core packages below the agent layer read the + * `dsh-scope` tag (`scopeOf`) instead, never this field. + */ + agent?: Agent } } @@ -51,6 +65,20 @@ export interface CreateAgentOptions { seed?: SessionEvent[] /** Per-agent options (model, …). */ agentOptions?: AgentOptions + /** + * Creation-time composition of the agent's scoped world. The factory runs it + * inside the agent's composite lifecycle effect — after the scope is minted + * and the agent registered, before `agent/session-start` fires and the loop + * starts — so everything it registers through `agentCtx` (scoped tools, + * prompt sections/variables, `restrict()`, listeners, `agentCtx.plugin(…)` + * profiles) exists before the first prompt assembly, and a THROWING setup + * unwinds inside the rollback boundary instead of leaking a half-created + * agent. **Setup registers, it never drives**: calling + * `send`/`steer`/`inject` here would open a turn before `agent/session-start` + * (the dev invariants flag a `turn/start` logged before session-start as a + * teaching error) — drive the agent after creation returns. + */ + setup?: (agentCtx: Context) => void } /** @@ -120,6 +148,13 @@ export class AgentRegistry extends Service { constructor(ctx: Context) { super(ctx, 'agents') + // The `ctx.agent` DX accessor: default `undefined` on every context, so a + // plain plugin context reads cleanly instead of hitting the Cordis + // unknown-property throw. Each Agent.ctx shadows it with an own property + // (own properties resolve before the context proxy is consulted), so the + // accessor body never needs to resolve a scope itself. Effect-scoped: + // unwinds with this service's fiber. + ctx.accessor('agent', { get: () => undefined }) } /** @@ -167,7 +202,11 @@ export class AgentRegistry extends Service { /** * Register a live agent. Throws if an agent with the same id is already * registered. Emits `agent/created` on registration and `agent/disposed` - * when the calling fiber is disposed. Returns the disposer. + * when the calling fiber is disposed — both with the agent's scope carrier + * (`scopeTarget(agent, agent)`): the subject is the agent in hand, so the + * emits are scope-filtered regardless of which context invoked `register` + * (calling through `agent.ctx` scopes EFFECTS; dispatch scoping always + * requires passing the carrier). Returns the disposer. * @param agent - the already-constructed agent to record in the store. * @returns the disposer that removes the agent and emits `agent/disposed`. */ @@ -196,12 +235,12 @@ export class AgentRegistry extends Service { // logging the listener bug and continuing is correct (mirrors the // guarded `agent/status` emit in dsh-agent-loop's ReactLoopAgent). try { - this.ctx.emit('agent/disposed', agent) + this.ctx.emit(scopeTarget(agent, agent), 'agent/disposed', agent) } catch (error: unknown) { this.ctx.logger.warn(`agent "${agent.id}": agent/disposed listener threw: ${String(error)}`) } } - this.ctx.emit('agent/created', agent) + this.ctx.emit(scopeTarget(agent, agent), 'agent/created', agent) }.bind(this), 'agents.register()') // ctx.effect's disposer returns Promise; our disposer API is // synchronous fire-and-forget — discard the (always-resolved) promise. diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 46bc52128a..ff6f79d039 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -44,6 +44,8 @@ */ import type { Branded } from '@deepseek-ai/dsh-brand' +import type { Context } from 'cordis' +import type { Scoped } from '@deepseek-ai/dsh-scope' import type { ContentBlock, LlmCallConfig, Message, MessageSource } from '@deepseek-ai/dsh-llm' import type {} from '@deepseek-ai/dsh-system-prompt' @@ -64,10 +66,14 @@ declare module '@deepseek-ai/dsh-system-prompt' { interface AssembleContext { /** * The agent this assembly is for. The agent loop passes it on every - * per-step `assemble({ agent })`; variable providers project per-agent - * facts from it (`options.model` → `{{model}}`, `session.header.cwd` → + * per-step assembly (via its `assembleContextFor(agent)` helper, which + * also sets the `scope` field to the same agent — the layer selector + * `dsh-system-prompt` reads); variable providers project per-agent facts + * from it (`options.model` → `{{model}}`, `session.header.cwd` → * `{{cwd}}`). Optional because a bare `assemble()` (tests, diagnostics) - * has no agent — providers must tolerate its absence. + * has no agent — providers must tolerate its absence. Never set `agent` + * without `scope`: the assembly would silently miss the agent's scoped + * sections/tools (the dev invariants flag it). */ agent?: Agent } @@ -175,6 +181,17 @@ export interface Agent { readonly options: AgentOptions readonly session: Session readonly status: AgentStatus + /** + * The agent's scope context (`@deepseek-ai/dsh-scope`, key = this agent). + * Registrations through it — tools, prompt sections/variables, event + * listeners, restrictions — are visible to THIS agent only and unwind when + * the agent is disposed; `agent.ctx.on('agent/…')` listeners fire only for + * this agent's dispatches (zero self-filtering). Service resolution through + * it flows through the loop plugin's dependency surface — handing out + * `agent.ctx` hands out that capability. Live for exactly the agent's + * lifetime: registrations after disposal throw Cordis's INACTIVE_EFFECT. + */ + readonly ctx: Context /** Queue a user message. Starts a turn when idle; otherwise waits for the next turn. */ send(content: ContentBlock[], options?: SendOptions): void @@ -259,34 +276,54 @@ declare module 'cordis' { * An agent was registered in the {@link AgentRegistry} and is ready to * receive messages. * @param agent - the newly registered agent, already resolvable in the registry. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/created'(agent: Agent): void + 'agent/created'(this: Scoped, agent: Agent): void /** * An agent was disposed and removed from the registry; its fiber and any * in-flight turn have been torn down. * @param agent - the agent that was torn down; its handle is now inert. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/disposed'(agent: Agent): void + 'agent/disposed'(this: Scoped, agent: Agent): void /** * Agent status changed (`idle` ⇄ `running`, or → `disposed`). Drive * lifecycle off this transition, never off a status you just requested — * `send()` does not flip status to `running` before it returns. * @param agent - the agent whose status flipped. * @param status - the status just entered (the transition's destination). + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/status'(agent: Agent, status: AgentStatus): void + 'agent/status'(this: Scoped, agent: Agent, status: AgentStatus): void /** * A message entered the agent's inbox (queued or steering). `source` is * the resolved source (defaults applied), not the caller's raw options. * @param agent - the agent whose inbox received the message. * @param content - the enqueued content blocks, verbatim. * @param info - the resolved source plus whether it entered as steering. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/queued'(agent: Agent, content: ContentBlock[], info: { source: MessageSource; steering: boolean }): void + 'agent/queued'(this: Scoped, agent: Agent, content: ContentBlock[], info: { source: MessageSource; steering: boolean }): void // ---- session lifecycle (emit) ---- /** @@ -299,9 +336,14 @@ declare module 'cordis' { * is deliberate (a bridge logs/injects, it does not gate startup). * @param agent - the agent whose session lifecycle began. * @param source - why the session started (fresh startup, resume, …). + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/session-start'(agent: Agent, source: SessionStartSource): void + 'agent/session-start'(this: Scoped, agent: Agent, source: SessionStartSource): void // Turn and step boundaries are NOT mirrored as agent/* emits: a consumer // that needs them reads the durable `turn/start`/`turn/end`/`step/start`/ @@ -334,6 +376,11 @@ declare module 'cordis' { * listener needs to measure pressure (the system prompt counts toward the * budget). `signal` cancels any in-flight work a listener starts (e.g. a * summarization model call). + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @param agent - the agent about to open the step. * @param turn - the already-open turn this step belongs to. * @param step - the number of the step about to start. @@ -346,7 +393,7 @@ declare module 'cordis' { // reads. Revisit if no second consumer appears: e.g. hand listeners a lazy // prompt provider, or move token-pressure measurement behind a // compaction-specific seam instead of the shared pre-step checkpoint. - 'agent/pre-step'(agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal): Promise | void + 'agent/pre-step'(this: Scoped, agent: Agent, turn: number, step: number, fullSystemPrompt: string, signal: AbortSignal): Promise | void /** * Waterfall: decide what happens to ONE drained queued message before it * becomes a `user/message` — allow (optionally rewriting the prompt bytes or @@ -357,9 +404,14 @@ declare module 'cordis' { * @param agent - the agent draining its inbox. * @param content - the drained message's blocks, as queued. * @param source - the message's resolved source. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode waterfall */ - 'agent/prompt-submit'(agent: Agent, content: ContentBlock[], source: MessageSource, next: () => Promise): Promise + 'agent/prompt-submit'(this: Scoped, agent: Agent, content: ContentBlock[], source: MessageSource, next: () => Promise): Promise /** * Waterfall: shape the step's call configuration — model switching, * sampling overrides — by returning a replacement {@link LlmCallConfig} @@ -380,9 +432,14 @@ declare module 'cordis' { * @param turn - the open turn number. * @param step - the step whose request this is. * @param config - the config the loop would use (frozen); return a replacement to switch. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode waterfall */ - 'agent/request'(agent: Agent, turn: number, step: number, config: LlmCallConfig, next: () => Promise): Promise + 'agent/request'(this: Scoped, agent: Agent, turn: number, step: number, config: LlmCallConfig, next: () => Promise): Promise /** * Waterfall: post-process the assembled assistant {@link Message} before * tool dispatch (validation, content rewriting, …). @@ -390,9 +447,14 @@ declare module 'cordis' { * @param turn - the open turn number. * @param step - the step that produced the message. * @param message - the assistant message as assembled from the stream. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode waterfall */ - 'agent/step-result'(agent: Agent, turn: number, step: number, message: Message, next: () => Promise): Promise + 'agent/step-result'(this: Scoped, agent: Agent, turn: number, step: number, message: Message, next: () => Promise): Promise /** * Waterfall: override the turn-continuation decision via a typed * {@link ContinuationDecision}. The loop's `defaultDecision` is `continue` @@ -403,9 +465,14 @@ declare module 'cordis' { * @param agent - the agent deciding whether to run another step. * @param turn - the turn being continued or stopped. * @param defaultDecision - what the loop would do absent an override. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode waterfall */ - 'agent/turn-continuation'(agent: Agent, turn: number, defaultDecision: ContinuationDecision, next: () => Promise): Promise + 'agent/turn-continuation'(this: Scoped, agent: Agent, turn: number, defaultDecision: ContinuationDecision, next: () => Promise): Promise // ---- error notifications (emit) ---- /** @@ -415,8 +482,13 @@ declare module 'cordis' { * @param turn - the turn in which the failure surfaced. * @param step - the step at which the failure surfaced. * @param error - the failure, verbatim. + * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered + * through `agent.ctx` fires only for that agent's dispatches; a listener on a + * plain plugin context fires for every agent. The dispatch `this` is the + * scope carrier (`Scoped`), built by the emitting side via + * `scopeTarget`/`agentEvents`. * @mode emit */ - 'agent/error'(agent: Agent, turn: number, step: number, error: Error): void + 'agent/error'(this: Scoped, agent: Agent, turn: number, step: number, error: Error): void } } diff --git a/packages/core/agent/tests/agent.spec.ts b/packages/core/agent/tests/agent.spec.ts index c344cd2a6f..43b4752f1b 100644 --- a/packages/core/agent/tests/agent.spec.ts +++ b/packages/core/agent/tests/agent.spec.ts @@ -10,6 +10,9 @@ function stubAgent(rawId: string): Agent { options: {}, session: new Session(SessionId(`${id}-session`)), status: 'idle', + // A bare context stands in for the agent scope: registry tests never + // register through it, they only need the field present. + ctx: new Context(), send() {}, steer() {}, inject() {}, diff --git a/packages/core/agent/tsconfig.json b/packages/core/agent/tsconfig.json index 7f4f457598..2692e1b7f7 100644 --- a/packages/core/agent/tsconfig.json +++ b/packages/core/agent/tsconfig.json @@ -17,6 +17,9 @@ { "path": "../../util/brand" }, + { + "path": "../../core/scope" + }, { "path": "../../llm/llm" },