diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 4c9d48e8d1..04050445d9 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -131,7 +131,7 @@ export interface Config { Depends on: [`AgentOptions`](../packages/core/agent/src/index.ts) · [`SessionId`](../packages/core/session/src/index.ts) -Source: [`packages/core/agent-loop/src/index.ts:323`](../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:332`](../packages/core/agent-loop/src/index.ts) ## `@deepseek-ai/dsh-bash-local` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index 504cc51e21..c65acef5c3 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -21,7 +21,7 @@ async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise = new Set([ class FactoryOwnership { private accepting = true private transactions = new Set() + private startupTasks = new Set>() constructor(private readonly fiber: Context['fiber']) {} @@ -56,12 +57,20 @@ class FactoryOwnership { return () => { this.transactions.delete(transaction) } } + /** Join config startup work that begins before an agent transaction exists. */ + trackStartup(task: Promise): void { + this.startupTasks.add(task) + const forget = () => { this.startupTasks.delete(task) } + void task.then(forget, forget) + } + async dispose(): Promise { this.accepting = false const reason = new Error('agent loop is not active') - await Promise.all( - [...this.transactions].map(transaction => transaction.disposeForFactory(reason)), - ) + await Promise.all([ + ...[...this.transactions].map(transaction => transaction.disposeForFactory(reason)), + ...this.startupTasks, + ]) } } @@ -370,9 +379,10 @@ export class AgentLoop extends Service implements AgentFactory { if (persistence === undefined) { this.create(configuredId, options, meta) } else { - void this.restoreOrCreateConfigured(ctx, persistence, configuredId, options, meta).catch((error: unknown) => { + const startup = this.restoreOrCreateConfigured(ctx, persistence, configuredId, options, meta).catch((error: unknown) => { ctx.logger.warn(`agent "${id}": config-driven restore of "${configuredId}" failed: ${String(error)}`) }) + this.ownership.trackStartup(startup) } continue } @@ -402,6 +412,7 @@ export class AgentLoop extends Service implements AgentFactory { meta: Pick, ): Promise { const exists = (await persistence.list()).some(header => header.id === sessionId) + if (!this.ownership.isActive()) return if (exists) { await this.resumeWith(ownerCtx, persistence, { resumeSessionId: sessionId, agentOptions }) return diff --git a/packages/core/agent-loop/tests/config-session-id.spec.ts b/packages/core/agent-loop/tests/config-session-id.spec.ts index 96f8ce8f6e..9410da7101 100644 --- a/packages/core/agent-loop/tests/config-session-id.spec.ts +++ b/packages/core/agent-loop/tests/config-session-id.spec.ts @@ -113,6 +113,31 @@ describe('config-driven session id', () => { await ctx.fiber.dispose() }) + it('joins an exact-id persistence lookup before AgentLoop disposal completes', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-exact-dispose-')) + dirs.push(root) + const ctx = await makeCoreContext() + await ctx.plugin(SessionPersistenceJsonl, { root }) + const listing = Promise.withResolvers>>() + vi.spyOn(ctx.sessionPersistence, 'list').mockReturnValue(listing.promise) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + + const loop = await ctx.plugin(AgentLoop, { + agents: [{ id: 'main', sessionId: SessionId('stdio-exact-dispose'), model: 'mock' }], + }) + let disposed = false + const disposal = loop.dispose().then(() => { disposed = true }) + await Promise.resolve() + expect(disposed).toBe(false) + + listing.resolve([]) + await disposal + expect(ctx.agents.get(SessionId('stdio-exact-dispose'))).toBeUndefined() + expect(warn).not.toHaveBeenCalled() + warn.mockRestore() + await ctx.fiber.dispose() + }) + it('identity-nests the deferred resume fiber under its labeled owner effect', async () => { const ctx = new Context() await ctx.plugin(LlmService)