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

fix(core): make initiator teardown reentrant-safe

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-19 15:12:18 +08:00
parent f66cde41d1
commit ee1a44793a
9 changed files with 195 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -26,7 +26,7 @@ The scoped-registration surface: `Agent.ctx` is the agent's scope context (`dsh-
- `ctx.agents.withInitiator(agent, operation)` — run with one exact Agent and preserve the operation's exact synchronous value or Promise.
- `ctx.agents.withoutInitiator(operation)` — hide an inherited initiator for unrelated process-local work.
The scope carries the `Agent` itself and is process-local. Ambient presence is neither liveness proof nor authorization; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. Teardown rejects new boundaries, lets injected dependents and returned-Promise boundaries drain, then disables the underlying `AsyncLocalStorage`; unreturned work remains owned by the subsystem that detached it. The [initiator-scope decision](../../../docs/rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md) owns the detailed boundary and teardown contract.
The scope carries the `Agent` itself and is process-local. Ambient presence is neither liveness proof nor authorization; explicit Agent fields remain authoritative at service, worker, process, persistence, and wire boundaries. Teardown rejects new boundaries, lets injected dependents and returned-Promise boundaries drain, then disables the underlying `AsyncLocalStorage`; unreturned work remains owned by the subsystem that detached it. If a boundary's inherited async chain starts an owning Cordis fiber's unload, that nested boundary chain is released from the drain so the unload cannot wait on itself; its continuations observe the disposed service after teardown. The [initiator-scope decision](../../../docs/rfc/implemented/architecture/2026-07-15-agent-initiator-scope.md) owns the detailed boundary and teardown contract.
#### Factory seam (creation)

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

@@ -5,7 +5,8 @@
* @module @deepseek-ai/dsh-agent
*/
import { Context, getTraceable, Service, symbols } from 'cordis'
import { Context, FiberState, getTraceable, Service, symbols } from 'cordis'
import type { Fiber } from 'cordis'
import { AsyncLocalStorage } from 'node:async_hooks'
import { isPromise } from 'node:util/types'
import { scopeTarget } from '@deepseek-ai/dsh-scope'
@@ -190,6 +191,12 @@ interface AgentEntry {
detachRequested: boolean
}
/** One tracked boundary plus its inherited nesting chain. */
interface InitiatorRun {
active: boolean
readonly parent: InitiatorRun | undefined
}
/** Plain holder prevents Cordis from tracing the factory field before the caller context is known. */
interface FactorySlot {
readonly target: AgentFactory
@@ -205,6 +212,7 @@ export class AgentRegistry extends Service {
private store = new Map<SessionId, AgentEntry>()
private factory: FactorySlot | undefined
private readonly initiators = new AsyncLocalStorage<Agent | undefined>()
private readonly initiatorRuns = new AsyncLocalStorage<InitiatorRun>()
private initiatorState: 'active' | 'closing' | 'disposed' = 'active'
private activeInitiatorRuns = 0
private initiatorDrain: PromiseWithResolvers<void> | undefined
@@ -219,6 +227,11 @@ export class AgentRegistry extends Service {
// accessor body never needs to resolve a scope itself. Effect-scoped:
// unwinds with this service's fiber.
ctx.accessor('agent', { get: () => undefined })
ctx.on('internal/status', (fiber) => {
if (fiber.state === FiberState.UNLOADING && this.hasLifecycleAncestor(fiber)) {
this.closeInitiators()
}
})
ctx.effect(function* (this: AgentRegistry) {
yield () => this.disposeInitiators()
yield () => { this.closeInitiators() }
@@ -249,6 +262,8 @@ export class AgentRegistry extends Service {
/**
* Run an operation with one exact Agent as its process-local initiator. The
* exact synchronous value or Promise returned by the operation is preserved.
* If its inherited async chain starts an owning-fiber unload, the nested
* boundary lineage is excluded from the drain so teardown cannot wait on itself.
* @param agent - initiating Agent to inherit; presence is neither liveness proof nor authorization.
* @param operation - synchronous or asynchronous operation to invoke.
* @returns the exact value returned by `operation`.
@@ -261,6 +276,8 @@ export class AgentRegistry extends Service {
/**
* Run an operation inside a boundary that hides any inherited initiating
* Agent. The exact synchronous value or Promise is preserved.
* If its inherited async chain starts an owning-fiber unload, the nested
* boundary lineage is excluded from the drain so teardown cannot wait on itself.
* @param operation - synchronous or asynchronous operation to invoke without an initiator.
* @returns the exact value returned by `operation`.
* @throws when the initiator scope is closing/disposed, or when `operation` throws.
@@ -537,42 +554,77 @@ export class AgentRegistry extends Service {
private disposeInitiators(): Promise<void> {
return (this.initiatorDisposal ??= (async () => {
this.closeInitiators()
this.releaseReentrantInitiatorRuns()
if (this.activeInitiatorRuns !== 0) {
this.initiatorDrain ??= Promise.withResolvers<void>()
await this.initiatorDrain.promise
}
this.initiatorState = 'disposed'
this.initiators.disable()
this.initiatorRuns.disable()
})())
}
/** Establish one tracked initiator or clearing boundary. */
private runWithInitiator<T>(agent: Agent | undefined, operation: () => T): T {
if (this.initiatorState !== 'active') throw new Error(DISPOSED_INITIATOR_MESSAGE)
const run: InitiatorRun = {
active: true,
parent: this.initiatorRuns.getStore(),
}
this.activeInitiatorRuns += 1
let result: T
try {
result = this.initiators.run(agent, operation)
result = this.initiatorRuns.run(run, () => this.initiators.run(agent, operation))
} catch (error: unknown) {
this.releaseInitiatorRun()
this.releaseInitiatorRun(run)
throw error
}
if (isPromise(result)) {
void result.then(
() => { this.releaseInitiatorRun() },
() => { this.releaseInitiatorRun() },
)
try {
void Promise.prototype.then.call(
result,
() => { this.releaseInitiatorRun(run) },
() => { this.releaseInitiatorRun(run) },
)
} catch {
// A branded Promise may expose a failing @@species. Observer setup did
// not attach, so preserve the exact return without leaking the run.
this.releaseInitiatorRun(run)
}
} else {
this.releaseInitiatorRun()
this.releaseInitiatorRun(run)
}
return result
}
/** Whether one unloading fiber owns this service's lifecycle. */
private hasLifecycleAncestor(candidate: Fiber): boolean {
let fiber = this.ctx.fiber
while (true) {
if (fiber === candidate) return true
const parent = fiber.parent.fiber
if (parent === fiber) return false
fiber = parent
}
}
private assertInitiatorsReadable(): void {
if (this.initiatorState === 'disposed') throw new Error(DISPOSED_INITIATOR_MESSAGE)
}
private releaseInitiatorRun(): void {
/** Exclude the boundary chain that initiated this teardown from its own drain. */
private releaseReentrantInitiatorRuns(): void {
let run = this.initiatorRuns.getStore()
while (run !== undefined) {
this.releaseInitiatorRun(run)
run = run.parent
}
}
private releaseInitiatorRun(run: InitiatorRun): void {
if (!run.active) return
run.active = false
this.activeInitiatorRuns -= 1
if (this.activeInitiatorRuns !== 0) return
this.initiatorDrain?.resolve()

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

@@ -23,6 +23,17 @@ async function harness(): Promise<{
}
}
/** Fail a lifecycle regression promptly instead of waiting for Vitest's suite timeout. */
async function promptly<T>(task: Promise<T>): Promise<T> {
const timeout = Promise.withResolvers<never>()
const timer = setTimeout(() => { timeout.reject(new Error('initiator teardown did not settle promptly')) }, 1000)
try {
return await Promise.race([task, timeout.promise])
} finally {
clearTimeout(timer)
}
}
describe('AgentRegistry initiator scope', () => {
it('reports an absent initiator and requires an active boundary', async () => {
const { service, dispose } = await harness()
@@ -52,6 +63,44 @@ describe('AgentRegistry initiator scope', () => {
await dispose()
})
it('tracks a branded Promise without calling its overridable then property', async () => {
const { service, dispose } = await harness()
const initiator = agent('overridden-then')
const release = Promise.withResolvers<boolean>()
void Object.defineProperty(release.promise, 'then', {
value: () => { throw new Error('overridden then called') },
})
const pending = service.withInitiator(initiator, () => release.promise)
expect(pending).toBe(release.promise)
let disposed = false
const disposal = dispose().then(() => { disposed = true })
await Promise.resolve()
expect(disposed).toBe(false)
release.resolve(true)
await new Promise<void>((resolve, reject) => {
void Promise.prototype.then.call(pending, resolve, reject)
})
await disposal
expect(disposed).toBe(true)
})
it('preserves a settled branded Promise when its species blocks observer construction', async () => {
const { service, dispose } = await harness()
const initiator = agent('invalid-species')
const promise = Promise.resolve()
const constructor = {}
Object.defineProperty(constructor, Symbol.species, {
get: () => { throw new Error('invalid species') },
})
void Object.defineProperty(promise, 'constructor', { value: constructor })
expect(service.withInitiator(initiator, () => promise)).toBe(promise)
await dispose()
})
it('isolates overlapping initiators', async () => {
const { service, dispose } = await harness()
const a = agent('a')
@@ -160,4 +209,57 @@ describe('AgentRegistry initiator scope', () => {
await disposal
expect(disposed).toBe(true)
})
it('does not self-deadlock when a boundary returns service disposal', async () => {
const { service, dispose } = await harness()
const initiator = agent('service-disposer')
const returned = service.withInitiator(initiator, dispose)
await promptly(returned)
expect(() => service.currentInitiator()).toThrow('agent initiator scope is disposed')
})
it('does not self-deadlock when nested boundaries return ancestor disposal', async () => {
const { ctx, service } = await harness()
const parent = agent('parent-disposer')
const child = agent('child-disposer')
let disposal: Promise<void> | undefined
const returned = service.withInitiator(parent, () => service.withInitiator(child, () => {
disposal = ctx.fiber.dispose()
return disposal
}))
expect(returned).toBe(disposal)
await promptly(returned)
expect(() => service.currentInitiator()).toThrow('agent initiator scope is disposed')
})
it('excludes an asynchronous teardown initiator while draining unrelated boundaries', async () => {
const { ctx, service } = await harness()
const initiator = agent('async-disposer')
const unrelated = agent('unrelated')
const release = Promise.withResolvers<boolean>()
const pending = service.withInitiator(unrelated, async () => {
await release.promise
expect(service.requireInitiator()).toBe(unrelated)
})
const returned = service.withInitiator(initiator, async () => {
await Promise.resolve()
await ctx.fiber.dispose()
})
let disposed = false
void returned.then(() => { disposed = true })
await Promise.resolve()
await Promise.resolve()
expect(disposed).toBe(false)
release.resolve(true)
await pending
await promptly(returned)
expect(() => service.currentInitiator()).toThrow('agent initiator scope is disposed')
})
})