diff --git a/packages/context/workspace-context/src/index.ts b/packages/context/workspace-context/src/index.ts index 6c2c5b95c7..d30a2f8b16 100644 --- a/packages/context/workspace-context/src/index.ts +++ b/packages/context/workspace-context/src/index.ts @@ -62,6 +62,14 @@ export function apply(ctx: Context, config: Config): void { ctx.on('agent/step', async (agent: Agent, _turn, _step, signal): Promise => { if (baselineLoaded.has(agent.session)) return + // The guard is mount-local, but the baseline is durable: a hot remount + // over a live session must fold the already-appended baseline from the + // log instead of injecting a duplicate. + if (agent.session.events.some(event => event.type === 'user/message' + && event.data.source.kind === 'plugin' && event.data.source.plugin === 'workspace-context')) { + baselineLoaded.add(agent.session) + return + } if (resolved.maxBytes <= 0 || !Number.isFinite(resolved.maxBytes)) { baselineLoaded.add(agent.session) return diff --git a/packages/context/workspace-context/tests/workspace-context.spec.ts b/packages/context/workspace-context/tests/workspace-context.spec.ts index a5fb046038..3e3285b1de 100644 --- a/packages/context/workspace-context/tests/workspace-context.spec.ts +++ b/packages/context/workspace-context/tests/workspace-context.spec.ts @@ -979,6 +979,32 @@ describe('workspace context request injection', () => { } }) + it('folds an already-appended baseline from the log after a plugin remount', async () => { + const root = await tempRepo() + const home = await tempRepo() + try { + await mkdir(join(root, '.git'), { recursive: true }) + await write(join(root, 'AGENTS.md'), 'repo rule') + const ctx = new Context() + await ctx.plugin(LocalFileSystem, { cwd: '/' }) + const fiber = await ctx.plugin(workspaceContext, { dshHome: home, maxBytes: 65536 }) + const agent = stubAgent(root) + await composeBaselinePrefix(ctx, agent) + + // Hot remount over the live session: the durable baseline survived, so + // the fresh mount must fold it from the log instead of appending a + // duplicate on its next step. + await fiber.dispose() + await ctx.plugin(workspaceContext, { dshHome: home, maxBytes: 65536 }) + await composeBaselinePrefix(ctx, agent) + + expect(agent.session.events.filter(event => event.type === 'user/message' && event.data.source.kind !== 'user')).toHaveLength(1) + } finally { + await rm(root, { recursive: true, force: true }) + await rm(home, { recursive: true, force: true }) + } + }) + it('tracks only baseline files that were actually included under the byte budget', async () => { const root = await tempRepo() const home = await tempRepo() diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index 33d96ae502..d6b1013b6d 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -186,8 +186,11 @@ export class ReactLoopAgent implements Agent { /** Resolve at idle quiescence: no run driving and no waking prompt waiting. */ async whenIdle(): Promise { - // `done` is replaced per activity, so re-reading it follows chained turns; - // a run failure still counts as quiescence for the waiter. + // `done` is replaced per activity, so re-reading it follows chained turns. + // Every driver failure today is contained before it can reject `done`, + // but the waiter must not gamble quiescence on that: a future escape + // still counts as settled activity. + /* v8 ignore next 3 -- the catch arm backstops rejection paths that are all currently contained */ while (this.abort !== undefined || this.queued.some(item => item.wakeup)) { await this.done.catch(() => undefined) } @@ -357,9 +360,18 @@ export class ReactLoopAgent implements Agent { if (!this.drainOutbox(turn)) break } } catch (caught: unknown) { - if (this.stepOpen) { - this.stepOpen = false - this.session.append('step/end', { turn, step }) + try { + if (this.stepOpen) { + this.stepOpen = false + this.session.append('step/end', { turn, step }) + } + } catch (closeError: unknown) { + // Contained like the finally's turn close: a persistently rejecting + // step boundary must not escape run(), or the post-finally tail would + // never publish the terminal status and observers would see a + // permanently running agent whose whenIdle() already resolved. + this.loopCtx.logger.warn(`agent "${this.id}": closing step ${turn}/${step} failed: ${errorChain(closeError)}`) + emitAgentEvent(this.loopCtx, this, 'agent/error', turn, step, closeError) } ({ reason, idle } = this.settle(turn, step, caught, signal)) } finally { diff --git a/packages/core/agent-loop/tests/coverage-edges.spec.ts b/packages/core/agent-loop/tests/coverage-edges.spec.ts index 344f03edf4..5bd855a27b 100644 --- a/packages/core/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts @@ -390,6 +390,29 @@ describe('post-turn continuation edges', () => { }) }) +describe('persistent step-close rejection', () => { + it('still publishes the terminal status when both step-close attempts are vetoed', async () => { + const adapter = new MockAdapter([textResponse('will not close')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('stepend-double-veto'), { provider: 'mock', model: 'mock' }) + // Persistently reject step/end: the catch's own close attempt fails too, + // and the contained failure must not strand status at running. + 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') + }) + const statuses: string[] = [] + ctx.on('agent/status', (subject, status) => { if (subject === agent) statuses.push(status) }) + + send(agent, 'go') + await agent.whenIdle() + + expect(agent.status).toBe('idle') + expect(statuses).toEqual(['running', 'idle']) + }) +}) + describe('tool result meta persistence', () => { it('records a presentationMeta payload on the tool/result event', async () => { const { defineTool } = await import('@deepseek-ai/dsh-tools') @@ -504,13 +527,12 @@ describe('driver bookkeeping edges', () => { 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') + // A throwing terminal-notification listener rejects the driver promise + // (the run's containment covers only session appends); the waiter's + // catch arm must treat that rejection as quiescence instead of + // propagating it. + ctx.on('agent/idle', (subject) => { + if (subject === agent) throw new Error('idle listener exploded') }) send(agent, 'one') diff --git a/packages/goal/goal-session/src/index.ts b/packages/goal/goal-session/src/index.ts index f6f0404fdc..2705a60209 100644 --- a/packages/goal/goal-session/src/index.ts +++ b/packages/goal/goal-session/src/index.ts @@ -358,6 +358,17 @@ export function apply(ctx: Context): void { state.attempt.turn = event.data.turn } return + case 'retry': + // A recovery policy (llm-retry) closed the round's failed turn + // and reopened its history: the attempt rides the retry turn, + // and the failed turn's provisional reason no longer settles + // the round — the retry's own outcome does. + if (state.attempt !== undefined && state.attempt.reason !== undefined + && state.attempt.reason.kind === 'error') { + state.attempt.turn = event.data.turn + state.attempt.reason = undefined + } + return default: // Injection and merge-extensible plugin triggers cannot admit a queued goal message. return @@ -415,7 +426,21 @@ export function apply(ctx: Context): void { requestDrive(state) return { kind: 'block', reason: STALE_ROUND_REASON } } - const decision = await next() + let decision: PromptDecision + try { + decision = await next() + } catch (error: unknown) { + // A throwing downstream hook drops the whole admission: the loop + // returns to idle without a turn, so a still-queued reservation would + // starve every later drive pass. Clear it and let the driver + // reschedule the round. + const attempt = state.attempt + if (attempt !== undefined && sameRound(source, attempt) && attempt.turn === undefined) { + state.attempt = undefined + requestDrive(state) + } + throw error + } if (decision.kind === 'block') { const attempt = state.attempt if (attempt !== undefined && sameRound(source, attempt)) state.attempt = undefined diff --git a/packages/goal/goal-session/tests/goal-session.spec.ts b/packages/goal/goal-session/tests/goal-session.spec.ts index 9af38b878c..2225bd2629 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -479,6 +479,101 @@ describe('same-session goal driving', () => { expect(test.adapter.requests).toHaveLength(0) }) + it('settles a goal round from its successful retry turn, not the failed original', async () => { + const test = await harness([ + new LlmError('transient', 'SERVER'), + textResponse('retry succeeded'), + ]) + // The llm-retry shape: schedule one retry for the failed goal-round request. + let retried = false + test.ctx.on('agent/request-error', async (subject) => { + if (!retried) { + retried = true + subject.retry() + } + }) + test.ctx.goals.create(test.agent, { objective: 'survive a transient failure', maxGoalRounds: 1 }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') + + // The retry turn's completed outcome settles the round: round-limit, not + // the failed original turn's turn-error. + expect(goal?.blockedReason?.code).toBe('round-limit') + expect(goal?.roundsStarted).toBe(1) + expect(test.adapter.requests).toHaveLength(2) + }) + + it('does not double-clear when a throwing hook already cancelled the round', async () => { + const test = await harness([]) + // The downstream hook cancels (pausing the goal and clearing the queued + // attempt through cancel-requested) and THEN throws: the catch finds no + // matching reservation and must not reschedule a paused goal. + let fired = false + test.ctx.on('agent/prompt-submit', async (agent, _content, source, _signal, next) => { + if (source.kind === 'goal' && !fired) { + fired = true + agent.cancel({ kind: 'user' }) + throw new Error('hook cancelled then exploded') + } + return next() + }) + test.ctx.goals.create(test.agent, { objective: 'cancel then throw' }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused') + await test.agent.whenIdle() + await new Promise((resolve) => { setImmediate(resolve) }) + + expect(goal?.roundsStarted).toBe(0) + expect(test.adapter.requests).toHaveLength(0) + expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'paused' }) + }) + + it('reschedules the round when a downstream admission hook throws', async () => { + const test = await harness([textResponse('second admission succeeded')]) + // Registered after goal-session's own listener: the throw propagates back + // through goal-session's next() await, dropping the whole admission. + let threw = false + test.ctx.on('agent/prompt-submit', async (_agent, _content, source, _signal, next) => { + if (source.kind === 'goal' && !threw) { + threw = true + throw new Error('downstream admission hook exploded') + } + return next() + }) + test.ctx.goals.create(test.agent, { objective: 'survive a throwing hook', maxGoalRounds: 1 }) + + // The cleared reservation lets the driver reschedule; the second + // admission passes and the round completes to its limit. + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') + expect(goal?.blockedReason?.code).toBe('round-limit') + expect(goal?.roundsStarted).toBe(1) + expect(test.adapter.requests).toHaveLength(1) + }) + + it('a retry turn on a non-goal failure leaves the goal reservation untouched', async () => { + const test = await harness([ + new LlmError('transient on human turn', 'SERVER'), + textResponse('human retry succeeded'), + textResponse('goal round ran'), + ]) + let retried = false + test.ctx.on('agent/request-error', async (subject) => { + if (!retried) { + retried = true + subject.retry() + } + }) + // A human prompt fails and retries while a goal is armed but its round + // is not yet reserved: the retry trigger must not adopt or clear + // anything (the attempt is absent), and the goal proceeds normally. + test.ctx.goals.create(test.agent, { objective: 'ignore foreign retries', maxGoalRounds: 1 }) + test.agent.followup({ content: [{ type: 'text', text: 'human work' }], source: { kind: 'user' } }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') + expect(goal?.blockedReason?.code).toBe('round-limit') + expect(goal?.roundsStarted).toBe(1) + }) + it('blocks the goal when a custom agent rejects the otherwise valid follow-up', async () => { const test = await harness([]) // Reject only the goal-sourced round follow-up, not the state-change injection diff --git a/packages/ui/tui/src/index.ts b/packages/ui/tui/src/index.ts index 753c93a6a0..f026022724 100644 --- a/packages/ui/tui/src/index.ts +++ b/packages/ui/tui/src/index.ts @@ -2802,16 +2802,14 @@ export function createTuiChat( // prompt and its attached context together instead of stranding the // snapshot in history for the next unrelated prompt. let cleanedUp = false - // Assigned after followup(); cleanup() can run earlier from the catch. - let detachDiscard: (() => void) | undefined = undefined const cleanup = (): void => { - // Both triggers detach themselves, so a second call needs a future - // third trigger; kept so adding one cannot double-release. + // Each trigger detaches both listeners, so a second call needs a + // future third trigger; kept so adding one cannot double-release. /* v8 ignore next -- unreachable idempotence guard, see above */ if (cleanedUp) return cleanedUp = true detachSubmit() - detachDiscard?.() + detachDiscard() } // Prepended so this wrapper is outermost: it observes the admission // whether a downstream hook allows or blocks, and detaches either way. @@ -2822,7 +2820,14 @@ export function createTuiChat( if (decision.kind !== 'allow') return decision return { ...decision, additionalContexts: [...decision.additionalContexts ?? [], attachedContext] } }, { prepend: true }) - let id: AgentMessageId + // Installed BEFORE followup(): admission runs synchronously inside it on + // the common path, and a listener registered after cleanup() already ran + // would never be released. The id lands before any discard can name it — + // discard is only ever emitted by a later cancel(). + let id: AgentMessageId | undefined + const detachDiscard = ctx.on('agent/inbox/discard', (subject, messages) => { + if (subject === agent && messages.some(message => message.id === id)) cleanup() + }) // followup() accepts any typed input and contains listener failures; // this guards a future synchronous throw so the wrapper cannot leak. /* v8 ignore start -- future-proofing guard, see above */ @@ -2833,10 +2838,6 @@ export function createTuiChat( throw error } /* v8 ignore stop */ - // A prompt discarded before admission (broad cancel) releases the wrapper. - detachDiscard = ctx.on('agent/inbox/discard', (subject, messages) => { - if (subject === agent && messages.some(message => message.id === id)) cleanup() - }) } /** Deliver a user turn to the agent: steer while running, send while idle, or report a disposed agent. */ diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index b002e165c1..fbe4accfdb 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -1923,6 +1923,52 @@ describe('pi-tui chat lifecycle and transcript', () => { await dispose(result) }) + it('releases the reference-admission wrapper on the ordinary allowed path', async () => { + const result = await setup({ + async configureContext(ctx) { + ctx.provide('tools', { get: () => undefined } as never) + await ctx.plugin(TestSessionQueryService) + await ctx.plugin(SessionReferenceService) + const source = ctx.sessions.create(SessionId('leak-source'), { meta: { cwd: process.cwd(), createdAt: 1 } }) + appendUser(source, 'source background') + }, + }) + const send = async (): Promise => { + result.terminal.send('@leak-source') + await vi.waitFor(() => { expect(result.terminal.output).toContain('Session · leak-source') }) + result.terminal.send('\t') + await tick() + result.terminal.send('\r') + } + await send() + await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(1) }) + await send() + await vi.waitFor(() => { expect(result.agent.sent).toHaveLength(2) }) + + // Both wrappers released on the allowed path: a discard for either prompt + // finds no armed listener, and an unrelated admission is untouched. The + // leak regression: a listener installed after its cleanup already ran + // would survive every future cleanup. + result.ctx.emit('agent/inbox/discard', result.agent, [{ + id: AgentMessageId('stub'), content: result.agent.sent[0]!, source: { kind: 'user' }, + }]) + const unrelated = await agentEvents(result.ctx, result.agent).waterfall( + 'agent/prompt-submit', [{ type: 'text', text: 'unrelated' }], { kind: 'user' }, + new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }), + ) + expect(unrelated.kind === 'allow' && unrelated.additionalContexts).toBeUndefined() + // Replaying either sent prompt attaches nothing: the one-shot wrappers + // are gone, not merely spent. + for (const sent of result.agent.sent) { + const replay = await agentEvents(result.ctx, result.agent).waterfall( + 'agent/prompt-submit', sent, { kind: 'user' }, + new AbortController().signal, () => Promise.resolve({ kind: 'allow' as const }), + ) + expect(replay.kind === 'allow' && replay.additionalContexts).toBeUndefined() + } + await dispose(result) + }) + it('discards the reference snapshot with its blocked or cancelled prompt', async () => { const result = await setup({ async configureContext(ctx) {