diff --git a/packages/acp/acp/src/index.ts b/packages/acp/acp/src/index.ts index 7fc7c939e4..9d15397186 100644 --- a/packages/acp/acp/src/index.ts +++ b/packages/acp/acp/src/index.ts @@ -275,14 +275,16 @@ export function apply(ctx: Context, config: AcpConfig): void { record.inflight = inflight try { record.agent.followup({ content: [{ type: 'text', text }], source: { kind: 'user' } }) + // The machine's send() contains listener failures and accepts + // any typed input; this guards a future synchronous throw so the + // slot cannot wedge. + /* v8 ignore start -- future-proofing guard, see above */ } catch (error: unknown) { record.inflight = undefined - // followup() throws only Errors (invalid input); the String arm - // is a defensive fallback for a non-Error throw. - /* v8 ignore next */ const detail = error instanceof Error ? error.message : String(error) throw internalError(`prompt was not queued: ${detail}`) } + /* v8 ignore stop */ // Admission is pre-turn and retries outlive their failed turn, so a // turnless slot settles only at quiescence: a held failure rejects // (no retry adopted the prompt); no turn at all means admission diff --git a/packages/acp/acp/tests/turns.spec.ts b/packages/acp/acp/tests/turns.spec.ts index 4458a6b221..ed00d89be5 100644 --- a/packages/acp/acp/tests/turns.spec.ts +++ b/packages/acp/acp/tests/turns.spec.ts @@ -166,4 +166,40 @@ describe('ACP prompt lifecycle', () => { .resolves.toEqual({ stopReason: 'end_turn' }) await vi.waitFor(() => { expect(messageText(harness!)).toBe('next') }) }) + + it('a retry turn adopts the prompt instead of rejecting at the failed turn end', async () => { + harness = await makeBridgeHarness({ script: [errorResponse('transient boom'), textResponse('recovered')] }) + // A recovery policy: schedule one retry for the failed request. + let retried = false + harness.ctx.on('agent/request-error', async (subject) => { + if (!retried) { + retried = true + subject.retry() + } + }) + const sessionId = await newSession(harness) + const result = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }) + expect(result.stopReason).toBe('end_turn') + await vi.waitFor(() => { expect(messageText(harness!)).toBe('recovered') }) + }) + + it('a failed turn with no retry still rejects, at quiescence', async () => { + harness = await makeBridgeHarness({ script: [errorResponse('terminal boom')] }) + let offered = 0 + harness.ctx.on('agent/request-error', async () => { offered += 1 }) + const sessionId = await newSession(harness) + await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })) + .rejects.toThrow(/turn failed: terminal boom/) + expect(offered).toBe(1) + }) + + it('an admission-blocked prompt settles cancelled instead of hanging', async () => { + harness = await makeBridgeHarness({ script: [] }) + harness.ctx.on('agent/prompt-submit', async () => ({ kind: 'block' as const, reason: 'policy said no' })) + const sessionId = await newSession(harness) + await expect(harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })) + .resolves.toEqual({ stopReason: 'cancelled' }) + // The blocked prompt opened no turn and streamed nothing. + expect(messageText(harness)).toBe('') + }) }) diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 56ab89dff2..50d01024e0 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -295,6 +295,9 @@ export class ReactLoopAgent implements Agent { if (outcome.continueTurn || this.outbox.some(item => 'id' in item)) continue break case 'request-failed': { + // step() reports request failures only after step/start commits + // and before its own step/end, so the step is always open here. + /* v8 ignore next -- unreachable false arm, see above */ if (this.stepOpen) { this.stepOpen = false this.session.append('step/end', { turn, step }) @@ -315,6 +318,9 @@ export class ReactLoopAgent implements Agent { `agent "${this.id}": request recovery failed at turn ${turn}, step ${step}: ${errorChain(recoveryError)}`, ) } finally { + // Nothing else writes the window while the waterfall runs: + // cancel() only flips `requested` and a second run cannot start. + /* v8 ignore next -- unreachable false arm, see above */ if (this.retryWindow === retryWindow) this.retryWindow = undefined } retry = recoveryCompleted @@ -342,6 +348,10 @@ export class ReactLoopAgent implements Agent { ({ reason, idle } = this.settle(turn, step, caught, signal)) } finally { try { + // Every step-close before this point clears the flag on both success + // and failure paths (step(), the request-failed branch, the catch), + // so the finally never finds a step still open. + /* v8 ignore next 4 -- unreachable last-resort step close, see above */ if (this.stepOpen) { this.stepOpen = false this.session.append('step/end', { turn, step }) @@ -357,6 +367,9 @@ export class ReactLoopAgent implements Agent { emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, error) } this.retryWindow = undefined + // cancel() aborts but never clears the slot, and no second run can + // install a controller while this one is still unwinding. + /* v8 ignore next -- unreachable false arm, see above */ if (this.abort === controller) this.abort = undefined signal.removeEventListener('abort', cancelRetry) } diff --git a/packages/core/agent-loop/src/index.ts b/packages/core/agent-loop/src/index.ts index 1fc67a30f9..46eb18fddd 100644 --- a/packages/core/agent-loop/src/index.ts +++ b/packages/core/agent-loop/src/index.ts @@ -328,6 +328,10 @@ export class AgentLoop extends Service implements AgentFactory { */ private prepare(ownerCtx: Context, id: SessionId, options: AgentOptions, session: Session, callerSignal?: AbortSignal): PreparedAgent { ownerCtx.fiber.assertActive() + // Every caller reaches prepare() synchronously from a service method + // whose Cordis dispatch already requires the live factory fiber, or + // re-checks ownership itself after its awaits (resume's load barrier). + /* v8 ignore next -- unreachable backstop, see above */ if (!this.ownership.isActive()) throw new Error('agent loop is not active') if (callerSignal?.aborted) { throw callerSignal.reason instanceof Error @@ -393,15 +397,21 @@ export class AgentLoop extends Service implements AgentFactory { abort.abort(new Error(`agent "${id}" setup aborted: owner disposed during setup`)) return dispose(true) }, `agentLoop.lifecycle(${id})`) + /* v8 ignore start -- ctx.effect throws only on an inactive fiber, which assertActive() above already rejected */ } catch (error: unknown) { untrack() callerSignal?.removeEventListener('abort', onCallerAbort) this.ownership.signal.removeEventListener('abort', onFactoryTeardown) throw error } + /* v8 ignore stop */ const assertLive = (): void => { if (!abort.signal.aborted) return + // Every fused abort source carries an Error reason: onCallerAbort and + // raceAbort wrap non-Error caller reasons, and the factory/lifecycle + // owners abort with constructed Errors. + /* v8 ignore next -- unreachable String() arm, see above */ throw abort.signal.reason instanceof Error ? abort.signal.reason : new Error(String(abort.signal.reason)) } try { diff --git a/packages/core/agent-loop/tests/coverage-edges.spec.ts b/packages/core/agent-loop/tests/coverage-edges.spec.ts index c7dbbd5c4a..344f03edf4 100644 --- a/packages/core/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts @@ -498,3 +498,48 @@ describe('unrenderable failure settlement', () => { } }) }) + +describe('driver bookkeeping edges', () => { + it('a whenIdle waiter survives a rejected driver promise', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('waiter-chain'), { provider: 'mock', model: 'mock' }) + // A persistent step/end veto escapes even the catch block's own close + // attempt, so the driver promise REJECTS; the waiter's catch arm must + // treat that rejection as quiescence instead of propagating it. + ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const event = args[1] as SessionEvent + if (event.type === 'step/end') throw new Error('step close permanently rejected') + }) + + send(agent, 'one') + // Entered while the run owns the abort slot, the waiter awaits the + // driver promise; its rejection must count as quiescence and resolve. + await expect(agent.whenIdle()).resolves.toBeUndefined() + }) + + it('a request failure that concludes recovery after step/end closed keeps the boundary balanced', async () => { + const { LlmError } = await import('@deepseek-ai/dsh-llm') + // The failure finish-chunk path returns request-failed AFTER step() has + // already appended step/end, so the request-failed branch's own + // step-close guard must see stepOpen === false and skip the append. + const adapter = new MockAdapter([ + [ + { type: 'usage' as const, usage: { inputTokens: 1, outputTokens: 0 } }, + { type: 'finish' as const, reason: { kind: 'error' as const, failure: { message: 'empty', code: 'EMPTY_RESPONSE' } } }, + ] satisfies StreamChunk[], + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('finish-after-close'), { provider: 'mock', model: 'mock' }) + void LlmError + + send(agent, 'go') + await agent.whenIdle() + + const types = agent.session.events.map(e => e.type) + expect(types.filter(t => t === 'step/end')).toHaveLength(1) + const end = agent.session.events.findLast(e => e.type === 'turn/end') + expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error') + }) +}) diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index c2261d8d89..1dd1bf77db 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -649,3 +649,110 @@ describe('creation and resume cancellation edges', () => { await disposal }) }) + +describe('configured-start failure edges', () => { + it('a non-Error mid-load abort reason is wrapped for the resume caller', async () => { + const sessionId = SessionId('resume-string-mid-abort') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + const gate = Promise.withResolvers() + gate.promise.catch(() => undefined) + const loadStarted = Promise.withResolvers() + ctx.sessionPersistence.load = () => { + loadStarted.resolve(undefined) + return gate.promise + } + const controller = new AbortController() + + const resuming = ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + signal: controller.signal, + }) + await loadStarted.promise + controller.abort('operator string reason') + + await expect(promptly(resuming)).rejects.toThrow(/creation aborted/) + expect(ctx.agents.get(sessionId)).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('a failing exact-id restore over an existing artifact stays loud', async () => { + const sessionId = SessionId('config-existing-corrupt') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + // The artifact exists (list reports it) but its load fails: this is + // corruption, not first creation — the failure must be reported, and no + // fresh same-id session may shadow the broken one. + ctx.sessionPersistence.load = () => Promise.reject(new Error('artifact corrupt')) + + const configured = new Context() + await configured.plugin(LlmService) + await configured.plugin(SessionStore) + await configured.plugin(SystemPrompt) + await configured.plugin(ToolRegistry) + await configured.plugin(AgentRegistry) + await configured.plugin(SessionPersistenceJsonl, { root }) + configured.llm.registerAdapter(['mock'], new MockAdapter([])) + configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id) + const configFailures: unknown[] = [] + configured.on('agent-loop/config-start-failed', (_id, error) => { configFailures.push(error) }) + const configWarnings: string[] = [] + const configWarn = configured.logger.warn.bind(configured.logger) + configured.logger.warn = ((...args: unknown[]) => { + if (typeof args[0] === 'string') configWarnings.push(args[0]) + return (configWarn as (...a: unknown[]) => unknown)(...args) + }) as typeof configured.logger.warn + const loop = await configured.plugin(AgentLoop, { + agents: [{ id: 'main', sessionId, provider: 'mock', model: 'mock' }], + }) + await expect.poll(() => configFailures.length).toBe(1) + expect(configFailures[0]).toBeInstanceOf(Error) + expect((configFailures[0] as Error).message).toBe('artifact corrupt') + expect(configWarnings.some(w => w.includes('config-driven restore'))).toBe(true) + expect(configured.agents.get(sessionId)).toBeUndefined() + + await loop.dispose() + await configured.fiber.dispose() + await ctx.fiber.dispose() + }) + + it('suppresses a configured-resume failure that lands after teardown', async () => { + const sessionId = SessionId('config-late-resume-failure') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + const gate = Promise.withResolvers() + gate.promise.catch(() => undefined) + const loadStarted = Promise.withResolvers() + ctx.sessionPersistence.load = () => { + loadStarted.resolve(undefined) + return gate.promise + } + const failures: unknown[] = [] + ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) }) + + const configured = new Context() + await configured.plugin(LlmService) + await configured.plugin(SessionStore) + await configured.plugin(SystemPrompt) + await configured.plugin(ToolRegistry) + await configured.plugin(AgentRegistry) + await configured.plugin(SessionPersistenceJsonl, { root }) + configured.llm.registerAdapter(['mock'], new MockAdapter([])) + configured.sessionPersistence.load = id => ctx.sessionPersistence.load(id) + configured.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) }) + const loop = await configured.plugin(AgentLoop, { + agents: [{ id: 'main', resumeSessionId: sessionId, provider: 'mock', model: 'mock' }], + }) + await loadStarted.promise + const disposal = loop.dispose() + gate.reject(new Error('late backend failure')) + await disposal + await new Promise(r => setTimeout(r, 20)) + + // Ownership deactivated before the failure landed: the report is dropped. + expect(failures).toEqual([]) + await configured.fiber.dispose() + await ctx.fiber.dispose() + }) +})