Simplify terminal streaming protocol
This commit is contained in:
Mohamed Boudra
@@ -1,131 +0,0 @@
|
||||
import { describe, expect, test, vi } from "vitest";
|
||||
import {
|
||||
TerminalStreamManager,
|
||||
type TerminalStreamChunk,
|
||||
} from "./daemon-client-terminal-stream-manager.js";
|
||||
|
||||
function createChunk(input: {
|
||||
streamId: number;
|
||||
offset: number;
|
||||
data: string;
|
||||
}): TerminalStreamChunk {
|
||||
const bytes = new TextEncoder().encode(input.data);
|
||||
return {
|
||||
streamId: input.streamId,
|
||||
offset: input.offset,
|
||||
endOffset: input.offset + bytes.byteLength,
|
||||
replay: false,
|
||||
data: bytes,
|
||||
};
|
||||
}
|
||||
|
||||
describe("TerminalStreamManager", () => {
|
||||
test("buffers chunks and flushes with ack when handler subscribes", () => {
|
||||
const sendAck = vi.fn();
|
||||
const manager = new TerminalStreamManager({ sendAck });
|
||||
const seen: string[] = [];
|
||||
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 7, offset: 4, data: "hello" }),
|
||||
});
|
||||
expect(sendAck).toHaveBeenCalledTimes(1);
|
||||
expect(sendAck).toHaveBeenCalledWith({ streamId: 7, offset: 9 });
|
||||
|
||||
manager.subscribe({
|
||||
streamId: 7,
|
||||
handler: (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
},
|
||||
});
|
||||
|
||||
expect(seen).toEqual(["hello"]);
|
||||
expect(sendAck).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
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", () => {
|
||||
const sendAck = vi.fn();
|
||||
const manager = new TerminalStreamManager({ sendAck });
|
||||
|
||||
manager.subscribe({
|
||||
streamId: 11,
|
||||
handler: () => {
|
||||
throw new Error("boom");
|
||||
},
|
||||
});
|
||||
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 11, offset: 0, data: "x" }),
|
||||
});
|
||||
|
||||
expect(sendAck).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
test("evicts oldest buffered chunks when max buffered chunk count is exceeded", () => {
|
||||
const sendAck = vi.fn();
|
||||
const manager = new TerminalStreamManager({
|
||||
sendAck,
|
||||
maxBufferedChunks: 2,
|
||||
maxBufferedBytes: 1024,
|
||||
});
|
||||
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 5, offset: 0, data: "A" }),
|
||||
});
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 5, offset: 1, data: "B" }),
|
||||
});
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 5, offset: 2, data: "C" }),
|
||||
});
|
||||
expect(sendAck).toHaveBeenNthCalledWith(1, { streamId: 5, offset: 1 });
|
||||
expect(sendAck).toHaveBeenNthCalledWith(2, { streamId: 5, offset: 2 });
|
||||
expect(sendAck).toHaveBeenNthCalledWith(3, { streamId: 5, offset: 3 });
|
||||
|
||||
const seen: string[] = [];
|
||||
manager.subscribe({
|
||||
streamId: 5,
|
||||
handler: (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
},
|
||||
});
|
||||
|
||||
expect(seen).toEqual(["B", "C"]);
|
||||
expect(sendAck).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
test("tracks explicit ack offsets and skips stale auto-acks", () => {
|
||||
const sendAck = vi.fn();
|
||||
const manager = new TerminalStreamManager({ sendAck });
|
||||
|
||||
manager.subscribe({
|
||||
streamId: 13,
|
||||
handler: () => {
|
||||
// no-op
|
||||
},
|
||||
});
|
||||
manager.noteAck({ streamId: 13, offset: 10 });
|
||||
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 13, offset: 0, data: "abc" }),
|
||||
});
|
||||
expect(sendAck).not.toHaveBeenCalled();
|
||||
|
||||
manager.receiveChunk({
|
||||
chunk: createChunk({ streamId: 13, offset: 11, data: "z" }),
|
||||
});
|
||||
expect(sendAck).toHaveBeenCalledTimes(1);
|
||||
expect(sendAck).toHaveBeenCalledWith({ streamId: 13, offset: 12 });
|
||||
});
|
||||
});
|
||||
@@ -1,173 +0,0 @@
|
||||
export type TerminalStreamChunk = {
|
||||
streamId: number;
|
||||
offset: number;
|
||||
endOffset: number;
|
||||
replay: boolean;
|
||||
data: Uint8Array;
|
||||
};
|
||||
|
||||
type TerminalChunkHandler = (chunk: TerminalStreamChunk) => void;
|
||||
|
||||
type BufferedTerminalStreamQueue = {
|
||||
chunks: TerminalStreamChunk[];
|
||||
bytes: number;
|
||||
};
|
||||
|
||||
type TerminalStreamAck = {
|
||||
streamId: number;
|
||||
offset: number;
|
||||
};
|
||||
|
||||
export type TerminalStreamManagerConfig = {
|
||||
sendAck: (ack: TerminalStreamAck) => void;
|
||||
maxBufferedChunks?: number;
|
||||
maxBufferedBytes?: number;
|
||||
};
|
||||
|
||||
const DEFAULT_MAX_BUFFERED_CHUNKS = 2048;
|
||||
const DEFAULT_MAX_BUFFERED_BYTES = 2 * 1024 * 1024;
|
||||
|
||||
export class TerminalStreamManager {
|
||||
private readonly handlers: Map<number, Set<TerminalChunkHandler>> = new Map();
|
||||
private readonly bufferedChunks: Map<number, BufferedTerminalStreamQueue> = new Map();
|
||||
private readonly ackOffsets: Map<number, number> = new Map();
|
||||
private readonly maxBufferedChunks: number;
|
||||
private readonly maxBufferedBytes: number;
|
||||
|
||||
constructor(private readonly config: TerminalStreamManagerConfig) {
|
||||
this.maxBufferedChunks = config.maxBufferedChunks ?? DEFAULT_MAX_BUFFERED_CHUNKS;
|
||||
this.maxBufferedBytes = config.maxBufferedBytes ?? DEFAULT_MAX_BUFFERED_BYTES;
|
||||
}
|
||||
|
||||
clearAll(): void {
|
||||
this.handlers.clear();
|
||||
this.bufferedChunks.clear();
|
||||
this.ackOffsets.clear();
|
||||
}
|
||||
|
||||
clearStream(input: { streamId: number }): void {
|
||||
this.handlers.delete(input.streamId);
|
||||
this.bufferedChunks.delete(input.streamId);
|
||||
this.ackOffsets.delete(input.streamId);
|
||||
}
|
||||
|
||||
subscribe(input: { streamId: number; handler: TerminalChunkHandler }): () => void {
|
||||
const { streamId, handler } = input;
|
||||
if (!this.handlers.has(streamId)) {
|
||||
this.handlers.set(streamId, new Set());
|
||||
}
|
||||
|
||||
const streamHandlers = this.handlers.get(streamId)!;
|
||||
streamHandlers.add(handler);
|
||||
this.flushBufferedChunks({ streamId, handler });
|
||||
|
||||
return () => {
|
||||
streamHandlers.delete(handler);
|
||||
if (streamHandlers.size === 0) {
|
||||
this.handlers.delete(streamId);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
receiveChunk(input: { chunk: TerminalStreamChunk }): void {
|
||||
const { chunk } = input;
|
||||
const streamHandlers = this.handlers.get(chunk.streamId);
|
||||
if (!streamHandlers || streamHandlers.size === 0) {
|
||||
this.bufferChunk({ chunk });
|
||||
this.maybeAckChunk({
|
||||
streamId: chunk.streamId,
|
||||
endOffset: chunk.endOffset,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
let delivered = false;
|
||||
for (const handler of streamHandlers) {
|
||||
try {
|
||||
handler(chunk);
|
||||
delivered = true;
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
if (delivered) {
|
||||
this.maybeAckChunk({
|
||||
streamId: chunk.streamId,
|
||||
endOffset: chunk.endOffset,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
noteAck(input: TerminalStreamAck): void {
|
||||
const normalizedOffset = Math.max(0, Math.floor(input.offset));
|
||||
const previousAck = this.ackOffsets.get(input.streamId) ?? -1;
|
||||
if (normalizedOffset > previousAck) {
|
||||
this.ackOffsets.set(input.streamId, normalizedOffset);
|
||||
}
|
||||
}
|
||||
|
||||
private flushBufferedChunks(input: { streamId: number; handler: TerminalChunkHandler }): void {
|
||||
const buffered = this.bufferedChunks.get(input.streamId);
|
||||
if (!buffered || buffered.chunks.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
for (const chunk of buffered.chunks) {
|
||||
try {
|
||||
input.handler(chunk);
|
||||
this.maybeAckChunk({
|
||||
streamId: input.streamId,
|
||||
endOffset: chunk.endOffset,
|
||||
});
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
this.bufferedChunks.delete(input.streamId);
|
||||
}
|
||||
|
||||
private bufferChunk(input: { chunk: TerminalStreamChunk }): void {
|
||||
const queue = this.bufferedChunks.get(input.chunk.streamId) ?? {
|
||||
chunks: [],
|
||||
bytes: 0,
|
||||
};
|
||||
|
||||
queue.chunks.push(input.chunk);
|
||||
queue.bytes += input.chunk.data.byteLength;
|
||||
|
||||
while (queue.chunks.length > this.maxBufferedChunks || queue.bytes > this.maxBufferedBytes) {
|
||||
const removed = queue.chunks.shift();
|
||||
if (!removed) {
|
||||
break;
|
||||
}
|
||||
queue.bytes -= removed.data.byteLength;
|
||||
if (queue.bytes < 0) {
|
||||
queue.bytes = 0;
|
||||
}
|
||||
}
|
||||
|
||||
this.bufferedChunks.set(input.chunk.streamId, queue);
|
||||
}
|
||||
|
||||
private maybeAckChunk(input: { streamId: number; endOffset: number }): void {
|
||||
if (!Number.isFinite(input.endOffset) || input.endOffset < 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const normalizedEndOffset = Math.floor(input.endOffset);
|
||||
const previousAck = this.ackOffsets.get(input.streamId) ?? -1;
|
||||
if (normalizedEndOffset <= previousAck) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.ackOffsets.set(input.streamId, normalizedEndOffset);
|
||||
try {
|
||||
this.config.sendAck({
|
||||
streamId: input.streamId,
|
||||
offset: normalizedEndOffset,
|
||||
});
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,12 +1,14 @@
|
||||
import { afterEach, describe, expect, expectTypeOf, test, vi } from "vitest";
|
||||
import { DaemonClient, type DaemonTransport } from "./daemon-client";
|
||||
import {
|
||||
BinaryMuxChannel,
|
||||
TerminalBinaryMessageType,
|
||||
asUint8Array,
|
||||
decodeBinaryMuxFrame,
|
||||
encodeBinaryMuxFrame,
|
||||
} from "../shared/binary-mux.js";
|
||||
decodeTerminalResizePayload,
|
||||
decodeTerminalSnapshotPayload,
|
||||
decodeTerminalStreamFrame,
|
||||
encodeTerminalSnapshotPayload,
|
||||
encodeTerminalStreamFrame,
|
||||
TerminalStreamOpcode,
|
||||
} from "../shared/terminal-stream-protocol.js";
|
||||
|
||||
expectTypeOf<"getGitDiff" extends keyof DaemonClient ? true : false>().toEqualTypeOf<false>();
|
||||
expectTypeOf<
|
||||
@@ -1224,7 +1226,7 @@ describe("DaemonClient", () => {
|
||||
});
|
||||
});
|
||||
|
||||
test("auto-acks terminal stream chunks after delivery", async () => {
|
||||
test("emits output events for the active terminal stream", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -1242,36 +1244,40 @@ describe("DaemonClient", () => {
|
||||
await connectPromise;
|
||||
|
||||
const seen: string[] = [];
|
||||
const decoder = new TextDecoder();
|
||||
const unsubscribe = client.onTerminalStreamData(7, (chunk) => {
|
||||
seen.push(decoder.decode(chunk.data));
|
||||
const unsubscribe = client.onTerminalStreamEvent((event) => {
|
||||
if (event.type !== "output") {
|
||||
return;
|
||||
}
|
||||
seen.push(new TextDecoder().decode(event.data));
|
||||
});
|
||||
|
||||
const payload = new TextEncoder().encode("hello");
|
||||
const subscribePromise = client.subscribeTerminal("term-1", "sub-1");
|
||||
mock.triggerMessage(
|
||||
encodeBinaryMuxFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.OutputUtf8,
|
||||
streamId: 7,
|
||||
offset: 10,
|
||||
payload,
|
||||
wrapSessionMessage({
|
||||
type: "subscribe_terminal_response",
|
||||
payload: {
|
||||
terminalId: "term-1",
|
||||
state: null,
|
||||
error: null,
|
||||
requestId: "sub-1",
|
||||
},
|
||||
}),
|
||||
);
|
||||
await subscribePromise;
|
||||
|
||||
mock.triggerMessage(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
payload: new TextEncoder().encode("hello"),
|
||||
}),
|
||||
);
|
||||
|
||||
expect(seen).toEqual(["hello"]);
|
||||
expect(mock.sent).toHaveLength(1);
|
||||
const ackBytes = asUint8Array(mock.sent[0]);
|
||||
expect(ackBytes).not.toBeNull();
|
||||
const ackFrame = decodeBinaryMuxFrame(ackBytes!);
|
||||
expect(ackFrame).not.toBeNull();
|
||||
expect(ackFrame!.channel).toBe(BinaryMuxChannel.Terminal);
|
||||
expect(ackFrame!.messageType).toBe(TerminalBinaryMessageType.Ack);
|
||||
expect(ackFrame!.streamId).toBe(7);
|
||||
expect(ackFrame!.offset).toBe(15);
|
||||
unsubscribe();
|
||||
});
|
||||
|
||||
test("handles terminal binary frames delivered as UTF-8 strings", async () => {
|
||||
test("emits snapshot events for the active terminal stream", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -1288,76 +1294,100 @@ describe("DaemonClient", () => {
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const seen: string[] = [];
|
||||
client.onTerminalStreamData(11, (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
const snapshots: unknown[] = [];
|
||||
client.onTerminalStreamEvent((event) => {
|
||||
if (event.type !== "snapshot") {
|
||||
return;
|
||||
}
|
||||
snapshots.push(event.state);
|
||||
});
|
||||
|
||||
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();
|
||||
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const subscribePromise = client.subscribeTerminal("term-1", "sub-2");
|
||||
mock.triggerMessage(
|
||||
encodeBinaryMuxFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.OutputUtf8,
|
||||
streamId: 19,
|
||||
offset: 0,
|
||||
payload: new TextEncoder().encode("buffered"),
|
||||
wrapSessionMessage({
|
||||
type: "subscribe_terminal_response",
|
||||
payload: {
|
||||
terminalId: "term-1",
|
||||
state: null,
|
||||
error: null,
|
||||
requestId: "sub-2",
|
||||
},
|
||||
}),
|
||||
);
|
||||
await subscribePromise;
|
||||
|
||||
const state = {
|
||||
rows: 1,
|
||||
cols: 5,
|
||||
grid: [[{ char: "h" }, { char: "e" }, { char: "l" }, { char: "l" }, { char: "o" }]],
|
||||
scrollback: [],
|
||||
cursor: { row: 0, col: 5 },
|
||||
};
|
||||
mock.triggerMessage(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Snapshot,
|
||||
payload: encodeTerminalSnapshotPayload(state),
|
||||
}),
|
||||
);
|
||||
|
||||
expect(mock.sent).toHaveLength(1);
|
||||
const bufferedAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!);
|
||||
expect(bufferedAck?.messageType).toBe(TerminalBinaryMessageType.Ack);
|
||||
expect(bufferedAck?.streamId).toBe(19);
|
||||
expect(bufferedAck?.offset).toBe(8);
|
||||
mock.sent.length = 0;
|
||||
|
||||
const seen: string[] = [];
|
||||
client.onTerminalStreamData(19, (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
});
|
||||
|
||||
expect(seen.join("")).toContain("buffered");
|
||||
expect(mock.sent).toHaveLength(0);
|
||||
expect(snapshots).toEqual([state]);
|
||||
});
|
||||
|
||||
test("stops delivering and acking after terminal_stream_exit", async () => {
|
||||
test("sends input and resize frames for the active terminal", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const subscribePromise = client.subscribeTerminal("term-1", "sub-3");
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "subscribe_terminal_response",
|
||||
payload: {
|
||||
terminalId: "term-1",
|
||||
state: null,
|
||||
error: null,
|
||||
requestId: "sub-3",
|
||||
},
|
||||
}),
|
||||
);
|
||||
await subscribePromise;
|
||||
mock.sent.length = 0;
|
||||
|
||||
client.sendTerminalInput("term-1", {
|
||||
type: "input",
|
||||
data: "echo hello\r",
|
||||
});
|
||||
client.sendTerminalInput("term-1", {
|
||||
type: "resize",
|
||||
rows: 24,
|
||||
cols: 80,
|
||||
});
|
||||
|
||||
const inputFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[0])!);
|
||||
const resizeFrame = decodeTerminalStreamFrame(asUint8Array(mock.sent[1])!);
|
||||
|
||||
expect(inputFrame?.opcode).toBe(TerminalStreamOpcode.Input);
|
||||
expect(new TextDecoder().decode(inputFrame?.payload ?? new Uint8Array())).toBe("echo hello\r");
|
||||
expect(resizeFrame?.opcode).toBe(TerminalStreamOpcode.Resize);
|
||||
expect(decodeTerminalResizePayload(resizeFrame?.payload ?? new Uint8Array())).toEqual({
|
||||
rows: 24,
|
||||
cols: 80,
|
||||
});
|
||||
});
|
||||
|
||||
test("ignores terminal stream frames after terminal_stream_exit", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -1375,114 +1405,55 @@ describe("DaemonClient", () => {
|
||||
await connectPromise;
|
||||
|
||||
const seen: string[] = [];
|
||||
const unsubscribe = client.onTerminalStreamData(23, (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
const unsubscribe = client.onTerminalStreamEvent((event) => {
|
||||
if (event.type !== "output") {
|
||||
return;
|
||||
}
|
||||
seen.push(new TextDecoder().decode(event.data));
|
||||
});
|
||||
|
||||
const subscribePromise = client.subscribeTerminal("term-1", "sub-4");
|
||||
mock.triggerMessage(
|
||||
encodeBinaryMuxFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.OutputUtf8,
|
||||
streamId: 23,
|
||||
offset: 0,
|
||||
wrapSessionMessage({
|
||||
type: "subscribe_terminal_response",
|
||||
payload: {
|
||||
terminalId: "term-1",
|
||||
state: null,
|
||||
error: null,
|
||||
requestId: "sub-4",
|
||||
},
|
||||
}),
|
||||
);
|
||||
await subscribePromise;
|
||||
|
||||
mock.triggerMessage(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
payload: new TextEncoder().encode("before-exit"),
|
||||
}),
|
||||
);
|
||||
expect(seen).toEqual(["before-exit"]);
|
||||
expect(mock.sent).toHaveLength(1);
|
||||
mock.sent.length = 0;
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "terminal_stream_exit",
|
||||
payload: {
|
||||
streamId: 23,
|
||||
terminalId: "term-1",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
mock.triggerMessage(
|
||||
encodeBinaryMuxFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.OutputUtf8,
|
||||
streamId: 23,
|
||||
offset: 12,
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Output,
|
||||
payload: new TextEncoder().encode("after-exit"),
|
||||
}),
|
||||
);
|
||||
|
||||
expect(seen).toEqual(["before-exit"]);
|
||||
expect(mock.sent).toHaveLength(1);
|
||||
const postExitAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!);
|
||||
expect(postExitAck?.messageType).toBe(TerminalBinaryMessageType.Ack);
|
||||
expect(postExitAck?.streamId).toBe(23);
|
||||
expect(postExitAck?.offset).toBe(22);
|
||||
unsubscribe();
|
||||
});
|
||||
|
||||
test("cleans local stream state even when detach reports success=false", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
const client = new DaemonClient({
|
||||
url: "ws://test",
|
||||
clientId: "clsk_unit_test",
|
||||
logger,
|
||||
reconnect: { enabled: false },
|
||||
transportFactory: () => mock.transport,
|
||||
});
|
||||
clients.push(client);
|
||||
|
||||
const connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const seen: string[] = [];
|
||||
client.onTerminalStreamData(31, (chunk) => {
|
||||
seen.push(new TextDecoder().decode(chunk.data));
|
||||
});
|
||||
|
||||
const detachPromise = client.detachTerminalStream(31, "detach-31");
|
||||
const detachRequest = JSON.parse(String(mock.sent[0])) as {
|
||||
type: "session";
|
||||
message: { type: string; streamId: number; requestId: string };
|
||||
};
|
||||
expect(detachRequest.message.type).toBe("detach_terminal_stream_request");
|
||||
expect(detachRequest.message.streamId).toBe(31);
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "detach_terminal_stream_response",
|
||||
payload: {
|
||||
streamId: 31,
|
||||
success: false,
|
||||
requestId: "detach-31",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const payload = await detachPromise;
|
||||
expect(payload.success).toBe(false);
|
||||
|
||||
mock.sent.length = 0;
|
||||
mock.triggerMessage(
|
||||
encodeBinaryMuxFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.OutputUtf8,
|
||||
streamId: 31,
|
||||
offset: 0,
|
||||
payload: new TextEncoder().encode("after-detach"),
|
||||
}),
|
||||
);
|
||||
|
||||
expect(seen).toEqual([]);
|
||||
expect(mock.sent).toHaveLength(1);
|
||||
const detachedAck = decodeBinaryMuxFrame(asUint8Array(mock.sent[0])!);
|
||||
expect(detachedAck?.messageType).toBe(TerminalBinaryMessageType.Ack);
|
||||
expect(detachedAck?.streamId).toBe(31);
|
||||
expect(detachedAck?.offset).toBe(12);
|
||||
});
|
||||
|
||||
test("parses canonical agent_stream tool_call payloads without crashing", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -44,10 +44,8 @@ import type {
|
||||
ListTerminalsResponse,
|
||||
CreateTerminalResponse,
|
||||
SubscribeTerminalResponse,
|
||||
TerminalOutput,
|
||||
TerminalState,
|
||||
KillTerminalResponse,
|
||||
AttachTerminalStreamResponse,
|
||||
DetachTerminalStreamResponse,
|
||||
TerminalInput,
|
||||
SessionInboundMessage,
|
||||
SessionOutboundMessage,
|
||||
@@ -63,18 +61,14 @@ import { getAgentProviderDefinition } from "../server/agent/provider-manifest.js
|
||||
import { isRelayClientWebSocketUrl } from "../shared/daemon-endpoints.js";
|
||||
import {
|
||||
asUint8Array,
|
||||
decodeBinaryMuxFrame,
|
||||
encodeBinaryMuxFrame,
|
||||
BinaryMuxChannel,
|
||||
TerminalBinaryFlags,
|
||||
TerminalBinaryMessageType,
|
||||
type BinaryMuxFrame,
|
||||
} from "../shared/binary-mux.js";
|
||||
decodeTerminalSnapshotPayload,
|
||||
decodeTerminalStreamFrame,
|
||||
encodeTerminalResizePayload,
|
||||
encodeTerminalStreamFrame,
|
||||
TerminalStreamOpcode,
|
||||
type TerminalStreamFrame,
|
||||
} from "../shared/terminal-stream-protocol.js";
|
||||
import { encodeTerminalKeyInput, type TerminalKeyInput } from "../shared/terminal-key-input.js";
|
||||
import {
|
||||
TerminalStreamManager,
|
||||
type TerminalStreamChunk,
|
||||
} from "./daemon-client-terminal-stream-manager.js";
|
||||
import {
|
||||
createRelayE2eeTransportFactory,
|
||||
createWebSocketTransportFactory,
|
||||
@@ -108,7 +102,10 @@ export type {
|
||||
WebSocketFactory,
|
||||
WebSocketLike,
|
||||
} from "./daemon-client-transport.js";
|
||||
export type { TerminalStreamChunk } from "./daemon-client-terminal-stream-manager.js";
|
||||
|
||||
export type TerminalStreamEvent =
|
||||
| { terminalId: string; type: "output"; data: Uint8Array }
|
||||
| { terminalId: string; type: "snapshot"; state: TerminalState };
|
||||
|
||||
export type ConnectionState =
|
||||
| { status: "idle" }
|
||||
@@ -244,10 +241,7 @@ type AgentPermissionResolvedPayload = AgentPermissionResolvedMessage["payload"];
|
||||
type ListTerminalsPayload = ListTerminalsResponse["payload"];
|
||||
type CreateTerminalPayload = CreateTerminalResponse["payload"];
|
||||
type SubscribeTerminalPayload = SubscribeTerminalResponse["payload"];
|
||||
type TerminalOutputPayload = TerminalOutput["payload"];
|
||||
type KillTerminalPayload = KillTerminalResponse["payload"];
|
||||
type AttachTerminalStreamPayload = AttachTerminalStreamResponse["payload"];
|
||||
type DetachTerminalStreamPayload = DetachTerminalStreamResponse["payload"];
|
||||
export type FetchAgentTimelinePayload = FetchAgentTimelineResponseMessage["payload"];
|
||||
|
||||
export type FetchAgentTimelineDirection = FetchAgentTimelinePayload["direction"];
|
||||
@@ -426,9 +420,10 @@ export class DaemonClient {
|
||||
{ cwd: string; compare: { mode: "uncommitted" | "base"; baseRef?: string } }
|
||||
>();
|
||||
private terminalDirectorySubscriptions = new Set<string>();
|
||||
private activeTerminalId: string | null = null;
|
||||
private readonly terminalStreamListeners = new Set<(event: TerminalStreamEvent) => void>();
|
||||
private logger: Logger;
|
||||
private pendingSendQueue: PendingSend[] = [];
|
||||
private terminalStreams: TerminalStreamManager;
|
||||
private readonly logConnectionPath: "direct" | "relay";
|
||||
private readonly logServerId: string | null;
|
||||
private readonly logClientIdHash: string;
|
||||
@@ -457,17 +452,6 @@ export class DaemonClient {
|
||||
Number.isFinite(this.config.runtimeGeneration)
|
||||
? this.config.runtimeGeneration
|
||||
: null;
|
||||
this.terminalStreams = new TerminalStreamManager({
|
||||
sendAck: (ack) => {
|
||||
this.sendBinaryFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.Ack,
|
||||
streamId: ack.streamId,
|
||||
offset: ack.offset,
|
||||
payload: new Uint8Array(0),
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -676,7 +660,7 @@ export class DaemonClient {
|
||||
this.disposeTransport(1000, "Client closed");
|
||||
this.clearWaiters(new Error("Daemon client closed"));
|
||||
this.rejectPendingSendQueue(new Error("Daemon client closed"));
|
||||
this.terminalStreams.clearAll();
|
||||
this.activeTerminalId = null;
|
||||
this.lastServerInfoMessage = null;
|
||||
this.updateConnectionState(
|
||||
{ status: "disposed" },
|
||||
@@ -800,7 +784,7 @@ export class DaemonClient {
|
||||
}
|
||||
}
|
||||
|
||||
private sendBinaryFrame(frame: BinaryMuxFrame): void {
|
||||
private sendBinaryFrame(frame: Uint8Array): void {
|
||||
if (!this.transport || this.connectionState.status !== "connected") {
|
||||
if (this.config.suppressSendErrors) {
|
||||
return;
|
||||
@@ -808,7 +792,7 @@ export class DaemonClient {
|
||||
throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
|
||||
}
|
||||
try {
|
||||
this.transport.send(encodeBinaryMuxFrame(frame));
|
||||
this.transport.send(frame);
|
||||
} catch (error) {
|
||||
if (this.config.suppressSendErrors) {
|
||||
return;
|
||||
@@ -2622,16 +2606,23 @@ export class DaemonClient {
|
||||
terminalId,
|
||||
requestId: resolvedRequestId,
|
||||
});
|
||||
return this.sendCorrelatedRequest({
|
||||
const payload = await this.sendCorrelatedRequest({
|
||||
requestId: resolvedRequestId,
|
||||
message,
|
||||
responseType: "subscribe_terminal_response",
|
||||
timeout: 10000,
|
||||
options: { skipQueue: true },
|
||||
});
|
||||
if (!payload.error) {
|
||||
this.activeTerminalId = terminalId;
|
||||
}
|
||||
return payload;
|
||||
}
|
||||
|
||||
unsubscribeTerminal(terminalId: string): void {
|
||||
if (this.activeTerminalId === terminalId) {
|
||||
this.activeTerminalId = null;
|
||||
}
|
||||
this.sendSessionMessage({
|
||||
type: "unsubscribe_terminal_request",
|
||||
terminalId,
|
||||
@@ -2639,6 +2630,29 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
sendTerminalInput(terminalId: string, message: TerminalInput["message"]): void {
|
||||
if (this.activeTerminalId === terminalId) {
|
||||
if (message.type === "input") {
|
||||
this.sendBinaryFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
payload: encodeUtf8String(message.data),
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
if (message.type === "resize") {
|
||||
this.sendBinaryFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Resize,
|
||||
payload: encodeTerminalResizePayload({
|
||||
rows: message.rows,
|
||||
cols: message.cols,
|
||||
}),
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
this.sendSessionMessage({
|
||||
type: "terminal_input",
|
||||
terminalId,
|
||||
@@ -2662,127 +2676,44 @@ export class DaemonClient {
|
||||
});
|
||||
}
|
||||
|
||||
async attachTerminalStream(
|
||||
terminalId: string,
|
||||
options?: {
|
||||
resumeOffset?: number;
|
||||
rows?: number;
|
||||
cols?: number;
|
||||
},
|
||||
requestId?: string,
|
||||
): Promise<AttachTerminalStreamPayload> {
|
||||
const resolvedRequestId = this.createRequestId(requestId);
|
||||
const message = SessionInboundMessageSchema.parse({
|
||||
type: "attach_terminal_stream_request",
|
||||
terminalId,
|
||||
requestId: resolvedRequestId,
|
||||
...(options?.resumeOffset !== undefined ? { resumeOffset: options.resumeOffset } : {}),
|
||||
...(options?.rows !== undefined ? { rows: options.rows } : {}),
|
||||
...(options?.cols !== undefined ? { cols: options.cols } : {}),
|
||||
});
|
||||
return this.sendCorrelatedRequest({
|
||||
requestId: resolvedRequestId,
|
||||
message,
|
||||
responseType: "attach_terminal_stream_response",
|
||||
timeout: 10000,
|
||||
options: { skipQueue: true },
|
||||
});
|
||||
onTerminalStreamEvent(handler: (event: TerminalStreamEvent) => void): () => void {
|
||||
this.terminalStreamListeners.add(handler);
|
||||
return () => {
|
||||
this.terminalStreamListeners.delete(handler);
|
||||
};
|
||||
}
|
||||
|
||||
async detachTerminalStream(
|
||||
streamId: number,
|
||||
requestId?: string,
|
||||
): Promise<DetachTerminalStreamPayload> {
|
||||
const resolvedRequestId = this.createRequestId(requestId);
|
||||
const message = SessionInboundMessageSchema.parse({
|
||||
type: "detach_terminal_stream_request",
|
||||
streamId,
|
||||
requestId: resolvedRequestId,
|
||||
});
|
||||
const payload = await this.sendCorrelatedRequest({
|
||||
requestId: resolvedRequestId,
|
||||
message,
|
||||
responseType: "detach_terminal_stream_response",
|
||||
timeout: 10000,
|
||||
options: { skipQueue: true },
|
||||
});
|
||||
this.terminalStreams.clearStream({ streamId });
|
||||
return payload;
|
||||
}
|
||||
|
||||
onTerminalStreamData(
|
||||
streamId: number,
|
||||
handler: (chunk: TerminalStreamChunk) => void,
|
||||
): () => void {
|
||||
return this.terminalStreams.subscribe({ streamId, handler });
|
||||
}
|
||||
|
||||
async waitForTerminalStreamData(
|
||||
streamId: number,
|
||||
predicate: (chunk: TerminalStreamChunk) => boolean,
|
||||
async waitForTerminalStreamEvent(
|
||||
predicate: (event: TerminalStreamEvent) => boolean,
|
||||
timeout = 5000,
|
||||
): Promise<TerminalStreamChunk> {
|
||||
return new Promise<TerminalStreamChunk>((resolve, reject) => {
|
||||
): Promise<TerminalStreamEvent> {
|
||||
return new Promise<TerminalStreamEvent>((resolve, reject) => {
|
||||
const timeoutHandle = setTimeout(() => {
|
||||
unsubscribe();
|
||||
reject(new Error(`Timeout waiting for terminal stream data (${timeout}ms)`));
|
||||
reject(new Error(`Timeout waiting for terminal stream event (${timeout}ms)`));
|
||||
}, timeout);
|
||||
|
||||
const unsubscribe = this.onTerminalStreamData(streamId, (chunk) => {
|
||||
if (!predicate(chunk)) {
|
||||
const unsubscribe = this.onTerminalStreamEvent((event) => {
|
||||
if (!predicate(event)) {
|
||||
return;
|
||||
}
|
||||
clearTimeout(timeoutHandle);
|
||||
unsubscribe();
|
||||
resolve(chunk);
|
||||
resolve(event);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
sendTerminalStreamInput(streamId: number, data: string | Uint8Array): void {
|
||||
const payload = typeof data === "string" ? encodeUtf8String(data) : data;
|
||||
this.sendBinaryFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.InputUtf8,
|
||||
streamId,
|
||||
offset: 0,
|
||||
payload,
|
||||
});
|
||||
}
|
||||
|
||||
sendTerminalStreamKey(streamId: number, input: TerminalKeyInput): void {
|
||||
sendTerminalStreamKey(input: TerminalKeyInput): void {
|
||||
const encoded = encodeTerminalKeyInput(input);
|
||||
if (!encoded) {
|
||||
return;
|
||||
}
|
||||
this.sendTerminalStreamInput(streamId, encoded);
|
||||
}
|
||||
|
||||
sendTerminalStreamAck(streamId: number, offset: number): void {
|
||||
const normalizedOffset = Math.max(0, Math.floor(offset));
|
||||
this.terminalStreams.noteAck({ streamId, offset: normalizedOffset });
|
||||
this.sendBinaryFrame({
|
||||
channel: BinaryMuxChannel.Terminal,
|
||||
messageType: TerminalBinaryMessageType.Ack,
|
||||
streamId,
|
||||
offset: normalizedOffset,
|
||||
payload: new Uint8Array(0),
|
||||
});
|
||||
}
|
||||
|
||||
async waitForTerminalOutput(terminalId: string, timeout = 5000): Promise<TerminalOutputPayload> {
|
||||
return this.waitFor(
|
||||
(msg) => {
|
||||
if (msg.type !== "terminal_output") {
|
||||
return null;
|
||||
}
|
||||
if (msg.payload.terminalId !== terminalId) {
|
||||
return null;
|
||||
}
|
||||
return msg.payload;
|
||||
},
|
||||
timeout,
|
||||
{ skipQueue: true },
|
||||
this.sendBinaryFrame(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
payload: encodeUtf8String(encoded),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
@@ -2890,7 +2821,7 @@ export class DaemonClient {
|
||||
|
||||
const rawBytes = asUint8Array(rawData);
|
||||
if (rawBytes) {
|
||||
const frame = decodeBinaryMuxFrame(rawBytes);
|
||||
const frame = decodeTerminalStreamFrame(rawBytes);
|
||||
if (frame) {
|
||||
this.handleBinaryFrame(frame);
|
||||
return;
|
||||
@@ -2922,21 +2853,42 @@ export class DaemonClient {
|
||||
this.handleSessionMessage(parsed.data.message);
|
||||
}
|
||||
|
||||
private handleBinaryFrame(frame: BinaryMuxFrame): void {
|
||||
if (
|
||||
frame.channel === BinaryMuxChannel.Terminal &&
|
||||
frame.messageType === TerminalBinaryMessageType.OutputUtf8
|
||||
) {
|
||||
const chunk: TerminalStreamChunk = {
|
||||
streamId: frame.streamId,
|
||||
offset: frame.offset,
|
||||
endOffset: frame.offset + (frame.payload?.byteLength ?? 0),
|
||||
replay: Boolean((frame.flags ?? 0) & TerminalBinaryFlags.Replay),
|
||||
data: frame.payload ?? new Uint8Array(0),
|
||||
};
|
||||
this.terminalStreams.receiveChunk({ chunk });
|
||||
private handleBinaryFrame(frame: TerminalStreamFrame): void {
|
||||
const terminalId = this.activeTerminalId;
|
||||
if (!terminalId) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (frame.opcode === TerminalStreamOpcode.Output) {
|
||||
this.emitTerminalStreamEvent({
|
||||
terminalId,
|
||||
type: "output",
|
||||
data: frame.payload,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if (frame.opcode === TerminalStreamOpcode.Snapshot) {
|
||||
const state = decodeTerminalSnapshotPayload(frame.payload);
|
||||
if (!state) {
|
||||
return;
|
||||
}
|
||||
this.emitTerminalStreamEvent({
|
||||
terminalId,
|
||||
type: "snapshot",
|
||||
state,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
private emitTerminalStreamEvent(event: TerminalStreamEvent): void {
|
||||
for (const listener of this.terminalStreamListeners) {
|
||||
try {
|
||||
listener(event);
|
||||
} catch {
|
||||
// no-op
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private updateConnectionState(
|
||||
@@ -2996,7 +2948,7 @@ export class DaemonClient {
|
||||
// and responses from the previous connection will never arrive.
|
||||
this.clearWaiters(new Error(reason ?? "Connection lost"));
|
||||
this.rejectPendingSendQueue(new Error(reason ?? "Connection lost"));
|
||||
this.terminalStreams.clearAll();
|
||||
this.activeTerminalId = null;
|
||||
this.lastServerInfoMessage = null;
|
||||
|
||||
if (wasDisposed) {
|
||||
@@ -3051,7 +3003,9 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
if (msg.type === "terminal_stream_exit") {
|
||||
this.terminalStreams.clearStream({ streamId: msg.payload.streamId });
|
||||
if (this.activeTerminalId === msg.payload.terminalId) {
|
||||
this.activeTerminalId = null;
|
||||
}
|
||||
}
|
||||
|
||||
if (this.rawMessageListeners.size > 0) {
|
||||
@@ -3153,14 +3107,6 @@ export class DaemonClient {
|
||||
}
|
||||
}
|
||||
|
||||
private async waitFor<T>(
|
||||
predicate: (msg: SessionOutboundMessage) => T | null,
|
||||
timeout = 30000,
|
||||
_options?: { skipQueue?: boolean },
|
||||
): Promise<T> {
|
||||
return this.waitForWithCancel(predicate, timeout, _options).promise;
|
||||
}
|
||||
|
||||
private waitForWithCancel<T>(
|
||||
predicate: (msg: SessionOutboundMessage) => T | null,
|
||||
timeout = 30000,
|
||||
|
||||
Reference in New Issue
Block a user