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

Merge remote-tracking branch 'origin/master' into worktree/agent-execution-context-rfc

Browse Source 
# Conflicts:
#	docs/architecture.md
#	docs/config-catalog.md
#	docs/cordis-catalog/services.md
#	docs/module-graph.md
#	examples/coding-agent/tests/code-mode.e2e.ts
#	packages/context/workspace-context/tests/workspace-context.spec.ts
#	packages/core/README.md
#	packages/core/agent-loop/README.md
#	packages/core/agent-loop/tests/agent.spec.ts
#	packages/core/agent-loop/tests/cancel.spec.ts
#	packages/core/agent-loop/tests/config-session-id.spec.ts
#	packages/core/agent-loop/tests/contract-regressions.spec.ts
#	packages/core/agent-loop/tests/coverage-edges.spec.ts
#	packages/core/agent-loop/tests/interception.spec.ts
#	packages/core/agent-loop/tests/loop.spec.ts
#	packages/core/agent-loop/tests/properties.spec.ts
#	packages/core/agent-loop/tests/request-cache.e2e.ts
#	packages/core/agent-loop/tests/request-reconstruction.spec.ts
#	packages/core/agent-loop/tests/resume.spec.ts
#	packages/core/agent-loop/tests/scope-lifecycle.spec.ts
#	packages/core/agent-loop/tests/tool-calls.spec.ts
#	packages/core/agent-loop/tests/tool-order.spec.ts
#	packages/core/agent-loop/tests/turn-stop.spec.ts
#	packages/workflow/workflow-workerthread/tests/workflow-workerthread.e2e.ts
#	website/zh-CN/api/harness/agent-loop.md
This commit is contained in:
Tianyi Cui
2026-07-19 11:44:50 +08:00
parent 0e4785cdb7 383a305bd4
commit cfe9b9617d
295 changed files with 5544 additions and 4182 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -8,28 +8,30 @@ Tracks live agents so UI, hook, and orchestrator plugins can find them without i
### Public API
`Agent.ctx` owns registrations visible only to that agent. `agentEvents()` couples event subjects to their scope carrier, and `assembleContextFor()` couples the agent and prompt scope. Creation and resume may compose this context through `setup`; the agent remains unpublished and must not be driven until creation resolves.
The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-scope`, key = the agent) — register tools/sections/variables/listeners through it for that agent alone, all unwound on disposal. `agentEvents(ctx, agent)` is the fused dispatcher for ordinary agent-subject operations (carrier + injected subject in one move); its notification mode invokes every listener and contains both synchronous throws and returned-promise rejections. The registry lifecycle pair reuses one stable routing carrier. `assembleContextFor(agent)` builds the per-agent assembly context (`agent` + `scope` together). `CreateAgentOptions.setup(agentCtx)` and `ResumeAgentOptions.setup(agentCtx)` compose a fresh or resumed agent's scoped world while both objects remain unpublished. Setup is trusted, composition-only same-process code: drive the agent only after creation resolves.
- `ctx.agents.register(agent: Agent): () => void` — record an **already-constructed** agent. Disposed with the calling fiber.
- Advanced factory lifecycle: `enter(agent)` publishes without announcing and returns an entry-bound detach; `announce(agent)` emits creation once. Detach during creation dispatch is deferred. Ordinary plugins use `register()`.
- `ctx.agents.get(id: AgentId): Agent | undefined`
- Advanced ordered lifecycle: `enter(agent, owner): () => void` enforces `agent.id === agent.session.id`, performs the authoritative ID collision check, and inserts without announcing; `owner` explicitly records the live creator-agent relation (or `undefined` for a root), independently of durable session lineage. `announce(agent)` emits `agent/created` exactly once. A detach requested synchronously by a creation listener is deferred until that dispatch unwinds, and every detach checks the captured entry object, so a stale capability cannot delete a later same-ID replacement. The async factory uses this split; ordinary plugins use `register()`.
- `ctx.agents.get(id: SessionId): Agent | undefined`
- `ctx.agents.isOwnedBy(id: SessionId, owner: Agent): boolean` — whether the exact live entry was created through that parent agent's scoped context; runtime ownership is independent of durable session lineage.
- `ctx.agents.list(): Agent[]`
- `ctx.agents.roots(): Agent[]` — live agents created without an owning agent context; a resumed lineage-bearing session can still be a runtime root.
#### Factory seam (creation)
The loop plugin registers `AgentFactory`, keeping consumers independent of its concrete package. Each call is traced through the caller's context so the caller owns the resulting transaction and handle.
Agent *creation* is provided by the plugin implementing `AgentFactory` (`dsh-agent-loop`), registered via `setFactory`. This keeps creation on the `dsh-agent` interface so consumers (UI, the ACP bridge) program against `ctx.agents` without depending on the concrete loop package. The registry canonicalizes an already traced Service to its concrete target and re-traces each call through the caller's context; this avoids nested Cordis shadows while passing an explicit caller-bound `ownerCtx` to plain factories.
- `ctx.agents.setFactory(factory: AgentFactory): () => void` — register the creation factory (the loop calls this on construction). Throws on a second factory; the slot clears on dispose.
- `ctx.agents.create(options)` creates and composes an unpublished session and agent, then atomically enters the registries and starts the loop. A creation-only signal cancels before publication; same-ID contenders arbitrate at entry and losers roll back.
- `ctx.agents.resume(options)` loads a persisted session and follows the same composition and publication boundary. It requires [session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md).
- `ctx.agents.create(options: CreateAgentOptions): Promise<AgentHandle>` — create a session and agent, await optional setup while unpublished, then publish through final `SessionStore.enter()` and `AgentRegistry.enter()` checks. Concurrent same-ID creation is unsupported: more than one operation may prepare, but only one can enter; every loser rolls its private scope/session/driver back. An optional creation-only `signal` cancels unpublished setup and is detached before the handle is returned; later cancellation uses `handle.dispose()` or `agent.cancel()`. Publication is rollback-covered and every delivered creation edge is paired during rollback. Rejects if no factory is registered.
- `ctx.agents.resume(options: ResumeAgentOptions): Promise<AgentHandle>` — load a persisted session ([session persistence](../../../docs/rfc/implemented/architecture/2026-06-14-session-persistence.md)), mint a fresh unpublished agent scope, await optional setup, and use the same final-entry publication sequence. Its optional `signal` is likewise creation-only. Rejects if no factory is registered or session persistence is unconfigured.
`AgentHandle = { agent, dispose }` is the consumer teardown capability; registry observers receive only the bare agent. Disposal stops and drains the loop and idle-injection flushes before unregistering the agent, detaching its session, and unwinding its scope. Caller and factory unload share that memoized boundary.
`AgentHandle = { agent: Agent; dispose(): Promise<void> }`. The disposer is a **consumer capability** — no observer holding the bare registry entry can tear the agent down. The caller fiber and the registered factory provider are structural co-owners: caller unload enforces structured ownership, while factory unload must stop old instances because their scoped dependency surface belongs to that provider. `dispose()` from any owner reaches one memoized quiescence boundary: it stops the loop, `await`s its exit plus every outstanding idle-injection flush (not just the `disposed` status flip), unregisters the agent, removes its session from the store, and finally unwinds its scoped world. This order captures every agent-started `session/flush` before the session is detached and keeps scoped listeners alive through those checkpoints. `ctx.agents.get(id)` still returns a bare `Agent`; the ACP bridge and in-process subagent backends hold consumer handles, while config-created agents are already owned by the loop fiber.
### Live events
`dsh-agent` declares the live `agent/*` coordination vocabulary so plugins do not depend on the concrete loop. Exact signatures, dispatch modes, scope-filtering rules, and payload contracts live in the generated [Cordis event catalog](../../../docs/cordis-catalog/events.md); the [architecture turn flow](../../../docs/architecture.md#turn-flow) shows their order relative to durable session events.
`agent/created` runs after setup and both registry entries; the following `agent/session-start` is the first supported startup injection point. `agent/disposed` means the exact entry left the registry. The loop quiesces its driver first; directly registered custom agents own any stronger ordering.
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 returning seam-specific decisions. `agent/pre-step` is a serial surface-mutation checkpoint, while `agent/turn-stop` is the terminal serial fold: it runs after ordinary continuation and steering folding, and a returned stop remains in force through turn close and flush so later steering cannot create an extra step or turn. Ordinary queued prompts remain intact. The full rationale is in the [agent-scope runtime-design RFC](../../../docs/rfc/implemented/architecture/2026-07-12-agent-scope-runtime-design.md#three-execution-boundaries-are-deliberately-one-way).
@@ -75,3 +77,4 @@ The handle every plugin programs against:
- **No public step-only abort** — `cancel()` clears ALL pending work (queued + steering + in-flight); an abort that preserves queued prompts returns only with a named consumer ([stop-surface RFC](../../../docs/rfc/implemented/simplification/2026-06-20-public-agent-stop-surface.md)).
- **`HookContext` carries exactly one `MessageSource`** — contributions from several plugins merged onto one tool call collapse under one source; mixed provenance is unrepresentable.
- **`SessionStartSource` reserves `'clear'`/`'compact'` with no emitter yet** — only `'startup'`/`'resume'` occur until the driving subsystems land (`TODO(compaction)`).
- **`agent/pre-step`'s `fullSystemPrompt`/`sessionPrefix` parameters are a flagged smell** — compaction is their only consumer; a lazy prompt provider or a compaction-specific pressure seam is the marked revisit.

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

@@ -9,7 +9,7 @@ import { Context, getTraceable, Service, symbols } from 'cordis'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
import type { Scoped } from '@deepseek-ai/dsh-scope'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import type { Agent, AgentId, AgentOptions } from './types.ts'
import type { Agent, AgentOptions } from './types.ts'
export * from './types.ts'
export { agentEvents, assembleContextFor } from './dispatch.ts'
@@ -31,23 +31,57 @@ declare module 'cordis' {
}
}
/** Options for creating an agent and its caller-named session. */
/**
* Options for programmatically creating an agent through the registry factory
* ({@link AgentRegistry.create}). The caller supplies the single live
* `sessionId` shared by the agent registry and session log (e.g. an
* ACP-generated id), plus optional session metadata (the validated `cwd`, fork
* lineage); the factory creates the session and agent under that identity.
*/
export interface CreateAgentOptions {
/** The agent's id (the registry handle). */
readonly agentId: AgentId
/** The live session's id (NOT derived from agentId). */
/** The live agent/session identity. */
readonly sessionId: SessionId
/** Durable session metadata, validated and detached before setup. */
/**
* Session creation metadata: validated absolute `cwd`, `parentSession`
* fork lineage, and the `seedLength` seed boundary. Mirrors the
* `cwd`/`parentSession`/`seedLength` fields of
* {@link CreateSessionOptions.meta} in dsh-session (the internal-only
* `createdAt`, used when reconstructing a persisted session, is deliberately
* excluded — a factory caller never sets it). This is durable session data,
* so the session boundary validates and snapshots it before asynchronous
* setup begins.
*/
readonly meta?: { readonly cwd?: string; readonly parentSession?: SessionId; readonly seedLength?: number }
/** Balanced contiguous event prefix for a forked session. */
/**
* Seed events to reconstruct the child session's log from (the fork lineage
* primitive). When present, the factory creates the session with this event
* prefix so `deriveMessages()`/`lastTurnNumber` continue from it — used by the
* in-process FORK subagent backend to seed a child with a balanced
* completed-turn prefix of the parent's log. The prefix MUST be contiguous
* from seq 0, carry only lossless-JSON data, and be balanced (no open
* turn/step, no dangling tool-call), or the session constructor (and the
* dev-mode invariants replay) reject it. The factory passes the raw seed to
* the session's durable validator/snapshot boundary. Absent for a fresh
* (spawn) child.
*/
readonly seed?: readonly SessionEvent[]
/** Per-agent options (model, …). */
readonly agentOptions?: AgentOptions
/** Optional creation-only cancellation signal; detached before the returned handle becomes visible. */
readonly signal?: AbortSignal
/**
* Compose the unpublished scoped context before lifecycle announcements.
* Failure rolls back without publishing either id; setup must not drive the agent.
* Creation-time composition of the agent's scoped world. The factory awaits
* setup after minting `agentCtx` but BEFORE inserting or announcing either
* the session or agent, so observers can never see a partially configured
* world. Everything registered through `agentCtx` (scoped tools, prompt
* sections/variables, `restrict()`, listeners, awaited child plugins) exists
* before `session/created`, `agent/created`, `agent/session-start`, and the
* first prompt assembly. A throw/rejection or owner disposal rolls the scope
* back without publishing either id.
*
* **Setup composes, it never drives**: the callback is trusted same-process
* code and receives the full scoped context, so this is a contract rather
* than a runtime restriction. Drive the agent only after creation resolves.
*/
readonly setup?: (agentCtx: Context) => Promise<void> | void
}
@@ -57,23 +91,41 @@ export interface CreateAgentOptions {
* ({@link AgentRegistry.resume}).
*/
export interface ResumeAgentOptions {
/** The agent's id (the registry handle). */
readonly agentId: AgentId
/** The persisted session id to load and resume on. */
/** The persisted session id to load and use as the live agent/session identity. */
readonly resumeSessionId: SessionId
/** Per-agent options (model, …). */
readonly agentOptions?: AgentOptions
/** Optional creation-only cancellation signal for persistence load/setup; detached before return. */
readonly signal?: AbortSignal
/** Compose after persistence load under the same unpublished rollback contract as create. */
/**
* Resume-time composition of the agent's fresh scoped world. Persistence is
* loaded first; the factory then mints `agentCtx` and awaits setup while the
* reconstructed session and agent remain unpublished. The callback has the
* same trusted composition-only contract as
* {@link CreateAgentOptions.setup}: all registrations exist before either
* creation announcement, and rejection or owner disposal rolls the
* transaction back without publishing either id.
*/
readonly setup?: (agentCtx: Context) => Promise<void> | void
}
/**
* Holder-owned agent capability. Disposal stops and drains the loop and idle
* flushes before unregistering the agent, detaching its session, and unwinding
* its scoped context. Provider unload reaches the same quiescence boundary;
* registry observers receive only the bare {@link Agent}.
* An owned agent plus its disposer, returned by {@link AgentRegistry.create} /
* {@link AgentRegistry.resume}. The disposer is a CAPABILITY: among consumers,
* only the holder can tear this agent down. The registered factory provider is
* also a structural owner because the scoped agent depends on that provider's
* service surface; provider unload stops and drains every live handle it made.
* `dispose()` stops the loop, awaits its exit and every outstanding
* idle-injection flush (quiescence — NOT just the `disposed`
* status flip), unregisters the agent, removes its session from the store, and
* finally unwinds its scoped world. This order captures every agent-started
* `session/flush` before the session is detached and keeps scoped listeners
* alive through those checkpoints.
*
* `ctx.agents.get(id)` still returns a bare {@link Agent} — the handle is
* exposed only to the consumer owner that created it; the structural provider
* reaches the same teardown internally. Config-created agents (the loop's own
* startup) are owned by the loop fiber and never need a handle.
*/
export interface AgentHandle {
agent: Agent
@@ -88,16 +140,30 @@ export interface AgentHandle {
*/
export interface AgentFactory {
/**
* Create and compose under caller ownership, publish and announce session then
* agent, emit session-start, and start the driver. Rollback pairs any creation
* announcement that began.
* Create a new agent on a caller-supplied session id. Async because creation
* awaits unpublished setup, inserts both session and agent, emits their
* creation notifications in order, emits `agent/session-start`, and only
* then starts the loop. The sequence is
* rollback-covered, but notifications delivered before a later listener
* failure remain observable; every agent or session creation announcement
* that began is paired by `agent/disposed` or `session/disposed` during
* rollback. The owner disposes the resolved handle to stop/drain,
* unregister, remove the session, and unwind the scope.
* The registry passes a context carrying the `create()` caller's fiber and
* scope as `ownerCtx`. The implementation attaches the unpublished
* transaction and resulting lifecycle to that owner; it must not infer
* ownership from the factory object's registration context.
* @param ownerCtx - caller-bound context that owns the transaction and live handle.
* @param options - agent/session identity, configuration, and optional setup.
* @returns the owned handle after setup, both announcements, and loop start complete.
*/
createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle>
/**
* Load, compose, publish, announce, and resume an agent under caller ownership.
* Load a persisted session and resume an agent on it. Async because it awaits
* both `ctx.sessionPersistence.load` and the optional unpublished setup
* transaction; must be called after that service exists (consumers inject
* `sessionPersistence`). Publication follows the same ordered boundary as
* {@link createAgent}.
* @param ownerCtx - caller-bound context that owns load, setup, and the live handle.
* @param options - persisted identity, configuration, and optional setup.
* @returns the owned handle after setup, both announcements, and loop start complete.
@@ -110,8 +176,10 @@ const NO_FACTORY_MESSAGE = 'no agent factory registered (load an agent-loop plug
/** All mutable lifecycle state for one exact registry entry. */
interface AgentEntry {
readonly id: AgentId
readonly id: SessionId
readonly agent: Agent
/** Runtime creator-agent ownership; independent of durable session lineage. */
readonly owner: Agent | undefined
readonly carrier: Scoped<Agent>
announced: boolean
announcing: boolean
@@ -131,33 +199,46 @@ interface FactorySlot {
* {@link setFactory}.
*/
export class AgentRegistry extends Service {
private store = new Map<AgentId, AgentEntry>()
// TODO(agent-entry-mirror): derive exact-object checks from store.get(agent.id)
// plus entry.agent identity; this WeakMap mirrors the authoritative id map.
private entries = new WeakMap<Agent, AgentEntry>()
private store = new Map<SessionId, AgentEntry>()
private factory: FactorySlot | undefined
constructor(ctx: Context) {
super(ctx, 'agents')
// Agent contexts shadow this plain-context default with an own property.
// 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 })
}
/**
* Register the effect-scoped creation factory, rejecting a duplicate. Service
* factories are retraced through each create/resume caller for ownership.
* Register the agent-creation factory (the loop calls this on construction,
* effect-scoped). A traced Cordis service is canonicalized to its concrete
* target; each create/resume call is then traced through that caller's
* context so ownership follows the caller without stacking proxy layers.
* Throws if a factory is already registered. Returns the disposer; on
* dispose the factory slot is cleared.
* @param factory - the loop-owned factory {@link create}/{@link resume} delegate to.
* @returns the exact Cordis effect disposer.
* @returns the disposer that clears the factory slot. The exact
* Cordis effect disposer (single-shot): composite (generator) effects may
* yield it directly — exact identity nests the teardown in order.
*/
setFactory(factory: AgentFactory): () => void {
const dispose = this.ctx.effect(() => {
if (this.factory !== undefined) throw new Error('an agent factory is already registered')
// Store the concrete service; calls are retraced through their owner.
// Avoid stacking two Cordis shadow layers when a caller passes a Service
// already read through a context. Calls are re-traced through their
// actual owner context below.
const target = (factory as AgentFactory & { [symbols.original]?: AgentFactory })[symbols.original] ?? factory
this.factory = { target }
return () => { this.factory = undefined }
}, 'agents.setFactory()')
// Return the exact disposer so composite effects preserve teardown order.
// The exact cordis effect disposer (the agents.register() convention): a
// caller's composite effect can yield it for in-order teardown; the
// loop's constructor effect returns it directly, identity-nesting the
// registration under that effect.
// eslint-disable-next-line @typescript-eslint/no-misused-promises -- synchronous cleanup; direct return preserves disposer identity
return dispose
}
@@ -169,14 +250,20 @@ export class AgentRegistry extends Service {
}
/**
* Create and publish an owned agent and session through the active factory.
* Rejects if no factory is registered or creation, setup, or publication fails.
* @param options - agent id, session id/seed/metadata, and agent options.
* Create and publish a new agent through the registered factory.
* Distinct from {@link register} (which records an already-constructed
* agent): this constructs the agent and its session. Rejects if no factory is
* registered or creation/setup fails. The resolved {@link AgentHandle} lets
* the owner tear down exactly this agent.
* @param options - shared identity, session seed/metadata, and agent options.
* @returns the handle after setup, rollback-covered publication, and loop start complete.
*/
async create(options: CreateAgentOptions): Promise<AgentHandle> {
const ownerCtx = this.ctx
// Bind service effects to this caller while preserving factory dependencies.
// Re-trace a Service-backed factory through the accessing context
// explicitly. This preserves AgentLoop's dependency origin while binding
// its effects to ownerCtx; plain factories receive ownerCtx as an explicit
// capability and need no Cordis tracker magic.
const { target } = this.requireFactory()
const receiver = getTraceable(ownerCtx, target)
// eslint-disable-next-line @typescript-eslint/unbound-method -- Reflect.apply intentionally supplies the caller-traced receiver
@@ -199,14 +286,26 @@ export class AgentRegistry extends Service {
}
/**
* Register a live agent in the calling effect scope, with scope-filtered
* creation and disposal events. Duplicate ids throw.
* 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 — 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 exact Cordis effect disposer for nested teardown ordering.
* @returns the EXACT Cordis effect disposer (single-shot; a repeat call
* returns undefined without awaiting an in-flight teardown). Exact
* identity is load-bearing: a composite (generator) effect that owns a
* teardown ORDER — the agent factory's lifecycle chain — must yield THIS
* function so Cordis nests the unregistration at that yield position;
* yielding a wrapper would leave it disposing as a concurrent sibling on
* owner unload, unregistering the agent (and emitting `agent/disposed`)
* while its final turn is still draining.
*/
register(agent: Agent): () => void {
const dispose = this.ctx.effect(function* (this: AgentRegistry) {
yield this.enter(agent)
yield this.enter(agent, this.ctx.agent)
this.announce(agent)
}.bind(this), 'agents.register()')
// eslint-disable-next-line @typescript-eslint/no-misused-promises -- synchronous cleanup; direct return preserves disposer identity
@@ -214,31 +313,48 @@ export class AgentRegistry extends Service {
}
/**
* Insert an unpublished agent for an ordered factory transaction.
* Insert an already-constructed agent without announcing it. This is the
* advanced ordered-lifecycle primitive used by the async agent factory: it
* first completes setup while the agent is unpublished, then assigns the
* returned detach closure into its pre-installed composite teardown before
* calling {@link announce}. Ordinary callers use {@link register}.
* @param agent - the prepared, unpublished agent.
* @returns an idempotent closure that removes this exact entry and emits the
* paired disposal edge; detachment during creation dispatch is deferred.
* @param owner - live agent whose scoped context created this agent, or
* undefined for a top-level runtime root. This is runtime ownership, not
* the resumed session's durable parent lineage.
* @returns an idempotent closure that removes this exact entry and emits
* `agent/disposed` with listener failures contained. When called from a
* synchronous `agent/created` listener, removal and disposal wait until
* that creation dispatch unwinds.
*/
enter(agent: Agent): () => void {
enter(agent: Agent, owner: Agent | undefined): () => void {
const id = agent.id
if (id !== agent.session.id) {
throw new Error(`agent id "${id}" does not match session id "${agent.session.id}"`)
}
const carrier = scopeTarget(agent, agent)
// Prepared transactions arbitrate identity at this publication boundary.
if (this.entries.has(agent) || this.store.has(id)) throw new Error(`agent "${id}" is already registered`)
// This is the authoritative collision boundary. Concurrent create/resume
// operations may both prepare, but only one exact entry can publish.
if (this.store.has(id)) throw new Error(`agent "${id}" is already registered`)
const entry: AgentEntry = {
id,
agent,
owner,
carrier,
announced: false,
announcing: false,
detachRequested: false,
}
this.store.set(id, entry)
this.entries.set(agent, entry)
let entered = true
const detach = (): void => {
if (!entered) return
entered = false
// Creation listeners observe one stable entry before paired disposal.
// Every callback reached by this creation dispatch must observe the same
// live entry, and disposal must follow creation. A listener may own
// the advanced detach capability, so make that ordering structural:
// visibility and the paired disposal are deferred until announce()'s
// synchronous dispatch has unwound.
if (entry.announcing) {
entry.detachRequested = true
return
@@ -256,7 +372,6 @@ export class AgentRegistry extends Service {
/* v8 ignore next -- enter() rejects replacement while this single-shot detach capability is live. */
if (this.store.get(entry.id) !== entry) return
this.store.delete(entry.id)
this.entries.delete(entry.agent)
// An insertion rolled back before announce was never externally created,
// so emitting disposed would invent an impossible lifecycle edge. Marking
// happens before the created emit: if a later created listener throws,
@@ -288,8 +403,8 @@ export class AgentRegistry extends Service {
* creation listener).
*/
announce(agent: Agent): void {
const entry = this.entries.get(agent)
if (entry === undefined || this.store.get(entry.id) !== entry) {
const entry = this.store.get(agent.id)
if (entry === undefined || entry.agent !== agent) {
throw new Error(`agent "${agent.id}" is not live in this registry`)
}
if (entry.announced || entry.announcing) {
@@ -318,13 +433,25 @@ export class AgentRegistry extends Service {
/**
* Look up a live agent.
* @param id - the agent id to look up.
* @param id - the shared agent/session id to look up.
* @returns the agent, or undefined when no live agent has that id.
*/
get(id: AgentId): Agent | undefined {
get(id: SessionId): Agent | undefined {
return this.store.get(id)?.agent
}
/**
* Test whether a live agent was created through one exact parent agent's
* scoped context. Runtime ownership is independent of durable session
* lineage and remains unambiguous when unrelated providers reuse an id.
* @param id - the candidate child agent's shared agent/session id.
* @param owner - the expected runtime creator agent.
* @returns true only while the exact child entry is live under that owner.
*/
isOwnedBy(id: SessionId, owner: Agent): boolean {
return this.store.get(id)?.owner === owner
}
/**
* All live agents, in registration order.
* @returns a fresh array; mutating it does not affect the registry.
@@ -332,6 +459,18 @@ export class AgentRegistry extends Service {
list(): Agent[] {
return [...this.store.values()].map(entry => entry.agent)
}
/**
* All live top-level agents in registration order. A top-level agent was
* created without an owning agent context; durable session lineage does not
* affect this runtime relation, so a resumed fork may still be a root.
* @returns a fresh array; mutating it does not affect the registry.
*/
roots(): Agent[] {
return [...this.store.values()]
.filter(entry => entry.owner === undefined)
.map(entry => entry.agent)
}
}
export default AgentRegistry

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

@@ -5,25 +5,11 @@
* @module @deepseek-ai/dsh-agent/types
*/
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 { ContextEnvelope, JsonValue, Session, SessionId } from '@deepseek-ai/dsh-session'
import type {} from '@deepseek-ai/dsh-system-prompt'
/** Identifies one live agent in the registry. */
export type AgentId = Branded<'AgentId'>
/**
* Brand a string as an {@link AgentId}.
* @param id - the raw agent id string.
* @returns the same string, branded (a compile-time cast — no runtime cost).
*/
export function AgentId(id: string): AgentId {
return id as AgentId
}
import type { ContextEnvelope, JsonValue, Session } from '@deepseek-ai/dsh-session'
declare module '@deepseek-ai/dsh-system-prompt' {
interface AssembleContext {
/** Agent for this assembly; absent on diagnostics. When present, `scope` must identify the same agent. */
@@ -94,9 +80,10 @@ export type ContinuationStop = Extract<ContinuationDecision, { action: 'stop' }>
/** Why a session lifecycle began; seeded creates are `startup`, while persisted loads are `resume`. */
export type SessionStartSource = 'startup' | 'resume' | 'clear' | 'compact'
/** Public agent handle; the concrete driver belongs to `@deepseek-ai/dsh-agent-loop`. */
/** Public agent handle; its concrete implementation is internal to `@deepseek-ai/dsh-agent-loop`. */
export interface Agent {
readonly id: AgentId
/** The single identity shared with {@link session}. */
readonly id: SessionId
readonly options: AgentOptions
readonly session: Session
readonly status: AgentStatus

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

@@ -2,15 +2,16 @@ import { describe, expect, expectTypeOf, it } from 'vitest'
import { Context, Service, symbols } from 'cordis'
import type { Events } from 'cordis'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import AgentRegistry, { AgentId, agentEvents } from '@deepseek-ai/dsh-agent'
import AgentRegistry, { agentEvents } from '@deepseek-ai/dsh-agent'
import type { Agent, AgentFactory, ContinuationStop, CreateAgentOptions, ResumeAgentOptions } from '@deepseek-ai/dsh-agent'
function stubAgent(rawId: string): Agent {
const id = AgentId(rawId)
const id = SessionId(rawId)
return {
id,
options: {},
session: new Session(SessionId(`${id}-session`)),
session: new Session(id),
status: 'idle',
ctx: new Context(),
send() {},
@@ -41,6 +42,7 @@ describe('AgentRegistry', () => {
const dispose = ctx.agents.register(agent)
expect(ctx.agents.get(agent.id)).toBe(agent)
expect(ctx.agents.list()).toEqual([agent])
expect(ctx.agents.roots()).toEqual([agent])
expect(() => ctx.agents.register(stubAgent('a1'))).toThrow(/already registered/)
dispose()
@@ -48,6 +50,37 @@ describe('AgentRegistry', () => {
expect(lifecycle).toEqual(['created:a1', 'disposed:a1'])
})
it('rejects an agent whose registry and session identities differ', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
const agent = { ...stubAgent('agent-id'), session: new Session(SessionId('session-id')) }
expect(() => ctx.agents.enter(agent, undefined))
.toThrow('agent id "agent-id" does not match session id "session-id"')
expect(ctx.agents.list()).toEqual([])
})
it('tracks runtime creator ownership separately from registry order', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
const root = stubAgent('root')
const child = stubAgent('child')
const detachRoot = ctx.agents.enter(root, undefined)
ctx.agents.announce(root)
const detachChild = ctx.agents.enter(child, root)
ctx.agents.announce(child)
expect(ctx.agents.list()).toEqual([root, child])
expect(ctx.agents.roots()).toEqual([root])
expect(ctx.agents.isOwnedBy(child.id, root)).toBe(true)
expect(ctx.agents.isOwnedBy(root.id, root)).toBe(false)
expect(ctx.agents.isOwnedBy(SessionId('missing'), root)).toBe(false)
detachChild()
expect(ctx.agents.isOwnedBy(child.id, root)).toBe(false)
detachRoot()
})
it('rolls an entry back and pairs a partially delivered creation when a listener throws', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
@@ -57,7 +90,7 @@ describe('AgentRegistry', () => {
ctx.on('agent/disposed', agent => void lifecycle.push(`disposed:${agent.id}`))
expect(() => ctx.agents.register(stubAgent('vetoed'))).toThrow('creation veto')
expect(ctx.agents.get(AgentId('vetoed'))).toBeUndefined()
expect(ctx.agents.get(SessionId('vetoed'))).toBeUndefined()
expect(lifecycle).toEqual(['created:vetoed', 'disposed:vetoed'])
})
@@ -93,7 +126,7 @@ describe('AgentRegistry', () => {
ctx.on('agent/disposed', agent => void lifecycle.push(`disposed:${agent.id}`))
const first = stubAgent('split')
const detachFirst = ctx.agents.enter(first)
const detachFirst = ctx.agents.enter(first, undefined)
expect(lifecycle).toEqual([])
ctx.agents.announce(first)
expect(() => { ctx.agents.announce(first) }).toThrow(/already announced/)
@@ -101,7 +134,7 @@ describe('AgentRegistry', () => {
detachFirst()
const replacement = stubAgent('split')
const detachReplacement = ctx.agents.enter(replacement)
const detachReplacement = ctx.agents.enter(replacement, undefined)
detachFirst()
expect(ctx.agents.get(replacement.id)).toBe(replacement)
expect(() => { ctx.agents.announce(first) }).toThrow(/not live/)
@@ -121,7 +154,7 @@ describe('AgentRegistry', () => {
})
ctx.on('agent/created', () => void order.push(`second:${ctx.agents.get(agent.id) === agent}`))
ctx.on('agent/disposed', () => void order.push('disposed'))
const detach = ctx.agents.enter(agent)
const detach = ctx.agents.enter(agent, undefined)
ctx.agents.announce(agent)
expect(order).toEqual(['first:true', 'after-detach:true', 'second:true', 'disposed'])
expect(ctx.agents.get(agent.id)).toBeUndefined()
@@ -158,11 +191,11 @@ describe('AgentRegistry factory seam', () => {
const factory: AgentFactory = {
async createAgent(ownerCtx, options) {
calls.create.push({ ownerCtx, options })
return { agent: stubAgent(options.agentId), dispose: () => Promise.resolve() }
return { agent: stubAgent(options.sessionId), dispose: () => Promise.resolve() }
},
async resume(ownerCtx, options) {
calls.resume.push({ ownerCtx, options })
return { agent: stubAgent(options.agentId), dispose: () => Promise.resolve() }
return { agent: stubAgent(options.resumeSessionId), dispose: () => Promise.resolve() }
},
}
return { factory, calls }
@@ -171,15 +204,15 @@ describe('AgentRegistry factory seam', () => {
it('requires a factory and delegates through the calling context', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
await expect(ctx.agents.create({ agentId: AgentId('a'), sessionId: SessionId('s') })).rejects.toThrow(/no agent factory/)
await expect(ctx.agents.create({ sessionId: SessionId('s') })).rejects.toThrow(/no agent factory/)
const { factory, calls } = stubFactory()
ctx.agents.setFactory(factory)
let callerFiber: Context['fiber'] | undefined
await ctx.plugin(Object.assign(async (inner: Context) => {
callerFiber = inner.fiber
await inner.agents.create({ agentId: AgentId('create'), sessionId: SessionId('create-s') })
await inner.agents.resume({ agentId: AgentId('resume'), resumeSessionId: SessionId('resume-s') })
await inner.agents.create({ sessionId: SessionId('create-s') })
await inner.agents.resume({ resumeSessionId: SessionId('resume-s') })
}, { inject: ['agents'] }))
expect(calls.create[0]?.ownerCtx.fiber).toBe(callerFiber)
expect(calls.resume[0]?.ownerCtx.fiber).toBe(callerFiber)
@@ -192,9 +225,9 @@ describe('AgentRegistry factory seam', () => {
inner.agents.setFactory(stubFactory().factory)
expect(() => inner.agents.setFactory(stubFactory().factory)).toThrow(/already registered/)
}, { inject: ['agents'] }))
await expect(ctx.agents.create({ agentId: AgentId('before'), sessionId: SessionId('before-s') })).resolves.toBeDefined()
await expect(ctx.agents.create({ sessionId: SessionId('before-s') })).resolves.toBeDefined()
await owner.dispose()
await expect(ctx.agents.create({ agentId: AgentId('after'), sessionId: SessionId('after-s') })).rejects.toThrow(/no agent factory/)
await expect(ctx.agents.create({ sessionId: SessionId('after-s') })).rejects.toThrow(/no agent factory/)
})
it('canonicalizes an already traced Service before tracing it for the caller', async () => {
@@ -214,18 +247,18 @@ describe('AgentRegistry factory seam', () => {
}
async createAgent(_ownerCtx: Context, options: CreateAgentOptions) {
this.calls().push('create')
return { agent: stubAgent(options.agentId), dispose: () => Promise.resolve() }
return { agent: stubAgent(options.sessionId), dispose: () => Promise.resolve() }
}
async resume(_ownerCtx: Context, options: ResumeAgentOptions) {
this.calls().push('resume')
return { agent: stubAgent(options.agentId), dispose: () => Promise.resolve() }
return { agent: stubAgent(options.resumeSessionId), dispose: () => Promise.resolve() }
}
}
await ctx.plugin(TracedFactory)
const traced = (ctx as Context & { tracedFactory: TracedFactory }).tracedFactory
ctx.agents.setFactory(traced)
await ctx.agents.create({ agentId: AgentId('create'), sessionId: SessionId('create-s') })
await ctx.agents.resume({ agentId: AgentId('resume'), resumeSessionId: SessionId('resume-s') })
await ctx.agents.create({ sessionId: SessionId('create-s') })
await ctx.agents.resume({ resumeSessionId: SessionId('resume-s') })
const raw = (traced as unknown as { [symbols.original]?: TracedFactory })[symbols.original]
expect(states.get(raw!)).toEqual(['create', 'resume'])
})