Fix transport queued message handling

Author: IM.codes

src/agent/transport-session-runtime.ts

@@ -155,8 +155,6 @@ export class TransportSessionRuntime implements SessionRuntime {
     if (!this._providerSessionId) {
       throw new Error('TransportSessionRuntime not initialized — call initialize() first');
     }
-    // Clear pending — user cancelled, they don't want queued messages either
-    this._pendingMessages = [];
     if (!this.provider.cancel) return;
     await this.provider.cancel(this._providerSessionId);
   }

src/daemon/command-handler.ts

@@ -1353,14 +1353,16 @@ async function handleSend(cmd: Record<string, unknown>, serverLink: ServerLink):
         try { serverLink.send({ type: 'command.ack', commandId: effectiveId, status: isLegacy ? 'accepted_legacy' : 'accepted', session: sessionName }); } catch {}
         return;
       }
-      timelineEmitter.emit(sessionName, 'user.message', { text });
       if (record?.agentType === 'qwen' && record.qwenAuthType === 'qwen-oauth') {
         recordQwenOAuthRequest();
         refreshQwenQuotaUsageLabels(serverLink);
       }
       // send() is synchronous: dispatches immediately if idle, queues if busy.
       // Status changes come from transport runtime's onStatusChange callback.
       const result = transportRuntime.send(text);
+      if (result === 'sent') {
+        timelineEmitter.emit(sessionName, 'user.message', { text });
+      }
       if (result === 'queued') {
         timelineEmitter.emit(sessionName, 'session.state', { state: 'queued', pendingCount: transportRuntime.pendingCount }, { source: 'daemon', confidence: 'high' });
       }

test/daemon/command-handler-transport-queue.test.ts

@@ -0,0 +1,178 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest';
+
+const { getSessionMock, getTransportRuntimeMock, emitMock } = vi.hoisted(() => ({
+  getSessionMock: vi.fn(),
+  getTransportRuntimeMock: vi.fn(),
+  emitMock: vi.fn(),
+}));
+
+vi.mock('../../src/store/session-store.js', () => ({
+  listSessions: vi.fn(() => []),
+  getSession: getSessionMock,
+  upsertSession: vi.fn(),
+  removeSession: vi.fn(),
+  updateSessionState: vi.fn(),
+}));
+
+vi.mock('../../src/agent/session-manager.js', () => ({
+  startProject: vi.fn(),
+  stopProject: vi.fn(),
+  teardownProject: vi.fn(),
+  getTransportRuntime: getTransportRuntimeMock,
+  launchTransportSession: vi.fn(),
+  isProviderSessionBound: vi.fn(() => false),
+  persistSessionRecord: vi.fn(),
+  relaunchSessionWithSettings: vi.fn(),
+}));
+
+vi.mock('../../src/agent/tmux.js', () => ({
+  sendKeys: vi.fn(),
+  sendKeysDelayedEnter: vi.fn(),
+  sendRawInput: vi.fn(),
+  resizeSession: vi.fn(),
+  sendKey: vi.fn(),
+  getPaneStartCommand: vi.fn(),
+}));
+
+vi.mock('../../src/router/message-router.js', () => ({
+  routeMessage: vi.fn(),
+}));
+
+vi.mock('../../src/daemon/terminal-streamer.js', () => ({
+  terminalStreamer: {
+    subscribe: vi.fn(),
+    unsubscribe: vi.fn(),
+    start: vi.fn(),
+    stop: vi.fn(),
+  },
+}));
+
+vi.mock('../../src/daemon/timeline-emitter.js', () => ({
+  timelineEmitter: {
+    emit: emitMock,
+    on: vi.fn(() => () => {}),
+    off: vi.fn(),
+    epoch: 0,
+    replay: vi.fn(() => ({ events: [], truncated: false })),
+  },
+}));
+
+vi.mock('../../src/daemon/timeline-store.js', () => ({
+  timelineStore: {
+    append: vi.fn(),
+    read: vi.fn(() => []),
+    clear: vi.fn(),
+  },
+}));
+
+vi.mock('../../src/daemon/subsession-manager.js', () => ({
+  startSubSession: vi.fn(),
+  stopSubSession: vi.fn(),
+  rebuildSubSessions: vi.fn(),
+  detectShells: vi.fn().mockResolvedValue([]),
+  readSubSessionResponse: vi.fn(),
+  subSessionName: (id: string) => `deck_sub_${id}`,
+}));
+
+vi.mock('../../src/daemon/p2p-orchestrator.js', () => ({
+  startP2pRun: vi.fn(),
+  cancelP2pRun: vi.fn(),
+  getP2pRun: vi.fn(() => undefined),
+  listP2pRuns: vi.fn(() => []),
+  serializeP2pRun: vi.fn(),
+}));
+
+vi.mock('../../src/daemon/repo-handler.js', () => ({
+  handleRepoCommand: vi.fn(),
+}));
+
+vi.mock('../../src/daemon/file-transfer-handler.js', () => ({
+  handleFileUpload: vi.fn(),
+  handleFileDownload: vi.fn(),
+  createProjectFileHandle: vi.fn(),
+  lookupAttachment: vi.fn(() => undefined),
+}));
+
+vi.mock('../../src/daemon/preview-relay.js', () => ({
+  handlePreviewCommand: vi.fn(),
+}));
+
+vi.mock('../../src/daemon/provider-sessions.js', () => ({
+  listProviderSessions: vi.fn(() => []),
+}));
+
+vi.mock('../../src/util/logger.js', () => ({
+  default: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  },
+}));
+
+vi.mock('../../src/util/imc-dir.js', () => ({
+  ensureImcDir: vi.fn().mockResolvedValue('/tmp/imc'),
+  imcSubDir: vi.fn((dir: string, sub: string) => `${dir}/.imc/${sub}`),
+}));
+
+import { handleWebCommand } from '../../src/daemon/command-handler.js';
+
+const flushAsync = () => new Promise<void>((resolve) => setTimeout(resolve, 0));
+
+describe('handleWebCommand transport queue behavior', () => {
+  const serverLink = {
+    send: vi.fn(),
+    sendBinary: vi.fn(),
+    sendTimelineEvent: vi.fn(),
+    daemonVersion: '0.1.0',
+  };
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    getSessionMock.mockReturnValue({
+      name: 'deck_transport_brain',
+      projectName: 'transport',
+      role: 'brain',
+      agentType: 'claude-code-sdk',
+      runtimeType: 'transport',
+      state: 'running',
+    });
+  });
+
+  it('does not emit a user.message for queued transport sends', async () => {
+    getTransportRuntimeMock.mockReturnValue({
+      send: vi.fn(() => 'queued'),
+      pendingCount: 2,
+    });
+
+    handleWebCommand({ type: 'session.send', session: 'deck_transport_brain', text: 'queued msg', commandId: 'cmd-queued' }, serverLink as any);
+    await flushAsync();
+
+    expect(emitMock).not.toHaveBeenCalledWith('deck_transport_brain', 'user.message', { text: 'queued msg' });
+    expect(emitMock).toHaveBeenCalledWith(
+      'deck_transport_brain',
+      'session.state',
+      { state: 'queued', pendingCount: 2 },
+      expect.any(Object),
+    );
+    expect(emitMock).toHaveBeenCalledWith('deck_transport_brain', 'command.ack', { commandId: 'cmd-queued', status: 'accepted' });
+  });
+
+  it('emits a user.message immediately for dispatched transport sends', async () => {
+    getTransportRuntimeMock.mockReturnValue({
+      send: vi.fn(() => 'sent'),
+      pendingCount: 0,
+    });
+
+    handleWebCommand({ type: 'session.send', session: 'deck_transport_brain', text: 'sent msg', commandId: 'cmd-sent' }, serverLink as any);
+    await flushAsync();
+
+    expect(emitMock).toHaveBeenCalledWith('deck_transport_brain', 'user.message', { text: 'sent msg' });
+    expect(emitMock).not.toHaveBeenCalledWith(
+      'deck_transport_brain',
+      'session.state',
+      expect.objectContaining({ state: 'queued' }),
+      expect.anything(),
+    );
+  });
+});

test/daemon/transport-session-runtime.test.ts

@@ -98,14 +98,27 @@ describe('TransportSessionRuntime', () => {
     expect(runtime.sending).toBe(false);
   });
 
-  it('cancel() delegates to provider.cancel and clears pending', () => {
+  it('cancel() delegates to provider.cancel and preserves pending', () => {
     runtime.send('first');
     runtime.send('queued1');
     runtime.send('queued2');
     expect(runtime.pendingCount).toBe(2);
 
     runtime.cancel();
     expect(mock.provider.cancel).toHaveBeenCalledWith('sess-1');
+    expect(runtime.pendingCount).toBe(2);
+  });
+
+  it('cancelled turns drain pending messages into the next turn', () => {
+    runtime.send('first');
+    runtime.send('queued1');
+    runtime.send('queued2');
+
+    runtime.cancel();
+    mock.fireError('sess-1', { code: 'CANCELLED', message: 'cancelled', recoverable: true });
+
+    expect(mock.provider.send).toHaveBeenCalledTimes(2);
+    expect(mock.provider.send).toHaveBeenNthCalledWith(2, 'sess-1', 'queued1\n\nqueued2', undefined, undefined);
     expect(runtime.pendingCount).toBe(0);
   });
 

test/daemon/transport-status-lifecycle.test.ts

@@ -200,18 +200,18 @@ describe('batched queuing', () => {
     expect(runtime.getStatus()).toBe('thinking');
   });
 
-  it('on cancel, pending messages are cleared (not drained)', () => {
+  it('on cancel, pending messages are drained into the next turn', () => {
     runtime.send('first');
-    runtime.send('will-be-dropped');
-    runtime.send('also-dropped');
+    runtime.send('still-queued');
+    runtime.send('send-after-stop');
 
     runtime.cancel();
     mock.fireCancelled('sess-1');
 
-    // Cancel clears pending → no drain, no second send
-    expect(mock.provider.send).toHaveBeenCalledTimes(1);
+    expect(mock.provider.send).toHaveBeenCalledTimes(2);
+    expect(mock.provider.send).toHaveBeenNthCalledWith(2, 'sess-1', 'still-queued\n\nsend-after-stop', undefined, undefined);
     expect(runtime.pendingCount).toBe(0);
-    expect(runtime.getStatus()).toBe('idle');
+    expect(runtime.getStatus()).toBe('thinking');
   });
 
   it('multiple turns with queuing: correct history order', () => {

web/src/components/SessionControls.tsx

@@ -837,7 +837,9 @@ export function SessionControls({ ws, activeSession, inputRef, onAfterAction, on
 
   const finalizeSend = useCallback((payload: PendingSendPayload, options?: { clearComposer?: boolean }) => {
     if (!ws || !activeSession) return;
-    // Transport sessions queue messages internally — no frontend notice needed
+    if (activeSession.runtimeType === 'transport' && activeSession.state === 'running') {
+      setQueuedNoticeVisible(true);
+    }
     quickData.recordHistory(payload.text, activeSession.name);
     try {
       ws.sendSessionCommand('send', { sessionName: activeSession.name, text: payload.text, ...payload.extra });

web/src/components/SessionPane.tsx

@@ -270,7 +270,9 @@ export function SessionPane({
           inputRef={inputRef}
           onAfterAction={onAfterAction}
           onSend={(_name, text) => {
-            addOptimisticUserMessage(text);
+            if (session.runtimeType !== 'transport') {
+              addOptimisticUserMessage(text);
+            }
             scrollToBottom();
           }}
           onStopProject={onStopProject}

web/src/styles.css

@@ -124,7 +124,22 @@ body {
 .p2p-settings-icon { font-size: 12px; line-height: 1; }
 .p2p-settings-label { font-size: 11px; font-weight: 600; letter-spacing: 0.01em; }
 .controls { padding: 6px 8px; display: flex; gap: 6px; align-items: center; }
-.controls-queued-hint { padding: 0 12px 8px; color: #fbbf24; font-size: 12px; line-height: 1.35; }
+.controls-queued-hint {
+  position: absolute;
+  left: 12px;
+  right: 12px;
+  bottom: calc(100% + 6px);
+  padding: 8px 10px;
+  color: #fbbf24;
+  font-size: 12px;
+  line-height: 1.35;
+  background: rgba(15, 23, 42, 0.92);
+  border: 1px solid rgba(251, 191, 36, 0.35);
+  border-radius: 10px;
+  box-shadow: 0 8px 24px rgba(0, 0, 0, 0.35);
+  pointer-events: none;
+  z-index: 3;
+}
 .attachment-badges { display: flex; flex-wrap: wrap; gap: 4px; padding: 4px 2px 2px; }
 .attachment-badge { display: inline-flex; align-items: center; gap: 4px; background: #1e293b; border: 1px solid #334155; border-radius: 12px; padding: 2px 8px 2px 6px; font-size: 12px; color: #94a3b8; max-width: 200px; }
 .attachment-badge-icon { font-size: 11px; flex-shrink: 0; }

web/test/components/SessionControls.test.tsx

@@ -688,7 +688,7 @@ describe('SessionControls', () => {
   });
 
 
-  it('sends message to running transport session without blocking (queuing is daemon-side)', () => {
+  it('shows a bottom queued notice when sending to a running transport session', () => {
     const ws = makeWs();
     const runningSession = makeSession({
       name: 'qwen-session',
@@ -714,8 +714,7 @@ describe('SessionControls', () => {
       sessionName: 'qwen-session',
       text: 'queued send',
     });
-    // No frontend queued notice — transport runtime queues internally
-    expect(screen.queryByText('transport_send_queued')).toBeNull();
+    expect(screen.getByText('transport_send_queued')).toBeDefined();
   });
 
   it('pressing Escape in a running transport input sends /stop command', () => {