diff --git a/packages/app/src/components/agent-input-area.tsx b/packages/app/src/components/agent-input-area.tsx index f405b31cb..622f2b48d 100644 --- a/packages/app/src/components/agent-input-area.tsx +++ b/packages/app/src/components/agent-input-area.tsx @@ -611,7 +611,7 @@ export function AgentInputArea({ ] ); - const cancelButton = isAgentRunning && !hasSendableContent ? ( + const cancelButton = isAgentRunning && !hasSendableContent && !isProcessing ? ( diff --git a/packages/app/src/components/message-input.tsx b/packages/app/src/components/message-input.tsx index 26926223d..66fd4fcba 100644 --- a/packages/app/src/components/message-input.tsx +++ b/packages/app/src/components/message-input.tsx @@ -75,6 +75,8 @@ export interface MessageInputProps { isAgentRunning?: boolean; /** Callback for queue button when agent is running */ onQueue?: (payload: MessagePayload) => void; + /** Optional handler used when submit button is in loading state. */ + onSubmitLoadingPress?: () => void; /** Intercept key press events before default handling. Return true to prevent default. */ onKeyPress?: (event: { key: string; preventDefault: () => void }) => boolean; } @@ -132,6 +134,7 @@ export const MessageInput = forwardRef( voiceAgentId, isAgentRunning = false, onQueue, + onSubmitLoadingPress, onKeyPress: onKeyPressCallback, }, ref @@ -576,6 +579,16 @@ export const MessageInput = forwardRef( const hasImages = images.length > 0; const hasSendableContent = value.trim().length > 0 || hasImages; const shouldShowSendButton = hasSendableContent || isSubmitLoading; + const canPressLoadingButton = + isSubmitLoading && typeof onSubmitLoadingPress === "function"; + const isSendButtonDisabled = + disabled || + (!canPressLoadingButton && (isSubmitDisabled || isSubmitLoading)); + const submitAccessibilityLabel = canPressLoadingButton + ? "Interrupt agent" + : isAgentRunning + ? "Send and interrupt" + : "Send message"; return ( @@ -724,7 +737,7 @@ export const MessageInput = forwardRef( {rightContent} - {shouldShowSendButton && isAgentRunning && onQueue && ( + {hasSendableContent && isAgentRunning && onQueue && ( ( )} {shouldShowSendButton && ( - {isSubmitLoading ? ( diff --git a/packages/app/src/components/sidebar-agent-list.tsx b/packages/app/src/components/sidebar-agent-list.tsx index 29e2d25b8..5aece480f 100644 --- a/packages/app/src/components/sidebar-agent-list.tsx +++ b/packages/app/src/components/sidebar-agent-list.tsx @@ -1032,17 +1032,19 @@ const styles = StyleSheet.create((theme) => ({ agentTitle: { flex: 1, fontSize: theme.fontSize.base, + fontWeight: theme.fontWeight.normal, color: theme.colors.foreground, - opacity: 0.85, + opacity: 1, }, agentTitleInactive: { - color: theme.colors.foregroundMuted, + opacity: 0.8, }, agentTitleSelected: { opacity: 1, }, relativeTimeText: { color: theme.colors.foregroundMuted, + opacity: 0.78, fontSize: theme.fontSize.xs, }, shortcutBadge: { @@ -1084,7 +1086,8 @@ const styles = StyleSheet.create((theme) => ({ opacity: 0.5, }, agentMetaProjectName: { - color: theme.colors.foregroundMuted, + color: theme.colors.foreground, + opacity: 0.78, fontSize: theme.fontSize.xs, flexShrink: 1, minWidth: 0, @@ -1101,7 +1104,8 @@ const styles = StyleSheet.create((theme) => ({ borderColor: theme.colors.borderAccent, }, agentMetaBranchText: { - color: theme.colors.foregroundMuted, + color: theme.colors.foreground, + opacity: 0.76, fontSize: theme.fontSize.xs, }, selectionBar: { diff --git a/packages/app/src/components/terminal-emulator.tsx b/packages/app/src/components/terminal-emulator.tsx index 44580b8b1..4ba89f3d6 100644 --- a/packages/app/src/components/terminal-emulator.tsx +++ b/packages/app/src/components/terminal-emulator.tsx @@ -26,6 +26,7 @@ interface TerminalEmulatorProps { meta: boolean; }) => Promise | void; onPendingModifiersConsumed?: () => Promise | void; + onOutputChunkConsumed?: (sequence: number) => Promise | void; pendingModifiers?: PendingTerminalModifiers; focusRequestToken?: number; } @@ -47,6 +48,7 @@ export default function TerminalEmulator({ onResize, onTerminalKey, onPendingModifiersConsumed, + onOutputChunkConsumed, pendingModifiers = { ctrl: false, shift: false, alt: false }, focusRequestToken = 0, }: TerminalEmulatorProps) { @@ -108,19 +110,30 @@ export default function TerminalEmulator({ useEffect(() => { const runtime = runtimeRef.current; - if (!runtime) { - return; - } - if (outputChunkSequence <= 0) { return; } - if (outputChunkText.length === 0) { - runtime.clear(); + + if (!runtime) { + onOutputChunkConsumed?.(outputChunkSequence); return; } - runtime.write({ text: outputChunkText }); - }, [outputChunkSequence, outputChunkText]); + + if (outputChunkText.length === 0) { + runtime.clear({ + onCommitted: () => { + onOutputChunkConsumed?.(outputChunkSequence); + }, + }); + return; + } + runtime.write({ + text: outputChunkText, + onCommitted: () => { + onOutputChunkConsumed?.(outputChunkSequence); + }, + }); + }, [onOutputChunkConsumed, outputChunkSequence, outputChunkText]); useEffect(() => { if (focusRequestToken <= 0) { diff --git a/packages/app/src/components/terminal-pane.tsx b/packages/app/src/components/terminal-pane.tsx index 5e8c3d8f1..3f493d971 100644 --- a/packages/app/src/components/terminal-pane.tsx +++ b/packages/app/src/components/terminal-pane.tsx @@ -25,6 +25,7 @@ import { TerminalOutputPump, type TerminalOutputChunk, } from "@/terminal/runtime/terminal-output-pump"; +import { TerminalOutputDeliveryQueue } from "@/terminal/runtime/terminal-output-delivery-queue"; import { TerminalStreamController, type TerminalStreamControllerStatus, @@ -72,6 +73,22 @@ type TerminalOutputChunkState = { text: string; }; +type PendingTerminalInput = + | { + type: "data"; + data: string; + } + | { + type: "key"; + input: { + key: string; + ctrl: boolean; + shift: boolean; + alt: boolean; + meta?: boolean; + }; + }; + const EMPTY_MODIFIERS: ModifierState = { ctrl: false, shift: false, @@ -121,6 +138,7 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { const lastReportedSizeRef = useRef<{ rows: number; cols: number } | null>(null); const streamControllerRef = useRef(null); const outputPumpRef = useRef(null); + const outputDeliveryQueueRef = useRef(null); const [selectedTerminalId, setSelectedTerminalId] = useState(null); const [selectedOutputChunk, setSelectedOutputChunk] = useState({ @@ -142,16 +160,33 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { ); const hoverOutTimeoutRef = useRef | null>(null); const selectedTerminalIdRef = useRef(selectedTerminalId); + const pendingTerminalInputRef = useRef([]); useEffect(() => { selectedTerminalIdRef.current = selectedTerminalId; }, [selectedTerminalId]); + useEffect(() => { + const outputDeliveryQueue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + setSelectedOutputChunk(chunk); + }, + }); + outputDeliveryQueueRef.current = outputDeliveryQueue; + + return () => { + if (outputDeliveryQueueRef.current === outputDeliveryQueue) { + outputDeliveryQueueRef.current = null; + } + outputDeliveryQueue.reset(); + }; + }, []); + useEffect(() => { const outputPump = new TerminalOutputPump({ maxOutputChars: MAX_OUTPUT_CHARS, onSelectedOutputChunk: (chunk: TerminalOutputChunk) => { - setSelectedOutputChunk(chunk); + outputDeliveryQueueRef.current?.enqueue(chunk); }, }); outputPumpRef.current = outputPump; @@ -405,6 +440,9 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { }, [client, handleStreamControllerStatus, isConnected]); useEffect(() => { + outputDeliveryQueueRef.current?.reset(); + pendingTerminalInputRef.current = []; + setSelectedOutputChunk({ sequence: 0, text: "" }); outputPumpRef.current?.setSelectedTerminal({ terminalId: selectedTerminalId, }); @@ -422,6 +460,9 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { activeStream && activeStream.terminalId === selectedTerminalId ? activeStream.streamId : null; + const getCurrentActiveStreamId = useCallback(() => { + return streamControllerRef.current?.getActiveStreamId() ?? null; + }, []); const selectedTerminal = useMemo( () => terminals.find((terminal) => terminal.id === selectedTerminalId) ?? null, @@ -431,6 +472,50 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { createTerminalMutation.mutate(); }, [createTerminalMutation]); + const enqueuePendingTerminalInput = useCallback((entry: PendingTerminalInput) => { + const queue = pendingTerminalInputRef.current; + queue.push(entry); + if (queue.length > 512) { + queue.splice(0, queue.length - 512); + } + }, []); + + const flushPendingTerminalInput = useCallback(() => { + if (!client) { + return; + } + const currentStreamId = getCurrentActiveStreamId(); + if (currentStreamId === null) { + return; + } + const queue = pendingTerminalInputRef.current; + if (queue.length === 0) { + return; + } + + const pending = queue.splice(0, queue.length); + terminalDebugLog({ + scope: "terminal-pane", + event: "input:queue:flush", + details: { + activeStreamId: currentStreamId, + count: pending.length, + }, + }); + + for (const entry of pending) { + if (entry.type === "data") { + client.sendTerminalStreamInput(currentStreamId, entry.data); + continue; + } + client.sendTerminalStreamKey(currentStreamId, entry.input); + } + }, [client, getCurrentActiveStreamId]); + + useEffect(() => { + flushPendingTerminalInput(); + }, [activeStreamId, flushPendingTerminalInput]); + const handleCloseTerminal = useCallback( (terminalId: string) => { if ( @@ -462,19 +547,30 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { meta?: boolean; } ): boolean => { - if (!client || activeStreamId === null) { + const currentStreamId = getCurrentActiveStreamId(); + if (!client || currentStreamId === null) { terminalDebugLog({ scope: "terminal-pane", - event: "input:key:drop-no-stream", + event: "input:key:queue-no-stream", details: { key: input.key, ctrl: input.ctrl, shift: input.shift, alt: input.alt, - activeStreamId, + activeStreamId: currentStreamId, }, }); - return false; + enqueuePendingTerminalInput({ + type: "key", + input: { + key: normalizeTerminalTransportKey(input.key), + ctrl: input.ctrl, + shift: input.shift, + alt: input.alt, + meta: input.meta, + }, + }); + return true; } const normalizedKey = normalizeTerminalTransportKey(input.key); @@ -486,10 +582,10 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { ctrl: input.ctrl, shift: input.shift, alt: input.alt, - activeStreamId, + activeStreamId: currentStreamId, }, }); - client.sendTerminalStreamKey(activeStreamId, { + client.sendTerminalStreamKey(currentStreamId, { key: normalizedKey, ctrl: input.ctrl, shift: input.shift, @@ -498,7 +594,7 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { }); return true; }, - [client, activeStreamId] + [client, enqueuePendingTerminalInput, getCurrentActiveStreamId] ); const handleTerminalData = useCallback( @@ -506,13 +602,14 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { if (data.length === 0) { return; } + const currentStreamId = getCurrentActiveStreamId(); terminalDebugLog({ scope: "terminal-pane", event: "input:data:received", details: { length: data.length, preview: summarizeTerminalText({ text: data, maxChars: 80 }), - activeStreamId, + activeStreamId: currentStreamId, }, }); @@ -541,15 +638,19 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { } } - if (!client || activeStreamId === null) { + if (!client || currentStreamId === null) { terminalDebugLog({ scope: "terminal-pane", - event: "input:data:drop-no-stream", + event: "input:data:queue-no-stream", details: { length: data.length, preview: summarizeTerminalText({ text: data, maxChars: 80 }), }, }); + enqueuePendingTerminalInput({ + type: "data", + data, + }); return; } terminalDebugLog({ @@ -558,19 +659,20 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { details: { length: data.length, preview: summarizeTerminalText({ text: data, maxChars: 80 }), - activeStreamId, + activeStreamId: currentStreamId, }, }); - client.sendTerminalStreamInput(activeStreamId, data); + client.sendTerminalStreamInput(currentStreamId, data); }, [ - activeStreamId, clearPendingModifiers, client, + getCurrentActiveStreamId, modifiers.alt, modifiers.ctrl, modifiers.shift, sendTerminalKey, + enqueuePendingTerminalInput, ] ); @@ -626,6 +728,10 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { clearPendingModifiers(); }, [clearPendingModifiers]); + const handleOutputChunkConsumed = useCallback((sequence: number) => { + outputDeliveryQueueRef.current?.consume({ sequence }); + }, []); + const toggleModifier = useCallback( (modifier: keyof ModifierState) => { setModifiers((current) => ({ ...current, [modifier]: !current[modifier] })); @@ -808,6 +914,7 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { onResize={handleTerminalResize} onTerminalKey={handleTerminalKey} onPendingModifiersConsumed={handlePendingModifiersConsumed} + onOutputChunkConsumed={handleOutputChunkConsumed} pendingModifiers={modifiers} focusRequestToken={focusRequestToken} /> @@ -819,7 +926,11 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) { )} {isAttaching ? ( - + ) : null} diff --git a/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts b/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts new file mode 100644 index 000000000..138a00fc8 --- /dev/null +++ b/packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts @@ -0,0 +1,160 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +import { TerminalEmulatorRuntime } from "./terminal-emulator-runtime"; + +type StubTerminal = { + write: (text: string, callback?: () => void) => void; + reset: () => void; + focus: () => void; +}; + +function createRuntimeWithTerminal(): { + runtime: TerminalEmulatorRuntime; + terminal: StubTerminal & { + resetCalls: number; + }; + writeCallbacks: Array<() => void>; + writeTexts: string[]; +} { + const runtime = new TerminalEmulatorRuntime(); + const writeCallbacks: Array<() => void> = []; + const writeTexts: string[] = []; + let resetCalls = 0; + + const terminal: StubTerminal & { resetCalls: number } = { + write: (text: string, callback?: () => void) => { + writeTexts.push(text); + if (callback) { + writeCallbacks.push(callback); + } + }, + reset: () => { + resetCalls += 1; + terminal.resetCalls = resetCalls; + }, + focus: () => {}, + resetCalls, + }; + + (runtime as unknown as { terminal: StubTerminal }).terminal = terminal; + + return { + runtime, + terminal, + writeCallbacks, + writeTexts, + }; +} + +describe("terminal-emulator-runtime", () => { + const originalWindow = (globalThis as { window?: unknown }).window; + + beforeEach(() => { + (globalThis as { window?: { __paseoTerminal?: unknown } }).window = { + __paseoTerminal: undefined, + }; + }); + + afterEach(() => { + (globalThis as { window?: unknown }).window = originalWindow; + vi.useRealTimers(); + }); + + it("processes write and clear operations in strict order", () => { + const { runtime, terminal, writeCallbacks, writeTexts } = createRuntimeWithTerminal(); + const committed: string[] = []; + + runtime.write({ + text: "first", + onCommitted: () => { + committed.push("first"); + }, + }); + runtime.clear({ + onCommitted: () => { + committed.push("clear"); + }, + }); + runtime.write({ + text: "second", + onCommitted: () => { + committed.push("second"); + }, + }); + + expect(writeTexts).toEqual(["first"]); + expect(terminal.resetCalls).toBe(0); + expect(committed).toEqual([]); + + writeCallbacks[0]?.(); + + expect(committed).toEqual(["first", "clear"]); + expect(terminal.resetCalls).toBe(1); + expect(writeTexts).toEqual(["first", "second"]); + + writeCallbacks[1]?.(); + expect(committed).toEqual(["first", "clear", "second"]); + }); + + it("falls back to timeout commit when xterm write callback does not fire", () => { + vi.useFakeTimers(); + const { runtime } = createRuntimeWithTerminal(); + const onCommitted = vi.fn(); + + runtime.write({ + text: "stuck", + onCommitted, + }); + + expect(onCommitted).not.toHaveBeenCalled(); + vi.advanceTimersByTime(5_000); + expect(onCommitted).toHaveBeenCalledTimes(1); + }); + + it("ignores stale duplicate write callbacks from a previous operation", () => { + const { runtime, writeCallbacks } = createRuntimeWithTerminal(); + const committed: string[] = []; + + runtime.write({ + text: "first", + onCommitted: () => { + committed.push("first"); + }, + }); + runtime.write({ + text: "second", + onCommitted: () => { + committed.push("second"); + }, + }); + + writeCallbacks[0]?.(); + expect(committed).toEqual(["first"]); + + writeCallbacks[0]?.(); + expect(committed).toEqual(["first"]); + + writeCallbacks[1]?.(); + expect(committed).toEqual(["first", "second"]); + }); + + it("commits pending output operations during unmount to avoid deadlock", () => { + const { runtime } = createRuntimeWithTerminal(); + const onCommittedA = vi.fn(); + const onCommittedB = vi.fn(); + + runtime.write({ + text: "a", + onCommitted: onCommittedA, + }); + runtime.write({ + text: "b", + onCommitted: onCommittedB, + }); + + runtime.unmount(); + + expect(onCommittedA).toHaveBeenCalledTimes(1); + expect(onCommittedB).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts b/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts index 8ca94ae55..6366de7a7 100644 --- a/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts +++ b/packages/app/src/terminal/runtime/terminal-emulator-runtime.ts @@ -51,6 +51,12 @@ type TerminalEmulatorRuntimeDisposables = { disposeTerminal: () => void; }; +type TerminalOutputOperation = { + type: "write" | "clear"; + text: string; + onCommitted?: () => void; +}; + declare global { interface Window { __paseoTerminal?: Terminal; @@ -59,6 +65,7 @@ declare global { const DEFAULT_TOUCH_SCROLL_LINE_HEIGHT_PX = 18; const FIT_TIMEOUT_DELAYS_MS = [0, 16, 48, 120, 250, 500, 1_000, 2_000]; +const OUTPUT_OPERATION_TIMEOUT_MS = 5_000; export class TerminalEmulatorRuntime { private callbacks: TerminalEmulatorRuntimeCallbacks = {}; @@ -71,8 +78,9 @@ export class TerminalEmulatorRuntime { private fitAddon: FitAddon | null = null; private lastSize: { rows: number; cols: number } | null = null; private cleanup: (() => void) | null = null; - private pendingWriteText = ""; - private isWriteFlushQueued = false; + private outputOperations: TerminalOutputOperation[] = []; + private inFlightOutputOperation: TerminalOutputOperation | null = null; + private inFlightOutputOperationTimeout: ReturnType | null = null; setCallbacks(input: { callbacks: TerminalEmulatorRuntimeCallbacks }): void { this.callbacks = input.callbacks; @@ -287,6 +295,8 @@ export class TerminalEmulatorRuntime { }); } + this.processOutputQueue(); + const disposables: TerminalEmulatorRuntimeDisposables = { disposeInput: () => { inputDisposable.dispose(); @@ -338,30 +348,39 @@ export class TerminalEmulatorRuntime { }; } - write(input: { text: string }): void { - if (!this.terminal || input.text.length === 0) { + write(input: { text: string; onCommitted?: () => void }): void { + if (input.text.length === 0) { + input.onCommitted?.(); return; } - this.pendingWriteText += input.text; + this.outputOperations.push({ + type: "write", + text: input.text, + ...(input.onCommitted ? { onCommitted: input.onCommitted } : {}), + }); terminalDebugLog({ scope: "emulator-runtime", event: "output:enqueue", details: { chunkLength: input.text.length, - queueLength: this.pendingWriteText.length, + queueLength: this.outputOperations.length, preview: summarizeTerminalText({ text: input.text, maxChars: 64 }), }, }); - this.scheduleWriteFlush(); + this.processOutputQueue(); } - clear(): void { - this.pendingWriteText = ""; + clear(input?: { onCommitted?: () => void }): void { + this.outputOperations.push({ + type: "clear", + text: "", + ...(input?.onCommitted ? { onCommitted: input.onCommitted } : {}), + }); terminalDebugLog({ scope: "emulator-runtime", event: "output:clear", }); - this.terminal?.reset(); + this.processOutputQueue(); } focus(): void { @@ -373,11 +392,19 @@ export class TerminalEmulatorRuntime { scope: "emulator-runtime", event: "mount:unmount", details: { - pendingWriteLength: this.pendingWriteText.length, + pendingWriteOperations: this.outputOperations.length, }, }); - this.pendingWriteText = ""; - this.isWriteFlushQueued = false; + this.clearInFlightOutputTimeout(); + const inFlightOperation = this.inFlightOutputOperation; + this.inFlightOutputOperation = null; + if (inFlightOperation?.onCommitted) { + inFlightOperation.onCommitted(); + } + const pendingOperations = this.outputOperations.splice(0, this.outputOperations.length); + for (const operation of pendingOperations) { + operation.onCommitted?.(); + } this.cleanup?.(); this.cleanup = null; @@ -389,24 +416,39 @@ export class TerminalEmulatorRuntime { this.lastSize = null; } - private scheduleWriteFlush(): void { - if (this.isWriteFlushQueued) { + private processOutputQueue(): void { + if (this.inFlightOutputOperation) { return; } - this.isWriteFlushQueued = true; - queueMicrotask(() => { - this.isWriteFlushQueued = false; - this.flushWriteQueue(); - }); - } - - private flushWriteQueue(): void { - if (!this.terminal || this.pendingWriteText.length === 0) { + const terminal = this.terminal; + if (!terminal) { return; } - const text = this.pendingWriteText; - this.pendingWriteText = ""; + + const operation = this.outputOperations.shift(); + if (!operation) { + return; + } + + this.inFlightOutputOperation = operation; + const finalizeOperation = (expectedOperation: TerminalOutputOperation) => { + if (this.inFlightOutputOperation !== expectedOperation) { + return; + } + this.inFlightOutputOperation = null; + this.clearInFlightOutputTimeout(); + expectedOperation.onCommitted?.(); + this.processOutputQueue(); + }; + + if (operation.type === "clear") { + terminal.reset(); + finalizeOperation(operation); + return; + } + + const text = operation.text; terminalDebugLog({ scope: "emulator-runtime", event: "output:flush", @@ -415,7 +457,33 @@ export class TerminalEmulatorRuntime { preview: summarizeTerminalText({ text, maxChars: 96 }), }, }); - this.terminal.write(text); + this.inFlightOutputOperationTimeout = setTimeout(() => { + terminalDebugLog({ + scope: "emulator-runtime", + event: "output:flush-timeout", + details: { + length: text.length, + preview: summarizeTerminalText({ text, maxChars: 96 }), + }, + }); + finalizeOperation(operation); + }, OUTPUT_OPERATION_TIMEOUT_MS); + + try { + terminal.write(text, () => { + finalizeOperation(operation); + }); + } catch { + finalizeOperation(operation); + } + } + + private clearInFlightOutputTimeout(): void { + if (!this.inFlightOutputOperationTimeout) { + return; + } + clearTimeout(this.inFlightOutputOperationTimeout); + this.inFlightOutputOperationTimeout = null; } private applyDocumentBoundsStyles(input: { root: HTMLDivElement }): () => void { diff --git a/packages/app/src/terminal/runtime/terminal-output-delivery-queue.test.ts b/packages/app/src/terminal/runtime/terminal-output-delivery-queue.test.ts new file mode 100644 index 000000000..d901e7e4c --- /dev/null +++ b/packages/app/src/terminal/runtime/terminal-output-delivery-queue.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, it, vi } from "vitest"; + +import { TerminalOutputDeliveryQueue } from "./terminal-output-delivery-queue"; + +describe("terminal-output-delivery-queue", () => { + it("retries in-flight delivery when consume is missing", () => { + vi.useFakeTimers(); + const delivered: Array<{ sequence: number; text: string }> = []; + const queue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + delivered.push(chunk); + }, + deliveryTimeoutMs: 100, + }); + + queue.enqueue({ sequence: 1, text: "a" }); + queue.enqueue({ sequence: 2, text: "b" }); + + expect(delivered).toEqual([{ sequence: 1, text: "a" }]); + + vi.advanceTimersByTime(100); + expect(delivered).toEqual([ + { sequence: 1, text: "a" }, + { sequence: 1, text: "a" }, + ]); + + queue.consume({ sequence: 1 }); + expect(delivered).toEqual([ + { sequence: 1, text: "a" }, + { sequence: 1, text: "a" }, + { sequence: 2, text: "b" }, + ]); + + vi.useRealTimers(); + }); + + it("delivers first chunk immediately and blocks later chunks until consumed", () => { + const delivered: Array<{ sequence: number; text: string }> = []; + const queue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + delivered.push(chunk); + }, + }); + + queue.enqueue({ sequence: 1, text: "a" }); + queue.enqueue({ sequence: 2, text: "b" }); + queue.enqueue({ sequence: 3, text: "c" }); + + expect(delivered).toEqual([{ sequence: 1, text: "a" }]); + + queue.consume({ sequence: 1 }); + expect(delivered).toEqual([ + { sequence: 1, text: "a" }, + { sequence: 3, text: "bc" }, + ]); + }); + + it("ignores stale consume acknowledgements", () => { + const delivered = vi.fn(); + const queue = new TerminalOutputDeliveryQueue({ onDeliver: delivered }); + + queue.enqueue({ sequence: 1, text: "x" }); + queue.consume({ sequence: 99 }); + queue.enqueue({ sequence: 2, text: "y" }); + + expect(delivered).toHaveBeenCalledTimes(1); + expect(delivered).toHaveBeenNthCalledWith(1, { sequence: 1, text: "x" }); + + queue.consume({ sequence: 1 }); + + expect(delivered).toHaveBeenCalledTimes(2); + expect(delivered).toHaveBeenNthCalledWith(2, { sequence: 2, text: "y" }); + }); + + it("resets in-flight and pending chunks", () => { + const delivered: Array<{ sequence: number; text: string }> = []; + const queue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + delivered.push(chunk); + }, + }); + + queue.enqueue({ sequence: 1, text: "hello" }); + queue.enqueue({ sequence: 2, text: " world" }); + + queue.reset(); + queue.enqueue({ sequence: 3, text: "next" }); + + expect(delivered).toEqual([ + { sequence: 1, text: "hello" }, + { sequence: 3, text: "next" }, + ]); + }); + + it("preserves empty chunk payloads for authoritative clears", () => { + const delivered: Array<{ sequence: number; text: string }> = []; + const queue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + delivered.push(chunk); + }, + }); + + queue.enqueue({ sequence: 1, text: "abc" }); + queue.consume({ sequence: 1 }); + queue.enqueue({ sequence: 2, text: "" }); + + expect(delivered).toEqual([ + { sequence: 1, text: "abc" }, + { sequence: 2, text: "" }, + ]); + }); + + it("treats clear chunks as a delivery barrier and drops pending stale text", () => { + const delivered: Array<{ sequence: number; text: string }> = []; + const queue = new TerminalOutputDeliveryQueue({ + onDeliver: (chunk) => { + delivered.push(chunk); + }, + }); + + queue.enqueue({ sequence: 1, text: "a" }); + queue.enqueue({ sequence: 2, text: "b" }); + queue.enqueue({ sequence: 3, text: "" }); + queue.enqueue({ sequence: 4, text: "c" }); + + expect(delivered).toEqual([{ sequence: 1, text: "a" }]); + + queue.consume({ sequence: 1 }); + expect(delivered).toEqual([ + { sequence: 1, text: "a" }, + { sequence: 3, text: "" }, + ]); + + queue.consume({ sequence: 3 }); + expect(delivered).toEqual([ + { sequence: 1, text: "a" }, + { sequence: 3, text: "" }, + { sequence: 4, text: "c" }, + ]); + }); +}); diff --git a/packages/app/src/terminal/runtime/terminal-output-delivery-queue.ts b/packages/app/src/terminal/runtime/terminal-output-delivery-queue.ts new file mode 100644 index 000000000..c6f89ab0e --- /dev/null +++ b/packages/app/src/terminal/runtime/terminal-output-delivery-queue.ts @@ -0,0 +1,151 @@ +import { summarizeTerminalText, terminalDebugLog } from "./terminal-debug"; + +export type TerminalOutputDeliveryChunk = { + sequence: number; + text: string; +}; + +export type TerminalOutputDeliveryQueueOptions = { + onDeliver: (chunk: TerminalOutputDeliveryChunk) => void; + deliveryTimeoutMs?: number; +}; + +const DEFAULT_DELIVERY_TIMEOUT_MS = 8_000; + +export class TerminalOutputDeliveryQueue { + private readonly pendingChunks: TerminalOutputDeliveryChunk[] = []; + private inFlightChunk: TerminalOutputDeliveryChunk | null = null; + private inFlightTimeout: ReturnType | null = null; + private lastSeenSequence = 0; + private readonly deliveryTimeoutMs: number; + + constructor(private readonly options: TerminalOutputDeliveryQueueOptions) { + this.deliveryTimeoutMs = + options.deliveryTimeoutMs ?? DEFAULT_DELIVERY_TIMEOUT_MS; + } + + enqueue(chunk: TerminalOutputDeliveryChunk): void { + if (chunk.sequence <= 0) { + return; + } + if (chunk.sequence <= this.lastSeenSequence) { + return; + } + this.lastSeenSequence = chunk.sequence; + + if (chunk.text.length === 0) { + this.pendingChunks.length = 0; + this.pendingChunks.push(chunk); + terminalDebugLog({ + scope: "output-delivery-queue", + event: "enqueue:clear", + details: { + sequence: chunk.sequence, + }, + }); + this.tryDeliver(); + return; + } + + const lastPendingChunk = this.pendingChunks[this.pendingChunks.length - 1]; + if (lastPendingChunk && lastPendingChunk.text.length > 0) { + lastPendingChunk.sequence = chunk.sequence; + lastPendingChunk.text += chunk.text; + } else { + this.pendingChunks.push(chunk); + } + terminalDebugLog({ + scope: "output-delivery-queue", + event: "enqueue:text", + details: { + sequence: chunk.sequence, + pendingCount: this.pendingChunks.length, + textLength: chunk.text.length, + preview: summarizeTerminalText({ text: chunk.text, maxChars: 80 }), + }, + }); + this.tryDeliver(); + } + + consume(input: { sequence: number }): void { + if (this.inFlightChunk?.sequence !== input.sequence) { + return; + } + + terminalDebugLog({ + scope: "output-delivery-queue", + event: "consume", + details: { + sequence: input.sequence, + }, + }); + this.clearInFlightTimeout(); + this.inFlightChunk = null; + this.tryDeliver(); + } + + reset(): void { + this.clearInFlightTimeout(); + this.pendingChunks.length = 0; + this.inFlightChunk = null; + this.lastSeenSequence = 0; + } + + private tryDeliver(): void { + if (this.inFlightChunk) { + return; + } + const nextChunk = this.pendingChunks.shift(); + if (!nextChunk) { + return; + } + + this.inFlightChunk = nextChunk; + terminalDebugLog({ + scope: "output-delivery-queue", + event: "deliver:start", + details: { + sequence: nextChunk.sequence, + pendingCount: this.pendingChunks.length, + }, + }); + this.deliverInFlightChunk(); + } + + private deliverInFlightChunk(): void { + const chunk = this.inFlightChunk; + if (!chunk) { + return; + } + + this.clearInFlightTimeout(); + this.inFlightTimeout = setTimeout(() => { + if (!this.inFlightChunk) { + return; + } + terminalDebugLog({ + scope: "output-delivery-queue", + event: "deliver:timeout-retry", + details: { + sequence: this.inFlightChunk.sequence, + pendingCount: this.pendingChunks.length, + timeoutMs: this.deliveryTimeoutMs, + }, + }); + this.deliverInFlightChunk(); + }, this.deliveryTimeoutMs); + + this.options.onDeliver({ + sequence: chunk.sequence, + text: chunk.text, + }); + } + + private clearInFlightTimeout(): void { + if (!this.inFlightTimeout) { + return; + } + clearTimeout(this.inFlightTimeout); + this.inFlightTimeout = null; + } +} diff --git a/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts b/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts index 90cf0c93e..0bcee67f0 100644 --- a/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts +++ b/packages/app/src/terminal/runtime/terminal-stream-controller.test.ts @@ -12,6 +12,7 @@ type FakeStreamSubscriber = (chunk: TerminalStreamControllerChunk) => void; class FakeTerminalStreamClient implements TerminalStreamControllerClient { private readonly streamSubscribers = new Map>(); + private readonly pendingChunksByStreamId = new Map(); public attachCalls: Array<{ terminalId: string; options?: { @@ -47,6 +48,14 @@ class FakeTerminalStreamClient implements TerminalStreamControllerClient { streamId: number, handler: (chunk: TerminalStreamControllerChunk) => void ): () => void { + const pendingChunks = this.pendingChunksByStreamId.get(streamId); + if (pendingChunks && pendingChunks.length > 0) { + for (const chunk of pendingChunks) { + handler(chunk); + } + this.pendingChunksByStreamId.delete(streamId); + } + const subscribers = this.streamSubscribers.get(streamId) ?? new Set(); subscribers.add(handler); this.streamSubscribers.set(streamId, subscribers); @@ -61,21 +70,44 @@ class FakeTerminalStreamClient implements TerminalStreamControllerClient { emitChunk(input: { streamId: number; + offset?: number; endOffset: number; + replay?: boolean; data: string; }): void { const subscribers = this.streamSubscribers.get(input.streamId); if (!subscribers || subscribers.size === 0) { return; } + const bytes = new TextEncoder().encode(input.data); const chunk: TerminalStreamControllerChunk = { + offset: input.offset ?? input.endOffset - bytes.byteLength, endOffset: input.endOffset, - data: new TextEncoder().encode(input.data), + replay: input.replay, + data: bytes, }; for (const subscriber of subscribers) { subscriber(chunk); } } + + bufferChunk(input: { + streamId: number; + offset?: number; + endOffset: number; + replay?: boolean; + data: string; + }): void { + const chunks = this.pendingChunksByStreamId.get(input.streamId) ?? []; + const bytes = new TextEncoder().encode(input.data); + chunks.push({ + offset: input.offset ?? input.endOffset - bytes.byteLength, + endOffset: input.endOffset, + replay: input.replay, + data: bytes, + }); + this.pendingChunksByStreamId.set(input.streamId, chunks); + } } function createControllerHarness(input?: { @@ -251,4 +283,203 @@ describe("terminal-stream-controller", () => { expect(harness.resets).toEqual(["term-reset"]); expect(harness.controller.getActiveStreamId()).toBe(12); }); + + it("delivers buffered replay chunks flushed synchronously during subscribe", async () => { + const harness = createControllerHarness(); + harness.client.nextAttachResponses.push({ + streamId: 17, + replayedFrom: 0, + currentOffset: 10, + reset: false, + error: null, + }); + harness.client.bufferChunk({ + streamId: 17, + offset: 0, + endOffset: 10, + replay: true, + data: "buffered-replay", + }); + + harness.controller.setTerminal({ terminalId: "term-buffered" }); + await flushAsyncWork(); + + expect(harness.chunks).toEqual([ + { terminalId: "term-buffered", text: "buffered-replay" }, + ]); + }); + + it("clears stale selected output when bootstrap replay starts before current offset", async () => { + const harness = createControllerHarness(); + harness.client.nextAttachResponses.push({ + streamId: 61, + replayedFrom: 0, + currentOffset: 20, + reset: false, + error: null, + }); + + harness.controller.setTerminal({ terminalId: "term-bootstrap-reset" }); + await flushAsyncWork(); + + expect(harness.resets).toEqual(["term-bootstrap-reset"]); + }); + + it("reattaches and replays from last contiguous offset when chunk offsets gap", async () => { + const harness = createControllerHarness(); + harness.client.nextAttachResponses.push({ + streamId: 31, + currentOffset: 0, + reset: false, + error: null, + }); + harness.client.nextAttachResponses.push({ + streamId: 32, + currentOffset: 8, + reset: false, + error: null, + }); + + harness.controller.setTerminal({ terminalId: "term-gap" }); + await flushAsyncWork(); + + harness.client.emitChunk({ + streamId: 31, + offset: 0, + endOffset: 2, + data: "ok", + }); + harness.client.emitChunk({ + streamId: 31, + offset: 4, + endOffset: 8, + data: "miss", + }); + await flushAsyncWork(); + + expect(harness.chunks).toEqual([{ terminalId: "term-gap", text: "ok" }]); + expect(harness.client.detachCalls).toContain(31); + expect(harness.client.attachCalls.length).toBe(2); + expect(harness.client.attachCalls[1]?.options).toEqual({ + resumeOffset: 2, + rows: 24, + cols: 80, + }); + expect(harness.controller.getActiveStreamId()).toBe(32); + }); + + it("does not treat replay range before currentOffset as a live gap", async () => { + const harness = createControllerHarness(); + harness.client.nextAttachResponses.push({ + streamId: 41, + replayedFrom: 50, + currentOffset: 100, + reset: false, + error: null, + }); + harness.client.bufferChunk({ + streamId: 41, + offset: 80, + endOffset: 100, + data: "replay-tail", + }); + + harness.controller.setTerminal({ terminalId: "term-replay" }); + await flushAsyncWork(); + + harness.client.emitChunk({ + streamId: 41, + offset: 100, + endOffset: 102, + data: "ok", + }); + + expect(harness.client.detachCalls).toEqual([]); + expect(harness.client.attachCalls).toHaveLength(1); + expect(harness.controller.getActiveStreamId()).toBe(41); + expect(harness.chunks).toEqual([ + { terminalId: "term-replay", text: "replay-tail" }, + { terminalId: "term-replay", text: "ok" }, + ]); + }); + + it("accepts clamped replay start after reconnect without entering a reconnect loop", async () => { + const harness = createControllerHarness(); + harness.client.nextAttachResponses.push({ + streamId: 71, + replayedFrom: 205, + currentOffset: 205, + reset: false, + error: null, + }); + harness.client.nextAttachResponses.push({ + streamId: 72, + replayedFrom: 396, + currentOffset: 978, + reset: true, + error: null, + }); + + harness.controller.setTerminal({ terminalId: "term-clamped-replay" }); + await flushAsyncWork(); + + harness.client.emitChunk({ + streamId: 71, + offset: 205, + endOffset: 299, + data: "first", + }); + harness.client.emitChunk({ + streamId: 71, + offset: 396, + endOffset: 493, + data: "gap", + }); + await flushAsyncWork(); + + expect(harness.client.attachCalls).toHaveLength(2); + expect(harness.client.attachCalls[1]?.options).toEqual({ + resumeOffset: 299, + rows: 24, + cols: 80, + }); + expect(harness.controller.getActiveStreamId()).toBe(72); + + harness.client.emitChunk({ + streamId: 72, + replay: true, + offset: 396, + endOffset: 493, + data: "replay-1", + }); + harness.client.emitChunk({ + streamId: 72, + replay: true, + offset: 493, + endOffset: 590, + data: "replay-2", + }); + harness.client.emitChunk({ + streamId: 72, + replay: true, + offset: 687, + endOffset: 784, + data: "replay-3", + }); + harness.client.emitChunk({ + streamId: 72, + replay: false, + offset: 978, + endOffset: 1075, + data: "live", + }); + await flushAsyncWork(); + + expect(harness.client.attachCalls).toHaveLength(2); + expect(harness.client.detachCalls.filter((streamId) => streamId === 71)).toHaveLength(1); + expect(harness.chunks.at(-1)).toEqual({ + terminalId: "term-clamped-replay", + text: "live", + }); + }); }); diff --git a/packages/app/src/terminal/runtime/terminal-stream-controller.ts b/packages/app/src/terminal/runtime/terminal-stream-controller.ts index 19dead0d2..bb82275f4 100644 --- a/packages/app/src/terminal/runtime/terminal-stream-controller.ts +++ b/packages/app/src/terminal/runtime/terminal-stream-controller.ts @@ -10,13 +10,16 @@ import { summarizeTerminalText, terminalDebugLog } from "./terminal-debug"; export type TerminalStreamControllerAttachPayload = { streamId: number | null; + replayedFrom?: number; currentOffset: number; reset: boolean; error?: string | null; }; export type TerminalStreamControllerChunk = { + offset: number; endOffset: number; + replay?: boolean; data: Uint8Array; }; @@ -71,6 +74,8 @@ type TerminalStreamControllerActiveStream = { terminalId: string; streamId: number; decoder: TextDecoder; + nextExpectedOffset: number | null; + catchUpEndOffset: number | null; unsubscribe: () => void; }; @@ -308,6 +313,37 @@ export class TerminalStreamController { const decoder = new TextDecoder(); const streamId = attachPayload.streamId; + const replayedFromOffset = + typeof attachPayload.replayedFrom === "number" + ? Math.max(0, Math.floor(attachPayload.replayedFrom)) + : null; + const currentOffset = Math.max( + 0, + Math.floor(attachPayload.currentOffset) + ); + const shouldResetForReplayBootstrap = + typeof resumeOffset !== "number" && + typeof replayedFromOffset === "number" && + replayedFromOffset < currentOffset; + if (shouldResetForReplayBootstrap) { + this.options.onReset?.({ terminalId: input.terminalId }); + } + const startExpectedOffset = + replayedFromOffset ?? + (typeof resumeOffset === "number" + ? Math.max(0, Math.floor(resumeOffset)) + : currentOffset); + const catchUpEndOffset = currentOffset; + + const activeStream: TerminalStreamControllerActiveStream = { + terminalId: input.terminalId, + streamId, + decoder, + nextExpectedOffset: startExpectedOffset, + catchUpEndOffset, + unsubscribe: () => {}, + }; + this.activeStream = activeStream; const unsubscribe = this.options.client.onTerminalStreamData(streamId, (chunk) => { this.handleChunk({ terminalId: input.terminalId, @@ -316,21 +352,22 @@ export class TerminalStreamController { decoder, }); }); - - this.activeStream = { - terminalId: input.terminalId, - streamId, - decoder, - unsubscribe, - }; + if (this.activeStream === activeStream) { + activeStream.unsubscribe = unsubscribe; + } else { + unsubscribe(); + } terminalDebugLog({ scope: "stream-controller", event: "attach:success", details: { terminalId: input.terminalId, streamId, + replayedFrom: attachPayload.replayedFrom ?? null, + requestedResumeOffset: resumeOffset ?? null, currentOffset: attachPayload.currentOffset, reset: attachPayload.reset, + bootstrapReset: shouldResetForReplayBootstrap, }, }); this.updateStatus({ @@ -390,9 +427,66 @@ export class TerminalStreamController { return; } + const chunkOffset = Number.isFinite(input.chunk.offset) + ? Math.max(0, Math.floor(input.chunk.offset)) + : 0; + const chunkEndOffset = Number.isFinite(input.chunk.endOffset) + ? Math.max(0, Math.floor(input.chunk.endOffset)) + : chunkOffset; + if (chunkEndOffset < chunkOffset) { + return; + } + + const expectedOffset = activeStream.nextExpectedOffset; + if (typeof expectedOffset === "number") { + if (chunkEndOffset <= expectedOffset) { + return; + } + if (chunkOffset !== expectedOffset) { + const catchUpEndOffset = activeStream.catchUpEndOffset; + const canSkipReplayGap = + chunkOffset > expectedOffset && + typeof catchUpEndOffset === "number" && + expectedOffset < catchUpEndOffset && + chunkOffset <= catchUpEndOffset; + + if (canSkipReplayGap) { + terminalDebugLog({ + scope: "stream-controller", + event: "stream:replay-gap-skip", + details: { + terminalId: input.terminalId, + streamId: input.streamId, + expectedOffset, + observedOffset: chunkOffset, + catchUpEndOffset, + replay: Boolean(input.chunk.replay), + }, + }); + activeStream.nextExpectedOffset = chunkOffset; + } else { + this.recoverFromStreamGap({ + terminalId: input.terminalId, + streamId: input.streamId, + expectedOffset, + observedOffset: chunkOffset, + }); + return; + } + } + } + + if ( + typeof activeStream.catchUpEndOffset === "number" && + chunkEndOffset >= activeStream.catchUpEndOffset + ) { + activeStream.catchUpEndOffset = null; + } + + activeStream.nextExpectedOffset = chunkEndOffset; updateTerminalResumeOffset({ terminalId: input.terminalId, - offset: input.chunk.endOffset, + offset: chunkEndOffset, resumeOffsetByTerminalId: this.resumeOffsetByTerminalId, }); @@ -406,7 +500,9 @@ export class TerminalStreamController { details: { terminalId: input.terminalId, streamId: input.streamId, - endOffset: input.chunk.endOffset, + offset: chunkOffset, + endOffset: chunkEndOffset, + replay: Boolean(input.chunk.replay), byteLength: input.chunk.data.byteLength, textLength: text.length, preview: summarizeTerminalText({ text, maxChars: 96 }), @@ -419,6 +515,57 @@ export class TerminalStreamController { }); } + private recoverFromStreamGap(input: { + terminalId: string; + streamId: number; + expectedOffset: number; + observedOffset: number; + }): void { + const activeStream = this.activeStream; + if (!activeStream) { + return; + } + if (activeStream.streamId !== input.streamId || activeStream.terminalId !== input.terminalId) { + return; + } + if (this.selectedTerminalId !== input.terminalId) { + return; + } + + terminalDebugLog({ + scope: "stream-controller", + event: "stream:gap-reconnect", + details: { + terminalId: input.terminalId, + streamId: input.streamId, + expectedOffset: input.expectedOffset, + observedOffset: input.observedOffset, + }, + }); + + updateTerminalResumeOffset({ + terminalId: input.terminalId, + offset: input.expectedOffset, + resumeOffsetByTerminalId: this.resumeOffsetByTerminalId, + }); + + this.attachGeneration += 1; + const generation = this.attachGeneration; + + void this.detachActiveStream({ shouldDetach: true }); + this.updateStatus({ + terminalId: input.terminalId, + streamId: null, + isAttaching: true, + error: + this.options.reconnectErrorMessage ?? DEFAULT_RECONNECT_ERROR_MESSAGE, + }); + void this.attachTerminal({ + terminalId: input.terminalId, + generation, + }); + } + private async detachActiveStream(input: { shouldDetach: boolean }): Promise { const activeStream = this.activeStream; if (!activeStream) { diff --git a/packages/server/src/client/daemon-client-terminal-stream-manager.test.ts b/packages/server/src/client/daemon-client-terminal-stream-manager.test.ts index 4652f728a..e6b170e63 100644 --- a/packages/server/src/client/daemon-client-terminal-stream-manager.test.ts +++ b/packages/server/src/client/daemon-client-terminal-stream-manager.test.ts @@ -40,7 +40,18 @@ describe("TerminalStreamManager", () => { expect(seen).toEqual(["hello"]); expect(sendAck).toHaveBeenCalledTimes(1); - expect(sendAck).toHaveBeenCalledWith({ streamId: 7, offset: 9 }); + }); + + test("acks buffered chunks even while no subscriber is attached", () => { + const sendAck = vi.fn(); + const manager = new TerminalStreamManager({ sendAck }); + + manager.receiveChunk({ + chunk: createChunk({ streamId: 2, offset: 0, data: "abc" }), + }); + + expect(sendAck).toHaveBeenCalledTimes(1); + expect(sendAck).toHaveBeenCalledWith({ streamId: 2, offset: 3 }); }); test("does not ack when every handler throws", () => { diff --git a/packages/server/src/client/daemon-client-terminal-stream-manager.ts b/packages/server/src/client/daemon-client-terminal-stream-manager.ts index 89a32232a..26310c3d6 100644 --- a/packages/server/src/client/daemon-client-terminal-stream-manager.ts +++ b/packages/server/src/client/daemon-client-terminal-stream-manager.ts @@ -79,8 +79,6 @@ export class TerminalStreamManager { const streamHandlers = this.handlers.get(chunk.streamId); if (!streamHandlers || streamHandlers.size === 0) { this.bufferChunk({ chunk }); - // Ack buffered chunks immediately to keep the daemon-side stream window - // moving even before a UI subscriber is attached. this.maybeAckChunk({ streamId: chunk.streamId, endOffset: chunk.endOffset, diff --git a/packages/server/src/client/daemon-client.test.ts b/packages/server/src/client/daemon-client.test.ts index d3ecaa215..c906fd95c 100644 --- a/packages/server/src/client/daemon-client.test.ts +++ b/packages/server/src/client/daemon-client.test.ts @@ -753,6 +753,48 @@ describe("DaemonClient", () => { unsubscribe(); }); + test("handles terminal binary frames delivered as UTF-8 strings", async () => { + const logger = createMockLogger(); + const mock = createMockTransport(); + + const client = new DaemonClient({ + url: "ws://test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); + + const connectPromise = client.connect(); + mock.triggerOpen(); + await connectPromise; + + const seen: string[] = []; + client.onTerminalStreamData(11, (chunk) => { + seen.push(new TextDecoder().decode(chunk.data)); + }); + + const payload = new TextEncoder().encode("ls\r\n"); + const frame = encodeBinaryMuxFrame({ + channel: BinaryMuxChannel.Terminal, + messageType: TerminalBinaryMessageType.OutputUtf8, + streamId: 11, + offset: 0, + payload, + }); + const frameAsString = new TextDecoder("utf-8", { fatal: true }).decode(frame); + + mock.triggerMessage(frameAsString); + + expect(seen).toEqual(["ls\r\n"]); + expect(mock.sent).toHaveLength(1); + const ackFrame = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!); + expect(ackFrame?.channel).toBe(BinaryMuxChannel.Terminal); + expect(ackFrame?.messageType).toBe(TerminalBinaryMessageType.Ack); + expect(ackFrame?.streamId).toBe(11); + expect(ackFrame?.offset).toBe(4); + }); + test("acks buffered terminal chunks when handler is attached", async () => { const logger = createMockLogger(); const mock = createMockTransport(); diff --git a/packages/server/src/server/agent/agent-management-mcp.ts b/packages/server/src/server/agent/agent-management-mcp.ts index 06fa277dd..dd0f45cd8 100644 --- a/packages/server/src/server/agent/agent-management-mcp.ts +++ b/packages/server/src/server/agent/agent-management-mcp.ts @@ -374,7 +374,9 @@ export async function createAgentManagementMcpServer( }); try { - agentManager.recordUserMessage(snapshot.id, trimmedPrompt); + agentManager.recordUserMessage(snapshot.id, trimmedPrompt, { + emitState: false, + }); } catch (error) { childLogger.error( { err: error, agentId: snapshot.id }, @@ -599,7 +601,9 @@ export async function createAgentManagementMcpServer( } try { - agentManager.recordUserMessage(agentId, prompt); + agentManager.recordUserMessage(agentId, prompt, { + emitState: false, + }); } catch (error) { childLogger.error( { err: error, agentId }, diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 86a74d620..dcd8f06ae 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, test } from "vitest"; +import { describe, expect, test, vi } from "vitest"; import { mkdtempSync } from "node:fs"; import { join } from "node:path"; import { tmpdir } from "node:os"; @@ -588,6 +588,74 @@ describe("AgentManager", () => { expect(refreshed?.runtimeInfo?.model).toBe("gpt-5.2-codex"); }); + test("keeps updatedAt monotonic when user message and run start happen in the same millisecond", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000120", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + const nowSpy = vi.spyOn(Date, "now").mockReturnValue(1_750_000_000_000); + try { + manager.recordUserMessage(snapshot.id, "hello"); + const afterMessage = manager.getAgent(snapshot.id); + expect(afterMessage).toBeDefined(); + const messageUpdatedAt = afterMessage!.updatedAt.getTime(); + + const stream = manager.streamAgent(snapshot.id, "hello"); + const afterRunStart = manager.getAgent(snapshot.id); + expect(afterRunStart).toBeDefined(); + expect(afterRunStart!.updatedAt.getTime()).toBeGreaterThan(messageUpdatedAt); + + await stream.return(undefined); + } finally { + nowSpy.mockRestore(); + } + }); + + test("recordUserMessage can skip emitting agent_state when run start will emit running", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); + const storagePath = join(workdir, "agents"); + const storage = new AgentStorage(storagePath, logger); + const manager = new AgentManager({ + clients: { + codex: new TestAgentClient(), + }, + registry: storage, + logger, + idFactory: () => "00000000-0000-4000-8000-000000000121", + }); + + const snapshot = await manager.createAgent({ + provider: "codex", + cwd: workdir, + }); + + const lifecycleUpdates: string[] = []; + manager.subscribe((event) => { + if (event.type !== "agent_state" || event.agent.id !== snapshot.id) { + return; + } + lifecycleUpdates.push(event.agent.lifecycle); + }); + lifecycleUpdates.length = 0; + + manager.recordUserMessage(snapshot.id, "hello", { emitState: false }); + + expect(lifecycleUpdates).toEqual([]); + }); + test("runAgent assembles finalText from trailing assistant chunks", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 08003deba..517144b42 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -307,6 +307,15 @@ export class AgentManager { this.onAgentAttention = callback; } + private touchUpdatedAt(agent: ManagedAgent): Date { + const nowMs = Date.now(); + const previousMs = agent.updatedAt.getTime(); + const nextMs = nowMs > previousMs ? nowMs : previousMs + 1; + const next = new Date(nextMs); + agent.updatedAt = next; + return next; + } + subscribe(callback: AgentSubscriber, options?: SubscribeOptions): () => void { const targetAgentId = options?.agentId == null @@ -847,7 +856,7 @@ export class AgentManager { recordUserMessage( agentId: string, text: string, - options?: { messageId?: string } + options?: { messageId?: string; emitState?: boolean } ): void { const agent = this.requireAgent(agentId); const item: AgentTimelineItem = { @@ -855,8 +864,8 @@ export class AgentManager { text, messageId: options?.messageId, }; - agent.updatedAt = new Date(); - agent.lastUserMessageAt = agent.updatedAt; + const updatedAt = this.touchUpdatedAt(agent); + agent.lastUserMessageAt = updatedAt; const row = this.recordTimeline(agent, item); this.dispatchStream(agentId, { type: "timeline", @@ -866,12 +875,14 @@ export class AgentManager { seq: row.seq, epoch: this.ensureTimelineState(agent).epoch, }); - this.emitState(agent); + if (options?.emitState !== false) { + this.emitState(agent); + } } async appendTimelineItem(agentId: string, item: AgentTimelineItem): Promise { const agent = this.requireAgent(agentId); - agent.updatedAt = new Date(); + this.touchUpdatedAt(agent); const row = this.recordTimeline(agent, item); this.dispatchStream(agentId, { type: "timeline", @@ -955,7 +966,7 @@ export class AgentManager { agent.lifecycle = "running"; // Bump updatedAt when lifecycle changes so downstream consumers can // deterministically order idle->running transitions. - agent.updatedAt = new Date(); + this.touchUpdatedAt(agent); self.emitState(agent); return streamForwarder; @@ -1527,7 +1538,7 @@ export class AgentManager { ): void { // Only update timestamp for live events, not history replay if (!options?.fromHistory) { - agent.updatedAt = new Date(); + this.touchUpdatedAt(agent); } let timelineRow: AgentTimelineRow | null = null; diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index 6b143c5c1..d0a306173 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -579,7 +579,9 @@ export async function createAgentMcpServer( }); try { - agentManager.recordUserMessage(snapshot.id, trimmedPrompt); + agentManager.recordUserMessage(snapshot.id, trimmedPrompt, { + emitState: false, + }); } catch (error) { childLogger.error( { err: error, agentId: snapshot.id }, @@ -806,7 +808,9 @@ export async function createAgentMcpServer( } try { - agentManager.recordUserMessage(agentId, prompt); + agentManager.recordUserMessage(agentId, prompt, { + emitState: false, + }); } catch (error) { childLogger.error( { err: error, agentId }, diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index f8b9228a3..8cf30a548 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -2311,7 +2311,10 @@ export class Session { const prompt = this.buildAgentPrompt(text, images); try { - this.agentManager.recordUserMessage(agentId, text, { messageId }); + this.agentManager.recordUserMessage(agentId, text, { + messageId, + emitState: false, + }); } catch (error) { this.sessionLogger.error( { err: error, agentId }, @@ -5244,7 +5247,10 @@ export class Session { await this.interruptAgentIfRunning(agentId); try { - this.agentManager.recordUserMessage(agentId, msg.text, { messageId: msg.messageId }); + this.agentManager.recordUserMessage(agentId, msg.text, { + messageId: msg.messageId, + emitState: false, + }); } catch (error) { this.sessionLogger.error( { err: error, agentId }, diff --git a/packages/server/src/shared/binary-mux.test.ts b/packages/server/src/shared/binary-mux.test.ts index 2af8d41b7..a246639bd 100644 --- a/packages/server/src/shared/binary-mux.test.ts +++ b/packages/server/src/shared/binary-mux.test.ts @@ -3,6 +3,7 @@ import { BinaryMuxChannel, TerminalBinaryFlags, TerminalBinaryMessageType, + asUint8Array, decodeBinaryMuxFrame, encodeBinaryMuxFrame, } from "./binary-mux.js"; @@ -40,4 +41,10 @@ describe("binary mux frame codec", () => { const tampered = encoded.slice(0, encoded.byteLength - 1); expect(decodeBinaryMuxFrame(tampered)).toBeNull(); }); + + it("converts UTF-8 string payloads to bytes", () => { + const bytes = asUint8Array("hello"); + expect(bytes).not.toBeNull(); + expect(Array.from(bytes ?? [])).toEqual(Array.from(new TextEncoder().encode("hello"))); + }); }); diff --git a/packages/server/src/shared/binary-mux.ts b/packages/server/src/shared/binary-mux.ts index 410797c8b..6b52c03c2 100644 --- a/packages/server/src/shared/binary-mux.ts +++ b/packages/server/src/shared/binary-mux.ts @@ -37,6 +37,19 @@ export interface BinaryMuxFrame { } export function asUint8Array(data: unknown): Uint8Array | null { + if (typeof data === "string") { + if (typeof TextEncoder !== "undefined") { + return new TextEncoder().encode(data); + } + if (typeof Buffer !== "undefined") { + return new Uint8Array(Buffer.from(data, "utf8")); + } + const out = new Uint8Array(data.length); + for (let i = 0; i < data.length; i += 1) { + out[i] = data.charCodeAt(i) & 0xff; + } + return out; + } if (data instanceof Uint8Array) { return data; }