fix(scope): harden lifecycle ownership foundation

Make Cordis construction and teardown ownership reentrancy-safe, then carry caller and provider ownership through reservation, setup, publication, quiescence, and sentinel retirement.

Stabilize registry carriers and factory/workflow boundaries, add adversarial lifecycle regressions, and align the rewritten RFC plus generated contracts with the enforced behavior.
This commit is contained in:
Tianyi Cui
2026-07-12 08:57:05 +08:00
parent 197f7237d2
commit c5b1a7941f
39 changed files with 2945 additions and 461 deletions

View File

@@ -1,12 +1,13 @@
/**
* 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.
* Fused scope-carrier dispatch for agent-subject operations, plus the assembly
* context builder. The sanctioned ordinary spelling is
* `agentEvents(ctx, agent).waterfall('agent/request', …)`: it builds the scope
* carrier ({@link scopeTarget} keyed by the agent) AND injects the subject as
* the first argument in one move, so a site cannot name a different subject.
* The registry lifecycle pair is the deliberate exception: `enter()` captures
* one stable carrier before commit and `announce()`/detach dispatch through it
* directly, preventing a mutable filter getter from changing or reentering the
* paired edges. The dev scoped-dispatch invariant checks both shapes.
*
* @module @deepseek-ai/dsh-agent/dispatch
*/

View File

@@ -5,11 +5,11 @@
* @module @deepseek-ai/dsh-agent
*/
import { Context, Service } from 'cordis'
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 { agentEvents } from './dispatch.ts'
export * from './types.ts'
export { agentEvents, assembleContextFor } from './dispatch.ts'
@@ -112,17 +112,21 @@ export interface ResumeAgentOptions {
/**
* An owned agent plus its disposer, returned by {@link AgentRegistry.create} /
* {@link AgentRegistry.resume}. The disposer is a CAPABILITY: only the holder
* can tear this agent down. `dispose()` stops the loop, awaits its exit and
* every outstanding idle-injection flush (quiescence — NOT just the `disposed`
* {@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 only
* for the OWNER that created it. Config-created agents (the loop's own startup)
* are owned by the loop fiber and never need a handle.
* `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
@@ -146,20 +150,62 @@ export interface AgentFactory {
* 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(options: CreateAgentOptions): Promise<AgentHandle>
createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<AgentHandle>
/**
* Load a persisted session and resume an agent on it. Async because it awaits
* both `ctx.sessionPersistence.load` and the optional unpublished setup
* transaction; must be called after that service exists (consumers inject
* `sessionPersistence`). Publication and drive unlocking follow 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.
*/
resume(options: ResumeAgentOptions): Promise<AgentHandle>
resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle>
}
/** One accepted factory target plus the callback identities captured at registration. */
interface AcceptedAgentFactory {
target: AgentFactory
createAgent: AgentFactory['createAgent']
resume: AgentFactory['resume']
}
/** Slot reservation while callback accessors are being captured. */
const ACCEPTING_FACTORY = Symbol('accepting agent factory')
/** Capture and validate the complete factory contract exactly once. */
function acceptAgentFactory(factory: unknown): AcceptedAgentFactory {
if ((typeof factory !== 'object' && typeof factory !== 'function') || factory === null) {
throw new TypeError('agent factory must be a non-null object or function')
}
// A service read through ctx is already a Cordis trace proxy. Retaining that
// proxy and tracing it again for each create() caller produces two shadow
// layers; raw-identity state (AgentLoop's private ownership controller is
// one example) then unwraps only to the inner proxy instead of its service.
// Canonicalize the one framework-produced layer at acceptance and capture
// callbacks from the concrete target. Plain objects expose no original.
const original: unknown = Reflect.get(factory, symbols.original)
const target = ((typeof original === 'object' || typeof original === 'function') && original !== null)
? original
: factory
const createAgent: unknown = Reflect.get(target, 'createAgent')
const resume: unknown = Reflect.get(target, 'resume')
if (typeof createAgent !== 'function') throw new TypeError('agent factory createAgent must be a function')
if (typeof resume !== 'function') throw new TypeError('agent factory resume must be a function')
return Object.freeze({
target: target as AgentFactory,
createAgent: createAgent as AgentFactory['createAgent'],
resume: resume as AgentFactory['resume'],
})
}
/** Thrown when create/resume is called before an agent factory is registered. */
@@ -187,6 +233,8 @@ export interface AgentRegistrationReservation {
/**
* Release the unpublished reservation; idempotent. The registry also
* releases it automatically when the fiber that called `reserve` disposes.
* This function is that exact Cordis effect disposer, so an ordered
* lifecycle may yield it by identity and place release after quiescence.
* @returns nothing.
*/
release(): void
@@ -201,13 +249,21 @@ export interface AgentRegistrationReservation {
*/
export class AgentRegistry extends Service {
private store = new Map<AgentId, Agent>()
/** Ids claimed across caller-code boundaries before their exact entry commits. */
private enteringIds = new Set<AgentId>()
/** The one accepted registry key for each live agent; never reread caller state. */
private acceptedIds = new WeakMap<Agent, AgentId>()
/** Unpublished identities held across factory setup/load transactions. */
private reservations = new Map<AgentId, AgentRegistrationReservation>()
/** Entries whose `agent/created` announcement phase began. */
private announced = new WeakSet<Agent>()
private factory: AgentFactory | undefined
/** Entries currently dispatching `agent/created`; detach waits for that dispatch to unwind. */
private announcing = new WeakSet<Agent>()
/** A detach requested reentrantly from `agent/created`. */
private pendingDetach = new WeakSet<Agent>()
/** Stable lifecycle dispatch carrier captured before an entry commits. */
private carriers = new WeakMap<Agent, Scoped<Agent>>()
private factory: AcceptedAgentFactory | typeof ACCEPTING_FACTORY | undefined
constructor(ctx: Context) {
super(ctx, 'agents')
@@ -234,39 +290,27 @@ export class AgentRegistry extends Service {
*/
reserve(id: AgentId): AgentRegistrationReservation {
if (typeof id !== 'string') throw new TypeError('agent id must be a string')
if (this.store.has(id) || this.reservations.has(id)) {
if (this.store.has(id) || this.reservations.has(id) || this.enteringIds.has(id)) {
throw new Error(`agent "${id}" is already registered or reserved`)
}
let active = true
const rawRelease = (): void => {
if (!active) return
active = false
this.reservations.delete(id)
}
let disposeEffect!: () => Promise<void> | void
const reservation: AgentRegistrationReservation = Object.freeze({
id,
release: () => {
rawRelease()
// Remove the now-inert ownership effect on manual transaction settle;
// its cleanup is the exact idempotent raw release above.
void disposeEffect()
},
})
// `release` is the exact effect disposer. A composite lifecycle can yield
// it by identity, moving automatic owner cleanup from a racing sibling to
// the transaction's final ordered position.
const release = this.ctx.effect(() => rawRelease, `agents.reserve(${id})`)
const reservation: AgentRegistrationReservation = Object.freeze({ id, release })
this.reservations.set(id, reservation)
try {
disposeEffect = this.ctx.effect(() => rawRelease, `agents.reserve(${id})`)
} catch (error: unknown) {
rawRelease()
throw error
}
return reservation
}
/**
* Register the agent-creation factory (the loop calls this on construction,
* effect-scoped). Throws if a factory is already registered. Returns the
* disposer; on dispose the factory slot is cleared.
* effect-scoped). The registry captures both callback identities once and
* later invokes them against the retained target receiver. 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 disposer that clears the factory slot. The exact
* Cordis effect disposer (single-shot): composite (generator) effects may
@@ -275,7 +319,17 @@ export class AgentRegistry extends Service {
setFactory(factory: AgentFactory): () => Promise<void> | void {
const dispose = this.ctx.effect(() => {
if (this.factory !== undefined) throw new Error('an agent factory is already registered')
this.factory = factory
// Claim the slot before reading caller-controlled method accessors. A
// getter may synchronously re-enter setFactory(); it must observe the
// registration in progress instead of installing a nested factory that
// the outer call would silently overwrite.
this.factory = ACCEPTING_FACTORY
try {
this.factory = acceptAgentFactory(factory)
} catch (error: unknown) {
this.factory = undefined
throw error
}
return () => { this.factory = undefined }
}, 'agents.setFactory()')
// The exact cordis effect disposer (the agents.register() convention): a
@@ -285,6 +339,13 @@ export class AgentRegistry extends Service {
return dispose
}
/** Return the accepted factory, excluding absence and reentrant acceptance. */
private requireFactory(): AcceptedAgentFactory {
const accepted = this.factory
if (accepted === undefined || accepted === ACCEPTING_FACTORY) throw new Error(NO_FACTORY_MESSAGE)
return accepted
}
/**
* Create and publish a new agent through the registered factory.
* Distinct from {@link register} (which records an already-constructed
@@ -295,8 +356,14 @@ export class AgentRegistry extends Service {
* @returns the handle after setup, rollback-covered publication, and loop start complete.
*/
async create(options: CreateAgentOptions): Promise<AgentHandle> {
if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE)
return this.factory.createAgent(options)
const accepted = this.requireFactory()
const ownerCtx = this.ctx
// 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 receiver = getTraceable(ownerCtx, accepted.target)
return Reflect.apply(accepted.createAgent, receiver, [ownerCtx, options])
}
/**
@@ -307,8 +374,10 @@ export class AgentRegistry extends Service {
* @returns the handle after setup, rollback-covered publication, and loop start complete.
*/
async resume(options: ResumeAgentOptions): Promise<AgentHandle> {
if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE)
return this.factory.resume(options)
const accepted = this.requireFactory()
const ownerCtx = this.ctx
const receiver = getTraceable(ownerCtx, accepted.target)
return Reflect.apply(accepted.resume, receiver, [ownerCtx, options])
}
/**
@@ -347,7 +416,9 @@ export class AgentRegistry extends Service {
* @param reservation - the exact unpublished-id capability, when a factory
* reserved this id across setup.
* @returns an idempotent closure that removes this exact entry and emits
* `agent/disposed` with listener failures contained.
* `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, reservation?: AgentRegistrationReservation): () => void {
const id = agent.id
@@ -361,39 +432,108 @@ export class AgentRegistry extends Service {
if (this.acceptedIds.has(agent)) {
throw new Error(`agent "${id}" is already registered`)
}
if (this.store.has(id)) {
if (this.store.has(id) || this.enteringIds.has(id)) {
throw new Error(`agent "${id}" is already registered`)
}
this.enteringIds.add(id)
let carrier: Scoped<Agent>
try {
// Registration accepts ownership of the public identity contract. Pin an
// own data slot from the one captured value so a custom JavaScript Agent
// with a getter or writable field cannot later present a different id to
// event listeners while the registry still owns the accepted key.
Object.defineProperty(agent, 'id', {
value: id,
enumerable: true,
writable: false,
configurable: false,
})
} catch {
// Only the engine's property-definition failure is swallowed; the stable
// public error below is the registration contract exposed to callers.
throw new TypeError('agent id must be installable as a stable own property')
try {
Object.defineProperty(agent, 'id', {
value: id,
enumerable: true,
writable: false,
configurable: false,
})
} catch {
// Only the engine's property-definition failure is normalized; filter
// construction below retains its own precise failure.
throw new TypeError('agent id must be installable as a stable own property')
}
// Capture one carrier for the paired lifecycle edges. Constructing it
// reads a custom Agent's Context.filter and is therefore caller code;
// the id claim above makes a same-id reentrant enter lose deterministically.
carrier = scopeTarget(agent, agent)
} finally {
// Kept through the entire caller-code window; the final commit below is
// synchronous and callback-free.
this.enteringIds.delete(id)
}
const currentReservation = this.reservations.get(id)
if (reservation === undefined) {
/* v8 ignore next 2 -- reserve() rejects enteringIds, so no callback in
* carrier construction can install a new same-id reservation */
if (currentReservation !== undefined) {
throw new Error(`agent "${id}" is reserved for unpublished creation`)
}
} else if (currentReservation !== reservation) {
throw new Error(`agent "${id}" registration reservation is not active for this id`)
}
/* v8 ignore next 2 -- the enteringIds claim blocks every public same-id
* commit until this callback-free final check has completed */
if (this.acceptedIds.has(agent) || this.store.has(id)) {
throw new Error(`agent "${id}" is already registered`)
}
this.store.set(id, agent)
this.acceptedIds.set(agent, id)
this.carriers.set(agent, carrier)
let entered = true
return () => {
const detach = (): void => {
if (!entered) return
entered = false
this.store.delete(id)
this.acceptedIds.delete(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,
// earlier listeners may already have observed it and must see disposal.
if (!this.announced.delete(agent)) return
agentEvents(this.ctx, agent).emit('agent/disposed')
// 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 (this.announcing.has(agent)) {
this.pendingDetach.add(agent)
return
}
this.detachEntered(agent, id)
}
return detach
}
/** Remove one exact entered agent and emit its paired disposal when announced. */
private detachEntered(agent: Agent, id: AgentId): void {
this.pendingDetach.delete(agent)
// A stale capability can never delete a later same-id lifecycle. The
// commit claim prevents this mismatch in normal operation; retain the
// exact-object guard as the final identity boundary.
/* v8 ignore next 1 -- the commit claim makes replacement impossible; this
* remains the exact-identity backstop against future mutation paths */
if (this.store.get(id) !== agent || this.acceptedIds.get(agent) !== id) return
this.store.delete(id)
this.acceptedIds.delete(agent)
const carrier = this.carriers.get(agent)
this.carriers.delete(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,
// earlier listeners may already have observed it and must see disposal.
if (!this.announced.delete(agent)) return
/* v8 ignore next -- enter commits the carrier with the exact store entry */
if (carrier === undefined) throw new Error(`agent "${id}" has no dispatch carrier`)
this.emitDisposed(agent, carrier, id)
}
/** Emit the paired disposal edge through the entry's stable carrier. */
private emitDisposed(agent: Agent, carrier: Scoped<Agent>, id: AgentId): void {
const args: unknown[] = [carrier, 'agent/disposed', agent]
for (const callback of this.ctx.events.dispatch('emit', args)) {
try {
const returned: unknown = callback(...args)
void Promise.resolve(returned).catch((error: unknown) => {
this.ctx.logger.warn(`agent "${id}": agent/disposed listener rejected: ${renderThrown(error)}`)
})
} catch (error: unknown) {
this.ctx.logger.warn(`agent "${id}": agent/disposed listener threw: ${renderThrown(error)}`)
}
}
}
@@ -409,21 +549,30 @@ export class AgentRegistry extends Service {
if (id === undefined || this.store.get(id) !== agent) {
throw new Error(`agent "${id ?? '<unknown>'}" is not live in this registry`)
}
if (this.announced.has(agent)) {
if (this.announced.has(agent) || this.announcing.has(agent)) {
throw new Error(`agent "${id}" was already announced`)
}
const carrier = this.carriers.get(agent)
/* v8 ignore next -- enter commits the carrier with the exact store entry */
if (carrier === undefined) throw new Error(`agent "${id}" has no dispatch carrier`)
// Mark before dispatch so a listener cannot recursively create a second
// lifecycle edge; detach still pairs a partially delivered first edge.
this.announcing.add(agent)
this.announced.add(agent)
const args: unknown[] = [scopeTarget(agent, agent), 'agent/created', agent]
for (const callback of this.ctx.events.dispatch('emit', args)) {
// A synchronous creation failure vetoes publication and rolls back.
// Returned-promise rejection happens after this synchronous boundary, so
// observe and report it instead of leaking an unhandled rejection.
const returned: unknown = callback(...args)
void Promise.resolve(returned).catch((error: unknown) => {
this.ctx.logger.warn(`agent "${id}": agent/created listener rejected: ${renderThrown(error)}`)
})
const args: unknown[] = [carrier, 'agent/created', agent]
try {
for (const callback of this.ctx.events.dispatch('emit', args)) {
// A synchronous creation failure vetoes publication and rolls back.
// Returned-promise rejection happens after this synchronous boundary, so
// observe and report it instead of leaking an unhandled rejection.
const returned: unknown = callback(...args)
void Promise.resolve(returned).catch((error: unknown) => {
this.ctx.logger.warn(`agent "${id}": agent/created listener rejected: ${renderThrown(error)}`)
})
}
} finally {
this.announcing.delete(agent)
if (this.pendingDetach.has(agent)) this.detachEntered(agent, id)
}
}

View File

@@ -291,7 +291,11 @@ declare module 'cordis' {
* to inject or queue work during startup. A synchronous listener throw
* vetoes publication and rollback emits the matching disposal edges;
* returned-promise rejection is observed and logged but cannot
* retroactively veto this synchronous boundary.
* retroactively veto this synchronous boundary. A synchronous listener
* that requests the advanced registry detach does not remove the entry
* immediately: removal and the paired `agent/disposed` edge wait until the
* creation dispatch unwinds, so no later creation listener observes a
* disposal that preceded its own creation callback.
* @param agent - the newly registered agent with its live session and completed setup.
* Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): a listener registered
* through `agent.ctx` fires only for that agent's dispatches; a listener on a
@@ -302,11 +306,12 @@ declare module 'cordis' {
*/
'agent/created'(this: Scoped<Agent>, agent: Agent): void
/**
* An agent was removed from the registry after its driver and any in-flight
* turn reached quiescence. Ordered teardown may still be detaching the
* session and unwinding the agent's scoped registrations when this
* notification runs.
* @param agent - the deregistered agent; its driving handle is now inert.
* An agent was removed from the registry. The concrete AgentLoop lifecycle
* emits this only after its driver and any in-flight turn reach quiescence;
* a custom agent registered through the public registry owns its own driver
* contract, which the registry cannot infer. Ordered teardown may still be
* detaching the session and unwinding scoped registrations when this runs.
* @param agent - the exact agent removed from 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
@@ -348,11 +353,12 @@ declare module 'cordis' {
/**
* The agent's session lifecycle began, fired once before its first turn.
* `source` says why ({@link SessionStartSource}: fresh startup, a resumed
* persisted session, …). A pure NOTIFICATION (emit, not waterfall): it
* carries no veto — a session-start listener that wants to seed context does
* so via `agent.inject()` (a `context/message` the first request sees), not
* by returning a decision. Cannot block the session from starting; that gap
* is deliberate (a bridge logs/injects, it does not gate startup).
* persisted session, …). A pure NOTIFICATION (emit, not waterfall): a
* listener cannot veto by returning a decision or throwing. A listener that
* wants to seed context does so via `agent.inject()` (a `context/message` the
* first request sees). A lifecycle owner can still dispose its structural
* ownership edge during this notification; publication rechecks liveness and
* then aborts before the driver starts.
* @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