fix: join exact session startup on teardown
This commit is contained in:
@@ -131,7 +131,7 @@ export interface Config {
|
|||||||
|
|
||||||
Depends on: [`AgentOptions`](../packages/core/agent/src/index.ts) · [`SessionId`](../packages/core/session/src/index.ts)
|
Depends on: [`AgentOptions`](../packages/core/agent/src/index.ts) · [`SessionId`](../packages/core/session/src/index.ts)
|
||||||
|
|
||||||
Source: [`packages/core/agent-loop/src/index.ts:324`](../packages/core/agent-loop/src/index.ts)
|
Source: [`packages/core/agent-loop/src/index.ts:333`](../packages/core/agent-loop/src/index.ts)
|
||||||
|
|
||||||
## `@deepseek-ai/dsh-bash-local`
|
## `@deepseek-ai/dsh-bash-local`
|
||||||
|
|
||||||
|
|||||||
@@ -19,7 +19,7 @@ async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise<Agent
|
|||||||
async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle>
|
async resume(ownerCtx: Context, options: ResumeAgentOptions): Promise<AgentHandle>
|
||||||
```
|
```
|
||||||
|
|
||||||
Source: [`packages/core/agent-loop/src/index.ts:339`](../../packages/core/agent-loop/src/index.ts)
|
Source: [`packages/core/agent-loop/src/index.ts:348`](../../packages/core/agent-loop/src/index.ts)
|
||||||
|
|
||||||
## `ctx.agents` — `AgentRegistry`
|
## `ctx.agents` — `AgentRegistry`
|
||||||
|
|
||||||
|
|||||||
@@ -45,6 +45,7 @@ const INACTIVE_STATES: ReadonlySet<FiberState> = new Set([
|
|||||||
class FactoryOwnership {
|
class FactoryOwnership {
|
||||||
private accepting = true
|
private accepting = true
|
||||||
private transactions = new Set<AgentCreationTransaction>()
|
private transactions = new Set<AgentCreationTransaction>()
|
||||||
|
private startupTasks = new Set<Promise<void>>()
|
||||||
|
|
||||||
constructor(private readonly fiber: Context['fiber']) {}
|
constructor(private readonly fiber: Context['fiber']) {}
|
||||||
|
|
||||||
@@ -57,12 +58,20 @@ class FactoryOwnership {
|
|||||||
return () => { this.transactions.delete(transaction) }
|
return () => { this.transactions.delete(transaction) }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Join config startup work that begins before an agent transaction exists. */
|
||||||
|
trackStartup(task: Promise<void>): void {
|
||||||
|
this.startupTasks.add(task)
|
||||||
|
const forget = () => { this.startupTasks.delete(task) }
|
||||||
|
void task.then(forget, forget)
|
||||||
|
}
|
||||||
|
|
||||||
async dispose(): Promise<void> {
|
async dispose(): Promise<void> {
|
||||||
this.accepting = false
|
this.accepting = false
|
||||||
const reason = new Error('agent loop is not active')
|
const reason = new Error('agent loop is not active')
|
||||||
await Promise.all(
|
await Promise.all([
|
||||||
[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
|
...[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
|
||||||
)
|
...this.startupTasks,
|
||||||
|
])
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -371,9 +380,10 @@ export class AgentLoop extends Service implements AgentFactory {
|
|||||||
if (persistence === undefined) {
|
if (persistence === undefined) {
|
||||||
this.create(configuredId, options, meta)
|
this.create(configuredId, options, meta)
|
||||||
} else {
|
} 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)}`)
|
ctx.logger.warn(`agent "${id}": config-driven restore of "${configuredId}" failed: ${String(error)}`)
|
||||||
})
|
})
|
||||||
|
this.ownership.trackStartup(startup)
|
||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
@@ -403,6 +413,7 @@ export class AgentLoop extends Service implements AgentFactory {
|
|||||||
meta: Pick<SessionHeader, 'cwd'>,
|
meta: Pick<SessionHeader, 'cwd'>,
|
||||||
): Promise<void> {
|
): Promise<void> {
|
||||||
const exists = (await persistence.list()).some(header => header.id === sessionId)
|
const exists = (await persistence.list()).some(header => header.id === sessionId)
|
||||||
|
if (!this.ownership.isActive()) return
|
||||||
if (exists) {
|
if (exists) {
|
||||||
await this.resumeWith(ownerCtx, persistence, { resumeSessionId: sessionId, agentOptions })
|
await this.resumeWith(ownerCtx, persistence, { resumeSessionId: sessionId, agentOptions })
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -113,6 +113,31 @@ describe('config-driven session id', () => {
|
|||||||
await ctx.fiber.dispose()
|
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<Awaited<ReturnType<typeof ctx.sessionPersistence.list>>>()
|
||||||
|
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 () => {
|
it('identity-nests the deferred resume fiber under its labeled owner effect', async () => {
|
||||||
const ctx = new Context()
|
const ctx = new Context()
|
||||||
await ctx.plugin(LlmService)
|
await ctx.plugin(LlmService)
|
||||||
|
|||||||
Reference in New Issue
Block a user