diff --git a/packages/context/workspace-context/src/state.ts b/packages/context/workspace-context/src/state.ts index 99a5254ed7..7bd093c342 100644 --- a/packages/context/workspace-context/src/state.ts +++ b/packages/context/workspace-context/src/state.ts @@ -117,8 +117,7 @@ function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null && !Array.isArray(value) } -function workspaceInstructionChanges(source: unknown): WorkspaceInstructionChange[] { - if (!isWorkspaceContextSource(source)) return [] +function workspaceInstructionChanges(source: { changes: unknown[] }): WorkspaceInstructionChange[] { const changes: WorkspaceInstructionChange[] = [] for (const value of source.changes) { if (!isRecord(value)) continue diff --git a/packages/context/workspace-context/tests/workspace-context.spec.ts b/packages/context/workspace-context/tests/workspace-context.spec.ts index 1e994500f6..a5fb046038 100644 --- a/packages/context/workspace-context/tests/workspace-context.spec.ts +++ b/packages/context/workspace-context/tests/workspace-context.spec.ts @@ -3260,6 +3260,29 @@ describe('workspace context pending state', () => { expect(versions.has(agent.session)).toBe(false) }) + it('keeps an unrelated scope\'s version fast path when a step-close discard empties only its own scope', () => { + const agent = stubAgent('/') + const pending = new WeakMap>() + const versions: InstructionVersionCache = new WeakMap() + agent.session.append('step/start', { turn: 1, step: 1 }) + commitPendingInstructionContexts(agent, [workspaceChangeContext('pkg', 'one')], pending) + versions.set(agent.session, new Map([ + ['pkg', { + path: join('pkg', 'AGENTS.md'), version: FsVersion('v1'), digest: 'one', trimmedDigest: 'one', + }], + ['other', { + path: join('other', 'AGENTS.md'), version: FsVersion('v2'), digest: 'two', trimmedDigest: 'two', + }], + ])) + + const ended = agent.session.append('step/end', { turn: 1, step: 1 }) + observeInstructionSessionEvent(agent.session, ended, pending, versions) + + expect(pending.has(agent.session)).toBe(false) + expect(versions.get(agent.session)?.has('pkg')).toBe(false) + expect(versions.get(agent.session)?.has('other')).toBe(true) + }) + it('rolls back only the exact current transition and releases empty session state', () => { const agent = stubAgent('/') const pending = new WeakMap>() @@ -3270,6 +3293,13 @@ describe('workspace context pending state', () => { expect(commitPendingInstructionContexts(agent, [{ content: [], source: { kind: 'plugin', plugin: 'workspace-context' }, }], pending)).toEqual([]) + // A workspace-instructions source whose change list filters to nothing + // must not mint per-session pending state. + expect(commitPendingInstructionContexts(agent, [{ + content: [], + source: { kind: 'workspace-instructions', changes: [] }, + }], pending)).toEqual([]) + expect(pending.has(agent.session)).toBe(false) const committed = commitPendingInstructionContexts(agent, [ workspaceChangeContext('first', 'one'), 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 00e9ded5cc..57e47deaa1 100644 --- a/packages/core/agent-loop/tests/config-session-id.spec.ts +++ b/packages/core/agent-loop/tests/config-session-id.spec.ts @@ -441,3 +441,36 @@ describe('config-driven session id', () => { await ctx.fiber.dispose() }) }) + +describe('startup reporting after factory teardown', () => { + it('suppresses the configured-restore failure report once the loop is disposed', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-disposed-report-')) + dirs.push(root) + const ctx = await makeCoreContext() + await ctx.plugin(SessionPersistenceJsonl, { root }) + ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')])) + + // A restore lookup that hangs until after the loop is gone: the eventual + // failure lands with ownership inactive and must be silently dropped. + const gate = Promise.withResolvers() + // The teardown path may drop the pending lookup without awaiting it. + gate.promise.catch(() => undefined) + vi.spyOn(ctx.sessionPersistence, 'list').mockReturnValue(gate.promise) + const failures: unknown[] = [] + ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) }) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + + const loop = await ctx.plugin(AgentLoop, { + agents: [{ id: 'main', sessionId: SessionId('config-disposed-report'), model: 'mock' }], + }) + const disposal = loop.dispose() + gate.reject(new Error('backend failed after teardown began')) + await disposal + + await new Promise(r => setTimeout(r, 20)) + expect(failures).toEqual([]) + expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('config-driven restore')) + warn.mockRestore() + await ctx.fiber.dispose() + }) +}) diff --git a/packages/core/agent-loop/tests/coverage-edges.spec.ts b/packages/core/agent-loop/tests/coverage-edges.spec.ts index 427005d907..c7dbbd5c4a 100644 --- a/packages/core/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/core/agent-loop/tests/coverage-edges.spec.ts @@ -254,3 +254,247 @@ describe('structured tool error propagation (the runtime-validation Agent Note, .toEqual({ name: 'HarnessError', code: 'BOOM' }) }) }) + +describe('retry() edges', () => { + it('throws while a turn runs with no request-error window open', async () => { + const adapter = new MockAdapter(['hang']) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('retry-busy'), { provider: 'mock', model: 'mock' }) + + send(agent, 'go') + // Wait until the hung request is in flight (the run owns this.abort). + await new Promise(r => setTimeout(r, 30)) + expect(() => { agent.retry() }).toThrow('cannot retry while busy') + agent.cancel({ kind: 'user' }) + await agent.whenIdle() + }) + + it('ignores a retry request arriving after the recovery window was aborted', async () => { + const { LlmError } = await import('@deepseek-ai/dsh-llm') + const adapter = new MockAdapter([ + () => { throw new LlmError('busy', 'RATE_LIMIT') }, + textResponse('never used'), + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('retry-after-cancel'), { provider: 'mock', model: 'mock' }) + ctx.on('agent/request-error', async (subject) => { + // Cancellation lands first; the window survives structurally but its + // signal is aborted, so the request must not arm a retry turn. + subject.cancel({ kind: 'user' }) + subject.retry() + }) + + send(agent, 'go') + await agent.whenIdle() + + // One failed request, no retry turn. + expect(adapter.requests).toHaveLength(1) + const ends = agent.session.events.filter(e => e.type === 'turn/end') + expect(ends).toHaveLength(1) + }) + + it('completed recovery does not retry when cancellation raced the waterfall', async () => { + const { LlmError } = await import('@deepseek-ai/dsh-llm') + const adapter = new MockAdapter([ + () => { throw new LlmError('busy', 'RATE_LIMIT') }, + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('retry-raced'), { provider: 'mock', model: 'mock' }) + ctx.on('agent/request-error', async (subject, _turn, _step, _error, _failure, signal, next) => { + await next() + // Recovery completes and requested the retry, but the turn signal + // aborts before the loop reads the window. + subject.retry() + subject.cancel({ kind: 'user' }) + expect(signal.aborted).toBe(true) + }) + + send(agent, 'go') + await agent.whenIdle() + + expect(adapter.requests).toHaveLength(1) + const end = agent.session.events.findLast(e => e.type === 'turn/end') + expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('aborted') + }) +}) + +describe('stream failure edges', () => { + it('rethrows a mid-stream throw that carries no adapter failure facts', async () => { + const adapter = new MockAdapter([textResponse('will be vetoed')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('stream-no-facts'), { provider: 'mock', model: 'mock' }) + let recoveries = 0 + ctx.on('agent/request-error', async () => { recoveries += 1 }) + // A pre-commit chunk veto throws INSIDE the stream-consumption try, but it + // is not an adapter-boundary failure, so llmFailureOf yields no facts. + let vetoed = false + ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const event = args[1] as SessionEvent + if (event.type === 'assistant/chunk' && !vetoed) { + vetoed = true + throw new Error('reject the first chunk') + } + }) + + send(agent, 'go') + await agent.whenIdle() + + // No facts -> not offered to recovery; the turn fails through settle(). + expect(recoveries).toBe(0) + const end = agent.session.events.findLast(e => e.type === 'turn/end') + expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error') + }) +}) + +describe('post-turn continuation edges', () => { + it('an agent/idle listener that enqueues a waking prompt preempts continueOrIdle', async () => { + const adapter = new MockAdapter([textResponse('one'), textResponse('two')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('idle-preempt'), { provider: 'mock', model: 'mock' }) + let injected = false + ctx.on('agent/idle', (subject) => { + if (subject !== agent || injected) return + injected = true + // kick() installs the next admission synchronously, so the following + // continueOrIdle() sees an abort owner and yields to it. + send(agent, 'follow-up from idle listener') + }) + + send(agent, 'go') + await agent.whenIdle() + + expect(adapter.requests).toHaveLength(2) + const starts = agent.session.events.filter(e => e.type === 'turn/start') + expect(starts).toHaveLength(2) + // The busy interval never broke between the turns: one running->idle cycle. + }) + + it('whenIdle resolves for a waiter whose awaited run fails', async () => { + const adapter = new MockAdapter([textResponse('unused')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('whenidle-reject'), { provider: 'mock', model: 'mock' }) + let rejected = false + ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const event = args[1] as SessionEvent + if (event.type === 'turn/start' && !rejected) { + rejected = true + throw new Error('veto turn start while a waiter is pending') + } + }) + + send(agent, 'go') + await expect(agent.whenIdle()).resolves.toBeUndefined() + expect(agent.status).toBe('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') + const adapter = new MockAdapter([ + toolCallResponse('c1', 'meta-tool', {}), + textResponse('done'), + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('tool-meta'), { provider: 'mock', model: 'mock' }) + ctx.tools.register(defineTool({ + name: 'meta-tool', + description: 'carries presentation meta', + parameters: {}, + output: { + schema: { type: 'string' }, + render: (_args, value) => [{ type: 'text', text: value }], + presentationMeta: () => ({ presentation: 'diff-card' }), + }, + async execute() { + return 'ran' + }, + })) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const result = agent.session.events.find(e => e.type === 'tool/result') + expect(result?.type === 'tool/result' && result.data.meta).toEqual({ presentation: 'diff-card' }) + }) +}) + +describe('turn close failure containment', () => { + it('a rejected turn/end append is contained: warn + agent/error, no retry', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('turnend-veto'), { provider: 'mock', model: 'mock' }) + let vetoed = false + ctx.on('internal/dispatch', (_mode, name, args) => { + if (name !== 'session/event') return + const event = args[1] as SessionEvent + if (event.type === 'turn/end' && !vetoed) { + vetoed = true + throw new Error('reject turn end') + } + }) + const errors: unknown[] = [] + ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) }) + + send(agent, 'go') + await agent.whenIdle() + + // The close failure is reported live; the machine still reaches idle. + expect(errors.map(e => e instanceof Error && e.message)).toContain('reject turn end') + expect(agent.status).toBe('idle') + expect(adapter.requests).toHaveLength(1) + }) +}) + +describe('recovery without a retry request', () => { + it('a completed recovery that never calls retry() leaves the failed turn terminal', async () => { + const { LlmError } = await import('@deepseek-ai/dsh-llm') + const adapter = new MockAdapter([ + () => { throw new LlmError('down', 'SERVICE_UNAVAILABLE') }, + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('recovery-no-retry'), { provider: 'mock', model: 'mock' }) + let recoveries = 0 + ctx.on('agent/request-error', async () => { recoveries += 1 }) + + send(agent, 'go') + await agent.whenIdle() + + expect(recoveries).toBe(1) + expect(adapter.requests).toHaveLength(1) + const end = agent.session.events.findLast(e => e.type === 'turn/end') + expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error') + }) +}) + +describe('unrenderable failure settlement', () => { + it('drops the rendered message when the error chain cannot be rendered', async () => { + const { LlmError } = await import('@deepseek-ai/dsh-llm') + const adapter = new MockAdapter([ + () => { + const error = new LlmError('will become hostile', 'SERVER') + // A hostile message getter makes errorChain collapse to its sentinel; + // settle() must then fall back to the failure facts alone. + Object.defineProperty(error, 'message', { + get() { throw new Error('hostile accessor') }, + }) + throw error + }, + ]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create(SessionId('unrenderable'), { provider: 'mock', model: 'mock' }) + + send(agent, 'go') + await agent.whenIdle() + + const end = agent.session.events.findLast(e => e.type === 'turn/end') + expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error') + if (end?.type === 'turn/end' && end.data.reason.kind === 'error') { + // The durable failure keeps the adapter facts' message, not the + // unrenderable chain. + expect(end.data.reason.failure?.message).not.toBe('') + } + }) +}) diff --git a/packages/core/agent-loop/tests/resume.spec.ts b/packages/core/agent-loop/tests/resume.spec.ts index 8a8d604773..c2261d8d89 100644 --- a/packages/core/agent-loop/tests/resume.spec.ts +++ b/packages/core/agent-loop/tests/resume.spec.ts @@ -555,3 +555,97 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume', await ctx.fiber.dispose() }) }) + +describe('creation and resume cancellation edges', () => { + it('rejects create() with a pre-aborted signal, including a non-Error reason', async () => { + const { ctx } = await persistentHarness(new MockAdapter([])) + + const errorReason = new AbortController() + errorReason.abort(new Error('caller gave up')) + await expect(promptly(ctx.agents.create({ + sessionId: SessionId('pre-aborted-error'), + agentOptions: { provider: 'mock', model: 'mock' }, + signal: errorReason.signal, + }))).rejects.toThrow('caller gave up') + + // A non-Error reason is wrapped into the creation-aborted error. + const stringReason = new AbortController() + stringReason.abort('operator string reason') + await expect(promptly(ctx.agents.create({ + sessionId: SessionId('pre-aborted-string'), + agentOptions: { provider: 'mock', model: 'mock' }, + signal: stringReason.signal, + }))).rejects.toThrow(/creation aborted/) + + expect(ctx.agents.get(SessionId('pre-aborted-error'))).toBeUndefined() + expect(ctx.agents.get(SessionId('pre-aborted-string'))).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('a non-Error abort reason arriving during setup is wrapped for the caller', async () => { + const { ctx } = await persistentHarness(new MockAdapter([])) + const controller = new AbortController() + const setupEntered = Promise.withResolvers() + const setupGate = Promise.withResolvers() + + const creating = ctx.agents.create({ + sessionId: SessionId('setup-string-abort'), + agentOptions: { provider: 'mock', model: 'mock' }, + signal: controller.signal, + async setup() { + setupEntered.resolve(undefined) + await setupGate.promise + }, + }) + await setupEntered.promise + controller.abort('mid-setup string reason') + setupGate.resolve(undefined) + + await expect(promptly(creating)).rejects.toThrow(/creation aborted/) + expect(ctx.agents.get(SessionId('setup-string-abort'))).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('resume with a pre-aborted caller signal rejects out of the load race', async () => { + const sessionId = SessionId('resume-pre-aborted') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + const controller = new AbortController() + controller.abort(new Error('resume abandoned')) + + await expect(promptly(ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + signal: controller.signal, + }))).rejects.toThrow('resume abandoned') + + expect(ctx.agents.get(sessionId)).toBeUndefined() + await ctx.fiber.dispose() + }) + + it('factory teardown during a hung resume load rejects with loop-inactive', async () => { + const sessionId = SessionId('resume-loop-teardown') + const root = await persistSession(sessionId) + const ctx = await mountPersistentHarness(root, new MockAdapter([])) + const snapshot = await ctx.sessionPersistence.load(sessionId) + const gate = Promise.withResolvers() + const loadStarted = Promise.withResolvers() + ctx.sessionPersistence.load = () => { + loadStarted.resolve(undefined) + return gate.promise + } + + const resuming = ctx.agents.resume({ + resumeSessionId: sessionId, + agentOptions: { provider: 'mock', model: 'mock' }, + }) + await loadStarted.promise + // Resolve the load only after teardown began: the post-load ownership + // check, not the abort race, must reject the wrapper. + const rejection = expect(promptly(resuming)).rejects.toThrow() + const disposal = ctx.fiber.dispose() + gate.resolve(structuredClone(snapshot)) + await rejection + await disposal + }) +}) diff --git a/packages/examples/cli-demo/src/cli.ts b/packages/examples/cli-demo/src/cli.ts index 34d578b259..2149e1d091 100644 --- a/packages/examples/cli-demo/src/cli.ts +++ b/packages/examples/cli-demo/src/cli.ts @@ -241,6 +241,10 @@ export async function runOneShot(ctx: Context, options: OneShotOptions): Promise resolveTurn() } const settleRejected = (error: Error): void => { + // The once-registered abort listener is the only rejecter, and a settled + // prompt makes targetTurn defined so onAbort skips rejection entirely; + // kept for symmetry with settleResolved. + /* v8 ignore next -- unreachable second settlement, see above */ if (firstTurnEnded) return firstTurnEnded = true rejectTurn(error) diff --git a/packages/goal/goal-session/tests/goal-session.spec.ts b/packages/goal/goal-session/tests/goal-session.spec.ts index 9e1e82fc4a..2b48be61d4 100644 --- a/packages/goal/goal-session/tests/goal-session.spec.ts +++ b/packages/goal/goal-session/tests/goal-session.spec.ts @@ -1,6 +1,6 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' -import type { Agent } from '@deepseek-ai/dsh-agent' +import type { Agent, PromptDecision } from '@deepseek-ai/dsh-agent' import { agentEvents } from '@deepseek-ai/dsh-agent' import AgentLoop from '@deepseek-ai/dsh-agent-loop' import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit' @@ -484,6 +484,22 @@ describe('same-session goal driving', () => { expect(test.adapter.requests).toHaveLength(0) }) + it('contains a mutation failure inside the scheduler loop and fails closed', async () => { + const test = await harness([textResponse('the only round')]) + // The only ctx.goals.block call in a completing one-round run is the + // driver's round-limit stop, so the mock fails exactly that drive pass. + vi.spyOn(test.ctx.goals, 'block').mockImplementationOnce(() => { + throw new Error('round-limit block failed') + }) + test.ctx.goals.create(test.agent, { objective: 'contain a driver failure', maxGoalRounds: 1 }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed') + + expect(goal).toMatchObject({ phase: 'active', roundsStarted: 1 }) + expect(goal?.blockedReason).toBeUndefined() + expect(test.adapter.requests).toHaveLength(1) + }) + it('contains synchronous scheduler startup failure', async () => { const test = await harness([]) vi.spyOn(test.ctx.agents, 'withoutInitiator').mockImplementationOnce(() => { @@ -674,6 +690,152 @@ describe('same-session goal driving', () => { expect(test.adapter.requests).toHaveLength(1) }) + it('leaves a queued reservation pending when the driver runs before its turn settles', async () => { + const test = await harness([textResponse('settled later')]) + let woken = false + test.ctx.on('agent/prompt-submit', async (_agent, _content, source, _signal, next) => { + if (source.kind === 'goal' && !woken) { + woken = true + // A concurrent driver pass must observe the still-unsettled attempt + // and yield rather than double-book or clear the reservation. + agentEvents(test.ctx, test.agent).emit('agent/status', 'idle') + await new Promise((resolve) => { setImmediate(resolve) }) + } + return next() + }) + test.ctx.goals.create(test.agent, { objective: 'wake early', maxGoalRounds: 1 }) + + 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('disarms instead of continuing when a plugin reports a post-turn persistence failure', async () => { + const test = await harness([textResponse('round one')]) + test.ctx.on('session/event', (session, event) => { + if (session === test.agent.session && event.type === 'turn/end') { + agentEvents(test.ctx, test.agent).emit('agent/error', event.data.turn, 1, new Error('post-turn flush failed')) + } + }) + test.ctx.goals.create(test.agent, { objective: 'stop when durability is lost', maxGoalRounds: 8 }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed') + await test.agent.whenIdle() + + expect(goal).toMatchObject({ phase: 'active', roundsStarted: 1 }) + expect(test.adapter.requests).toHaveLength(1) + }) + + it('ignores a post-turn failure reported for a retired agent', async () => { + const test = await harness([textResponse('ordinary work')]) + const handle = await test.ctx.agents.create({ + sessionId: SessionId('goal-session-retired'), + agentOptions: { provider: 'mock', model: 'mock' }, + }) + handle.agent.followup({ content: [{ type: 'text', text: 'one ordinary turn' }], source: { kind: 'user' } }) + await handle.agent.whenIdle() + const closed = handle.agent.session.events.findLast(event => event.type === 'turn/end') + if (closed?.type !== 'turn/end') throw new Error('expected a closed turn') + await handle.dispose() + const warn = vi.spyOn(test.ctx.logger, 'warn') + + agentEvents(test.ctx, handle.agent).emit('agent/error', closed.data.turn, 1, new Error('late flush failure')) + + expect(test.ctx.agents.get(handle.agent.id)).toBeUndefined() + expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('goal-session')) + }) + + it('ignores the failed outcome of a round made stale by human work queued at turn start', async () => { + const test = await harness([new Error('round one broke'), textResponse('human answer')]) + let queued = false + test.ctx.on('session/event', (session, event) => { + if (session !== test.agent.session || queued) return + if (event.type === 'turn/start' && event.data.trigger.kind === 'message' + && event.data.trigger.source.kind === 'goal') { + queued = true + test.agent.followup({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } }) + } + }) + test.ctx.goals.create(test.agent, { objective: 'survive a stale failure', maxGoalRounds: 1 }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked') + + // The stale round's turn-error never blocks the goal; only the durable + // round budget does, after the interleaved human turn ran. + expect(goal?.blockedReason?.code).toBe('round-limit') + expect(test.adapter.requests).toHaveLength(2) + expect(requestText(test.adapter.requests[1]!)).toContain('human interleaved') + }) + + it('waits for work queued by a pause observer before considering the next round', async () => { + const test = await harness(['hang', textResponse('inspection answer')]) + test.ctx.on('goal/changed', (agent, change) => { + if (agent === test.agent && change.operation === 'pause') { + agent.followup({ content: [{ type: 'text', text: 'inspect the pause' }], source: { kind: 'user' } }) + } + }) + test.ctx.goals.create(test.agent, { objective: 'pause then inspect' }) + await waitForRequests(test.adapter, 1) + + test.agent.cancel({ kind: 'user' }) + await waitForRequests(test.adapter, 2) + await test.agent.whenIdle() + + expect(test.ctx.goals.get(test.agent)).toMatchObject({ + phase: 'paused', + roundsStarted: 1, + activation: 'disarmed', + }) + expect(requestText(test.adapter.requests[1]!)).toContain('inspect the pause') + }) + + it('does not re-block a goal the downstream veto already saw cancelled', async () => { + const test = await harness([]) + let vetoed = false + test.ctx.on('agent/prompt-submit', (agent, _content, source, _signal, next) => { + if (source.kind === 'goal' && !vetoed) { + vetoed = true + agent.cancel({ kind: 'user' }) + return Promise.resolve({ kind: 'block', reason: 'cancelled by policy' }) + } + return next() + }) + test.ctx.goals.create(test.agent, { objective: 'veto after cancellation' }) + + const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused') + await test.agent.whenIdle() + + // Cancellation already cleared the reservation and paused the goal, so the + // veto neither touches an absent attempt nor blocks the paused goal. + expect(goal).toMatchObject({ roundsStarted: 0, activation: 'disarmed' }) + expect(goal?.blockedReason).toBeUndefined() + expect(test.adapter.requests).toHaveLength(0) + }) + + it('awaits an unadmitted reservation stuck in admission during teardown without cancelling', async () => { + const test = await harness([]) + let release: (() => void) | undefined + test.ctx.on('agent/prompt-submit', async (_agent, _content, source, _signal, next) => { + if (source.kind === 'goal' && release === undefined) { + await new Promise((resolve) => { release = resolve }) + } + return next() + }) + test.ctx.goals.create(test.agent, { objective: 'unload during admission' }) + await vi.waitFor(() => { expect(release).toBeDefined() }) + + const disposal = Promise.resolve(test.driver.dispose()) + await waitForGoal(test.ctx, test.agent, goal => goal?.activation === 'disarmed') + release?.() + await disposal + + expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'active', roundsStarted: 0 }) + expect(test.adapter.requests).toHaveLength(0) + expect(test.agent.session.events.some(event => event.type === 'turn/start')).toBe(false) + }) + it('ignores session events without an exact owning agent and retires disposed agent state', async () => { const test = await harness([]) const orphan = test.ctx.sessions.create(SessionId('goal-session-orphan')) diff --git a/packages/goal/goal/src/fold.ts b/packages/goal/goal/src/fold.ts index c9dc578ce4..cfb8994748 100644 --- a/packages/goal/goal/src/fold.ts +++ b/packages/goal/goal/src/fold.ts @@ -358,6 +358,9 @@ export function applyGoalEvent(state: GoalFoldState, event: SessionEvent): GoalC if (source === undefined) return undefined // A goal-sourced message without a change must be a positive-round // admitted continuation prompt; round zero owes a durable source change. + /* v8 ignore next 3 -- decodeGoalEvent returns the change or fails loud for every + round-zero goal source, so only positive rounds reach here; the guard keeps + replay fail-loud against a decoder change */ if (source.round === 0) { throw new Error(`goal source at session event ${event.seq} lacks goal change data`) } diff --git a/packages/goal/goal/tests/goal.spec.ts b/packages/goal/goal/tests/goal.spec.ts index d0f0264068..44adca93e0 100644 --- a/packages/goal/goal/tests/goal.spec.ts +++ b/packages/goal/goal/tests/goal.spec.ts @@ -821,6 +821,12 @@ describe('goal replay validation', () => { expect(() => foldGoal(oneChange(change, { source: { kind: 'goal', goalId: change.goal.id, revision: 1, round: -1 }, }))).toThrow('source is invalid') + expect(() => foldGoal(oneChange(change, { + source: { kind: 'goal', goalId: GoalId('goal-imposter'), revision: 1, round: 0, change }, + }))).toThrow('mismatched source attribution') + expect(() => foldGoal(oneChange(change, { + source: { kind: 'goal', goalId: change.goal.id, revision: 2, round: 0, change }, + }))).toThrow('mismatched source attribution') expect(() => foldGoal(oneChange(change, { content: [{ type: 'text', text: 'wrong' }] }))).toThrow('model-visible content') }) diff --git a/packages/host/runtime/tests/api-proxy-cold.spec.ts b/packages/host/runtime/tests/api-proxy-cold.spec.ts index 3d4ba8e15a..20b806334a 100644 --- a/packages/host/runtime/tests/api-proxy-cold.spec.ts +++ b/packages/host/runtime/tests/api-proxy-cold.spec.ts @@ -1,8 +1,9 @@ /** * Cold-session and degenerate-composition paths of the host ApiProxy: * sessions.list merging persisted-but-unattached summaries (mtime source, - * createdAt fallbacks, lineage projection) and the resume error split when - * the composition has no persistence gate and no agent factory. + * createdAt fallbacks, lineage projection), the resume error split when + * the composition has no persistence gate and no agent factory, and the + * agent-busy mapping of a synchronous prompt rejection. */ import { mkdtempSync, writeFileSync, utimesSync } from 'node:fs' @@ -12,6 +13,7 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' import SessionStore from '@deepseek-ai/dsh-session' import AgentRegistry from '@deepseek-ai/dsh-agent' +import type { Agent } from '@deepseek-ai/dsh-agent' import UserInteractionService from '@deepseek-ai/dsh-user-interaction' import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session' import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc' @@ -95,3 +97,38 @@ describe('degenerate composition (no persistence, no factory)', () => { } }) }) + +describe('sessions.prompt synchronous rejection', () => { + it('maps a synchronous send throw (disposed/invalid input) to agent-busy with the reason attached', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + await ctx.plugin(AgentRegistry) + await ctx.plugin(UserInteractionService) + const session = ctx.sessions.create(sid('session-throwing')) + // A live structural stub whose delivery verbs throw synchronously, the + // shape a disposed loop presents at this seam. + ctx.agents.register({ + id: session.id, + session, + status: 'idle', + ctx, + followup: () => { throw new Error('agent "session-throwing" lifecycle disposed') }, + steer: () => { throw new Error('agent "session-throwing" lifecycle disposed') }, + } as unknown as Agent) + const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp' }) + + for (const mode of ['queue', 'steer'] as const) { + const response = await api.sessions.prompt(request({ + sessionId: session.id, mode, content: [{ type: 'text' as const, text: 'x' }], + })) + expect(response.result.ok).toBe(false) + if (!response.result.ok) { + expect(response.result.error.code).toBe('agent-busy') + expect(response.result.error.message).toBe('prompt rejected') + expect(response.result.error.details).toEqual({ + reason: 'Error: agent "session-throwing" lifecycle disposed', + }) + } + } + }) +}) diff --git a/packages/llm/llm-retry/tests/invariant.spec.ts b/packages/llm/llm-retry/tests/invariant.spec.ts index f04f8f36a2..1a9b3f7446 100644 --- a/packages/llm/llm-retry/tests/invariant.spec.ts +++ b/packages/llm/llm-retry/tests/invariant.spec.ts @@ -63,6 +63,50 @@ describe('llm-retry invariants', () => { }).toThrow(message) }) + it('rejects a retry record appended after its turn already closed', async () => { + const ctx = await setup() + const closed = closeStep(ctx, 'retry-invariant-closed-turn') + closed.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } }) + expect(() => { + closed.append('llm/retry', { + turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure, + }) + }).toThrow(/inside an open turn/) + }) + + it('starts a fresh chain when the turn before a retry trigger did not fail structurally', async () => { + const ctx = await setup() + const session = closeStep(ctx, 'retry-invariant-completed-predecessor') + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } }) + session.append('step/start', { turn: 2, step: 1 }) + session.append('step/end', { turn: 2, step: 1 }) + expect(() => { + session.append('llm/retry', { + turn: 2, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure, + }) + }).not.toThrow() + }) + + it('walks the chain across non-boundary events and stops at an unmatched turn start', async () => { + const ctx = await setup() + // The failed predecessor's turn/start is outside this log prefix (e.g. a + // truncated replay): the chain walk must stop rather than loop or throw. + const session = ctx.sessions.create(SessionId('retry-invariant-unmatched-start')) + session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } }) + // A durable non-boundary record between the turns exercises the walk over + // non-turn/end events. + session.append('todo/write', { todos: [] }) + session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } }) + session.append('step/start', { turn: 2, step: 1 }) + session.append('step/end', { turn: 2, step: 1 }) + expect(() => { + session.append('llm/retry', { + turn: 2, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure, + }) + }).not.toThrow() + }) + it('requires an open turn and its latest closed step', async () => { const ctx = await setup() const absent = ctx.sessions.create(SessionId('retry-invariant-no-turn')) @@ -158,4 +202,18 @@ describe('llm-retry invariants', () => { await ctx.plugin(InvariantService) await expect(ctx.plugin(RetryInvariant)).rejects.toThrow(/inside an open turn/) }) + + it('accepts a valid mixed pre-existing history on late registration', async () => { + const ctx = new Context() + await ctx.plugin(SessionStore) + const session = ctx.sessions.create(SessionId('retry-invariant-late-valid')) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('llm/retry', { + turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure, + }) + await ctx.plugin(InvariantService) + await expect(ctx.plugin(RetryInvariant)).resolves.toBeDefined() + }) }) diff --git a/packages/llm/llm-retry/tests/retry.spec.ts b/packages/llm/llm-retry/tests/retry.spec.ts index d037b62e20..aafb61ad55 100644 --- a/packages/llm/llm-retry/tests/retry.spec.ts +++ b/packages/llm/llm-retry/tests/retry.spec.ts @@ -117,6 +117,23 @@ afterEach(async () => { context = undefined }) +describe('config validation', () => { + it.each([ + [{ maxTransientRetries: 1.5 }, /maxTransientRetries must be a non-negative integer/], + [{ maxTransientRetries: -1 }, /maxTransientRetries must be a non-negative integer/], + [{ initialDelayMs: 0 }, /initialDelayMs must be a positive finite number/], + [{ initialDelayMs: Number.NaN }, /initialDelayMs must be a positive finite number/], + [{ maxDelayMs: 0 }, /maxDelayMs must be a positive finite number/], + [{ initialDelayMs: 600, maxDelayMs: 500 }, /initialDelayMs must be less than or equal to maxDelayMs/], + [{ jitterRatio: Number.NaN }, /jitterRatio must be between 0 and 1/], + [{ retryableCodes: [] }, /retryableCodes must not be empty/], + [{ retryableCodes: ['SERVER', ''] }, /retryableCodes must contain only non-empty strings/], + [{ retryableCodes: ['SERVER', 'SERVER'] }, /retryableCodes must not contain duplicates/], + ] satisfies [retry.Config, RegExp][])('rejects invalid config %j at load', (config, message) => { + expect(() => { retry.apply(new Context(), config) }).toThrow(message) + }) +}) + describe('bounded transient retry policy', () => { it('records the scheduled delay before opening a fresh request attempt', async () => { vi.useFakeTimers() @@ -375,6 +392,125 @@ describe('bounded transient retry policy', () => { expect(vi.getTimerCount()).toBe(0) }) + it('keeps the consumed budget when an unowned session logs an assistant message', async () => { + vi.useFakeTimers() + const adapter = new ScriptedAdapter([ + new LlmError('busy one', 'SERVER'), + new LlmError('busy two', 'SERVER'), + ]) + ;({ ctx: context } = await harness(adapter, { maxTransientRetries: 1 })) + const agent = context.agentLoop.create(SessionId('retry-foreign-session'), { provider: 'mock', model: 'mock' }) + const scheduled = waitForRetry(context, agent, 1) + + agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }) + await scheduled + + // A session no agent owns completes a response; the agent's consecutive- + // failure sequence must not reset from that foreign success. + const foreign = context.sessions.create(SessionId('retry-foreign-session-other')) + foreign.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + foreign.append('step/start', { turn: 1, step: 1 }) + foreign.append('assistant/message', { + turn: 1, + step: 1, + content: [{ type: 'text', text: 'foreign' }], + provenance: { provider: 'mock', model: 'mock' }, + }, { surfaceOp: 'append' }) + + const idle = waitForIdle(context, agent) + await vi.advanceTimersByTimeAsync(500) + await idle + + expect(adapter.requests).toHaveLength(2) + expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1) + expect(agent.session.events.at(-1)).toMatchObject({ + type: 'turn/end', + data: { reason: { kind: 'error', failure: { message: 'busy two', code: 'SERVER' } } }, + }) + }) + + it('drops a scheduled retry when cancellation lands between its durable record and its wait', async () => { + vi.useFakeTimers() + const adapter = new ScriptedAdapter([ + new LlmError('busy', 'SERVER'), + textResponse('must not run'), + ]) + ;({ ctx: context } = await harness(adapter)) + const agent = context.agentLoop.create(SessionId('retry-cancel-at-record'), { provider: 'mock', model: 'mock' }) + // The durable record commits synchronously before the cancellable wait; a + // user cancel observed at that exact point must skip the wait entirely. + const dispose = context.on('session/event', (session, event) => { + if (session === agent.session && event.type === 'llm/retry') { + dispose() + agent.cancel({ kind: 'user' }) + } + }) + const idle = waitForIdle(context, agent) + + agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }) + await idle + await vi.advanceTimersByTimeAsync(60_000) + + expect(adapter.requests).toHaveLength(1) + expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1) + expect(vi.getTimerCount()).toBe(0) + }) + + it('schedules nothing when an upstream recovery listener already cancelled the turn', async () => { + vi.useFakeTimers() + const adapter = new ScriptedAdapter([ + new LlmError('busy', 'SERVER'), + textResponse('must not run'), + ]) + ;({ ctx: context } = await harness(adapter, {}, (ctx) => { + // Registered before the retry plugin, so it wraps the policy: it cancels + // the turn, then delegates into a policy that sees an aborted signal. + ctx.on('agent/request-error', (agent, _turn, _step, _error, _failure, _signal, next) => { + agent.cancel({ kind: 'user' }) + return next() + }) + })) + const agent = context.agentLoop.create(SessionId('retry-upstream-cancel'), { provider: 'mock', model: 'mock' }) + const idle = waitForIdle(context, agent) + + agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }) + await idle + await vi.advanceTimersByTimeAsync(60_000) + + expect(adapter.requests).toHaveLength(1) + expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false) + expect(vi.getTimerCount()).toBe(0) + }) + + it('does nothing when its captured listener resumes after plugin disposal', async () => { + vi.useFakeTimers() + const adapter = new ScriptedAdapter([ + new LlmError('busy', 'SERVER'), + textResponse('must not run'), + ]) + const holder: { dispose?: () => Promise } = {} + const mounted = await harness(adapter, {}, (ctx) => { + // An upstream listener captured in the same waterfall disposes the retry + // plugin before delegating; the stale downstream callback must bail. + ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _signal, next) => { + await holder.dispose?.() + return next() + }) + }) + context = mounted.ctx + holder.dispose = () => mounted.retryFiber.dispose() + const agent = context.agentLoop.create(SessionId('retry-stale-listener'), { provider: 'mock', model: 'mock' }) + const idle = waitForIdle(context, agent) + + agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } }) + await idle + await vi.advanceTimersByTimeAsync(60_000) + + expect(adapter.requests).toHaveLength(1) + expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false) + expect(vi.getTimerCount()).toBe(0) + }) + it('aborts and drains a captured backoff before plugin disposal completes', async () => { vi.useFakeTimers() const adapter = new ScriptedAdapter([ diff --git a/packages/support/acp-snapshot/tests/normalize.spec.ts b/packages/support/acp-snapshot/tests/normalize.spec.ts index 37d23411e6..9489c79244 100644 --- a/packages/support/acp-snapshot/tests/normalize.spec.ts +++ b/packages/support/acp-snapshot/tests/normalize.spec.ts @@ -1,6 +1,7 @@ import { describe, expect, it } from 'vitest' import { type NormalizeContext, + extractSnapshotSpillPaths, normalizeSessionLog, normalizeStdout, scrubRequestHeaders, @@ -319,6 +320,24 @@ describe('normalizeSessionLog', () => { }) }) +describe('extractSnapshotSpillPaths', () => { + it('maps each spill filename to its full matched path, last match wins per name', () => { + const log = [ + 'Full formatted result stored at: /tmp/dsh-acp-snapshot-spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt. Use read with offset/limit, or grep this path to search within it.', + 'stale copy at /tmp/dsh-acp-snap-012345678/session-aaaaaaaaaaaa/bbbbbbbbbbbb-grep.txt then', + 'fresh copy at /tmp/dsh-acp-snap-012345678/session-cccccccccccc/dddddddddddd-grep.txt then', + ].join('\n') + expect(extractSnapshotSpillPaths(log)).toEqual(new Map([ + ['bash.txt', '/tmp/dsh-acp-snapshot-spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt'], + ['grep.txt', '/tmp/dsh-acp-snap-012345678/session-cccccccccccc/dddddddddddd-grep.txt'], + ])) + }) + + it('returns an empty map when the log carries no snapshot spill paths', () => { + expect(extractSnapshotSpillPaths('no spill paths here, only /tmp/other.txt\n')).toEqual(new Map()) + }) +}) + describe('scrubRequestHeaders', () => { const headerLine = JSON.stringify({ type: 'session', version: 0, id: 's', createdAt: 1, cwd: '/w' }) const headerEvent = (header: object) => diff --git a/packages/support/acp-snapshot/tests/suite.spec.ts b/packages/support/acp-snapshot/tests/suite.spec.ts index 14bcb8178c..1e1f2d93c8 100644 --- a/packages/support/acp-snapshot/tests/suite.spec.ts +++ b/packages/support/acp-snapshot/tests/suite.spec.ts @@ -455,6 +455,29 @@ describe('refreshFixtureReplacements', () => { { from: '/new', to: '/old' }, ]) }) + + it('stabilizes moved snapshot spill paths by filename while skipping unchanged or unmatched names', () => { + const spill = (session: string, hash: string, name: string): string => + `/tmp/dsh-acp-snapshot-spill/session-${session}/${hash}-${name}` + const record = (text: string): string => + `${JSON.stringify({ type: 'session', id: 'same', cwd: '/same' })}\n` + + `${JSON.stringify({ type: 'tool/result', data: { content: [{ type: 'text', text: `stored at: ${text} ` }] } })}\n` + const freshBash = spill('aaaaaaaaaaaa', 'bbbbbbbbbbbb', 'bash.txt') + const oldBash = spill('cccccccccccc', 'dddddddddddd', 'bash.txt') + const shared = spill('eeeeeeeeeeee', 'ffffffffffff', 'grep.txt') + const orphan = spill('111111111111', '222222222222', 'orphan.txt') + const logs: HarvestedLog[] = [{ + id: 'diagnostic', + createdAt: 1, + content: record(`${freshBash} and ${shared}`), + }] + const fixtures = [record(`${oldBash} and ${shared} and ${orphan}`)] + // bash.txt moved → replaced; grep.txt is identical and orphan.txt has no + // fresh counterpart → both skipped. + expect(refreshFixtureReplacements(logs, fixtures)).toEqual([ + { from: freshBash, to: oldBash }, + ]) + }) }) describe('stabilizeRefreshLog', () => { diff --git a/packages/ui/acp/tests/turns.spec.ts b/packages/ui/acp/tests/turns.spec.ts index 451e50aa90..d8ac198ef3 100644 --- a/packages/ui/acp/tests/turns.spec.ts +++ b/packages/ui/acp/tests/turns.spec.ts @@ -14,6 +14,17 @@ import { } from './harness.ts' import { SessionId } from '@deepseek-ai/dsh-session' +declare module '@deepseek-ai/dsh-session' { + interface SessionEventMap { + /** Test-only log-only event driven through appendOutOfBand below. */ + 'test/acp-out-of-band': { note: string } + } + + interface OutOfBandSessionEventMap { + 'test/acp-out-of-band': true + } +} + /** Boilerplate: initialize + create one session, returning its id. */ async function newSession(h: BridgeHarness, clientCapabilities: Record = {}): Promise { await h.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities }) @@ -342,6 +353,43 @@ describe('acp bridge — turn outcomes', () => { expect(text).toContain('real answer') }) + it('an out-of-band injection TURN while the prompt is queued neither captures nor settles it', async () => { + // Unlike the idle inject above (which appends context without a turn), a + // log-only out-of-band append on a closed log opens a real synthetic + // injection-triggered turn. Its turn/start must NOT capture inflight.turn + // (only message-triggered turns own the prompt) and its turn/end must not + // settle the prompt — the prompt settles on its OWN later message turn. + harness = await makeBridgeHarness({ storageDir, script: [textResponse('real answer')] }) + const sessionId = await newSession(harness) + const agent = harness.ctx.agents.get(SessionId(sessionId))! + let appended: Promise | undefined + const sessions = harness.ctx.sessions + harness.ctx.on('agent/inbox/enqueue', (subject) => { + if (subject === agent && appended === undefined) { + // Fires synchronously inside followup(), after the bridge installed the + // in-flight slot but before the prompt's own turn starts; the synthetic + // turn/start + turn/end land in that window. + appended = sessions.appendOutOfBand( + agent.session, + 'test/acp-out-of-band', + { note: 'log-only' }, + { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } }, + ) + } + }) + const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }) + expect(res.stopReason).toBe('end_turn') + await appended + // The synthetic injection turn precedes the prompt's own message turn. + const triggers = agent.session.events.flatMap(e => e.type === 'turn/start' ? [e.data.trigger.kind] : []) + expect(triggers).toEqual(['injection', 'message']) + const text = harness.updates + .filter(u => u.sessionUpdate === 'agent_message_chunk') + .map(u => (u.content.type === 'text' ? u.content.text : '')) + .join('') + expect(text).toContain('real answer') + }) + it('rejects a second prompt while one is in flight', async () => { harness = await makeBridgeHarness({ storageDir, script: ['hang'] }) const sessionId = await newSession(harness) diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index 2dc8955e73..9ec3c57a02 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -1431,6 +1431,32 @@ describe('pi-tui chat lifecycle and transcript', () => { expect(result.terminal.output).toContain('— Enter sends steering, Esc cancels') expect(result.terminal.output).not.toContain('queued') + // A cancellation discards queued steering: the badge clears without drains. + submitSteering('third') + submitSteering('fourth') + await tick() + expect(result.terminal.output).toContain('2 queued') + const discarded = result.agent.steeredIds.splice(0).map(id => ({ + id, content: [{ type: 'text' as const, text: 'discarded' }], source: { kind: 'user' as const }, + })) + // Another agent's dequeue/discard, and ones naming no pending id, leave + // the badge alone. + result.ctx.emit('agent/inbox/dequeue', other, discarded[0]!) + result.ctx.emit('agent/inbox/dequeue', result.agent, { + id: AgentMessageId('never-queued'), content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, + }) + result.ctx.emit('agent/inbox/discard', other, discarded) + result.ctx.emit('agent/inbox/discard', result.agent, [ + { id: AgentMessageId('never-queued'), content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } }, + ]) + await tick() + expect(result.terminal.output).toContain('2 queued') + result.terminal.output = '' + result.ctx.emit('agent/inbox/discard', result.agent, discarded) + await tick() + expect(result.terminal.output).toContain('— Enter sends steering, Esc cancels') + expect(result.terminal.output).not.toContain('queued') + await dispose(result) }) @@ -1811,6 +1837,18 @@ describe('pi-tui chat lifecycle and transcript', () => { // /reload without a Loader in the context degrades to a warning. expect(result.terminal.output).toContain('/reload needs the cordis Loader') expect(result.exit).toHaveBeenCalledWith(0) + + // The exit above left the TUI disposed (the mocked runtime.exit returns): + // a message submitted now is refused instead of reaching the agent. The + // refusal notice lands in the transcript, but the stopped UI no longer + // paints, so assert the refusal through the agent surface. + const sentBefore = result.agent.sent.length + const steeredBefore = result.agent.steered.length + result.terminal.send('after shutdown') + result.terminal.send('\r') + await tick() + expect(result.agent.sent).toHaveLength(sentBefore) + expect(result.agent.steered).toHaveLength(steeredBefore) await result.controller.dispose() await result.ctx.fiber.dispose() @@ -2148,6 +2186,7 @@ describe('pi-tui chat lifecycle and transcript', () => { expect(result.terminal.output).not.toContain('secret full snapshot payload') const invalidCards: [JsonValue, string][] = [ + ['plain-string-source', 'invalid-shape'], [{ kind: 'other' }, 'invalid-kind'], [{ kind: 'session-reference', references: [null] }, 'invalid-entry'], [{ kind: 'session-reference', references: [{}] }, 'invalid-fields'],