fix: coordinate overlapping configured reloads
This commit is contained in:
@@ -41,7 +41,7 @@ interface Config {
|
||||
}
|
||||
```
|
||||
|
||||
Agents listed in config are auto-created at startup. `cwd` seeds a fresh config-created session; a materialized exact `sessionId` remount and an explicit `resumeSessionId` keep the persisted session header. While the factory is active, a declarative lookup, resume, setup, or publication failure is contained, logged, and emitted as `agent-loop/config-start-failed(sessionId, error)` because no live `Agent` exists for an `agent/*` signal; cancellation caused by factory teardown is silent. Config agents have no per-agent persona field: they use `dsh-system-prompt`'s deployment default, while programmatic factory callers can register an agent-scoped `deployment:persona` shadow in `setup`. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from `assembleContextFor(agent)` — the helper couples the typed agent with its matching scope selector. These are runtime facts of the agents THIS loop drives, unlike the `harness:identity` and default `deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
|
||||
Agents listed in config are auto-created at startup. `cwd` seeds a fresh config-created session; a materialized exact `sessionId` remount and an explicit `resumeSessionId` keep the persisted session header. An overlapping remount waits for an already-disposed same-id agent to finish detaching both registries before it inspects persistence, so asynchronous teardown cannot strand the configured identity. While the factory is active, a declarative lookup, resume, setup, or publication failure is contained, logged, and emitted as `agent-loop/config-start-failed(sessionId, error)` because no live `Agent` exists for an `agent/*` signal; cancellation caused by factory teardown is silent. Config agents have no per-agent persona field: they use `dsh-system-prompt`'s deployment default, while programmatic factory callers can register an agent-scoped `deployment:persona` shadow in `setup`. The plugin registers the built-in `model`/`cwd` prompt variables on `ctx.systemPrompt`, resolved per step from `assembleContextFor(agent)` — the helper couples the typed agent with its matching scope selector. These are runtime facts of the agents THIS loop drives, unlike the `harness:identity` and default `deployment:persona` sections, which live on `dsh-system-prompt` so they survive a swapped loop plugin.
|
||||
|
||||
### Exported concrete class
|
||||
|
||||
|
||||
@@ -53,6 +53,7 @@ function renderThrown(value: unknown): string {
|
||||
/** Factory-level ownership of every preparing or live transaction. */
|
||||
class FactoryOwnership {
|
||||
private accepting = true
|
||||
private readonly inactive = Promise.withResolvers<void>()
|
||||
private transactions = new Set<AgentCreationTransaction>()
|
||||
private startupTasks = new Set<Promise<void>>()
|
||||
|
||||
@@ -74,8 +75,14 @@ class FactoryOwnership {
|
||||
void task.then(forget, forget)
|
||||
}
|
||||
|
||||
/** Resolve `task`, or stop waiting when factory teardown begins. */
|
||||
async waitWhileActive(task: Promise<void>): Promise<void> {
|
||||
await Promise.race([task, this.inactive.promise])
|
||||
}
|
||||
|
||||
async dispose(): Promise<void> {
|
||||
this.accepting = false
|
||||
this.inactive.resolve()
|
||||
const reason = new Error('agent loop is not active')
|
||||
await Promise.all([
|
||||
...[...this.transactions].map(transaction => transaction.disposeForFactory(reason)),
|
||||
@@ -455,6 +462,8 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
agentOptions: AgentOptions,
|
||||
meta: Pick<SessionHeader, 'cwd'>,
|
||||
): Promise<void> {
|
||||
await this.waitForDrainingConfiguredIdentity(ownerCtx, sessionId)
|
||||
if (!this.ownership.isActive()) return
|
||||
const exists = (await persistence.list()).some(header => header.id === sessionId)
|
||||
if (!this.ownership.isActive()) return
|
||||
if (exists) {
|
||||
@@ -464,6 +473,32 @@ export class AgentLoop extends Service implements AgentFactory {
|
||||
this.create(sessionId, agentOptions, meta)
|
||||
}
|
||||
|
||||
/** Wait for an already-disposed same-id lifecycle to finish registry teardown. */
|
||||
private async waitForDrainingConfiguredIdentity(ownerCtx: Context, sessionId: SessionId): Promise<void> {
|
||||
const current = ownerCtx.agents.get(sessionId)
|
||||
if (current?.status !== 'disposed') return
|
||||
|
||||
const released = Promise.withResolvers<void>()
|
||||
const checkReleased = (): void => {
|
||||
if (ownerCtx.agents.get(sessionId) === undefined && ownerCtx.sessions.get(sessionId) === undefined) {
|
||||
released.resolve()
|
||||
}
|
||||
}
|
||||
const disposeAgentListener = ownerCtx.on('agent/disposed', (agent) => {
|
||||
if (agent.id === sessionId) checkReleased()
|
||||
})
|
||||
const disposeSessionListener = ownerCtx.on('session/disposed', (session) => {
|
||||
if (session.id === sessionId) checkReleased()
|
||||
})
|
||||
try {
|
||||
checkReleased()
|
||||
await this.ownership.waitWhileActive(released.promise)
|
||||
} finally {
|
||||
disposeAgentListener()
|
||||
disposeSessionListener()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an agent and session under one caller-supplied identity, owned by
|
||||
* the accessing fiber. Constructor-driven config calls mint a fresh combined
|
||||
|
||||
@@ -92,6 +92,50 @@ describe('config-driven session id', () => {
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('waits for a draining exact-id lifecycle during an overlapping reload', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-exact-overlap-'))
|
||||
dirs.push(root)
|
||||
const ctx = await makeCoreContext()
|
||||
await ctx.plugin(SessionPersistenceJsonl, { root })
|
||||
const sessionId = SessionId('stdio-exact-overlap')
|
||||
const config = { agents: [{ id: 'main', sessionId, model: 'mock' }] }
|
||||
const firstLoop = await ctx.plugin(AgentLoop, config)
|
||||
await expect.poll(() => ctx.agents.get(sessionId)).toBeDefined()
|
||||
const first = ctx.agents.get(sessionId) as ReactLoopAgent
|
||||
|
||||
const flushGate = Promise.withResolvers<undefined>()
|
||||
let flushStarted = false
|
||||
ctx.on('session/flush', (session) => {
|
||||
if (session !== first.session) return
|
||||
flushStarted = true
|
||||
return flushGate.promise
|
||||
})
|
||||
first.inject([{ type: 'text', text: 'persist before replacement' }], {
|
||||
source: { kind: 'plugin', plugin: 'test' },
|
||||
})
|
||||
expect(flushStarted).toBe(true)
|
||||
|
||||
const firstDisposal = firstLoop.dispose()
|
||||
await expect.poll(() => first.status).toBe('disposed')
|
||||
const failures: unknown[] = []
|
||||
ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
|
||||
const secondLoop = await ctx.plugin(AgentLoop, config)
|
||||
await new Promise(resolve => setTimeout(resolve, 0))
|
||||
expect(ctx.agents.get(sessionId)).toBe(first)
|
||||
expect(failures).toEqual([])
|
||||
|
||||
flushGate.resolve(undefined)
|
||||
await firstDisposal
|
||||
await expect.poll(() => ctx.agents.get(sessionId)).toBeDefined()
|
||||
const second = ctx.agents.get(sessionId) as ReactLoopAgent
|
||||
expect(second).not.toBe(first)
|
||||
expect(JSON.stringify(second.session.deriveMessages())).toContain('persist before replacement')
|
||||
expect(failures).toEqual([])
|
||||
|
||||
await secondLoop.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
})
|
||||
|
||||
it('contains an exact-id persistence lookup failure', async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-exact-failure-'))
|
||||
dirs.push(root)
|
||||
|
||||
Reference in New Issue
Block a user