test: add manual terminal perf harness
This commit is contained in:
Mohamed Boudra
@@ -19,6 +19,13 @@ export type TerminalPerfDaemonClient = {
|
||||
terminal: { id: string; name: string; cwd: string } | null;
|
||||
error: string | null;
|
||||
}>;
|
||||
createAgent(options: {
|
||||
provider: string;
|
||||
cwd: string;
|
||||
title?: string;
|
||||
modeId?: string;
|
||||
}): Promise<{ id: string; status: string }>;
|
||||
sendAgentMessage(agentId: string, text: string): Promise<void>;
|
||||
subscribeTerminal(
|
||||
terminalId: string,
|
||||
): Promise<{ terminalId: string; slot: number; error: null } | { error: string }>;
|
||||
|
||||
@@ -0,0 +1,350 @@
|
||||
import { performance } from "node:perf_hooks";
|
||||
import type { Page, TestInfo } from "@playwright/test";
|
||||
import { test, expect } from "./fixtures";
|
||||
import { TerminalE2EHarness, type TerminalInstance } from "./helpers/terminal-dsl";
|
||||
import {
|
||||
installTerminalKeystrokeStressProbe,
|
||||
readTerminalKeystrokeStressReport,
|
||||
resetTerminalKeystrokeStressProbe,
|
||||
type LatencyStats,
|
||||
} from "./helpers/terminal-probes";
|
||||
import { waitForTerminalContent } from "./helpers/terminal-perf";
|
||||
|
||||
const INPUT_TEXT = buildStressText(600);
|
||||
const BIG_DIFF_BYTES = 256_000;
|
||||
const SMALL_AGENT_STREAM_UPDATES = 1000;
|
||||
const STRESS_TIMEOUT_MS = 15_000;
|
||||
const RUN_MANUAL_TERMINAL_PERF = process.env.PASEO_TERMINAL_PERF_E2E === "1";
|
||||
const terminalPerfDescribe = RUN_MANUAL_TERMINAL_PERF ? test.describe : test.describe.skip;
|
||||
|
||||
type DaemonEchoReport = {
|
||||
inputTextLength: number;
|
||||
inputFrameCount: number;
|
||||
outputEventCount: number;
|
||||
echoedBytes: number;
|
||||
sendToOutputMs: LatencyStats;
|
||||
firstSendAt: number;
|
||||
lastOutputAt: number;
|
||||
};
|
||||
|
||||
terminalPerfDescribe("Terminal keystroke stress", () => {
|
||||
let harness: TerminalE2EHarness;
|
||||
|
||||
test.beforeAll(async () => {
|
||||
harness = await TerminalE2EHarness.create({ tempPrefix: "terminal-key-stress-" });
|
||||
});
|
||||
|
||||
test.afterAll(async () => {
|
||||
await harness?.cleanup();
|
||||
});
|
||||
|
||||
test("logs daemon-only and app keystroke echo latency under burst input", async ({
|
||||
page,
|
||||
}, testInfo) => {
|
||||
test.setTimeout(75_000);
|
||||
|
||||
const daemonTerminal = await harness.createTerminal({ name: "daemon-keystroke-baseline" });
|
||||
try {
|
||||
const daemonReport = await measureDaemonBurstEcho(harness, daemonTerminal, INPUT_TEXT);
|
||||
await attachJson(testInfo, "daemon-keystroke-baseline", daemonReport);
|
||||
console.log("[terminal-stress-daemon]", JSON.stringify(daemonReport));
|
||||
|
||||
expect(daemonReport.echoedBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
} finally {
|
||||
await harness.killTerminal(daemonTerminal.id);
|
||||
}
|
||||
|
||||
await installTerminalKeystrokeStressProbe(page);
|
||||
const appBaselineReport = await measureAppBurstEcho({
|
||||
page,
|
||||
harness,
|
||||
terminalName: "app-keystroke-stress",
|
||||
});
|
||||
await attachJson(testInfo, "app-keystroke-stress", appBaselineReport);
|
||||
console.log("[terminal-stress-app]", JSON.stringify(appBaselineReport));
|
||||
|
||||
expect(appBaselineReport.keydownCount).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBaselineReport.inputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBaselineReport.outputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBaselineReport.keydownToXtermCommitMs?.count ?? 0).toBeGreaterThan(0);
|
||||
|
||||
const appSmallChunksReport = await measureAppBurstEcho({
|
||||
page,
|
||||
harness,
|
||||
terminalName: "app-keystroke-stress-small-agent-chunks",
|
||||
agentStreamUpdateCount: SMALL_AGENT_STREAM_UPDATES,
|
||||
});
|
||||
await attachJson(testInfo, "app-keystroke-stress-small-agent-chunks", appSmallChunksReport);
|
||||
console.log("[terminal-stress-app-small-agent-chunks]", JSON.stringify(appSmallChunksReport));
|
||||
|
||||
expect(appSmallChunksReport.keydownCount).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appSmallChunksReport.inputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appSmallChunksReport.outputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appSmallChunksReport.keydownToXtermCommitMs?.count ?? 0).toBeGreaterThan(0);
|
||||
expect(appSmallChunksReport.agentStreamTextMessageCount).toBeGreaterThanOrEqual(
|
||||
SMALL_AGENT_STREAM_UPDATES,
|
||||
);
|
||||
|
||||
const appBigDiffReport = await measureAppBurstEcho({
|
||||
page,
|
||||
harness,
|
||||
terminalName: "app-keystroke-stress-big-diff",
|
||||
bigDiffBytes: BIG_DIFF_BYTES,
|
||||
});
|
||||
await attachJson(testInfo, "app-keystroke-stress-big-diff", appBigDiffReport);
|
||||
console.log("[terminal-stress-app-big-diff]", JSON.stringify(appBigDiffReport));
|
||||
|
||||
expect(appBigDiffReport.keydownCount).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBigDiffReport.inputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBigDiffReport.outputFramePayloadBytes).toBeGreaterThanOrEqual(INPUT_TEXT.length);
|
||||
expect(appBigDiffReport.keydownToXtermCommitMs?.count ?? 0).toBeGreaterThan(0);
|
||||
expect(appBigDiffReport.largeAgentStreamTextMessageCount).toBeGreaterThanOrEqual(1);
|
||||
expect(appBigDiffReport.largestAgentStreamTextMessageBytes).toBeGreaterThanOrEqual(
|
||||
BIG_DIFF_BYTES,
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
async function measureAppBurstEcho(input: {
|
||||
page: Page;
|
||||
harness: TerminalE2EHarness;
|
||||
terminalName: string;
|
||||
bigDiffBytes?: number;
|
||||
agentStreamUpdateCount?: number;
|
||||
}) {
|
||||
const appTerminal = await input.harness.createTerminal({ name: input.terminalName });
|
||||
try {
|
||||
await input.harness.openTerminal(input.page, { terminalId: appTerminal.id });
|
||||
await input.harness.setupPrompt(input.page);
|
||||
|
||||
const agent =
|
||||
input.bigDiffBytes === undefined && input.agentStreamUpdateCount === undefined
|
||||
? null
|
||||
: await input.harness.client.createAgent({
|
||||
provider: "mock",
|
||||
cwd: input.harness.tempRepo.path,
|
||||
title: "Large WebSocket payload",
|
||||
modeId: "load-test",
|
||||
});
|
||||
|
||||
const bigDiffBytes = input.bigDiffBytes;
|
||||
const agentStreamUpdateCount = input.agentStreamUpdateCount;
|
||||
const startAgentLoad =
|
||||
agent === null
|
||||
? null
|
||||
: () => {
|
||||
if (bigDiffBytes !== undefined) {
|
||||
return emitLargeDiffAgentPayload(input.harness, {
|
||||
agentId: agent.id,
|
||||
bytes: bigDiffBytes,
|
||||
});
|
||||
}
|
||||
if (agentStreamUpdateCount !== undefined) {
|
||||
return emitRapidAgentStreamUpdates(input.harness, {
|
||||
agentId: agent.id,
|
||||
count: agentStreamUpdateCount,
|
||||
});
|
||||
}
|
||||
return Promise.resolve();
|
||||
};
|
||||
|
||||
const terminal = input.harness.terminalSurface(input.page);
|
||||
await terminal.press("Control+c");
|
||||
await input.page.waitForTimeout(200);
|
||||
await resetTerminalKeystrokeStressProbe(input.page);
|
||||
|
||||
const activeAgentLoadPromise = startAgentLoad === null ? Promise.resolve() : startAgentLoad();
|
||||
|
||||
await terminal.pressSequentially(INPUT_TEXT, { delay: 0 });
|
||||
await waitForAppStressEcho(input.page, INPUT_TEXT);
|
||||
await waitForAppProbePayload(input.page, INPUT_TEXT.length);
|
||||
if (input.bigDiffBytes !== undefined) {
|
||||
await waitForLargeAgentStreamMessage(input.page, input.bigDiffBytes);
|
||||
}
|
||||
if (input.agentStreamUpdateCount !== undefined) {
|
||||
await waitForAgentStreamMessages(input.page, input.agentStreamUpdateCount);
|
||||
}
|
||||
await activeAgentLoadPromise;
|
||||
|
||||
return readTerminalKeystrokeStressReport(input.page, INPUT_TEXT);
|
||||
} finally {
|
||||
await input.harness.killTerminal(appTerminal.id);
|
||||
}
|
||||
}
|
||||
|
||||
async function emitRapidAgentStreamUpdates(
|
||||
harness: TerminalE2EHarness,
|
||||
input: { agentId: string; count: number },
|
||||
): Promise<void> {
|
||||
await harness.client.sendAgentMessage(input.agentId, `emit ${input.count} agent stream updates`);
|
||||
}
|
||||
|
||||
async function emitLargeDiffAgentPayload(
|
||||
harness: TerminalE2EHarness,
|
||||
input: { agentId: string; bytes: number },
|
||||
): Promise<void> {
|
||||
await harness.client.sendAgentMessage(
|
||||
input.agentId,
|
||||
`emit ${input.bytes} byte large diff agent stream update`,
|
||||
);
|
||||
}
|
||||
|
||||
async function measureDaemonBurstEcho(
|
||||
harness: TerminalE2EHarness,
|
||||
terminal: TerminalInstance,
|
||||
inputText: string,
|
||||
): Promise<DaemonEchoReport> {
|
||||
await harness.client.subscribeTerminal(terminal.id);
|
||||
|
||||
const outputTimesByByte: number[] = [];
|
||||
let outputEventCount = 0;
|
||||
let echoedBytes = 0;
|
||||
const decoder = new TextDecoder();
|
||||
const unsubscribe = harness.client.onTerminalStreamEvent((event) => {
|
||||
if (event.terminalId !== terminal.id || event.type !== "output" || !event.data) {
|
||||
return;
|
||||
}
|
||||
outputEventCount += 1;
|
||||
const text = decoder.decode(event.data);
|
||||
const now = performance.now();
|
||||
const previousEchoedBytes = echoedBytes;
|
||||
echoedBytes += text.length;
|
||||
for (let index = previousEchoedBytes; index < echoedBytes; index += 1) {
|
||||
outputTimesByByte[index] = now;
|
||||
}
|
||||
});
|
||||
|
||||
try {
|
||||
await new Promise((resolve) => setTimeout(resolve, 250));
|
||||
echoedBytes = 0;
|
||||
outputTimesByByte.length = 0;
|
||||
|
||||
const sendTimes: number[] = [];
|
||||
for (const char of inputText) {
|
||||
sendTimes.push(performance.now());
|
||||
harness.client.sendTerminalInput(terminal.id, {
|
||||
type: "input",
|
||||
data: char,
|
||||
});
|
||||
}
|
||||
|
||||
await waitForDaemonEchoBytes({
|
||||
getEchoedBytes: () => echoedBytes,
|
||||
expectedBytes: inputText.length,
|
||||
timeoutMs: STRESS_TIMEOUT_MS,
|
||||
});
|
||||
|
||||
const latencies = sendTimes.map((sentAt, index) => outputTimesByByte[index]! - sentAt);
|
||||
return {
|
||||
inputTextLength: inputText.length,
|
||||
inputFrameCount: sendTimes.length,
|
||||
outputEventCount,
|
||||
echoedBytes,
|
||||
sendToOutputMs: summarizeLatency(latencies),
|
||||
firstSendAt: sendTimes[0] ?? 0,
|
||||
lastOutputAt: outputTimesByByte[inputText.length - 1] ?? 0,
|
||||
};
|
||||
} finally {
|
||||
unsubscribe();
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForDaemonEchoBytes(input: {
|
||||
getEchoedBytes: () => number;
|
||||
expectedBytes: number;
|
||||
timeoutMs: number;
|
||||
}): Promise<void> {
|
||||
const deadline = Date.now() + input.timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
if (input.getEchoedBytes() >= input.expectedBytes) {
|
||||
return;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
throw new Error(
|
||||
`Timed out waiting for daemon echo bytes: ${input.getEchoedBytes()}/${input.expectedBytes}`,
|
||||
);
|
||||
}
|
||||
|
||||
async function waitForAppStressEcho(page: Page, text: string): Promise<void> {
|
||||
const tail = text.slice(-80);
|
||||
await waitForTerminalContent(page, (content) => content.includes(tail), STRESS_TIMEOUT_MS);
|
||||
}
|
||||
|
||||
async function waitForAppProbePayload(page: Page, expectedBytes: number): Promise<void> {
|
||||
const deadline = Date.now() + STRESS_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
const report = await readTerminalKeystrokeStressReport(page, INPUT_TEXT);
|
||||
if (report.outputFramePayloadBytes >= expectedBytes) {
|
||||
return;
|
||||
}
|
||||
await page.waitForTimeout(25);
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForLargeAgentStreamMessage(page: Page, expectedBytes: number): Promise<void> {
|
||||
const deadline = Date.now() + STRESS_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
const report = await readTerminalKeystrokeStressReport(page, INPUT_TEXT);
|
||||
if (report.largestAgentStreamTextMessageBytes >= expectedBytes) {
|
||||
return;
|
||||
}
|
||||
await page.waitForTimeout(25);
|
||||
}
|
||||
const report = await readTerminalKeystrokeStressReport(page, INPUT_TEXT);
|
||||
throw new Error(
|
||||
`Timed out waiting for large agent_stream message: largest=${report.largestAgentStreamTextMessageBytes}, expected=${expectedBytes}`,
|
||||
);
|
||||
}
|
||||
|
||||
async function waitForAgentStreamMessages(page: Page, expectedCount: number): Promise<void> {
|
||||
const deadline = Date.now() + STRESS_TIMEOUT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
const report = await readTerminalKeystrokeStressReport(page, INPUT_TEXT);
|
||||
if (report.agentStreamTextMessageCount >= expectedCount) {
|
||||
return;
|
||||
}
|
||||
await page.waitForTimeout(25);
|
||||
}
|
||||
const report = await readTerminalKeystrokeStressReport(page, INPUT_TEXT);
|
||||
throw new Error(
|
||||
`Timed out waiting for agent_stream messages: count=${report.agentStreamTextMessageCount}, expected=${expectedCount}`,
|
||||
);
|
||||
}
|
||||
|
||||
async function attachJson(testInfo: TestInfo, name: string, value: unknown): Promise<void> {
|
||||
await testInfo.attach(name, {
|
||||
body: JSON.stringify(value, null, 2),
|
||||
contentType: "application/json",
|
||||
});
|
||||
}
|
||||
|
||||
function buildStressText(length: number): string {
|
||||
const alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
|
||||
let output = "";
|
||||
while (output.length < length) {
|
||||
output += alphabet;
|
||||
}
|
||||
return output.slice(0, length);
|
||||
}
|
||||
|
||||
function summarizeLatency(values: number[]): LatencyStats {
|
||||
const sorted = [...values].sort((a, b) => a - b);
|
||||
const percentile = (p: number) => {
|
||||
const index = Math.ceil((p / 100) * sorted.length) - 1;
|
||||
return sorted[Math.max(0, Math.min(sorted.length - 1, index))] ?? 0;
|
||||
};
|
||||
const total = values.reduce((sum, value) => sum + value, 0);
|
||||
return {
|
||||
count: values.length,
|
||||
minMs: round2(sorted[0] ?? 0),
|
||||
p50Ms: round2(percentile(50)),
|
||||
p95Ms: round2(percentile(95)),
|
||||
maxMs: round2(sorted[sorted.length - 1] ?? 0),
|
||||
avgMs: round2(total / values.length),
|
||||
};
|
||||
}
|
||||
|
||||
function round2(value: number): number {
|
||||
return Math.round(value * 100) / 100;
|
||||
}
|
||||
@@ -16,8 +16,10 @@ const LINE_COUNT = 50_000;
|
||||
const THROUGHPUT_BUDGET_MS = 30_000;
|
||||
const KEYSTROKE_SAMPLE_COUNT = 20;
|
||||
const KEYSTROKE_P95_BUDGET_MS = 150;
|
||||
const RUN_MANUAL_TERMINAL_PERF = process.env.PASEO_TERMINAL_PERF_E2E === "1";
|
||||
const terminalPerfDescribe = RUN_MANUAL_TERMINAL_PERF ? test.describe : test.describe.skip;
|
||||
|
||||
test.describe("Terminal wire performance", () => {
|
||||
terminalPerfDescribe("Terminal wire performance", () => {
|
||||
let client: TerminalPerfDaemonClient;
|
||||
let tempRepo: { path: string; cleanup: () => Promise<void> };
|
||||
let workspaceId: string;
|
||||
|
||||
@@ -89,6 +89,16 @@ type ActiveTurn = {
|
||||
completed: Promise<AgentRunResult>;
|
||||
};
|
||||
|
||||
type LargeAgentStreamPayloadRequest = {
|
||||
bytes: number;
|
||||
kind: "diff" | "file" | "image";
|
||||
};
|
||||
|
||||
type AgentStreamStressRequest = {
|
||||
count: number;
|
||||
coalesced: boolean;
|
||||
};
|
||||
|
||||
function resolveModelProfile(modelId: string | null | undefined): {
|
||||
modelId: string;
|
||||
durationMs: number;
|
||||
@@ -115,6 +125,52 @@ function promptToText(prompt: AgentPromptInput): string {
|
||||
.trim();
|
||||
}
|
||||
|
||||
function parseLargeAgentStreamPayloadPrompt(
|
||||
prompt: AgentPromptInput,
|
||||
): LargeAgentStreamPayloadRequest | null {
|
||||
const text = promptToText(prompt);
|
||||
const match =
|
||||
/emit\s+(\d+)\s+(?:byte\s+)?(?:large\s+)?(diff|file|image)\s+agent stream (?:update|payload)/i.exec(
|
||||
text,
|
||||
);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const bytes = Number(match[1]);
|
||||
if (!Number.isFinite(bytes) || bytes <= 0) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
bytes: Math.min(bytes, 1_000_000),
|
||||
kind: match[2]?.toLowerCase() as LargeAgentStreamPayloadRequest["kind"],
|
||||
};
|
||||
}
|
||||
|
||||
function parseAgentStreamStressPrompt(prompt: AgentPromptInput): AgentStreamStressRequest | null {
|
||||
const text = promptToText(prompt);
|
||||
const match = /emit\s+(\d+)\s+(coalesced\s+)?agent stream updates/i.exec(text);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const count = Number(match[1]);
|
||||
if (!Number.isFinite(count) || count <= 0) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
count: Math.min(count, 5_000),
|
||||
coalesced: Boolean(match[2]),
|
||||
};
|
||||
}
|
||||
|
||||
function buildRepeatedPayload(bytes: number, prefix: string): string {
|
||||
const line = `${prefix} ${"x".repeat(96)}\n`;
|
||||
let output = "";
|
||||
while (output.length < bytes) {
|
||||
output += line;
|
||||
}
|
||||
return output.slice(0, bytes);
|
||||
}
|
||||
|
||||
function buildMarkdownDocument(iteration: number, prompt: AgentPromptInput): string {
|
||||
const promptPreview = promptToText(prompt).slice(0, 160) || "No prompt text supplied.";
|
||||
return [
|
||||
@@ -282,7 +338,15 @@ export class MockLoadTestAgentSession implements AgentSession {
|
||||
};
|
||||
this.activeTurn = turn;
|
||||
|
||||
this.schedule(turn, 0);
|
||||
const largePayload = parseLargeAgentStreamPayloadPrompt(prompt);
|
||||
const stress = parseAgentStreamStressPrompt(prompt);
|
||||
if (largePayload) {
|
||||
this.scheduleLargePayloadTurn(turn, largePayload);
|
||||
} else if (stress) {
|
||||
this.scheduleStressTurn(turn, stress);
|
||||
} else {
|
||||
this.schedule(turn, 0);
|
||||
}
|
||||
return { turnId };
|
||||
}
|
||||
|
||||
@@ -380,6 +444,147 @@ export class MockLoadTestAgentSession implements AgentSession {
|
||||
turn.timer.unref?.();
|
||||
}
|
||||
|
||||
private scheduleLargePayloadTurn(
|
||||
turn: ActiveTurn,
|
||||
largePayload: LargeAgentStreamPayloadRequest,
|
||||
): void {
|
||||
turn.timer = setTimeout(() => {
|
||||
this.emitLargePayloadTurn(turn, largePayload);
|
||||
}, 0);
|
||||
turn.timer.unref?.();
|
||||
}
|
||||
|
||||
private scheduleStressTurn(turn: ActiveTurn, stress: AgentStreamStressRequest): void {
|
||||
turn.timer = setTimeout(() => {
|
||||
this.emitStressTurn(turn, stress);
|
||||
}, 0);
|
||||
turn.timer.unref?.();
|
||||
}
|
||||
|
||||
private emitStressTurn(turn: ActiveTurn, stress: AgentStreamStressRequest): void {
|
||||
if (this.activeTurn !== turn) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.clearTurnTimer(turn);
|
||||
this.emit({
|
||||
type: "turn_started",
|
||||
provider: this.provider,
|
||||
turnId: turn.turnId,
|
||||
});
|
||||
|
||||
for (let index = 0; index < stress.count; index += 1) {
|
||||
this.emitTimeline(
|
||||
turn.turnId,
|
||||
stress.coalesced
|
||||
? {
|
||||
type: "assistant_message",
|
||||
text: `stress-update-${index}`,
|
||||
}
|
||||
: {
|
||||
type: "todo",
|
||||
items: [{ text: `stress-update-${index}`, completed: index % 2 === 0 }],
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
this.activeTurn = null;
|
||||
const usage = {
|
||||
inputTokens: 1,
|
||||
outputTokens: stress.count,
|
||||
contextWindowUsedTokens: stress.count,
|
||||
contextWindowMaxTokens: 128_000,
|
||||
};
|
||||
this.emit({
|
||||
type: "turn_completed",
|
||||
provider: this.provider,
|
||||
turnId: turn.turnId,
|
||||
usage,
|
||||
});
|
||||
turn.resolve({
|
||||
sessionId: this.id,
|
||||
finalText: "Synthetic agent stream stress complete",
|
||||
usage,
|
||||
timeline: [],
|
||||
canceled: false,
|
||||
});
|
||||
}
|
||||
|
||||
private emitLargePayloadTurn(
|
||||
turn: ActiveTurn,
|
||||
largePayload: LargeAgentStreamPayloadRequest,
|
||||
): void {
|
||||
if (this.activeTurn !== turn) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.clearTurnTimer(turn);
|
||||
this.emit({
|
||||
type: "turn_started",
|
||||
provider: this.provider,
|
||||
turnId: turn.turnId,
|
||||
});
|
||||
|
||||
const payload = buildRepeatedPayload(largePayload.bytes, largePayload.kind);
|
||||
if (largePayload.kind === "diff") {
|
||||
this.emitTimeline(
|
||||
turn.turnId,
|
||||
createToolCall({
|
||||
turnId: turn.turnId,
|
||||
iteration: 0,
|
||||
name: "edit",
|
||||
status: "completed",
|
||||
detail: {
|
||||
type: "edit",
|
||||
filePath: "src/large-diff.ts",
|
||||
unifiedDiff: `diff --git a/src/large-diff.ts b/src/large-diff.ts\n${payload}`,
|
||||
},
|
||||
}),
|
||||
);
|
||||
} else if (largePayload.kind === "file") {
|
||||
this.emitTimeline(
|
||||
turn.turnId,
|
||||
createToolCall({
|
||||
turnId: turn.turnId,
|
||||
iteration: 0,
|
||||
name: "read",
|
||||
status: "completed",
|
||||
detail: {
|
||||
type: "read",
|
||||
filePath: "src/large-file.txt",
|
||||
content: payload,
|
||||
},
|
||||
}),
|
||||
);
|
||||
} else {
|
||||
this.emitTimeline(turn.turnId, {
|
||||
type: "assistant_message",
|
||||
text: `data:image/png;base64,${payload}`,
|
||||
});
|
||||
}
|
||||
|
||||
this.activeTurn = null;
|
||||
const usage = {
|
||||
inputTokens: 1,
|
||||
outputTokens: largePayload.bytes,
|
||||
contextWindowUsedTokens: largePayload.bytes,
|
||||
contextWindowMaxTokens: 128_000,
|
||||
};
|
||||
this.emit({
|
||||
type: "turn_completed",
|
||||
provider: this.provider,
|
||||
turnId: turn.turnId,
|
||||
usage,
|
||||
});
|
||||
turn.resolve({
|
||||
sessionId: this.id,
|
||||
finalText: "Synthetic large payload complete",
|
||||
usage,
|
||||
timeline: [],
|
||||
canceled: false,
|
||||
});
|
||||
}
|
||||
|
||||
private tick(turn: ActiveTurn): void {
|
||||
if (this.activeTurn !== turn) {
|
||||
return;
|
||||
|
||||
@@ -376,6 +376,16 @@ function getFrameText(frame: TerminalStreamFrame): string {
|
||||
return "";
|
||||
}
|
||||
|
||||
function sendRawTerminalInput(input: { ws: WebSocket; slot: number; text: string }): void {
|
||||
input.ws.send(
|
||||
encodeTerminalStreamFrame({
|
||||
opcode: TerminalStreamOpcode.Input,
|
||||
slot: input.slot,
|
||||
payload: new Uint8Array(Buffer.from(input.text, "utf8")),
|
||||
}),
|
||||
);
|
||||
}
|
||||
|
||||
async function subscribeRawTerminal(
|
||||
ws: WebSocket,
|
||||
terminalId: string,
|
||||
|
||||
@@ -181,6 +181,107 @@ function buildToolCallForPrompt(provider: string, prompt: string) {
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseAgentStreamStressPrompt(prompt: string): {
|
||||
count: number;
|
||||
coalesced: boolean;
|
||||
} | null {
|
||||
const match = /emit\s+(\d+)\s+(coalesced\s+)?agent stream updates/i.exec(prompt);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const count = Number(match[1]);
|
||||
if (!Number.isFinite(count) || count <= 0) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
count: Math.min(count, 5000),
|
||||
coalesced: Boolean(match[2]),
|
||||
};
|
||||
}
|
||||
|
||||
function parseLargeAgentStreamPayloadPrompt(prompt: string): {
|
||||
bytes: number;
|
||||
kind: "diff" | "file" | "image";
|
||||
} | null {
|
||||
const match =
|
||||
/emit\s+(\d+)\s+(?:byte\s+)?(?:large\s+)?(diff|file|image)\s+agent stream (?:update|payload)/i.exec(
|
||||
prompt,
|
||||
);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const bytes = Number(match[1]);
|
||||
if (!Number.isFinite(bytes) || bytes <= 0) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
bytes: Math.min(bytes, 1_000_000),
|
||||
kind: match[2]?.toLowerCase() as "diff" | "file" | "image",
|
||||
};
|
||||
}
|
||||
|
||||
function buildRepeatedPayload(bytes: number, prefix: string): string {
|
||||
const line = `${prefix} ${"x".repeat(96)}\n`;
|
||||
let output = "";
|
||||
while (output.length < bytes) {
|
||||
output += line;
|
||||
}
|
||||
return output.slice(0, bytes);
|
||||
}
|
||||
|
||||
function buildLargeTimelineItem(input: {
|
||||
bytes: number;
|
||||
kind: "diff" | "file" | "image";
|
||||
callId: string;
|
||||
provider: AgentStreamEvent["provider"];
|
||||
}): AgentStreamEvent {
|
||||
const payload = buildRepeatedPayload(input.bytes, input.kind);
|
||||
if (input.kind === "diff") {
|
||||
return {
|
||||
type: "timeline",
|
||||
provider: input.provider,
|
||||
item: {
|
||||
type: "tool_call",
|
||||
name: "apply_patch",
|
||||
callId: input.callId,
|
||||
status: "completed",
|
||||
detail: {
|
||||
type: "edit",
|
||||
filePath: "src/large-diff.ts",
|
||||
unifiedDiff: `diff --git a/src/large-diff.ts b/src/large-diff.ts\n${payload}`,
|
||||
},
|
||||
error: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
if (input.kind === "file") {
|
||||
return {
|
||||
type: "timeline",
|
||||
provider: input.provider,
|
||||
item: {
|
||||
type: "tool_call",
|
||||
name: "read_file",
|
||||
callId: input.callId,
|
||||
status: "completed",
|
||||
detail: {
|
||||
type: "read",
|
||||
filePath: "src/large-file.txt",
|
||||
content: payload,
|
||||
},
|
||||
error: null,
|
||||
},
|
||||
};
|
||||
}
|
||||
return {
|
||||
type: "timeline",
|
||||
provider: input.provider,
|
||||
item: {
|
||||
type: "assistant_message",
|
||||
text: `data:image/png;base64,${payload}`,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
class FakeAgentSession implements AgentSession {
|
||||
readonly capabilities = TEST_CAPABILITIES;
|
||||
readonly id: string;
|
||||
@@ -372,6 +473,57 @@ class FakeAgentSession implements AgentSession {
|
||||
await this.appendHistoryEvent(turnStarted);
|
||||
this.notifySubscribers(turnStarted);
|
||||
|
||||
const stress = parseAgentStreamStressPrompt(textPrompt);
|
||||
if (stress !== null) {
|
||||
for (let index = 0; index < stress.count; index += 1) {
|
||||
const stressUpdate: AgentStreamEvent = {
|
||||
type: "timeline",
|
||||
provider: this.providerName,
|
||||
item: stress.coalesced
|
||||
? {
|
||||
type: "assistant_message",
|
||||
text: `stress-update-${index}`,
|
||||
}
|
||||
: {
|
||||
type: "todo",
|
||||
items: [{ text: `stress-update-${index}`, completed: index % 2 === 0 }],
|
||||
},
|
||||
};
|
||||
await this.appendHistoryEvent(stressUpdate);
|
||||
this.notifySubscribers(stressUpdate);
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
}
|
||||
|
||||
const completed: AgentStreamEvent = {
|
||||
type: "turn_completed",
|
||||
provider: this.providerName,
|
||||
usage: { inputTokens: 1, outputTokens: stress.count },
|
||||
};
|
||||
await this.appendHistoryEvent(completed);
|
||||
this.notifySubscribers(completed);
|
||||
return;
|
||||
}
|
||||
|
||||
const largePayload = parseLargeAgentStreamPayloadPrompt(textPrompt);
|
||||
if (largePayload !== null) {
|
||||
const largeUpdate = buildLargeTimelineItem({
|
||||
...largePayload,
|
||||
callId: randomUUID(),
|
||||
provider: this.providerName,
|
||||
});
|
||||
await this.appendHistoryEvent(largeUpdate);
|
||||
this.notifySubscribers(largeUpdate);
|
||||
|
||||
const completed: AgentStreamEvent = {
|
||||
type: "turn_completed",
|
||||
provider: this.providerName,
|
||||
usage: { inputTokens: 1, outputTokens: largePayload.bytes },
|
||||
};
|
||||
await this.appendHistoryEvent(completed);
|
||||
this.notifySubscribers(completed);
|
||||
return;
|
||||
}
|
||||
|
||||
const tool = buildToolCallForPrompt(this.providerName, textPrompt);
|
||||
if (tool) {
|
||||
const needsPermission = this.needsPermissionForTool(tool.name, tool.input ?? {});
|
||||
|
||||
Reference in New Issue
Block a user