Add metrics collection and terminal performance tests
This commit is contained in:
@@ -448,12 +448,15 @@ export default async function globalSetup() {
|
|||||||
PASEO_LISTEN: `0.0.0.0:${port}`,
|
PASEO_LISTEN: `0.0.0.0:${port}`,
|
||||||
PASEO_RELAY_ENDPOINT: `127.0.0.1:${relayPort}`,
|
PASEO_RELAY_ENDPOINT: `127.0.0.1:${relayPort}`,
|
||||||
PASEO_CORS_ORIGINS: `http://localhost:${metroPort}`,
|
PASEO_CORS_ORIGINS: `http://localhost:${metroPort}`,
|
||||||
// Use OpenAI speech providers in e2e to avoid local model bootstrapping delays.
|
PASEO_DICTATION_ENABLED: openAiUsable ? "1" : "0",
|
||||||
PASEO_DICTATION_ENABLED: "1",
|
PASEO_VOICE_MODE_ENABLED: openAiUsable ? "1" : "0",
|
||||||
PASEO_VOICE_MODE_ENABLED: "1",
|
...(openAiUsable
|
||||||
PASEO_DICTATION_STT_PROVIDER: dictationProvider,
|
? {
|
||||||
PASEO_VOICE_STT_PROVIDER: "openai",
|
PASEO_DICTATION_STT_PROVIDER: "openai",
|
||||||
PASEO_VOICE_TTS_PROVIDER: "openai",
|
PASEO_VOICE_STT_PROVIDER: "openai",
|
||||||
|
PASEO_VOICE_TTS_PROVIDER: "openai",
|
||||||
|
}
|
||||||
|
: {}),
|
||||||
...(localModelsDir ? { PASEO_LOCAL_MODELS_DIR: localModelsDir } : {}),
|
...(localModelsDir ? { PASEO_LOCAL_MODELS_DIR: localModelsDir } : {}),
|
||||||
NODE_ENV: "development",
|
NODE_ENV: "development",
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -0,0 +1,228 @@
|
|||||||
|
import type { Page } from "@playwright/test";
|
||||||
|
import path from "node:path";
|
||||||
|
import { pathToFileURL } from "node:url";
|
||||||
|
import { randomUUID } from "node:crypto";
|
||||||
|
import { buildHostWorkspaceRoute } from "../../src/utils/host-routes";
|
||||||
|
|
||||||
|
export type TerminalPerfDaemonClient = {
|
||||||
|
connect(): Promise<void>;
|
||||||
|
close(): Promise<void>;
|
||||||
|
createTerminal(
|
||||||
|
cwd: string,
|
||||||
|
name?: string,
|
||||||
|
): Promise<{
|
||||||
|
terminal: { id: string; name: string; cwd: string } | null;
|
||||||
|
error: string | null;
|
||||||
|
}>;
|
||||||
|
subscribeTerminal(
|
||||||
|
terminalId: string,
|
||||||
|
): Promise<{ terminalId: string; slot: number; error: null } | { error: string }>;
|
||||||
|
sendTerminalInput(
|
||||||
|
terminalId: string,
|
||||||
|
message: { type: "input"; data: string } | { type: "resize"; rows: number; cols: number },
|
||||||
|
): void;
|
||||||
|
onTerminalStreamEvent(
|
||||||
|
handler: (event: { terminalId: string; type: string; data?: Uint8Array }) => void,
|
||||||
|
): () => void;
|
||||||
|
killTerminal(terminalId: string): Promise<{ error: string | null }>;
|
||||||
|
};
|
||||||
|
|
||||||
|
function getDaemonWsUrl(): string {
|
||||||
|
const daemonPort = process.env.E2E_DAEMON_PORT;
|
||||||
|
if (!daemonPort) {
|
||||||
|
throw new Error("E2E_DAEMON_PORT is not set.");
|
||||||
|
}
|
||||||
|
return `ws://127.0.0.1:${daemonPort}/ws`;
|
||||||
|
}
|
||||||
|
|
||||||
|
function getServerId(): string {
|
||||||
|
const serverId = process.env.E2E_SERVER_ID;
|
||||||
|
if (!serverId) {
|
||||||
|
throw new Error("E2E_SERVER_ID is not set.");
|
||||||
|
}
|
||||||
|
return serverId;
|
||||||
|
}
|
||||||
|
|
||||||
|
async function loadDaemonClientConstructor(): Promise<
|
||||||
|
new (config: { url: string; clientId: string; clientType: "cli" }) => TerminalPerfDaemonClient
|
||||||
|
> {
|
||||||
|
const repoRoot = path.resolve(process.cwd(), "../..");
|
||||||
|
const moduleUrl = pathToFileURL(
|
||||||
|
path.join(repoRoot, "packages/server/dist/server/server/exports.js"),
|
||||||
|
).href;
|
||||||
|
const mod = (await import(moduleUrl)) as {
|
||||||
|
DaemonClient: new (config: {
|
||||||
|
url: string;
|
||||||
|
clientId: string;
|
||||||
|
clientType: "cli";
|
||||||
|
}) => TerminalPerfDaemonClient;
|
||||||
|
};
|
||||||
|
return mod.DaemonClient;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function connectTerminalClient(): Promise<TerminalPerfDaemonClient> {
|
||||||
|
const DaemonClient = await loadDaemonClientConstructor();
|
||||||
|
const client = new DaemonClient({
|
||||||
|
url: getDaemonWsUrl(),
|
||||||
|
clientId: `terminal-perf-${randomUUID()}`,
|
||||||
|
clientType: "cli",
|
||||||
|
});
|
||||||
|
await client.connect();
|
||||||
|
return client;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function buildTerminalWorkspaceUrl(cwd: string, terminalId: string): string {
|
||||||
|
const serverId = getServerId();
|
||||||
|
const route = buildHostWorkspaceRoute(serverId, cwd);
|
||||||
|
return `${route}?open=${encodeURIComponent(`terminal:${terminalId}`)}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function getTerminalBufferText(page: Page): Promise<string> {
|
||||||
|
return page.evaluate(() => {
|
||||||
|
const term = (window as any).__paseoTerminal;
|
||||||
|
if (!term) {
|
||||||
|
return "";
|
||||||
|
}
|
||||||
|
const buf = term.buffer.active;
|
||||||
|
const lines: string[] = [];
|
||||||
|
for (let i = 0; i < buf.length; i++) {
|
||||||
|
const line = buf.getLine(i);
|
||||||
|
if (line) {
|
||||||
|
lines.push(line.translateToString(true));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return lines.join("\n");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function waitForTerminalContent(
|
||||||
|
page: Page,
|
||||||
|
predicate: (text: string) => boolean,
|
||||||
|
timeout: number,
|
||||||
|
): Promise<void> {
|
||||||
|
const deadline = Date.now() + timeout;
|
||||||
|
while (Date.now() < deadline) {
|
||||||
|
const text = await getTerminalBufferText(page);
|
||||||
|
if (predicate(text)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
await page.waitForTimeout(50);
|
||||||
|
}
|
||||||
|
throw new Error(`Terminal content did not match predicate within ${timeout}ms`);
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function navigateToTerminal(
|
||||||
|
page: Page,
|
||||||
|
input: { cwd: string; terminalId: string },
|
||||||
|
): Promise<void> {
|
||||||
|
// Boot the app at the workspace route directly.
|
||||||
|
// The fixtures.ts beforeEach addInitScript seeds localStorage on every navigation,
|
||||||
|
// so the daemon registry is already configured when the app starts.
|
||||||
|
const workspaceRoute = buildHostWorkspaceRoute(getServerId(), input.cwd);
|
||||||
|
await page.goto(workspaceRoute);
|
||||||
|
|
||||||
|
// Wait for daemon connection (sidebar shows host label)
|
||||||
|
await page.getByText("localhost", { exact: true }).first().waitFor({ state: "visible", timeout: 15_000 });
|
||||||
|
|
||||||
|
// The workspace should now query listTerminals and discover our terminal.
|
||||||
|
// Click the terminal tab if it auto-appeared, or wait for it.
|
||||||
|
const terminalSurface = page.locator('[data-testid="terminal-surface"]');
|
||||||
|
const surfaceVisible = await terminalSurface.isVisible().catch(() => false);
|
||||||
|
|
||||||
|
if (!surfaceVisible) {
|
||||||
|
// Terminal tab might not be focused — look for it in the tab row and click it
|
||||||
|
const terminalTab = page.locator(`[data-testid="workspace-tab-terminal:${input.terminalId}"]`);
|
||||||
|
const tabExists = await terminalTab.isVisible({ timeout: 5_000 }).catch(() => false);
|
||||||
|
|
||||||
|
if (tabExists) {
|
||||||
|
await terminalTab.click();
|
||||||
|
} else {
|
||||||
|
// Terminal tab not yet created — click "New terminal tab" to create one through the UI
|
||||||
|
const newTerminalBtn = page.getByRole("button", { name: "New terminal tab" });
|
||||||
|
await newTerminalBtn.waitFor({ state: "visible", timeout: 10_000 });
|
||||||
|
await newTerminalBtn.click();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Wait for terminal surface to be visible
|
||||||
|
await terminalSurface.waitFor({ state: "visible", timeout: 15_000 });
|
||||||
|
|
||||||
|
// Wait for loading overlay to disappear (terminal attached)
|
||||||
|
await page
|
||||||
|
.locator('[data-testid="terminal-attach-loading"]')
|
||||||
|
.waitFor({ state: "hidden", timeout: 10_000 })
|
||||||
|
.catch(() => {
|
||||||
|
// overlay may never appear if attachment is instant
|
||||||
|
});
|
||||||
|
|
||||||
|
await terminalSurface.click();
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function setupDeterministicPrompt(page: Page, sentinel?: string): Promise<void> {
|
||||||
|
const tag = sentinel ?? `READY_${Date.now()}`;
|
||||||
|
const terminal = page.locator('[data-testid="terminal-surface"]');
|
||||||
|
|
||||||
|
await terminal.pressSequentially(`echo ${tag}\n`, { delay: 0 });
|
||||||
|
await waitForTerminalContent(page, (text) => text.includes(tag), 10_000);
|
||||||
|
|
||||||
|
await terminal.pressSequentially("export PS1='$ '\n", { delay: 0 });
|
||||||
|
await page.waitForTimeout(300);
|
||||||
|
}
|
||||||
|
|
||||||
|
export type LatencySample = {
|
||||||
|
char: string;
|
||||||
|
latencyMs: number;
|
||||||
|
};
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Measures keystroke echo round-trip latency.
|
||||||
|
*
|
||||||
|
* Starts a high-resolution timer on the browser keydown event (capture phase)
|
||||||
|
* and stops it when xterm.js finishes parsing the echoed write. This measures
|
||||||
|
* the full path: keydown → WebSocket → daemon PTY echo → WebSocket → xterm render.
|
||||||
|
*/
|
||||||
|
export async function measureKeystrokeLatency(page: Page, char: string): Promise<number> {
|
||||||
|
await page.evaluate(() => {
|
||||||
|
const term = (window as any).__paseoTerminal;
|
||||||
|
if (!term) {
|
||||||
|
throw new Error("__paseoTerminal not available");
|
||||||
|
}
|
||||||
|
|
||||||
|
const state = ((window as any).__perfKeystroke = {
|
||||||
|
promise: null as Promise<number> | null,
|
||||||
|
});
|
||||||
|
|
||||||
|
state.promise = new Promise<number>((resolve, reject) => {
|
||||||
|
const timeout = setTimeout(() => {
|
||||||
|
document.removeEventListener("keydown", onKeyDown, true);
|
||||||
|
reject(new Error("keystroke echo timeout (5s)"));
|
||||||
|
}, 5000);
|
||||||
|
|
||||||
|
function onKeyDown() {
|
||||||
|
document.removeEventListener("keydown", onKeyDown, true);
|
||||||
|
const start = performance.now();
|
||||||
|
const disposable = term.onWriteParsed(() => {
|
||||||
|
clearTimeout(timeout);
|
||||||
|
disposable.dispose();
|
||||||
|
resolve(performance.now() - start);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
document.addEventListener("keydown", onKeyDown, true);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
await page.keyboard.press(char);
|
||||||
|
|
||||||
|
return page.evaluate(() => (window as any).__perfKeystroke.promise);
|
||||||
|
}
|
||||||
|
|
||||||
|
export function computePercentile(samples: number[], p: number): number {
|
||||||
|
const sorted = [...samples].sort((a, b) => a - b);
|
||||||
|
const index = Math.ceil((p / 100) * sorted.length) - 1;
|
||||||
|
return sorted[Math.max(0, index)];
|
||||||
|
}
|
||||||
|
|
||||||
|
export function round2(value: number): number {
|
||||||
|
return Math.round(value * 100) / 100;
|
||||||
|
}
|
||||||
@@ -0,0 +1,160 @@
|
|||||||
|
import { test, expect } from "./fixtures";
|
||||||
|
import { createTempGitRepo } from "./helpers/workspace";
|
||||||
|
import {
|
||||||
|
connectTerminalClient,
|
||||||
|
navigateToTerminal,
|
||||||
|
setupDeterministicPrompt,
|
||||||
|
waitForTerminalContent,
|
||||||
|
measureKeystrokeLatency,
|
||||||
|
computePercentile,
|
||||||
|
round2,
|
||||||
|
type TerminalPerfDaemonClient,
|
||||||
|
type LatencySample,
|
||||||
|
} from "./helpers/terminal-perf";
|
||||||
|
|
||||||
|
const LINE_COUNT = 50_000;
|
||||||
|
const THROUGHPUT_BUDGET_MS = 30_000;
|
||||||
|
const KEYSTROKE_SAMPLE_COUNT = 20;
|
||||||
|
const KEYSTROKE_P95_BUDGET_MS = 150;
|
||||||
|
|
||||||
|
test.describe("Terminal wire performance", () => {
|
||||||
|
let client: TerminalPerfDaemonClient;
|
||||||
|
let tempRepo: { path: string; cleanup: () => Promise<void> };
|
||||||
|
|
||||||
|
test.beforeAll(async () => {
|
||||||
|
tempRepo = await createTempGitRepo("perf-");
|
||||||
|
client = await connectTerminalClient();
|
||||||
|
});
|
||||||
|
|
||||||
|
test.afterAll(async () => {
|
||||||
|
if (client) {
|
||||||
|
await client.close();
|
||||||
|
}
|
||||||
|
if (tempRepo) {
|
||||||
|
await tempRepo.cleanup();
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("throughput: bulk terminal output renders within budget", async ({ page }, testInfo) => {
|
||||||
|
test.setTimeout(90_000);
|
||||||
|
|
||||||
|
const result = await client.createTerminal(tempRepo.path, "throughput");
|
||||||
|
if (!result.terminal) {
|
||||||
|
throw new Error(`Failed to create terminal: ${result.error}`);
|
||||||
|
}
|
||||||
|
const terminalId = result.terminal.id;
|
||||||
|
|
||||||
|
try {
|
||||||
|
await navigateToTerminal(page, { cwd: tempRepo.path, terminalId });
|
||||||
|
await setupDeterministicPrompt(page);
|
||||||
|
|
||||||
|
const sentinel = `PERF_DONE_${Date.now()}`;
|
||||||
|
const terminal = page.locator('[data-testid="terminal-surface"]');
|
||||||
|
const startMs = Date.now();
|
||||||
|
|
||||||
|
await terminal.pressSequentially(`seq 1 ${LINE_COUNT}; echo ${sentinel}\n`, { delay: 0 });
|
||||||
|
|
||||||
|
await waitForTerminalContent(page, (text) => text.includes(sentinel), THROUGHPUT_BUDGET_MS + 15_000);
|
||||||
|
|
||||||
|
const elapsedMs = Date.now() - startMs;
|
||||||
|
|
||||||
|
// seq 1 N outputs each number on its own line
|
||||||
|
const estimatedBytes = Array.from(
|
||||||
|
{ length: LINE_COUNT },
|
||||||
|
(_, i) => String(i + 1).length + 1,
|
||||||
|
).reduce((a, b) => a + b, 0);
|
||||||
|
const throughputMBps = estimatedBytes / (1024 * 1024) / (elapsedMs / 1000);
|
||||||
|
|
||||||
|
const report = {
|
||||||
|
lineCount: LINE_COUNT,
|
||||||
|
estimatedBytes,
|
||||||
|
elapsedMs,
|
||||||
|
throughputMBps: round2(throughputMBps),
|
||||||
|
};
|
||||||
|
|
||||||
|
await testInfo.attach("throughput-report", {
|
||||||
|
body: JSON.stringify(report, null, 2),
|
||||||
|
contentType: "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
console.log(
|
||||||
|
`[perf] Throughput: ${report.throughputMBps} MB/s — ${LINE_COUNT} lines in ${elapsedMs}ms`,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(elapsedMs, `${LINE_COUNT} lines should render within ${THROUGHPUT_BUDGET_MS}ms`).toBeLessThan(
|
||||||
|
THROUGHPUT_BUDGET_MS,
|
||||||
|
);
|
||||||
|
} finally {
|
||||||
|
await client.killTerminal(terminalId).catch(() => {});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
test("keystroke latency: echo round-trip under budget", async ({ page }, testInfo) => {
|
||||||
|
test.setTimeout(60_000);
|
||||||
|
|
||||||
|
const result = await client.createTerminal(tempRepo.path, "latency");
|
||||||
|
if (!result.terminal) {
|
||||||
|
throw new Error(`Failed to create terminal: ${result.error}`);
|
||||||
|
}
|
||||||
|
const terminalId = result.terminal.id;
|
||||||
|
|
||||||
|
try {
|
||||||
|
await navigateToTerminal(page, { cwd: tempRepo.path, terminalId });
|
||||||
|
await setupDeterministicPrompt(page);
|
||||||
|
|
||||||
|
// Ensure clean prompt state
|
||||||
|
const terminal = page.locator('[data-testid="terminal-surface"]');
|
||||||
|
await terminal.press("Control+c");
|
||||||
|
await page.waitForTimeout(200);
|
||||||
|
|
||||||
|
const samples: LatencySample[] = [];
|
||||||
|
const chars = "abcdefghijklmnopqrst";
|
||||||
|
|
||||||
|
for (let i = 0; i < KEYSTROKE_SAMPLE_COUNT; i++) {
|
||||||
|
const char = chars[i % chars.length];
|
||||||
|
const latencyMs = await measureKeystrokeLatency(page, char);
|
||||||
|
samples.push({ char, latencyMs });
|
||||||
|
await page.waitForTimeout(50);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Clean up typed characters
|
||||||
|
await terminal.press("Control+c");
|
||||||
|
|
||||||
|
const latencies = samples.map((s) => s.latencyMs);
|
||||||
|
const p50 = computePercentile(latencies, 50);
|
||||||
|
const p95 = computePercentile(latencies, 95);
|
||||||
|
const max = Math.max(...latencies);
|
||||||
|
const min = Math.min(...latencies);
|
||||||
|
const avg = latencies.reduce((a, b) => a + b, 0) / latencies.length;
|
||||||
|
|
||||||
|
const report = {
|
||||||
|
sampleCount: KEYSTROKE_SAMPLE_COUNT,
|
||||||
|
p50Ms: round2(p50),
|
||||||
|
p95Ms: round2(p95),
|
||||||
|
maxMs: round2(max),
|
||||||
|
minMs: round2(min),
|
||||||
|
avgMs: round2(avg),
|
||||||
|
samples: samples.map((s) => ({
|
||||||
|
char: s.char,
|
||||||
|
latencyMs: round2(s.latencyMs),
|
||||||
|
})),
|
||||||
|
};
|
||||||
|
|
||||||
|
await testInfo.attach("latency-report", {
|
||||||
|
body: JSON.stringify(report, null, 2),
|
||||||
|
contentType: "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
console.log(
|
||||||
|
`[perf] Keystroke latency — p50: ${report.p50Ms}ms, p95: ${report.p95Ms}ms, max: ${report.maxMs}ms`,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(
|
||||||
|
p95,
|
||||||
|
`Keystroke p95 latency should be under ${KEYSTROKE_P95_BUDGET_MS}ms`,
|
||||||
|
).toBeLessThan(KEYSTROKE_P95_BUDGET_MS);
|
||||||
|
} finally {
|
||||||
|
await client.killTerminal(terminalId).catch(() => {});
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -231,6 +231,16 @@ export type ManagedAgent =
|
|||||||
| ManagedAgentError
|
| ManagedAgentError
|
||||||
| ManagedAgentClosed;
|
| ManagedAgentClosed;
|
||||||
|
|
||||||
|
export interface AgentMetricsSnapshot {
|
||||||
|
total: number;
|
||||||
|
byLifecycle: Record<string, number>;
|
||||||
|
withActiveForegroundTurn: number;
|
||||||
|
timelineStats: {
|
||||||
|
totalItems: number;
|
||||||
|
maxItemsPerAgent: number;
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
type ActiveManagedAgent =
|
type ActiveManagedAgent =
|
||||||
| ManagedAgentInitializing
|
| ManagedAgentInitializing
|
||||||
| ManagedAgentIdle
|
| ManagedAgentIdle
|
||||||
@@ -345,6 +355,37 @@ export class AgentManager {
|
|||||||
this.onAgentAttention = callback;
|
this.onAgentAttention = callback;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public getMetricsSnapshot(): AgentMetricsSnapshot {
|
||||||
|
const byLifecycle: Record<string, number> = {};
|
||||||
|
let withActiveForegroundTurn = 0;
|
||||||
|
let totalItems = 0;
|
||||||
|
let maxItemsPerAgent = 0;
|
||||||
|
|
||||||
|
for (const agent of this.agents.values()) {
|
||||||
|
byLifecycle[agent.lifecycle] = (byLifecycle[agent.lifecycle] ?? 0) + 1;
|
||||||
|
|
||||||
|
if (agent.activeForegroundTurnId !== null) {
|
||||||
|
withActiveForegroundTurn++;
|
||||||
|
}
|
||||||
|
|
||||||
|
const len = agent.timeline.length;
|
||||||
|
totalItems += len;
|
||||||
|
if (len > maxItemsPerAgent) {
|
||||||
|
maxItemsPerAgent = len;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return {
|
||||||
|
total: this.agents.size,
|
||||||
|
byLifecycle,
|
||||||
|
withActiveForegroundTurn,
|
||||||
|
timelineStats: {
|
||||||
|
totalItems,
|
||||||
|
maxItemsPerAgent,
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
private touchUpdatedAt(agent: ManagedAgent): Date {
|
private touchUpdatedAt(agent: ManagedAgent): Date {
|
||||||
const nowMs = Date.now();
|
const nowMs = Date.now();
|
||||||
const previousMs = agent.updatedAt.getTime();
|
const previousMs = agent.updatedAt.getTime();
|
||||||
|
|||||||
@@ -291,6 +291,8 @@ export type SessionRuntimeMetrics = {
|
|||||||
checkoutDiffFallbackRefreshTargetCount: number;
|
checkoutDiffFallbackRefreshTargetCount: number;
|
||||||
terminalDirectorySubscriptionCount: number;
|
terminalDirectorySubscriptionCount: number;
|
||||||
terminalSubscriptionCount: number;
|
terminalSubscriptionCount: number;
|
||||||
|
inflightRequests: number;
|
||||||
|
peakInflightRequests: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
type FetchAgentsRequestMessage = Extract<SessionInboundMessage, { type: "fetch_agents_request" }>;
|
type FetchAgentsRequestMessage = Extract<SessionInboundMessage, { type: "fetch_agents_request" }>;
|
||||||
@@ -610,6 +612,8 @@ export class Session {
|
|||||||
private readonly activeTerminalStreams = new Map<number, ActiveTerminalStream>();
|
private readonly activeTerminalStreams = new Map<number, ActiveTerminalStream>();
|
||||||
private readonly terminalIdToSlot = new Map<string, number>();
|
private readonly terminalIdToSlot = new Map<string, number>();
|
||||||
private nextTerminalSlot = 0;
|
private nextTerminalSlot = 0;
|
||||||
|
private inflightRequests = 0;
|
||||||
|
private peakInflightRequests = 0;
|
||||||
private readonly checkoutDiffSubscriptions = new Map<string, { targetKey: string }>();
|
private readonly checkoutDiffSubscriptions = new Map<string, { targetKey: string }>();
|
||||||
private readonly checkoutDiffTargets = new Map<string, CheckoutDiffWatchTarget>();
|
private readonly checkoutDiffTargets = new Map<string, CheckoutDiffWatchTarget>();
|
||||||
private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>();
|
private readonly workspaceGitWatchTargets = new Map<string, WorkspaceGitWatchTarget>();
|
||||||
@@ -754,6 +758,8 @@ export class Session {
|
|||||||
checkoutDiffFallbackRefreshTargetCount,
|
checkoutDiffFallbackRefreshTargetCount,
|
||||||
terminalDirectorySubscriptionCount: this.subscribedTerminalDirectories.size,
|
terminalDirectorySubscriptionCount: this.subscribedTerminalDirectories.size,
|
||||||
terminalSubscriptionCount: this.activeTerminalStreams.size,
|
terminalSubscriptionCount: this.activeTerminalStreams.size,
|
||||||
|
inflightRequests: this.inflightRequests,
|
||||||
|
peakInflightRequests: this.peakInflightRequests,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1454,6 +1460,11 @@ export class Session {
|
|||||||
* Main entry point for processing session messages
|
* Main entry point for processing session messages
|
||||||
*/
|
*/
|
||||||
public async handleMessage(msg: SessionInboundMessage): Promise<void> {
|
public async handleMessage(msg: SessionInboundMessage): Promise<void> {
|
||||||
|
this.inflightRequests++;
|
||||||
|
if (this.inflightRequests > this.peakInflightRequests) {
|
||||||
|
this.peakInflightRequests = this.inflightRequests;
|
||||||
|
}
|
||||||
|
try {
|
||||||
this.sessionLogger.trace({ inbound: msg }, "inbound message");
|
this.sessionLogger.trace({ inbound: msg }, "inbound message");
|
||||||
try {
|
try {
|
||||||
switch (msg.type) {
|
switch (msg.type) {
|
||||||
@@ -1779,6 +1790,13 @@ export class Session {
|
|||||||
},
|
},
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
} finally {
|
||||||
|
this.inflightRequests--;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
public resetPeakInflight(): void {
|
||||||
|
this.peakInflightRequests = this.inflightRequests;
|
||||||
}
|
}
|
||||||
|
|
||||||
public handleBinaryFrame(frame: TerminalStreamFrame): void {
|
public handleBinaryFrame(frame: TerminalStreamFrame): void {
|
||||||
|
|||||||
@@ -197,6 +197,7 @@ type WebSocketRuntimeCounters = {
|
|||||||
hostRejected: number;
|
hostRejected: number;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
const SLOW_REQUEST_THRESHOLD_MS = 500;
|
||||||
const EXTERNAL_SESSION_DISCONNECT_GRACE_MS = 90_000;
|
const EXTERNAL_SESSION_DISCONNECT_GRACE_MS = 90_000;
|
||||||
const HELLO_TIMEOUT_MS = 15_000;
|
const HELLO_TIMEOUT_MS = 15_000;
|
||||||
const WS_CLOSE_HELLO_TIMEOUT = 4001;
|
const WS_CLOSE_HELLO_TIMEOUT = 4001;
|
||||||
@@ -275,6 +276,7 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
};
|
};
|
||||||
private readonly inboundMessageCounts = new Map<string, number>();
|
private readonly inboundMessageCounts = new Map<string, number>();
|
||||||
private readonly inboundSessionRequestCounts = new Map<string, number>();
|
private readonly inboundSessionRequestCounts = new Map<string, number>();
|
||||||
|
private readonly requestLatencies = new Map<string, number[]>();
|
||||||
private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
|
private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
|
||||||
|
|
||||||
constructor(
|
constructor(
|
||||||
@@ -1056,7 +1058,21 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
|
|
||||||
if (message.type === "session") {
|
if (message.type === "session") {
|
||||||
this.recordInboundSessionRequestType(message.message.type);
|
this.recordInboundSessionRequestType(message.message.type);
|
||||||
|
const startMs = performance.now();
|
||||||
await activeConnection.session.handleMessage(message.message);
|
await activeConnection.session.handleMessage(message.message);
|
||||||
|
const durationMs = performance.now() - startMs;
|
||||||
|
this.recordRequestLatency(message.message.type, durationMs);
|
||||||
|
|
||||||
|
if (durationMs >= SLOW_REQUEST_THRESHOLD_MS) {
|
||||||
|
activeConnection.connectionLogger.warn(
|
||||||
|
{
|
||||||
|
requestType: message.message.type,
|
||||||
|
durationMs: Math.round(durationMs),
|
||||||
|
inflightRequests: activeConnection.session.getRuntimeMetrics().inflightRequests,
|
||||||
|
},
|
||||||
|
"ws_slow_request",
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
} catch (error) {
|
} catch (error) {
|
||||||
const err = error instanceof Error ? error : new Error(String(error));
|
const err = error instanceof Error ? error : new Error(String(error));
|
||||||
@@ -1147,10 +1163,49 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
this.incrementCount(this.inboundSessionRequestCounts, type);
|
this.incrementCount(this.inboundSessionRequestCounts, type);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private recordRequestLatency(type: string, durationMs: number): void {
|
||||||
|
let latencies = this.requestLatencies.get(type);
|
||||||
|
if (!latencies) {
|
||||||
|
latencies = [];
|
||||||
|
this.requestLatencies.set(type, latencies);
|
||||||
|
}
|
||||||
|
latencies.push(durationMs);
|
||||||
|
}
|
||||||
|
|
||||||
private getTopCounts(map: Map<string, number>, limit: number): Array<[string, number]> {
|
private getTopCounts(map: Map<string, number>, limit: number): Array<[string, number]> {
|
||||||
return [...map.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit);
|
return [...map.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private computeLatencyStats(): Array<{
|
||||||
|
type: string;
|
||||||
|
count: number;
|
||||||
|
minMs: number;
|
||||||
|
maxMs: number;
|
||||||
|
p50Ms: number;
|
||||||
|
totalMs: number;
|
||||||
|
}> {
|
||||||
|
const stats: Array<{
|
||||||
|
type: string;
|
||||||
|
count: number;
|
||||||
|
minMs: number;
|
||||||
|
maxMs: number;
|
||||||
|
p50Ms: number;
|
||||||
|
totalMs: number;
|
||||||
|
}> = [];
|
||||||
|
for (const [type, latencies] of this.requestLatencies) {
|
||||||
|
if (latencies.length === 0) continue;
|
||||||
|
latencies.sort((a, b) => a - b);
|
||||||
|
const count = latencies.length;
|
||||||
|
const minMs = Math.round(latencies[0]!);
|
||||||
|
const maxMs = Math.round(latencies[count - 1]!);
|
||||||
|
const p50Ms = Math.round(latencies[Math.floor(count / 2)]!);
|
||||||
|
const totalMs = Math.round(latencies.reduce((sum, v) => sum + v, 0));
|
||||||
|
stats.push({ type, count, minMs, maxMs, p50Ms, totalMs });
|
||||||
|
}
|
||||||
|
stats.sort((a, b) => b.totalMs - a.totalMs);
|
||||||
|
return stats.slice(0, 15);
|
||||||
|
}
|
||||||
|
|
||||||
private collectSessionRuntimeMetrics(): SessionRuntimeMetrics {
|
private collectSessionRuntimeMetrics(): SessionRuntimeMetrics {
|
||||||
const uniqueConnections = new Set<SessionConnection>(this.externalSessionsByKey.values());
|
const uniqueConnections = new Set<SessionConnection>(this.externalSessionsByKey.values());
|
||||||
let checkoutDiffTargetCount = 0;
|
let checkoutDiffTargetCount = 0;
|
||||||
@@ -1159,6 +1214,8 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
let checkoutDiffFallbackRefreshTargetCount = 0;
|
let checkoutDiffFallbackRefreshTargetCount = 0;
|
||||||
let terminalDirectorySubscriptionCount = 0;
|
let terminalDirectorySubscriptionCount = 0;
|
||||||
let terminalSubscriptionCount = 0;
|
let terminalSubscriptionCount = 0;
|
||||||
|
let inflightRequests = 0;
|
||||||
|
let peakInflightRequests = 0;
|
||||||
|
|
||||||
for (const connection of uniqueConnections) {
|
for (const connection of uniqueConnections) {
|
||||||
const sessionMetrics = connection.session.getRuntimeMetrics();
|
const sessionMetrics = connection.session.getRuntimeMetrics();
|
||||||
@@ -1169,6 +1226,9 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
sessionMetrics.checkoutDiffFallbackRefreshTargetCount;
|
sessionMetrics.checkoutDiffFallbackRefreshTargetCount;
|
||||||
terminalDirectorySubscriptionCount += sessionMetrics.terminalDirectorySubscriptionCount;
|
terminalDirectorySubscriptionCount += sessionMetrics.terminalDirectorySubscriptionCount;
|
||||||
terminalSubscriptionCount += sessionMetrics.terminalSubscriptionCount;
|
terminalSubscriptionCount += sessionMetrics.terminalSubscriptionCount;
|
||||||
|
inflightRequests += sessionMetrics.inflightRequests;
|
||||||
|
peakInflightRequests = Math.max(peakInflightRequests, sessionMetrics.peakInflightRequests);
|
||||||
|
connection.session.resetPeakInflight();
|
||||||
}
|
}
|
||||||
|
|
||||||
return {
|
return {
|
||||||
@@ -1178,6 +1238,8 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
checkoutDiffFallbackRefreshTargetCount,
|
checkoutDiffFallbackRefreshTargetCount,
|
||||||
terminalDirectorySubscriptionCount,
|
terminalDirectorySubscriptionCount,
|
||||||
terminalSubscriptionCount,
|
terminalSubscriptionCount,
|
||||||
|
inflightRequests,
|
||||||
|
peakInflightRequests,
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1192,6 +1254,8 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
connection.sockets.size === 0 && connection.externalDisconnectCleanupTimeout !== null,
|
connection.sockets.size === 0 && connection.externalDisconnectCleanupTimeout !== null,
|
||||||
).length;
|
).length;
|
||||||
const sessionMetrics = this.collectSessionRuntimeMetrics();
|
const sessionMetrics = this.collectSessionRuntimeMetrics();
|
||||||
|
const latencyStats = this.computeLatencyStats();
|
||||||
|
const agentSnapshot = this.agentManager.getMetricsSnapshot();
|
||||||
|
|
||||||
this.logger.info(
|
this.logger.info(
|
||||||
{
|
{
|
||||||
@@ -1210,6 +1274,8 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
inboundMessageTypesTop: this.getTopCounts(this.inboundMessageCounts, 12),
|
inboundMessageTypesTop: this.getTopCounts(this.inboundMessageCounts, 12),
|
||||||
inboundSessionRequestTypesTop: this.getTopCounts(this.inboundSessionRequestCounts, 20),
|
inboundSessionRequestTypesTop: this.getTopCounts(this.inboundSessionRequestCounts, 20),
|
||||||
runtime: sessionMetrics,
|
runtime: sessionMetrics,
|
||||||
|
latency: latencyStats,
|
||||||
|
agents: agentSnapshot,
|
||||||
},
|
},
|
||||||
"ws_runtime_metrics",
|
"ws_runtime_metrics",
|
||||||
);
|
);
|
||||||
@@ -1221,6 +1287,7 @@ export class VoiceAssistantWebSocketServer {
|
|||||||
}
|
}
|
||||||
this.inboundMessageCounts.clear();
|
this.inboundMessageCounts.clear();
|
||||||
this.inboundSessionRequestCounts.clear();
|
this.inboundSessionRequestCounts.clear();
|
||||||
|
this.requestLatencies.clear();
|
||||||
this.runtimeWindowStartedAt = now;
|
this.runtimeWindowStartedAt = now;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user