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

fix: complete scoped lifecycle simplification

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-12 23:15:07 +08:00
parent ed5304fb6d
commit 738054562d
16 changed files with 261 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/agent-loop/src/index.ts
+10 -20
View File
@@ -54,7 +54,6 @@ class FactoryOwnership {
}
track(transaction: AgentCreationTransaction): () => void {
if (!this.isActive()) throw new Error('agent loop is not active')
this.transactions.add(transaction)
return () => { this.transactions.delete(transaction) }
}
@@ -62,12 +61,9 @@ class FactoryOwnership {
async dispose(): Promise<void> {
this.accepting = false
const reason = new Error('agent loop is not active')
const results = await Promise.allSettled(
await Promise.all(
[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
)
const errors = results.flatMap(result => result.status === 'rejected' ? [result.reason as unknown] : [])
if (errors.length === 1) throw errors[0]
if (errors.length > 1) throw new AggregateError(errors, 'agent loop transaction disposal failed')
}
}
@@ -101,8 +97,6 @@ class AgentCreationTransaction {
private detachAgent: (() => void) | undefined
private publishing = false
private cleanupTask: Promise<void> | undefined
private finished = false
private wrapperFinished = false
private ownerFollowing = true
private readonly ownerDispose: () => Promise<void> | void
private readonly untrackFactory: () => void
@@ -120,20 +114,17 @@ class AgentCreationTransaction {
ownerCtx.fiber.assertActive()
this.ownerAgent = ownerCtx.agent
this.ownerFiber = ownerCtx.fiber
if (!ownership.isActive()) throw new Error('agent loop is not active')
this.ownerDispose = ownerCtx.effect(() => () => {
if (!this.ownerFollowing) return
return this.dispose(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
}, `agentLoop.owner(${id})`)
this.untrackFactory = ownership.track(this)
try {
this.ownerDispose = ownerCtx.effect(() => () => {
if (!this.ownerFollowing) return
return this.dispose(new Error(`agent "${id}" setup aborted: owner disposed during setup`))
}, `agentLoop.owner(${id})`)
} catch (error: unknown) {
this.untrackFactory()
throw error
}
if (signal === undefined) {
this.abortListener = undefined
} else {
this.abortListener = () => {
/* v8 ignore next 3 -- transaction teardown contains callback/driver failures; rejection is a future-drift backstop. */
void this.dispose(signalAbortError(id, signal)).catch((error: unknown) => {
this.loopCtx.logger.error(error)
})
@@ -168,6 +159,7 @@ class AgentCreationTransaction {
return await Promise.race([
Promise.resolve(operation),
this.deactivation.promise.then(() => {
/* v8 ignore next -- deactivate() assigns failure before resolving deactivation. */
throw this.failure ?? new Error(`agent "${this.id}" creation deactivated`)
}),
])
@@ -229,9 +221,11 @@ class AgentCreationTransaction {
publish(source: SessionStartSource): AgentHandle {
this.assertActive()
const driver = this.driver
/* v8 ignore next -- publish() is private and every caller invokes prepare() first. */
if (driver === undefined) throw new Error(`agent "${this.id}" is not prepared`)
const agent = driver.agent
const session = this.session
/* v8 ignore next -- prepare() assigns the session before it can produce the driver above. */
if (session === undefined) throw new Error(`agent "${this.id}" has no prepared session`)
this.publishing = true
try {
@@ -274,8 +268,6 @@ class AgentCreationTransaction {
/** Complete ownership bookkeeping after every resource reached quiescence. */
private finish(): void {
if (this.finished) return
this.finished = true
this.untrackFactory()
this.ownerFollowing = false
void this.ownerDispose()
@@ -310,8 +302,6 @@ class AgentCreationTransaction {
/** Mark the public create/resume continuation settled and detach its creation-only signal. */
finishWrapper(): void {
if (this.wrapperFinished) return
this.wrapperFinished = true
if (this.signal !== undefined && this.abortListener !== undefined) {
this.signal.removeEventListener('abort', this.abortListener)
}
packages/core/agent-loop/tests/agent.spec.ts
+14
View File
@@ -49,6 +49,20 @@ function send(agent: ReactLoopAgent, text: string) {
}
describe('ReactLoopAgent', () => {
it('rejects access before context binding and a second driver for one session', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
const session = ctx.sessions.create(SessionId('exclusive-driver'))
const prepared = prepareReactLoopAgent(ctx, AgentId('first-driver'), { model: 'mock' }, session)
expect(() => prepared.agent.ctx).toThrow('context is not bound')
expect(() => prepareReactLoopAgent(ctx, AgentId('second-driver'), { model: 'mock' }, session))
.toThrow('already has a concrete agent driver')
await prepared.dispose()
await ctx.fiber.dispose()
})
it('borrows caller options and binds its scoped context exactly once', async () => {
const ctx = await harness(new MockAdapter([textResponse('unused')]))
const options = { model: 'mock' }
packages/core/agent-loop/tests/resume.spec.ts
+22
View File
@@ -69,7 +69,29 @@ async function promptly<T>(task: Promise<T>): Promise<T> {
}
}
/** Throw an arbitrary callback value to exercise the public unknown-error boundary. */
function throwUnknown(value: unknown): never {
throw value
}
describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
it('normalizes a non-Error resume publication failure for rollback and rethrows it', async () => {
const sessionId = SessionId('unknown-resume-failure-s')
const root = await persistSession(sessionId)
const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
const failure = { source: 'resume' }
ctx.on('session/created', () => throwUnknown(failure))
await expect(ctx.agents.resume({
agentId: AgentId('unknown-resume-failure'),
resumeSessionId: sessionId,
})).rejects.toBe(failure)
expect(ctx.agents.get(AgentId('unknown-resume-failure'))).toBeUndefined()
expect(ctx.sessions.get(sessionId)).toBeUndefined()
await ctx.fiber.dispose()
})
it('createAgent uses the caller-supplied sessionId (not ${id}-session)', async () => {
const adapter = new MockAdapter([textResponse('hi')])
const { ctx } = await persistentHarness(adapter)
packages/core/agent-loop/tests/scope-lifecycle.spec.ts
+90
View File
@@ -40,6 +40,11 @@ function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
const text = (t: string): ContentBlock[] => [{ type: 'text', text: t }]
/** Throw an arbitrary callback value to exercise the public unknown-error boundary. */
function throwUnknown(value: unknown): never {
throw value
}
/** Invoke the exact lifecycle effect to exercise same-stack reentrant teardown. */
function disposeCurrentLifecycle(ownerCtx: Context): void {
const lifecycle = [...ownerCtx.fiber._disposables]
@@ -52,6 +57,91 @@ function disposeCurrentLifecycle(ownerCtx: Context): void {
}
describe('agent scope lifecycle', () => {
it('rejects an already-aborted creation signal before publishing either identity', async () => {
const ctx = await harness()
const reason = new Error('cancelled before creation')
const controller = new AbortController()
controller.abort(reason)
await expect(ctx.agents.create({
agentId: AgentId('pre-aborted'),
sessionId: SessionId('pre-aborted-s'),
signal: controller.signal,
})).rejects.toBe(reason)
expect(ctx.agents.get(AgentId('pre-aborted'))).toBeUndefined()
expect(ctx.sessions.get(SessionId('pre-aborted-s'))).toBeUndefined()
const valueController = new AbortController()
valueController.abort('plain cancellation reason')
await expect(ctx.agents.create({
agentId: AgentId('pre-aborted-value'),
sessionId: SessionId('pre-aborted-value-s'),
signal: valueController.signal,
})).rejects.toMatchObject({
message: 'agent "pre-aborted-value" creation aborted',
cause: 'plain cancellation reason',
})
expect(ctx.agents.get(AgentId('pre-aborted-value'))).toBeUndefined()
expect(ctx.sessions.get(SessionId('pre-aborted-value-s'))).toBeUndefined()
await ctx.fiber.dispose()
})
it('joins cleanup when an abort lands reentrantly during scope preparation', async () => {
const ctx = await harness()
const reason = new Error('cancelled while preparing')
const controller = new AbortController()
let aborted = false
ctx.on('internal/plugin', (fiber) => {
if (aborted || fiber.name !== 'scope') return
aborted = true
controller.abort(reason)
})
await expect(ctx.agents.create({
agentId: AgentId('prepare-abort'),
sessionId: SessionId('prepare-abort-s'),
signal: controller.signal,
})).rejects.toBe(reason)
expect(ctx.agents.get(AgentId('prepare-abort'))).toBeUndefined()
expect(ctx.sessions.get(SessionId('prepare-abort-s'))).toBeUndefined()
await ctx.fiber.dispose()
})
it('normalizes non-Error create failures for rollback while rethrowing the original value', async () => {
const ctx = await harness()
let thrown: unknown
ctx.on('session/created', () => {
if (thrown === undefined) return
const value = thrown
thrown = undefined
throwUnknown(value)
})
const createFailure = { source: 'create' }
thrown = createFailure
let createCaught: unknown
try {
ctx.agentLoop.create(AgentId('unknown-create'))
} catch (error: unknown) {
createCaught = error
}
expect(createCaught).toBe(createFailure)
const ownedFailure = { source: 'createAgent' }
thrown = ownedFailure
await expect(ctx.agents.create({
agentId: AgentId('unknown-owned-create'),
sessionId: SessionId('unknown-owned-create-s'),
})).rejects.toBe(ownedFailure)
expect(ctx.agents.get(AgentId('unknown-create'))).toBeUndefined()
expect(ctx.agents.get(AgentId('unknown-owned-create'))).toBeUndefined()
await ctx.fiber.dispose()
})
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' })
packages/core/agent/src/index.ts
+1
View File
@@ -364,6 +364,7 @@ export class AgentRegistry extends Service {
entry.detachRequested = false
// A stale capability can never delete a later same-id lifecycle. The
// captured entry identity is the final boundary.
/* 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)
packages/core/session/src/index.ts
+1
View File
@@ -731,6 +731,7 @@ export class SessionStore extends Service {
entry.detachRequested = false
// A stale capability cannot remove observers or storage belonging to a
// later same-id lifecycle.
/* 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)
attachments.delete(entry.session)
packages/core/tools/src/index.ts
+2 -1
View File
@@ -843,7 +843,8 @@ export class ToolRegistry extends Service {
// invariant assertion as well as protection against future layer changes.
if (this.codeTransport !== undefined) {
visible.set(RUN_CODE_NAME, this.codeTransport)
if (this.codeTransport.ownerFinal === true) ownerFinalNames.add(RUN_CODE_NAME)
// createRunCodeTool() owns this internal transport and always marks it owner-final.
ownerFinalNames.add(RUN_CODE_NAME)
}
return { visible, knownNames, restrictableNames, ownerFinalNames }
}
packages/core/tools/tests/scoped.spec.ts
+15
View File
@@ -103,6 +103,21 @@ describe('scoped tool registration', () => {
.toThrow(/owner-final tool "reserved" cannot be registered while a scoped shadow exists/)
})
it('restores global and scoped owner-final tools removed by assembly middleware', async () => {
const ctx = await mount()
const { scope, key } = await mintAgentScope(ctx, 'owner-final')
ctx.tools.register({ ...tool('required'), ownerFinal: true })
scope.ctx.tools.register({ ...tool('scoped-required'), ownerFinal: true })
ctx.on('system-prompt/assemble', async assembly => ({
...assembly,
tools: assembly.tools.filter(schema => !schema.name.includes('required')),
}))
expect((await ctx.systemPrompt.assemble()).tools.map(schema => schema.name)).toContain('required')
expect((await ctx.systemPrompt.assemble({ scope: key })).tools.map(schema => schema.name))
.toEqual(expect.arrayContaining(['required', 'scoped-required']))
})
it('disposing the scope unwinds its registrations and leaves no residue', async () => {
const ctx = await mount()
const { scope, key } = await mintAgentScope(ctx, 'a')
packages/skill/skill/src/index.ts
-1
View File
@@ -527,7 +527,6 @@ function waitWithAbort<T>(promise: Promise<T>, signal: AbortSignal | undefined):
reject(toError(error))
},
)
if (signal.aborted) onAbort()
})
}
packages/subagent/subagent-acp/src/run.ts
+5
View File
@@ -340,6 +340,11 @@ export async function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpe
cancelSettled.then((): SubagentResult => ({ output: collectOutput(), stopReason: 'aborted' })),
])
} catch (error: unknown) {
// A deterministic cancellation resolves `cancelSettled` before its
// best-effort ACP cancel can reject the prompt. This fallback is only for
// a process/pipe rejection already queued when the abort event fires; its
// first-outcome ordering cannot be forced without a timing-dependent test.
/* v8 ignore next */
if (flags.cancelled) return { output: collectOutput(), stopReason: 'aborted' }
// The seam contract: result resolves (never rejects) on a child-level
// failure. Startup failures were already rejected before publication;
packages/subagent/subagent-acp/tests/subagent-acp.spec.ts
+16 -1
View File
@@ -127,7 +127,9 @@ describe('dsh-subagent-acp', () => {
const result = await run.result
expect(result.stopReason).toBe('completed')
expect(text(result.output)).toBe('hello from acp child')
await run.dispose()
const disposal = run.dispose()
expect(run.dispose()).toBe(disposal)
await disposal
})
it('maps a max_tokens stop reason', async () => {
@@ -470,6 +472,19 @@ describe('dsh-subagent-acp', () => {
await run.dispose()
})
it('logs a flattened child failure through the registered provider', async () => {
const ctx = await setup({ MOCK_CRASH_ON_PROMPT: '1' })
const warnings: string[] = []
ctx.logger.warn = ((message: unknown) => { warnings.push(String(message)) }) as typeof ctx.logger.warn
const run = await ctx.subagents.start('acp', request())
const result = await run.result
expect(result.stopReason).toBe('error')
expect(warnings).toEqual([
expect.stringContaining('subagent-acp "acp": child run failed (error):'),
])
await run.dispose()
})
it('resolves error (never rejects) even when the onError sink itself throws', async () => {
// onError is a caller-supplied callback boundary: its own exception must be
// contained, or it would reject `result` and break the seam's "result never
packages/support/subagent-mock/tests/subagent-mock.spec.ts
+20
View File
@@ -32,6 +32,7 @@ describe('dsh-subagent-mock', () => {
structured: undefined,
stopReason: 'completed',
})
await run.dispose()
})
it('registers under a configurable name', async () => {
@@ -75,6 +76,25 @@ describe('dsh-subagent-mock', () => {
await expect(run.result).resolves.toMatchObject({ stopReason: 'aborted' })
})
it('rejects an already-aborted request before starting publication', async () => {
const ctx = await mount()
const controller = new AbortController()
controller.abort()
await expect(ctx.subagents.start('mock', baseRequest({ signal: controller.signal })))
.rejects.toThrow('mock subagent start aborted before publication')
})
it('rejects when cancellation wins the asynchronous publication handoff', async () => {
const ctx = await mount()
const controller = new AbortController()
const pending = ctx.subagents.start('mock', baseRequest({ signal: controller.signal }))
controller.abort()
await expect(pending).rejects.toThrow('mock subagent start aborted before publication')
})
it('unregisters the provider when the owning fiber is disposed (HMR safety)', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
packages/ui/user-approval/src/index.ts
-3
View File
@@ -451,9 +451,6 @@ export class ApprovalService extends Service {
resolve('cancelled')
}
signal.addEventListener('abort', onAbort, { once: true })
// Abort can win after the initial check but before listener installation.
// Recheck at the settlement boundary so that edge still cancels.
if (signal.aborted) onAbort()
void answer.then((outcome) => {
signal.removeEventListener('abort', onAbort)
// After an abort won the race this resolve is a settled-promise no-op:
packages/ui/user-approval/tests/approval.spec.ts
-20
View File
@@ -281,26 +281,6 @@ describe('ApprovalService.request', () => {
expect(appended[1]?.data).toMatchObject({ outcome: 'cancelled' })
})
it('does not miss an abort between the initial check and listener installation', async () => {
const ctx = await mounted()
const { agent, appended } = fakeAgent()
const answer = Promise.withResolvers<ApprovalOutcome>()
ctx.on('approval/request', () => answer.promise)
const controller = new AbortController()
const addEventListener = controller.signal.addEventListener.bind(controller.signal)
const add = vi.spyOn(controller.signal, 'addEventListener').mockImplementation((type, listener, options) => {
controller.abort()
addEventListener(type, listener, options)
})
await expect(ctx.approval.request(requestOf(agent, { signal: controller.signal }))).resolves.toBe('cancelled')
answer.resolve('allowed-once')
await Promise.resolve()
expect(add).toHaveBeenCalledOnce()
expect(appended[1]?.data).toMatchObject({ outcome: 'cancelled' })
})
it('resolves cancelled when the signal aborts mid-question and discards the late answer', async () => {
const ctx = await mounted()
const { agent, appended } = fakeAgent()
packages/workflow/workflow-workerthread/src/host.ts
+4 -4
View File
@@ -468,13 +468,13 @@ export class WorkerRun implements WorkflowRun {
.catch((error: unknown) => {
this.ctx.logger.warn(`workflow-workerthread: child dispose failed: ${renderThrown(error)}`)
})
.then(() => { this.finishChild(callId, record) })
.then(() => { this.finishChild(callId) })
return record.disposal
}
/** Drop an exact child record and release quiescence waiters when all work ends. */
private finishChild(callId: number, record: ChildRecord): void {
if (this.children.get(callId) === record) this.children.delete(callId)
/** Drop a child record and release quiescence waiters when all work ends. */
private finishChild(callId: number): void {
this.children.delete(callId)
this.notifyChildQuiescence()
}
packages/workflow/workflow-workerthread/tests/workflow-workerthread.spec.ts
+61
View File
@@ -1049,6 +1049,67 @@ describe('dsh-workflow-workerthread', () => {
await ctx.fiber.dispose()
})
it('refuses and disposes a provider run that becomes ready after its real worker dies', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)
const requested = Promise.withResolvers<SubagentStartRequest>()
const ready = Promise.withResolvers<SubagentRun>()
let disposeCalls = 0
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => ctx.logger)
const provider: SubagentProvider = {
name: 'late-ready',
capabilities: { outputSchema: true, depthLimit: true, toolFilter: true, persona: false },
inheritsParentContext: false,
start: (request) => {
requested.resolve(request)
// Model a backend whose independent startup boundary cannot be
// interrupted promptly. The host must still reject ownership if the
// worker dies before this promise transfers the ready run.
return ready.promise
},
}
ctx.subagents.registerProvider(provider)
await ctx.plugin(WorkerWorkflowEngine, { provider: 'late-ready', maxConcurrentAgents: 1 })
const lifecycle: string[] = []
ctx.on('workflow/agent-start', () => { lifecycle.push('start') })
ctx.on('workflow/agent-end', () => { lifecycle.push('end') })
const handle = ctx.workflows.start({
...scripted("return await agent('pending startup')"),
parent: fakeParent(),
})
const request = await requested.promise
const worker = (handle as unknown as { worker: Worker }).worker
// Kill the actual Worker while provider startup is independently
// pending. Death closes admission and aborts the shared signal, but this
// deliberately uncooperative provider still fulfills afterward.
await worker.terminate()
const result = await handle.result
expect(result.stopReason).toBe('error')
expect(result.error).toContain('exit code')
expect(request.signal.aborted).toBe(true)
expect(request.signal.reason).toBe('workflow worker gone')
ready.resolve({
id: AgentId('late-ready-child'),
result: Promise.resolve({ output: [], stopReason: 'aborted' }),
dispose: () => {
disposeCalls += 1
return Promise.reject(new Error('late ready dispose failed'))
},
})
await waitFor(() => {
expect(disposeCalls).toBe(1)
expect(warn).toHaveBeenCalledWith(expect.stringContaining('refused child dispose failed: Error: late ready dispose failed'))
}, 1000)
expect(lifecycle).toEqual([])
await handle.dispose()
expect(disposeCalls).toBe(1)
await ctx.fiber.dispose()
})
it('a worker that exits before settling reports an error result and reaps its children', async () => {
const ctx = new Context()
await ctx.plugin(SubagentService)