jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Harden terminal streaming and worktree setup flow

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-15 19:05:27 +07:00
parent 75dc802a3a
commit 0103b4da96
21 changed files with 1297 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/app/src/components/agent-input-area.tsx
+2 -1
View File
@@ -611,7 +611,7 @@ export function AgentInputArea({
]
);
const cancelButton = isAgentRunning && !hasSendableContent ? (
const cancelButton = isAgentRunning && !hasSendableContent && !isProcessing ? (
<Tooltip delayDuration={0} enabledOnDesktop enabledOnMobile={false}>
<TooltipTrigger
onPress={handleCancelAgent}
@@ -755,6 +755,7 @@ export function AgentInputArea({
voiceAgentId={agentId}
isAgentRunning={isAgentRunning}
onQueue={handleQueue}
onSubmitLoadingPress={isAgentRunning ? handleCancelAgent : undefined}
onKeyPress={handleCommandKeyPress}
/>
</View>
packages/app/src/components/message-input.tsx
+24 -16
View File
@@ -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<MessageInputRef, MessageInputProps>(
voiceAgentId,
isAgentRunning = false,
onQueue,
onSubmitLoadingPress,
onKeyPress: onKeyPressCallback,
},
ref
@@ -576,6 +579,16 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
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 (
<View style={styles.container} testID="message-input-root">
@@ -724,7 +737,7 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
</TooltipContent>
</Tooltip>
{rightContent}
{shouldShowSendButton && isAgentRunning && onQueue && (
{hasSendableContent && isAgentRunning && onQueue && (
<Tooltip delayDuration={0} enabledOnDesktop enabledOnMobile={false}>
<TooltipTrigger
onPress={handleQueueMessage}
@@ -748,22 +761,17 @@ export const MessageInput = forwardRef<MessageInputRef, MessageInputProps>(
)}
{shouldShowSendButton && (
<Tooltip delayDuration={0} enabledOnDesktop enabledOnMobile={false}>
<TooltipTrigger
onPress={handleSendMessage}
disabled={
isSubmitDisabled ||
isSubmitLoading ||
disabled
<TooltipTrigger
onPress={
canPressLoadingButton ? onSubmitLoadingPress : handleSendMessage
}
accessibilityLabel={isAgentRunning ? "Send and interrupt" : "Send message"}
accessibilityRole="button"
style={[
styles.sendButton,
(isSubmitDisabled ||
isSubmitLoading ||
disabled) &&
styles.buttonDisabled,
]}
disabled={isSendButtonDisabled}
accessibilityLabel={submitAccessibilityLabel}
accessibilityRole="button"
style={[
styles.sendButton,
isSendButtonDisabled && styles.buttonDisabled,
]}
>
{isSubmitLoading ? (
<ActivityIndicator size="small" color="white" />
packages/app/src/components/sidebar-agent-list.tsx
+8 -4
View File
@@ -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: {
packages/app/src/components/terminal-emulator.tsx
+21 -8
View File
@@ -26,6 +26,7 @@ interface TerminalEmulatorProps {
meta: boolean;
}) => Promise<void> | void;
onPendingModifiersConsumed?: () => Promise<void> | void;
onOutputChunkConsumed?: (sequence: number) => Promise<void> | 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) {
packages/app/src/components/terminal-pane.tsx
+126 -15
View File
@@ -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<TerminalStreamController | null>(null);
const outputPumpRef = useRef<TerminalOutputPump | null>(null);
const outputDeliveryQueueRef = useRef<TerminalOutputDeliveryQueue | null>(null);
const [selectedTerminalId, setSelectedTerminalId] = useState<string | null>(null);
const [selectedOutputChunk, setSelectedOutputChunk] = useState<TerminalOutputChunkState>({
@@ -142,16 +160,33 @@ export function TerminalPane({ serverId, cwd }: TerminalPaneProps) {
);
const hoverOutTimeoutRef = useRef<ReturnType<typeof setTimeout> | null>(null);
const selectedTerminalIdRef = useRef<string | null>(selectedTerminalId);
const pendingTerminalInputRef = useRef<PendingTerminalInput[]>([]);
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 ? (
<View style={styles.attachOverlay} testID="terminal-attach-loading">
<View
style={styles.attachOverlay}
pointerEvents="none"
testID="terminal-attach-loading"
>
<ActivityIndicator size="small" color={theme.colors.foregroundMuted} />
</View>
) : null}
packages/app/src/terminal/runtime/terminal-emulator-runtime.test.ts
+160
View File
@@ -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);
});
});
packages/app/src/terminal/runtime/terminal-emulator-runtime.ts
+95 -27
View File
@@ -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<typeof setTimeout> | 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 {
packages/app/src/terminal/runtime/terminal-output-delivery-queue.test.ts
+141
View File
@@ -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" },
]);
});
});
packages/app/src/terminal/runtime/terminal-output-delivery-queue.ts
+151
View File
@@ -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<typeof setTimeout> | 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;
}
}
packages/app/src/terminal/runtime/terminal-stream-controller.test.ts
+232 -1
View File
@@ -12,6 +12,7 @@ type FakeStreamSubscriber = (chunk: TerminalStreamControllerChunk) => void;
class FakeTerminalStreamClient implements TerminalStreamControllerClient {
private readonly streamSubscribers = new Map<number, Set<FakeStreamSubscriber>>();
private readonly pendingChunksByStreamId = new Map<number, TerminalStreamControllerChunk[]>();
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",
});
});
});
packages/app/src/terminal/runtime/terminal-stream-controller.ts
+156 -9
View File
@@ -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<void> {
const activeStream = this.activeStream;
if (!activeStream) {
packages/server/src/client/daemon-client-terminal-stream-manager.test.ts
+12 -1
View File
@@ -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", () => {
packages/server/src/client/daemon-client-terminal-stream-manager.ts
-2
View File
@@ -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,
packages/server/src/client/daemon-client.test.ts
+42
View File
@@ -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();
packages/server/src/server/agent/agent-management-mcp.ts
+6 -2
View File
@@ -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 },
packages/server/src/server/agent/agent-manager.test.ts
+69 -1
View File
@@ -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");
packages/server/src/server/agent/agent-manager.ts
+18 -7
View File
@@ -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<void> {
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;
packages/server/src/server/agent/mcp-server.ts
+6 -2
View File
@@ -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 },
packages/server/src/server/session.ts
+8 -2
View File
@@ -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 },
packages/server/src/shared/binary-mux.test.ts
+7
View File
@@ -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")));
});
});
packages/server/src/shared/binary-mux.ts
+13
View File
@@ -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;
}