From b58cf7ec2fdb3ba7d88ec097766dd47f2a188a2a Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 14 Jul 2026 14:24:21 +0800 Subject: [PATCH] fix: coordinate overlapping configured reloads --- docs/config-catalog.md | 2 +- docs/cordis-catalog/events.md | 2 +- docs/cordis-catalog/services.md | 2 +- docs/event-producer-consumer.md | 2 +- packages/core/agent-loop/README.md | 2 +- packages/core/agent-loop/src/index.ts | 35 +++++++++++++++ .../tests/config-session-id.spec.ts | 44 +++++++++++++++++++ packages/ui/stdio-agent/src/stdio-chat.ts | 13 +++++- .../ui/stdio-agent/tests/stdio-chat.spec.ts | 12 +++-- 9 files changed, 103 insertions(+), 11 deletions(-) diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 173806572a..9bebe62380 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:354`](../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:361`](../packages/core/agent-loop/src/index.ts) ## `@deepseek-ai/dsh-bash-local` diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index df3aec8b12..f913b95c36 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -185,7 +185,7 @@ A declarative agent entry failed before it could publish a live agent. Consumers 'agent-loop/config-start-failed'(sessionId: SessionId, error: unknown): void ``` -Source: [`packages/core/agent-loop/src/index.ts:349`](../../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:356`](../../packages/core/agent-loop/src/index.ts) ## `approval/*` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index d3e8b74edf..e4e68fe17f 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -19,7 +19,7 @@ async createAgent(ownerCtx: Context, options: CreateAgentOptions): Promise ``` -Source: [`packages/core/agent-loop/src/index.ts:369`](../../packages/core/agent-loop/src/index.ts) +Source: [`packages/core/agent-loop/src/index.ts:376`](../../packages/core/agent-loop/src/index.ts) ## `ctx.agents` โ€” `AgentRegistry` diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index cc155ebb16..2b62ec93eb 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -7,7 +7,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac | Event | Mode | Declared in | Dispatchers | Listeners | | --- | --- | --- | --- | --- | -| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:349`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) | +| `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:356`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/created` | `emit` | [`packages/core/agent/src/types.ts:304`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`jsonrpc`](../packages/ui/jsonrpc), [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:319`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/error` | `emit` | [`packages/core/agent/src/types.ts:593`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index e27e8b625a..881d60027a 100644 --- a/packages/core/agent-loop/README.md +++ b/packages/core/agent-loop/README.md @@ -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 diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index 06c108d324..3274e36587 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -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() private transactions = new Set() private startupTasks = new Set>() @@ -74,8 +75,14 @@ class FactoryOwnership { void task.then(forget, forget) } + /** Resolve `task`, or stop waiting when factory teardown begins. */ + async waitWhileActive(task: Promise): Promise { + await Promise.race([task, this.inactive.promise]) + } + async dispose(): Promise { 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, ): Promise { + 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 { + const current = ownerCtx.agents.get(sessionId) + if (current?.status !== 'disposed') return + + const released = Promise.withResolvers() + 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 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 53782fe03c..fda5681c29 100644 --- a/packages/core/agent-loop/tests/config-session-id.spec.ts +++ b/packages/core/agent-loop/tests/config-session-id.spec.ts @@ -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() + 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) diff --git a/packages/ui/stdio-agent/src/stdio-chat.ts b/packages/ui/stdio-agent/src/stdio-chat.ts index 39d59dfdc0..4505f19b98 100644 --- a/packages/ui/stdio-agent/src/stdio-chat.ts +++ b/packages/ui/stdio-agent/src/stdio-chat.ts @@ -67,6 +67,15 @@ function isTTYPair(input: Readable, output: Writable): boolean { return Boolean((input as { isTTY?: boolean }).isTTY && (output as { isTTY?: boolean }).isTTY) } +/** Render an arbitrary failure without allowing hostile coercion to escape the UI boundary. */ +function renderThrown(value: unknown): string { + try { + return String(value) + } catch { + return '' + } +} + interface PendingQuestion { request: AskUserQuestionRequest questionIndex: number @@ -230,7 +239,7 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt queuedInput.length = 0 submittedWork = sawRunning if (dropped > 0) { - ctx.logger.error(`ui-stdio: main agent failed to start; dropped queued stdin (${dropped} line(s)): ${String(error)}`) + ctx.logger.error(`ui-stdio: main agent failed to start; dropped queued stdin (${dropped} line(s)): ${renderThrown(error)}`) } maybeExit() }) @@ -390,7 +399,7 @@ export function createStdioChat(ctx: Context, config: Config, runtime: StdioRunt const text = line.trim() if (!text) return if (failedStartup !== undefined) { - ctx.logger.error(`ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): ${String(failedStartup.error)}`) + ctx.logger.error(`ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): ${renderThrown(failedStartup.error)}`) return } const agent = target diff --git a/packages/ui/stdio-agent/tests/stdio-chat.spec.ts b/packages/ui/stdio-agent/tests/stdio-chat.spec.ts index 11e64ac19c..c303577c48 100644 --- a/packages/ui/stdio-agent/tests/stdio-chat.spec.ts +++ b/packages/ui/stdio-agent/tests/stdio-chat.spec.ts @@ -83,6 +83,10 @@ function chunkEvent(chunk: StreamChunk): SessionEvent { const CONFIG: Config = { welcome: 'hi there', sessionId: 'main' } +function unrenderableFailure(): unknown { + return { [Symbol.toPrimitive](): never { throw new Error('coercion escaped') } } +} + async function setup(config: Config = CONFIG, runtimeOver: Partial = {}) { const ctx = new Context() await ctx.plugin(AgentRegistry) @@ -753,14 +757,14 @@ describe('createStdioChat input', () => { it('drops later input after the configured startup fails', async () => { const { ctx, input } = await setup() const error = vi.spyOn(ctx.logger, 'error').mockImplementation(() => {}) - const failure = new Error('persisted session is corrupt') + const failure = unrenderableFailure() ctx.emit('agent-loop/config-start-failed', SessionId('main'), failure) input.feed('cannot run') await new Promise(r => setImmediate(r)) expect(error).toHaveBeenCalledWith( - 'ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): Error: persisted session is corrupt', + 'ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): ', ) }) @@ -844,11 +848,11 @@ describe('createStdioChat EOF exit', () => { await flushExit() expect(exit).not.toHaveBeenCalled() - ctx.emit('agent-loop/config-start-failed', SessionId('main'), new Error('missing persisted session')) + ctx.emit('agent-loop/config-start-failed', SessionId('main'), unrenderableFailure()) await flushExit() expect(error).toHaveBeenCalledWith( - 'ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): Error: missing persisted session', + 'ui-stdio: main agent failed to start; dropped queued stdin (1 line(s)): ', ) expect(exit).toHaveBeenCalledWith(0) })