Skip to content
Merged
Show file tree
Hide file tree
Changes from 6 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,12 @@
* renders AgentComposer with `key={activeView}`, so switching tabs unmounts
* the composer; any queue held only in local component state is discarded
* and the message is never delivered.
*
* Also covers #10148 -- delivery must not depend on the keyed composer
* being mounted: queued follow-ups are flushed when the agent settles to
* idle even while the user is on another teammate tab, and the queue is
* dropped when the agent reaches a terminal status (it can never be
* delivered then, and a permanent "queued" display would remain otherwise).
*/

import { render } from 'ink-testing-library';
Expand Down Expand Up @@ -88,7 +94,7 @@ function makeFakeAgent(): AgentInteractive {
} as unknown as AgentInteractive;
}

describe('AgentComposer queued follow-ups (#10069)', () => {
describe('Agent View queued follow-ups (#10069, #10148)', () => {
let agentA: AgentInteractive;
let agentB: AgentInteractive;
const streamingByAgent = new Map<AgentInteractive, typeof IDLE>();
Expand Down Expand Up @@ -191,17 +197,55 @@ describe('AgentComposer queued follow-ups (#10069)', () => {
expect(agentA.enqueueMessage).toHaveBeenCalledTimes(1);
});

it('flushes when returning to a tab after the agent becomes idle', async () => {
it('delivers queued follow-ups while the user is on another tab (#10148)', async () => {
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
streamingByAgent.set(agentA, BUSY);
const app = await renderWithView('agent-a');

submitCapture.current!('follow-up while away');
await switchTo(app, 'agent-a');
expect(app.lastFrame()).toContain('follow-up while away');

// The user keeps working on teammate B while A finishes its round.
await switchTo(app, 'agent-b');
streamingByAgent.set(agentA, IDLE);
await switchTo(app, 'agent-a');
await switchTo(app, 'agent-b');

// Delivery must not wait for the user to revisit A's tab: A's composer
// is unmounted here, only the provider persists (#10148).
expect(agentA.enqueueMessage).toHaveBeenCalledTimes(1);
expect(agentA.enqueueMessage).toHaveBeenCalledWith('follow-up while away');

// Revisiting A later must not re-deliver.
await switchTo(app, 'agent-a');
expect(agentA.enqueueMessage).toHaveBeenCalledTimes(1);
expect(app.lastFrame()).not.toContain('follow-up while away');
});

it('drops the queue when the agent reaches a terminal status (#10148)', async () => {
streamingByAgent.set(agentA, BUSY);
const app = await renderWithView('agent-a');

submitCapture.current!('never delivered');
await switchTo(app, 'agent-a');
expect(app.lastFrame()).toContain('never delivered');

// A fails while the user is on B; queued follow-ups can never arrive.
await switchTo(app, 'agent-b');
streamingByAgent.set(agentA, {
status: AgentStatus.FAILED,
streamingState: StreamingState.Idle,
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
isInputActive: false,
elapsedTime: 0,
lastPromptTokenCount: 0,
});
await switchTo(app, 'agent-b');

expect(agentA.enqueueMessage).not.toHaveBeenCalled();

// The undeliverable queue must be dropped, not shown as "queued" forever.
await switchTo(app, 'agent-a');
expect(app.lastFrame()).not.toContain('never delivered');
expect(agentA.enqueueMessage).not.toHaveBeenCalled();
});

it('joins multiple queued follow-ups into one prompt after a tab switch', async () => {
Expand Down
39 changes: 7 additions & 32 deletions packages/cli/src/ui/components/agent-view/AgentComposer.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,9 @@
*/

import { Box, Text, useStdin } from 'ink';
import { useCallback, useEffect, useMemo, useRef } from 'react';
import { useCallback, useEffect, useMemo } from 'react';
import {
AgentStatus,
isTerminalStatus,
ApprovalMode,
APPROVAL_MODES,
} from '@qwen-code/qwen-code-core';
Expand Down Expand Up @@ -71,7 +70,6 @@ export const AgentComposer: React.FC<AgentComposerProps> = ({ agentId }) => {
setAgentInputBufferText,
setAgentTabBarFocused,
setAgentApprovalMode,
setAgentMessageQueue,
appendToAgentMessageQueue,
} = useAgentViewActions();
const agent = agents.get(agentId);
Expand Down Expand Up @@ -198,39 +196,16 @@ export const AgentComposer: React.FC<AgentComposerProps> = ({ agentId }) => {
[buffer, agentTabBarFocused, setAgentTabBarFocused],
);

// ── Message queue (accumulate while streaming, flush as one prompt on idle) ──
// ── Message queue display ──
//
// The queue lives in AgentViewContext (keyed by agentId), not in local
// state: the layout keys this component by the active view, so switching
// teammate tabs unmounts it and a local queue would be silently dropped
// before the flush below ever runs (#10069).
// Queued follow-ups live in AgentViewContext (keyed by agentId) and are
// delivered by the provider's always-mounted per-agent flusher, not here:
// the layout keys this component by the active view, so a flush effect in
// this component would only run while the agent's tab is focused (#10069,
// #10148).

const messageQueue = agentMessageQueues.get(agentId) ?? EMPTY_MESSAGE_QUEUE;

// When agent becomes idle (and not terminal), flush queued messages.
const flushedQueueRef = useRef<readonly string[] | null>(null);
useEffect(() => {
if (
streamingState === StreamingState.Idle &&
messageQueue.length > 0 &&
status !== undefined &&
!isTerminalStatus(status)
) {
if (flushedQueueRef.current === messageQueue) return;
flushedQueueRef.current = messageQueue;
const combined = messageQueue.join('\n');
setAgentMessageQueue(agentId, []);
interactiveAgent?.enqueueMessage(combined);
}
}, [
streamingState,
messageQueue,
interactiveAgent,
status,
agentId,
setAgentMessageQueue,
]);

const handleSubmit = useCallback(
(text: string) => {
const trimmed = text.trim();
Expand Down
59 changes: 29 additions & 30 deletions packages/cli/src/ui/contexts/AgentViewContext.test.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,29 @@ function makeConfig(): Config {
} as unknown as Config;
}

/**
* Minimal AgentInteractive stub. The provider mounts a per-agent queue
* flusher that derives streaming state via useAgentStreamingState, so the
* stub must cover that surface even in storage tests. Status stays
* undefined so the flusher never delivers here (delivery is covered by
* AgentComposer.queuedMessages.test.tsx).
*/
function makeInteractiveAgent(): AgentInteractive {
return {
getCore: () => ({
runtimeContext: {
getApprovalMode: () => ApprovalMode.DEFAULT,
setApprovalMode: vi.fn(),
},
}),
getStatus: () => undefined,
getPendingApprovals: () => new Map(),
getLastPromptTokenCount: () => 0,
getEventEmitter: () => undefined,
enqueueMessage: vi.fn(),
} as unknown as AgentInteractive;
}

describe('AgentViewProvider in-process bridges', () => {
// Regression guard. The team bridge (useTeamInProcess) was authored but
// never mounted in the provider, so teammate TEAMMATE_JOINED events never
Expand All @@ -59,11 +82,7 @@ describe('AgentViewProvider in-process bridges', () => {

it('clears embedded shell focus when switching agent tabs', async () => {
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

function Probe() {
const state = useAgentViewState();
Expand Down Expand Up @@ -113,11 +132,7 @@ describe('AgentViewProvider in-process bridges', () => {
// state change lands in its own commit (the production focus seed is
// a keypress in a commit well after the tab switch).
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

const probeActions: {
registerAgent?: (
Expand Down Expand Up @@ -174,11 +189,7 @@ describe('AgentViewProvider in-process bridges', () => {
describe('AgentViewProvider per-agent message queues', () => {
it('stores queues per agent and clears them when emptied or unregistered', async () => {
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

const probeActions: {
registerAgent?: (
Expand Down Expand Up @@ -237,11 +248,7 @@ describe('AgentViewProvider per-agent message queues', () => {

it('appends queued messages without losing same-batch updates', async () => {
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

const probeActions: {
registerAgent?: (
Expand Down Expand Up @@ -287,11 +294,7 @@ describe('AgentViewProvider per-agent message queues', () => {
// manager detaching while the user submits), the append must not
// resurrect the queue entry the delete just removed.
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

const probeActions: {
registerAgent?: (
Expand Down Expand Up @@ -334,11 +337,7 @@ describe('AgentViewProvider per-agent message queues', () => {

it('clears all queued messages when all agents unregister', async () => {
const config = makeConfig();
const interactiveAgent = {
getCore: () => ({
runtimeContext: { getApprovalMode: () => ApprovalMode.DEFAULT },
}),
} as AgentInteractive;
const interactiveAgent = makeInteractiveAgent();

const probeActions: {
registerAgent?: (
Expand Down
73 changes: 72 additions & 1 deletion packages/cli/src/ui/contexts/AgentViewContext.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,15 @@ import {
useState,
} from 'react';
import {
isTerminalStatus,
type AgentInteractive,
type ApprovalMode,
type Config,
} from '@qwen-code/qwen-code-core';
import { useArenaInProcess } from '../hooks/useArenaInProcess.js';
import { useAgentStreamingState } from '../hooks/useAgentStreamingState.js';
import { useTeamInProcess } from '../hooks/useTeamInProcess.js';
import { StreamingState } from '../types.js';

// ─── Types ──────────────────────────────────────────────────

Expand Down Expand Up @@ -60,7 +63,9 @@ export interface AgentViewState {
* Queued follow-up messages per agent (keyed by agentId). Held here —
* not in the composer — because the layout keys AgentComposer by the
* active view, so switching teammate tabs unmounts the composer and any
* component-local queue would be silently discarded (#10069).
* component-local queue would be silently discarded (#10069). Delivery
* also lives here (AgentQueueFlusher) so queues flush even while the
* agent's tab is unfocused (#10148).
*/
agentMessageQueues: ReadonlyMap<string, readonly string[]>;
}
Expand Down Expand Up @@ -134,6 +139,66 @@ export function useAgentViewActions(): AgentViewActions {
return useContext(AgentViewActionsContext) ?? DEFAULT_ACTIONS;
}

// ─── Queue delivery ─────────────────────────────────────────

// Shared empty queue identity so agents without queued messages don't
// allocate on every render.
const EMPTY_MESSAGE_QUEUE: readonly string[] = [];

/**
* Always-mounted delivery for one registered agent's queued follow-ups.
*
* AgentViewProvider mounts one flusher per registered agentId. Delivery
* cannot live in AgentComposer: the layout renders it keyed by the active
* view, so switching teammate tabs unmounts the composer while the queues
* persist — an agent that settles to idle (or a terminal status) while
* unfocused would otherwise keep accepted-but-undelivered messages forever
* (#10148).
*/
function AgentQueueFlusher({ agentId }: { agentId: string }) {
const { agents, agentMessageQueues } = useAgentViewState();
const { setAgentMessageQueue } = useAgentViewActions();
const interactiveAgent = agents.get(agentId)?.interactiveAgent;
const { status, streamingState } = useAgentStreamingState(interactiveAgent);
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
const messageQueue = agentMessageQueues.get(agentId) ?? EMPTY_MESSAGE_QUEUE;

// Dedupe by queue identity: effects run twice per commit under
// StrictMode, and the clear below only lands on the next render — without
// this the same queue would be delivered twice.
const flushedQueueRef = useRef<readonly string[] | null>(null);

useEffect(() => {
if (status !== undefined && isTerminalStatus(status)) {
// The agent can never accept these messages now; drop them so the
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
// display doesn't show undeliverable "queued" follow-ups forever.
if (messageQueue.length > 0) {
setAgentMessageQueue(agentId, []);
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
}
return;
}
if (
streamingState === StreamingState.Idle &&
messageQueue.length > 0 &&
status !== undefined
) {
if (flushedQueueRef.current === messageQueue) return;
flushedQueueRef.current = messageQueue;
const combined = messageQueue.join('\n');
setAgentMessageQueue(agentId, []);
interactiveAgent?.enqueueMessage(combined);
Comment thread
yiliang114 marked this conversation as resolved.
Resolved
}
}, [
streamingState,
messageQueue,
interactiveAgent,
status,
agentId,
setAgentMessageQueue,
]);

return null;
}

// ─── Provider ───────────────────────────────────────────────

interface AgentViewProviderProps {
Expand Down Expand Up @@ -385,6 +450,12 @@ export function AgentViewProvider({
return (
<AgentViewStateContext.Provider value={state}>
<AgentViewActionsContext.Provider value={actions}>
{/* Always-mounted queue delivery, one flusher per registered agent
— delivery must not depend on the keyed composer being mounted
(#10148). */}
{[...agents.keys()].map((agentId) => (
<AgentQueueFlusher key={agentId} agentId={agentId} />
))}
{children}
</AgentViewActionsContext.Provider>
</AgentViewStateContext.Provider>
Expand Down
Loading