Add app diagnostic report (#1728)
* feat(diagnostics): add app diagnostic report * fix(diagnostics): guard app diagnostic runs * fix(diagnostics): include websocket runtime metrics * fix(diagnostics): type websocket metric snapshot * fix(diagnostics): include daemon shell env * fix(loop): make stop cancellation test deterministic
This commit is contained in:
Mohamed Boudra
@@ -635,6 +635,7 @@ describe("LoopService", () => {
|
||||
|
||||
test("stops a running loop and cancels the active worker", async () => {
|
||||
let release: (() => void) | null = null;
|
||||
const cancelledAgentIds: string[] = [];
|
||||
const blocker = new Promise<void>((resolve) => {
|
||||
release = resolve;
|
||||
});
|
||||
@@ -653,6 +654,11 @@ describe("LoopService", () => {
|
||||
registry: storage,
|
||||
logger,
|
||||
});
|
||||
const cancelAgentRun = manager.cancelAgentRun.bind(manager);
|
||||
manager.cancelAgentRun = async (agentId) => {
|
||||
cancelledAgentIds.push(agentId);
|
||||
return cancelAgentRun(agentId);
|
||||
};
|
||||
const service = new LoopService({
|
||||
paseoHome,
|
||||
agentManager: manager,
|
||||
@@ -667,14 +673,26 @@ describe("LoopService", () => {
|
||||
verifyChecks: ["test -f never.txt"],
|
||||
});
|
||||
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
const stopped = await service.stopLoop(loop.id);
|
||||
release?.();
|
||||
const workerAgentId = await waitForActiveWorkerRun(service, manager, loop.id);
|
||||
const stopPromise = service.stopLoop(loop.id);
|
||||
let cancelWaitError: unknown;
|
||||
try {
|
||||
await waitForCancelledAgent(cancelledAgentIds, workerAgentId);
|
||||
} catch (error) {
|
||||
cancelWaitError = error;
|
||||
} finally {
|
||||
release?.();
|
||||
}
|
||||
const stopped = await stopPromise;
|
||||
if (cancelWaitError) {
|
||||
throw cancelWaitError;
|
||||
}
|
||||
|
||||
expect(stopped.status).toBe("stopped");
|
||||
const finalLoop = await service.inspectLoop(loop.id);
|
||||
expect(finalLoop.status).toBe("stopped");
|
||||
expect(finalLoop.iterations[0]?.status).toBe("stopped");
|
||||
expect(cancelledAgentIds).toEqual([workerAgentId]);
|
||||
expect(finalLoop.logs.some((entry) => entry.text.includes("Stop requested"))).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -692,3 +710,34 @@ async function waitForLoopCompletion(service: LoopService, loopId: string): Prom
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
}
|
||||
|
||||
async function waitForActiveWorkerRun(
|
||||
service: LoopService,
|
||||
manager: AgentManager,
|
||||
loopId: string,
|
||||
): Promise<string> {
|
||||
const deadline = Date.now() + 2_000;
|
||||
while (Date.now() < deadline) {
|
||||
const loop = await service.inspectLoop(loopId);
|
||||
const workerAgentId = loop.activeWorkerAgentId;
|
||||
if (workerAgentId && manager.getAgent(workerAgentId)?.activeForegroundTurnId) {
|
||||
return workerAgentId;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
throw new Error("Timed out waiting for loop worker run to start");
|
||||
}
|
||||
|
||||
async function waitForCancelledAgent(
|
||||
cancelledAgentIds: readonly string[],
|
||||
agentId: string,
|
||||
): Promise<void> {
|
||||
const deadline = Date.now() + 2_000;
|
||||
while (Date.now() < deadline) {
|
||||
if (cancelledAgentIds.includes(agentId)) {
|
||||
return;
|
||||
}
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
throw new Error("Timed out waiting for loop worker cancellation");
|
||||
}
|
||||
|
||||
@@ -145,6 +145,7 @@ import { WorkspaceFilesSession } from "./session/files/workspace-files-session.j
|
||||
import { AgentConfigSession } from "./session/agent-config/agent-config-session.js";
|
||||
import { ProjectConfigSession } from "./session/project-config/project-config-session.js";
|
||||
import { DaemonSession, type DaemonRuntimeConfig } from "./session/daemon/daemon-session.js";
|
||||
import type { DaemonWebSocketRuntimeDiagnosticSnapshot } from "./session/daemon/diagnostics.js";
|
||||
import { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import { PushTokenStore } from "./push/token-store.js";
|
||||
import {
|
||||
@@ -463,6 +464,7 @@ export interface SessionOptions {
|
||||
serverId?: string;
|
||||
daemonVersion?: string;
|
||||
daemonRuntimeConfig?: DaemonRuntimeConfig;
|
||||
getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
|
||||
}
|
||||
|
||||
export type SessionLifecycleIntent =
|
||||
@@ -632,6 +634,7 @@ export class Session {
|
||||
serverId,
|
||||
daemonVersion,
|
||||
daemonRuntimeConfig,
|
||||
getWebSocketRuntimeMetrics,
|
||||
} = options;
|
||||
this.clientId = clientId;
|
||||
this.appVersion = appVersion ?? null;
|
||||
@@ -779,7 +782,11 @@ export class Session {
|
||||
serverId,
|
||||
daemonVersion,
|
||||
daemonRuntimeConfig,
|
||||
getWebSocketRuntimeMetrics,
|
||||
listProviderAvailability: () => this.agentManager.listProviderAvailability(),
|
||||
listAgents: () => this.agentManager.listAgents(),
|
||||
listProjects: () => this.projectRegistry.list(),
|
||||
listWorkspaces: () => this.workspaceRegistry.list(),
|
||||
logger: this.sessionLogger,
|
||||
});
|
||||
this.daemonConfigStore = daemonConfigStore;
|
||||
@@ -1501,6 +1508,8 @@ export class Session {
|
||||
return this.daemonSession.handleGetStatusRequest(msg);
|
||||
case "daemon.get_pairing_offer.request":
|
||||
return this.daemonSession.handleGetPairingOfferRequest(msg);
|
||||
case "diagnostics.request":
|
||||
return this.daemonSession.handleDiagnosticsRequest(msg);
|
||||
case "set_daemon_config_request":
|
||||
this.emit({
|
||||
type: "set_daemon_config_response",
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import { mkdtempSync, realpathSync, rmSync } from "node:fs";
|
||||
import { mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, describe, expect, test } from "vitest";
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
type DaemonRuntimeConfig,
|
||||
type DaemonSessionHost,
|
||||
} from "./daemon-session.js";
|
||||
import type { DaemonWebSocketRuntimeDiagnosticSnapshot } from "./diagnostics.js";
|
||||
import type { ProviderAvailability } from "../../agent/agent-manager.js";
|
||||
import type { SessionOutboundMessage } from "../../messages.js";
|
||||
|
||||
@@ -25,24 +26,38 @@ function makeHome(): string {
|
||||
return home;
|
||||
}
|
||||
|
||||
function restoreEnv(name: string, value: string | undefined): void {
|
||||
if (value === undefined) {
|
||||
delete process.env[name];
|
||||
return;
|
||||
}
|
||||
process.env[name] = value;
|
||||
}
|
||||
|
||||
function makeSubsystem(overrides: {
|
||||
serverId?: string;
|
||||
daemonVersion?: string;
|
||||
daemonRuntimeConfig?: DaemonRuntimeConfig;
|
||||
listProviderAvailability?: () => Promise<ProviderAvailability[]>;
|
||||
getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
|
||||
}) {
|
||||
const emitted: SessionOutboundMessage[] = [];
|
||||
const host: DaemonSessionHost = { emit: (msg) => emitted.push(msg) };
|
||||
const paseoHome = makeHome();
|
||||
const subsystem = new DaemonSession({
|
||||
host,
|
||||
paseoHome: makeHome(),
|
||||
paseoHome,
|
||||
serverId: overrides.serverId,
|
||||
daemonVersion: overrides.daemonVersion,
|
||||
daemonRuntimeConfig: overrides.daemonRuntimeConfig,
|
||||
listAgents: () => [],
|
||||
listProjects: async () => [],
|
||||
listWorkspaces: async () => [],
|
||||
listProviderAvailability: overrides.listProviderAvailability ?? (async () => []),
|
||||
getWebSocketRuntimeMetrics: overrides.getWebSocketRuntimeMetrics,
|
||||
logger: pino({ level: "silent" }),
|
||||
});
|
||||
return { subsystem, emitted };
|
||||
return { subsystem, emitted, paseoHome };
|
||||
}
|
||||
|
||||
describe("DaemonSession", () => {
|
||||
@@ -168,4 +183,177 @@ describe("DaemonSession", () => {
|
||||
expect(message.payload.url.startsWith("https://app.example.test")).toBe(true);
|
||||
expect(typeof message.payload.qr).toBe("string");
|
||||
});
|
||||
|
||||
test("diagnostics includes a log tail and redacts connection secrets", async () => {
|
||||
const { subsystem, emitted, paseoHome } = makeSubsystem({
|
||||
serverId: "srv-1",
|
||||
daemonVersion: "1.2.3",
|
||||
daemonRuntimeConfig: {
|
||||
listen: "127.0.0.1:6767",
|
||||
relay: {
|
||||
enabled: true,
|
||||
endpoint: "relay.secret.test:443",
|
||||
publicEndpoint: "relay.secret.test:443",
|
||||
useTls: true,
|
||||
publicUseTls: true,
|
||||
},
|
||||
},
|
||||
});
|
||||
writeFileSync(
|
||||
join(paseoHome, "daemon.log"),
|
||||
"first line\nrelay.secret.test:443 token=super-secret paseo://pairing-secret\n",
|
||||
);
|
||||
|
||||
await subsystem.handleDiagnosticsRequest({ type: "diagnostics.request", requestId: "d-1" });
|
||||
|
||||
expect(emitted).toHaveLength(1);
|
||||
const message = emitted[0];
|
||||
expect(message.type).toBe("diagnostics.response");
|
||||
if (message.type !== "diagnostics.response") {
|
||||
throw new Error("expected diagnostics response");
|
||||
}
|
||||
expect(message.payload.requestId).toBe("d-1");
|
||||
expect(message.payload.diagnostic).toContain("Daemon log tail");
|
||||
expect(message.payload.diagnostic).toContain("first line");
|
||||
expect(message.payload.diagnostic).not.toContain("relay.secret.test:443");
|
||||
expect(message.payload.diagnostic).not.toContain("super-secret");
|
||||
expect(message.payload.diagnostic).not.toContain("pairing-secret");
|
||||
});
|
||||
|
||||
test("diagnostics includes the PATH and shell visible to the daemon", async () => {
|
||||
const originalPath = process.env.PATH;
|
||||
const originalShell = process.env.SHELL;
|
||||
const originalComSpec = process.env.ComSpec;
|
||||
const originalCOMSPEC = process.env.COMSPEC;
|
||||
try {
|
||||
process.env.PATH = "/opt/paseo-test/bin:/usr/bin";
|
||||
process.env.SHELL = "/bin/paseo-test-shell";
|
||||
delete process.env.ComSpec;
|
||||
delete process.env.COMSPEC;
|
||||
|
||||
const { subsystem, emitted } = makeSubsystem({});
|
||||
|
||||
await subsystem.handleDiagnosticsRequest({ type: "diagnostics.request", requestId: "d-env" });
|
||||
|
||||
expect(emitted).toHaveLength(1);
|
||||
const message = emitted[0];
|
||||
expect(message.type).toBe("diagnostics.response");
|
||||
if (message.type !== "diagnostics.response") {
|
||||
throw new Error("expected diagnostics response");
|
||||
}
|
||||
expect(message.payload.diagnostic).toContain("PATH: /opt/paseo-test/bin:/usr/bin");
|
||||
expect(message.payload.diagnostic).toContain("Shell: SHELL=/bin/paseo-test-shell");
|
||||
} finally {
|
||||
restoreEnv("PATH", originalPath);
|
||||
restoreEnv("SHELL", originalShell);
|
||||
restoreEnv("ComSpec", originalComSpec);
|
||||
restoreEnv("COMSPEC", originalCOMSPEC);
|
||||
}
|
||||
});
|
||||
|
||||
test("diagnostics includes the last flushed websocket runtime metrics", async () => {
|
||||
const { subsystem, emitted } = makeSubsystem({
|
||||
getWebSocketRuntimeMetrics: () => ({
|
||||
collectedAt: "2026-01-02T03:04:05.000Z",
|
||||
windowMs: 30_000,
|
||||
final: false,
|
||||
sessions: {
|
||||
activeConnections: 2,
|
||||
externalSessionKeys: 3,
|
||||
reconnectGraceSessions: 1,
|
||||
},
|
||||
sockets: {
|
||||
activeSockets: 2,
|
||||
pendingConnections: 1,
|
||||
},
|
||||
counters: {
|
||||
connectedAwaitingHello: 1,
|
||||
helloResumed: 0,
|
||||
helloNew: 2,
|
||||
pendingDisconnected: 0,
|
||||
sessionDisconnectedWaitingReconnect: 0,
|
||||
sessionSocketDisconnectedAttached: 0,
|
||||
sessionCleanup: 0,
|
||||
validationFailed: 0,
|
||||
binaryBeforeHelloRejected: 0,
|
||||
pendingMessageRejectedBeforeHello: 0,
|
||||
missingConnectionForMessage: 0,
|
||||
unexpectedHelloOnActiveConnection: 0,
|
||||
relayExternalSocketAttached: 0,
|
||||
originRejected: 0,
|
||||
hostRejected: 0,
|
||||
},
|
||||
inboundMessageTypesTop: [["session", 4]],
|
||||
inboundSessionRequestTypesTop: [["diagnostics.request", 2]],
|
||||
outboundMessageTypesTop: [["session_message", 5]],
|
||||
outboundSessionMessageTypesTop: [["diagnostics.response", 2]],
|
||||
outboundAgentStreamTypesTop: [["timeline:message", 3]],
|
||||
outboundAgentStreamAgentsTop: [["agent-1", 3]],
|
||||
outboundBinaryFrameTypesTop: [["binary", 1]],
|
||||
bufferedAmount: {
|
||||
p95: 128,
|
||||
max: 256,
|
||||
},
|
||||
eventLoopDelay: {
|
||||
p50Ms: 1,
|
||||
p99Ms: 4,
|
||||
maxMs: 7,
|
||||
},
|
||||
runtime: {
|
||||
inflightRequests: 1,
|
||||
peakInflightRequests: 3,
|
||||
terminalSubscriptionCount: 4,
|
||||
terminalDirectorySubscriptionCount: 5,
|
||||
checkoutDiffTargetCount: 6,
|
||||
checkoutDiffSubscriptionCount: 7,
|
||||
checkoutDiffWatcherCount: 8,
|
||||
checkoutDiffFallbackRefreshTargetCount: 9,
|
||||
},
|
||||
latency: [
|
||||
{
|
||||
type: "diagnostics.request",
|
||||
count: 2,
|
||||
minMs: 3,
|
||||
maxMs: 7,
|
||||
p50Ms: 4,
|
||||
totalMs: 11,
|
||||
},
|
||||
],
|
||||
agents: {
|
||||
total: 10,
|
||||
byLifecycle: {
|
||||
idle: 8,
|
||||
running: 2,
|
||||
},
|
||||
withActiveForegroundTurn: 2,
|
||||
timelineStats: {
|
||||
totalItems: 42,
|
||||
maxItemsPerAgent: 12,
|
||||
},
|
||||
},
|
||||
}),
|
||||
});
|
||||
|
||||
await subsystem.handleDiagnosticsRequest({ type: "diagnostics.request", requestId: "d-2" });
|
||||
|
||||
expect(emitted).toHaveLength(1);
|
||||
const message = emitted[0];
|
||||
expect(message.type).toBe("diagnostics.response");
|
||||
if (message.type !== "diagnostics.response") {
|
||||
throw new Error("expected diagnostics response");
|
||||
}
|
||||
expect(message.payload.diagnostic).toContain("WebSocket runtime metrics");
|
||||
expect(message.payload.diagnostic).toContain("Collected at: 2026-01-02T03:04:05.000Z");
|
||||
expect(message.payload.diagnostic).toContain(
|
||||
"Sessions: active=2, externalKeys=3, reconnectGrace=1",
|
||||
);
|
||||
expect(message.payload.diagnostic).toContain(
|
||||
"Latency: diagnostics.request count=2 p50=4ms max=7ms total=11ms",
|
||||
);
|
||||
expect(message.payload.diagnostic).toContain("Inbound session requests: diagnostics.request=2");
|
||||
expect(message.payload.diagnostic).toContain(
|
||||
"Checkout diff: targets=6, subscriptions=7, watchers=8, fallbackRefreshTargets=9",
|
||||
);
|
||||
expect(message.payload.diagnostic).toContain("Agent lifecycle: idle=8, running=2");
|
||||
});
|
||||
});
|
||||
|
||||
@@ -3,6 +3,12 @@ import type { ProviderAvailability } from "../../agent/agent-manager.js";
|
||||
import type { SessionInboundMessage, SessionOutboundMessage } from "../../messages.js";
|
||||
import { getPidLockInfo } from "../../pid-lock.js";
|
||||
import { generateLocalPairingOffer } from "../../pairing-offer.js";
|
||||
import {
|
||||
collectDaemonDiagnostics,
|
||||
type DaemonWebSocketRuntimeDiagnosticSnapshot,
|
||||
} from "./diagnostics.js";
|
||||
import type { ManagedAgent } from "../../agent/agent-manager.js";
|
||||
import type { PersistedProjectRecord, PersistedWorkspaceRecord } from "../../workspace-registry.js";
|
||||
|
||||
export interface DaemonRuntimeConfig {
|
||||
listen: string | null;
|
||||
@@ -26,7 +32,11 @@ export interface DaemonSessionOptions {
|
||||
serverId: string | undefined;
|
||||
daemonVersion: string | undefined;
|
||||
daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
|
||||
listAgents: () => ManagedAgent[];
|
||||
listProjects: () => Promise<PersistedProjectRecord[]>;
|
||||
listWorkspaces: () => Promise<PersistedWorkspaceRecord[]>;
|
||||
listProviderAvailability: () => Promise<ProviderAvailability[]>;
|
||||
getWebSocketRuntimeMetrics?: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
|
||||
logger: pino.Logger;
|
||||
}
|
||||
|
||||
@@ -43,7 +53,11 @@ export class DaemonSession {
|
||||
private readonly serverId: string | undefined;
|
||||
private readonly daemonVersion: string | undefined;
|
||||
private readonly daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
|
||||
private readonly listAgents: () => ManagedAgent[];
|
||||
private readonly listProjects: () => Promise<PersistedProjectRecord[]>;
|
||||
private readonly listWorkspaces: () => Promise<PersistedWorkspaceRecord[]>;
|
||||
private readonly listProviderAvailability: () => Promise<ProviderAvailability[]>;
|
||||
private readonly getWebSocketRuntimeMetrics: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
|
||||
private readonly logger: pino.Logger;
|
||||
|
||||
constructor(options: DaemonSessionOptions) {
|
||||
@@ -52,7 +66,11 @@ export class DaemonSession {
|
||||
this.serverId = options.serverId;
|
||||
this.daemonVersion = options.daemonVersion;
|
||||
this.daemonRuntimeConfig = options.daemonRuntimeConfig;
|
||||
this.listAgents = options.listAgents;
|
||||
this.listProjects = options.listProjects;
|
||||
this.listWorkspaces = options.listWorkspaces;
|
||||
this.listProviderAvailability = options.listProviderAvailability;
|
||||
this.getWebSocketRuntimeMetrics = options.getWebSocketRuntimeMetrics ?? (() => null);
|
||||
this.logger = options.logger;
|
||||
}
|
||||
|
||||
@@ -136,4 +154,41 @@ export class DaemonSession {
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async handleDiagnosticsRequest(
|
||||
msg: Extract<SessionInboundMessage, { type: "diagnostics.request" }>,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const diagnostic = await collectDaemonDiagnostics({
|
||||
paseoHome: this.paseoHome,
|
||||
serverId: this.serverId,
|
||||
daemonVersion: this.daemonVersion,
|
||||
daemonRuntimeConfig: this.daemonRuntimeConfig,
|
||||
listAgents: this.listAgents,
|
||||
listProjects: this.listProjects,
|
||||
listWorkspaces: this.listWorkspaces,
|
||||
listProviderAvailability: this.listProviderAvailability,
|
||||
getWebSocketRuntimeMetrics: this.getWebSocketRuntimeMetrics,
|
||||
logger: this.logger,
|
||||
});
|
||||
this.host.emit({
|
||||
type: "diagnostics.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
diagnostic,
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.error({ err: error }, "Failed to handle diagnostics request");
|
||||
this.host.emit({
|
||||
type: "diagnostics.response",
|
||||
payload: {
|
||||
requestId: msg.requestId,
|
||||
diagnostic: `Paseo diagnostics\n Error: ${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,547 @@
|
||||
import { open, statfs } from "node:fs/promises";
|
||||
import { cpus, freemem, loadavg, platform, release, totalmem, type } from "node:os";
|
||||
import path from "node:path";
|
||||
|
||||
import type pino from "pino";
|
||||
|
||||
import type { ManagedAgent, ProviderAvailability } from "../../agent/agent-manager.js";
|
||||
import type { WebSocketRuntimeDiagnosticSnapshot } from "../../websocket/runtime-metrics.js";
|
||||
import type { PersistedProjectRecord, PersistedWorkspaceRecord } from "../../workspace-registry.js";
|
||||
import { execCommand } from "../../../utils/spawn.js";
|
||||
import type { DaemonRuntimeConfig } from "./daemon-session.js";
|
||||
|
||||
interface DiagnosticEntry {
|
||||
label: string;
|
||||
value: string;
|
||||
}
|
||||
|
||||
export interface DaemonDiagnosticsOptions {
|
||||
paseoHome: string;
|
||||
serverId: string | undefined;
|
||||
daemonVersion: string | undefined;
|
||||
daemonRuntimeConfig: DaemonRuntimeConfig | undefined;
|
||||
listAgents: () => ManagedAgent[];
|
||||
listProjects: () => Promise<PersistedProjectRecord[]>;
|
||||
listWorkspaces: () => Promise<PersistedWorkspaceRecord[]>;
|
||||
listProviderAvailability: () => Promise<ProviderAvailability[]>;
|
||||
getWebSocketRuntimeMetrics: () => DaemonWebSocketRuntimeDiagnosticSnapshot | null;
|
||||
logger: pino.Logger;
|
||||
}
|
||||
|
||||
interface DiagnosticWebSocketRuntimeMetrics {
|
||||
terminalDirectorySubscriptionCount: number;
|
||||
terminalSubscriptionCount: number;
|
||||
inflightRequests: number;
|
||||
peakInflightRequests: number;
|
||||
checkoutDiffTargetCount: number;
|
||||
checkoutDiffSubscriptionCount: number;
|
||||
checkoutDiffWatcherCount: number;
|
||||
checkoutDiffFallbackRefreshTargetCount: number;
|
||||
}
|
||||
|
||||
interface DiagnosticAgentRuntimeMetrics {
|
||||
total: number;
|
||||
byLifecycle: Record<string, number>;
|
||||
withActiveForegroundTurn: number;
|
||||
timelineStats: {
|
||||
totalItems: number;
|
||||
maxItemsPerAgent: number;
|
||||
};
|
||||
}
|
||||
|
||||
export type DaemonWebSocketRuntimeDiagnosticSnapshot = WebSocketRuntimeDiagnosticSnapshot<
|
||||
DiagnosticWebSocketRuntimeMetrics,
|
||||
DiagnosticAgentRuntimeMetrics
|
||||
>;
|
||||
|
||||
const TOOL_TIMEOUT_MS = 3_000;
|
||||
const TOOL_OUTPUT_LIMIT = 512;
|
||||
const LOG_TAIL_LINES = 80;
|
||||
const LOG_TAIL_MAX_BYTES = 64 * 1024;
|
||||
|
||||
export async function collectDaemonDiagnostics(options: DaemonDiagnosticsOptions): Promise<string> {
|
||||
const sections: string[] = [
|
||||
formatSection("Paseo diagnostics", [
|
||||
{ label: "Collected at", value: new Date().toISOString() },
|
||||
{ label: "Server ID", value: options.serverId ?? "unknown" },
|
||||
{ label: "Daemon version", value: options.daemonVersion ?? "unknown" },
|
||||
]),
|
||||
];
|
||||
|
||||
sections.push(
|
||||
await safeSection("Daemon process", () => collectProcessEntries(options), options.logger),
|
||||
);
|
||||
sections.push(
|
||||
await safeSection("Runtime config", () => collectRuntimeConfigEntries(options), options.logger),
|
||||
);
|
||||
sections.push(await safeSection("System", collectSystemEntries, options.logger));
|
||||
sections.push(await safeSection("Disk", () => collectDiskEntries(options), options.logger));
|
||||
sections.push(await safeSection("Agents", () => collectAgentEntries(options), options.logger));
|
||||
sections.push(
|
||||
await safeSection("Workspaces", () => collectWorkspaceEntries(options), options.logger),
|
||||
);
|
||||
sections.push(
|
||||
await safeSection("Providers", () => collectProviderEntries(options), options.logger),
|
||||
);
|
||||
sections.push(
|
||||
await safeSection(
|
||||
"WebSocket runtime metrics",
|
||||
() => collectWebSocketRuntimeEntries(options),
|
||||
options.logger,
|
||||
),
|
||||
);
|
||||
sections.push(await safeSection("Tools", collectToolEntries, options.logger));
|
||||
sections.push(await safeLogTailSection(options));
|
||||
|
||||
return redactDiagnostic(sections.filter(Boolean).join("\n\n"), options);
|
||||
}
|
||||
|
||||
async function safeSection(
|
||||
title: string,
|
||||
collect: () => DiagnosticEntry[] | Promise<DiagnosticEntry[]>,
|
||||
logger: pino.Logger,
|
||||
): Promise<string> {
|
||||
try {
|
||||
return formatSection(title, await collect());
|
||||
} catch (error) {
|
||||
logger.debug({ err: error, title }, "diagnostic section failed");
|
||||
return formatSection(title, [{ label: "Error", value: toErrorMessage(error) }]);
|
||||
}
|
||||
}
|
||||
|
||||
function formatSection(title: string, entries: DiagnosticEntry[]): string {
|
||||
return [title, ...entries.map((entry) => ` ${entry.label}: ${entry.value}`)].join("\n");
|
||||
}
|
||||
|
||||
function collectProcessEntries(options: DaemonDiagnosticsOptions): DiagnosticEntry[] {
|
||||
const memory = process.memoryUsage();
|
||||
return [
|
||||
{ label: "PID", value: String(process.pid) },
|
||||
{ label: "Node", value: process.version },
|
||||
{ label: "Node path", value: process.execPath },
|
||||
{ label: "PATH", value: getEnvValue("PATH", "Path") ?? "unset" },
|
||||
{ label: "Shell", value: formatDaemonShell() },
|
||||
{ label: "Uptime", value: formatDurationMs(process.uptime() * 1000) },
|
||||
{ label: "Paseo home", value: options.paseoHome },
|
||||
{ label: "RSS", value: formatBytes(memory.rss) },
|
||||
{
|
||||
label: "Heap used",
|
||||
value: `${formatBytes(memory.heapUsed)} / ${formatBytes(memory.heapTotal)}`,
|
||||
},
|
||||
{ label: "External", value: formatBytes(memory.external) },
|
||||
];
|
||||
}
|
||||
|
||||
function collectRuntimeConfigEntries(options: DaemonDiagnosticsOptions): DiagnosticEntry[] {
|
||||
const relay = options.daemonRuntimeConfig?.relay ?? null;
|
||||
return [
|
||||
{ label: "Listen", value: formatListenKind(options.daemonRuntimeConfig?.listen ?? null) },
|
||||
{ label: "Relay enabled", value: relay ? String(relay.enabled) : "false" },
|
||||
{ label: "Relay endpoint configured", value: relay?.endpoint ? "true" : "false" },
|
||||
{ label: "Relay public endpoint configured", value: relay?.publicEndpoint ? "true" : "false" },
|
||||
{ label: "Relay TLS", value: relay ? String(relay.useTls) : "n/a" },
|
||||
{ label: "Relay public TLS", value: relay ? String(relay.publicUseTls) : "n/a" },
|
||||
];
|
||||
}
|
||||
|
||||
function collectSystemEntries(): DiagnosticEntry[] {
|
||||
const loads = loadavg();
|
||||
return [
|
||||
{ label: "OS", value: `${type()} ${release()}` },
|
||||
{ label: "Platform", value: `${platform()} ${process.arch}` },
|
||||
{ label: "CPU cores", value: String(cpus().length) },
|
||||
{ label: "Load avg", value: loads.map((value) => value.toFixed(2)).join(", ") },
|
||||
{ label: "Memory free", value: `${formatBytes(freemem())} / ${formatBytes(totalmem())}` },
|
||||
];
|
||||
}
|
||||
|
||||
async function collectDiskEntries(options: DaemonDiagnosticsOptions): Promise<DiagnosticEntry[]> {
|
||||
const stats = await statfs(options.paseoHome);
|
||||
const freeBytes = stats.bavail * stats.bsize;
|
||||
const totalBytes = stats.blocks * stats.bsize;
|
||||
return [
|
||||
{ label: "Path", value: options.paseoHome },
|
||||
{ label: "Free", value: `${formatBytes(freeBytes)} / ${formatBytes(totalBytes)}` },
|
||||
];
|
||||
}
|
||||
|
||||
function collectAgentEntries(options: DaemonDiagnosticsOptions): DiagnosticEntry[] {
|
||||
const agents = options.listAgents();
|
||||
return [
|
||||
{ label: "Total", value: String(agents.length) },
|
||||
{ label: "By provider", value: formatCountMap(countBy(agents, (agent) => agent.provider)) },
|
||||
{ label: "By lifecycle", value: formatCountMap(countBy(agents, (agent) => agent.lifecycle)) },
|
||||
{
|
||||
label: "Pending permissions",
|
||||
value: String(
|
||||
agents.reduce((total, agent) => total + (agent.pendingPermissions?.size ?? 0), 0),
|
||||
),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
async function collectWorkspaceEntries(
|
||||
options: DaemonDiagnosticsOptions,
|
||||
): Promise<DiagnosticEntry[]> {
|
||||
const [projects, workspaces] = await Promise.all([
|
||||
options.listProjects(),
|
||||
options.listWorkspaces(),
|
||||
]);
|
||||
const activeProjects = projects.filter((project) => !project.archivedAt);
|
||||
const activeWorkspaces = workspaces.filter((workspace) => !workspace.archivedAt);
|
||||
return [
|
||||
{ label: "Projects", value: `${activeProjects.length} active / ${projects.length} total` },
|
||||
{
|
||||
label: "Workspaces",
|
||||
value: `${activeWorkspaces.length} active / ${workspaces.length} total`,
|
||||
},
|
||||
{
|
||||
label: "Workspaces by kind",
|
||||
value: formatCountMap(countBy(activeWorkspaces, (workspace) => workspace.kind)),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
async function collectProviderEntries(
|
||||
options: DaemonDiagnosticsOptions,
|
||||
): Promise<DiagnosticEntry[]> {
|
||||
const providers = await options.listProviderAvailability();
|
||||
return [
|
||||
{ label: "Total", value: String(providers.length) },
|
||||
{
|
||||
label: "Available",
|
||||
value: String(providers.filter((provider) => provider.available).length),
|
||||
},
|
||||
{
|
||||
label: "Unavailable",
|
||||
value:
|
||||
providers
|
||||
.filter((provider) => !provider.available)
|
||||
.map((provider) =>
|
||||
provider.error ? `${provider.provider} (${provider.error})` : provider.provider,
|
||||
)
|
||||
.join(", ") || "none",
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
async function collectToolEntries(): Promise<DiagnosticEntry[]> {
|
||||
const [git, gh] = await Promise.all([
|
||||
checkTool("git", ["--version"]),
|
||||
checkTool("gh", ["--version"]),
|
||||
]);
|
||||
return [
|
||||
{ label: "git", value: git },
|
||||
{ label: "gh", value: gh },
|
||||
];
|
||||
}
|
||||
|
||||
function collectWebSocketRuntimeEntries(options: DaemonDiagnosticsOptions): DiagnosticEntry[] {
|
||||
const snapshot = options.getWebSocketRuntimeMetrics();
|
||||
if (!snapshot) {
|
||||
return [{ label: "Status", value: "no runtime metrics window has been flushed yet" }];
|
||||
}
|
||||
|
||||
const runtime = snapshot.runtime;
|
||||
const agents = snapshot.agents;
|
||||
|
||||
return [
|
||||
{ label: "Collected at", value: snapshot.collectedAt },
|
||||
{ label: "Window", value: formatDurationMs(snapshot.windowMs) },
|
||||
{ label: "Final", value: String(snapshot.final) },
|
||||
{
|
||||
label: "Sessions",
|
||||
value: [
|
||||
`active=${snapshot.sessions.activeConnections}`,
|
||||
`externalKeys=${snapshot.sessions.externalSessionKeys}`,
|
||||
`reconnectGrace=${snapshot.sessions.reconnectGraceSessions}`,
|
||||
].join(", "),
|
||||
},
|
||||
{
|
||||
label: "Sockets",
|
||||
value: [
|
||||
`active=${snapshot.sockets.activeSockets}`,
|
||||
`pending=${snapshot.sockets.pendingConnections}`,
|
||||
].join(", "),
|
||||
},
|
||||
{
|
||||
label: "Runtime requests",
|
||||
value: [
|
||||
`inflight=${formatNumberMetric(runtime.inflightRequests)}`,
|
||||
`peakInflight=${formatNumberMetric(runtime.peakInflightRequests)}`,
|
||||
].join(", "),
|
||||
},
|
||||
{
|
||||
label: "Terminal subscriptions",
|
||||
value: [
|
||||
`terminals=${formatNumberMetric(runtime.terminalSubscriptionCount)}`,
|
||||
`directories=${formatNumberMetric(runtime.terminalDirectorySubscriptionCount)}`,
|
||||
].join(", "),
|
||||
},
|
||||
{
|
||||
label: "Checkout diff",
|
||||
value: [
|
||||
`targets=${formatNumberMetric(runtime.checkoutDiffTargetCount)}`,
|
||||
`subscriptions=${formatNumberMetric(runtime.checkoutDiffSubscriptionCount)}`,
|
||||
`watchers=${formatNumberMetric(runtime.checkoutDiffWatcherCount)}`,
|
||||
`fallbackRefreshTargets=${formatNumberMetric(
|
||||
runtime.checkoutDiffFallbackRefreshTargetCount,
|
||||
)}`,
|
||||
].join(", "),
|
||||
},
|
||||
{
|
||||
label: "Buffered amount",
|
||||
value: `p95=${formatBytes(snapshot.bufferedAmount.p95)}, max=${formatBytes(
|
||||
snapshot.bufferedAmount.max,
|
||||
)}`,
|
||||
},
|
||||
{ label: "Event loop delay", value: formatEventLoopDelay(snapshot.eventLoopDelay) },
|
||||
{ label: "Latency", value: formatLatencyStats(snapshot.latency) },
|
||||
{ label: "Inbound messages", value: formatTopCounts(snapshot.inboundMessageTypesTop) },
|
||||
{
|
||||
label: "Inbound session requests",
|
||||
value: formatTopCounts(snapshot.inboundSessionRequestTypesTop),
|
||||
},
|
||||
{ label: "Outbound messages", value: formatTopCounts(snapshot.outboundMessageTypesTop) },
|
||||
{
|
||||
label: "Outbound session messages",
|
||||
value: formatTopCounts(snapshot.outboundSessionMessageTypesTop),
|
||||
},
|
||||
{ label: "Agent streams", value: formatTopCounts(snapshot.outboundAgentStreamTypesTop) },
|
||||
{ label: "Agent stream agents", value: formatTopCounts(snapshot.outboundAgentStreamAgentsTop) },
|
||||
{ label: "Binary frames", value: formatTopCounts(snapshot.outboundBinaryFrameTypesTop) },
|
||||
{ label: "Counters", value: formatNonZeroNumberRecord(snapshot.counters) },
|
||||
{
|
||||
label: "Agent metrics",
|
||||
value: [
|
||||
`total=${formatNumberMetric(agents.total)}`,
|
||||
`activeForegroundTurns=${formatNumberMetric(agents.withActiveForegroundTurn)}`,
|
||||
].join(", "),
|
||||
},
|
||||
{ label: "Agent lifecycle", value: formatNumberRecord(agents.byLifecycle) },
|
||||
{
|
||||
label: "Agent timelines",
|
||||
value: [
|
||||
`items=${formatNumberMetric(agents.timelineStats?.totalItems)}`,
|
||||
`maxPerAgent=${formatNumberMetric(agents.timelineStats?.maxItemsPerAgent)}`,
|
||||
].join(", "),
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function formatTopCounts(counts: Array<[string, number]>): string {
|
||||
if (counts.length === 0) return "none";
|
||||
return counts.map(([key, count]) => `${key}=${count}`).join(", ");
|
||||
}
|
||||
|
||||
function formatLatencyStats(stats: DaemonWebSocketRuntimeDiagnosticSnapshot["latency"]): string {
|
||||
if (stats.length === 0) return "none";
|
||||
return stats
|
||||
.map(
|
||||
(stat) =>
|
||||
`${stat.type} count=${stat.count} p50=${formatMilliseconds(
|
||||
stat.p50Ms,
|
||||
)} max=${formatMilliseconds(stat.maxMs)} total=${formatMilliseconds(stat.totalMs)}`,
|
||||
)
|
||||
.join("; ");
|
||||
}
|
||||
|
||||
function formatEventLoopDelay(
|
||||
stats: DaemonWebSocketRuntimeDiagnosticSnapshot["eventLoopDelay"],
|
||||
): string {
|
||||
if (!stats) return "unavailable";
|
||||
return `p50=${formatMilliseconds(stats.p50Ms)}, p99=${formatMilliseconds(
|
||||
stats.p99Ms,
|
||||
)}, max=${formatMilliseconds(stats.maxMs)}`;
|
||||
}
|
||||
|
||||
function formatNumberRecord(record: object | undefined): string {
|
||||
if (!record) return "unknown";
|
||||
const entries = Object.entries(record).filter(([, value]) => Number.isFinite(value));
|
||||
if (entries.length === 0) return "none";
|
||||
return entries
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([key, value]) => `${key}=${value}`)
|
||||
.join(", ");
|
||||
}
|
||||
|
||||
function formatNonZeroNumberRecord(record: object): string {
|
||||
const entries = Object.entries(record).filter(
|
||||
([, value]) => Number.isFinite(value) && value !== 0,
|
||||
);
|
||||
if (entries.length === 0) return "none";
|
||||
return entries
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([key, value]) => `${key}=${value}`)
|
||||
.join(", ");
|
||||
}
|
||||
|
||||
function formatNumberMetric(value: number | undefined): string {
|
||||
if (typeof value !== "number" || !Number.isFinite(value)) return "unknown";
|
||||
return String(value);
|
||||
}
|
||||
|
||||
function formatMilliseconds(ms: number): string {
|
||||
if (!Number.isFinite(ms) || ms < 0) return "unknown";
|
||||
return `${Math.round(ms)}ms`;
|
||||
}
|
||||
|
||||
async function checkTool(command: string, args: string[]): Promise<string> {
|
||||
try {
|
||||
const result = await execCommand(command, args, {
|
||||
timeout: TOOL_TIMEOUT_MS,
|
||||
maxBuffer: TOOL_OUTPUT_LIMIT * 2,
|
||||
});
|
||||
const output = truncateForDiagnostic(
|
||||
(result.stdout || result.stderr).trim(),
|
||||
TOOL_OUTPUT_LIMIT,
|
||||
);
|
||||
return output || "ok";
|
||||
} catch (error) {
|
||||
return `error: ${truncateForDiagnostic(toErrorMessage(error), TOOL_OUTPUT_LIMIT)}`;
|
||||
}
|
||||
}
|
||||
|
||||
async function safeLogTailSection(options: DaemonDiagnosticsOptions): Promise<string> {
|
||||
const logPath = path.join(options.paseoHome, "daemon.log");
|
||||
try {
|
||||
const tail = await tailFile(logPath, LOG_TAIL_LINES, LOG_TAIL_MAX_BYTES);
|
||||
return ["Daemon log tail", ` Path: ${logPath}`, tail ? tail : " No log lines found"].join(
|
||||
"\n",
|
||||
);
|
||||
} catch (error) {
|
||||
options.logger.debug({ err: error, logPath }, "diagnostic log tail failed");
|
||||
return ["Daemon log tail", ` Path: ${logPath}`, ` Error: ${toErrorMessage(error)}`].join(
|
||||
"\n",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
async function tailFile(filePath: string, lines: number, maxBytes: number): Promise<string> {
|
||||
const handle = await open(filePath, "r");
|
||||
try {
|
||||
const stats = await handle.stat();
|
||||
const length = Math.min(stats.size, maxBytes);
|
||||
const buffer = Buffer.alloc(length);
|
||||
await handle.read(buffer, 0, length, stats.size - length);
|
||||
return buffer
|
||||
.toString("utf8")
|
||||
.split("\n")
|
||||
.filter(Boolean)
|
||||
.slice(-lines)
|
||||
.map((line) => ` ${line}`)
|
||||
.join("\n");
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
}
|
||||
|
||||
function formatListenKind(listen: string | null): string {
|
||||
if (!listen) return "not configured";
|
||||
if (listen.startsWith("unix://") || listen.startsWith("/")) return "local socket";
|
||||
if (listen.startsWith("pipe://") || listen.startsWith("\\\\.\\pipe\\")) return "local pipe";
|
||||
return "direct TCP";
|
||||
}
|
||||
|
||||
function countBy<T>(
|
||||
items: T[],
|
||||
getKey: (item: T) => string | null | undefined,
|
||||
): Map<string, number> {
|
||||
const counts = new Map<string, number>();
|
||||
for (const item of items) {
|
||||
const key = getKey(item) || "unknown";
|
||||
counts.set(key, (counts.get(key) ?? 0) + 1);
|
||||
}
|
||||
return counts;
|
||||
}
|
||||
|
||||
function formatCountMap(counts: Map<string, number>): string {
|
||||
if (counts.size === 0) return "none";
|
||||
return [...counts.entries()]
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([key, count]) => `${key}=${count}`)
|
||||
.join(", ");
|
||||
}
|
||||
|
||||
function formatDurationMs(ms: number): string {
|
||||
const seconds = Math.max(0, Math.round(ms / 1000));
|
||||
const hours = Math.floor(seconds / 3600);
|
||||
const minutes = Math.floor((seconds % 3600) / 60);
|
||||
const remainingSeconds = seconds % 60;
|
||||
if (hours > 0) return `${hours}h ${minutes}m ${remainingSeconds}s`;
|
||||
if (minutes > 0) return `${minutes}m ${remainingSeconds}s`;
|
||||
return `${remainingSeconds}s`;
|
||||
}
|
||||
|
||||
function formatBytes(bytes: number): string {
|
||||
if (!Number.isFinite(bytes) || bytes < 0) return "unknown";
|
||||
const units = ["B", "KiB", "MiB", "GiB", "TiB"];
|
||||
let value = bytes;
|
||||
let unitIndex = 0;
|
||||
while (value >= 1024 && unitIndex < units.length - 1) {
|
||||
value /= 1024;
|
||||
unitIndex += 1;
|
||||
}
|
||||
return `${value.toFixed(unitIndex === 0 ? 0 : 1)} ${units[unitIndex]}`;
|
||||
}
|
||||
|
||||
function truncateForDiagnostic(value: string, maxLength: number): string {
|
||||
const trimmed = value.trim();
|
||||
if (trimmed.length <= maxLength) return trimmed;
|
||||
return `${trimmed.slice(0, maxLength)}...(truncated)`;
|
||||
}
|
||||
|
||||
function formatDaemonShell(): string {
|
||||
const shell = getEnvValue("SHELL");
|
||||
if (shell) return `SHELL=${shell}`;
|
||||
const comspec = getEnvValue("ComSpec", "COMSPEC");
|
||||
if (comspec) return `ComSpec=${comspec}`;
|
||||
return "unset";
|
||||
}
|
||||
|
||||
function getEnvValue(...names: string[]): string | null {
|
||||
for (const name of names) {
|
||||
const value = process.env[name];
|
||||
if (value) return value;
|
||||
}
|
||||
|
||||
const lowerNames = new Set(names.map((name) => name.toLowerCase()));
|
||||
for (const [key, value] of Object.entries(process.env)) {
|
||||
if (value && lowerNames.has(key.toLowerCase())) return value;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function toErrorMessage(error: unknown): string {
|
||||
if (error instanceof Error) return error.message;
|
||||
if (typeof error === "string") return error;
|
||||
return String(error);
|
||||
}
|
||||
|
||||
export function redactDiagnostic(
|
||||
value: string,
|
||||
options?: Partial<DaemonDiagnosticsOptions>,
|
||||
): string {
|
||||
let redacted = value;
|
||||
const sensitiveValues = [
|
||||
options?.daemonRuntimeConfig?.listen,
|
||||
options?.daemonRuntimeConfig?.relay?.endpoint,
|
||||
options?.daemonRuntimeConfig?.relay?.publicEndpoint,
|
||||
].filter((item): item is string => Boolean(item));
|
||||
|
||||
for (const sensitive of sensitiveValues) {
|
||||
redacted = redacted.split(sensitive).join("[redacted]");
|
||||
}
|
||||
|
||||
return redacted
|
||||
.replace(/paseo:\/\/\S+/gi, "paseo://[redacted]")
|
||||
.replace(
|
||||
/([?&](?:password|token|secret|key|publicKey|daemonPublicKeyB64)=)[^&\s"']+/gi,
|
||||
"$1[redacted]",
|
||||
)
|
||||
.replace(
|
||||
/((?:password|token|secret|authorization|api[_-]?key|daemonPublicKeyB64|relayKey)\s*[:=]\s*)("[^"]+"|'[^']+'|[^\s,}]+)/gi,
|
||||
"$1[redacted]",
|
||||
);
|
||||
}
|
||||
@@ -3,7 +3,7 @@ import type { IncomingMessage, Server as HTTPServer } from "http";
|
||||
import { basename, join } from "path";
|
||||
import { hostname as getHostname } from "node:os";
|
||||
import { monitorEventLoopDelay } from "node:perf_hooks";
|
||||
import type { AgentManager } from "./agent/agent-manager.js";
|
||||
import type { AgentManager, AgentMetricsSnapshot } from "./agent/agent-manager.js";
|
||||
import type { AgentStorage } from "./agent/agent-storage.js";
|
||||
import type { DownloadTokenStore } from "./file-download/token-store.js";
|
||||
import type { TerminalManager } from "../terminal/terminal-manager.js";
|
||||
@@ -60,6 +60,7 @@ import {
|
||||
import {
|
||||
WebSocketRuntimeMetricsWindow,
|
||||
type WebSocketRuntimeCounters,
|
||||
type WebSocketRuntimeDiagnosticSnapshot,
|
||||
} from "./websocket/runtime-metrics.js";
|
||||
import { ProviderUsageService } from "../services/quota-fetcher/service.js";
|
||||
|
||||
@@ -81,6 +82,11 @@ interface WebSocketServerConfig {
|
||||
}
|
||||
|
||||
type WebSocketRuntimeMetrics = SessionRuntimeMetrics & CheckoutDiffMetrics;
|
||||
type WebSocketRuntimeDiagnosticPayload = WebSocketRuntimeDiagnosticSnapshot<
|
||||
WebSocketRuntimeMetrics,
|
||||
AgentMetricsSnapshot
|
||||
>;
|
||||
type WebSocketRuntimeMetricsLogPayload = Omit<WebSocketRuntimeDiagnosticPayload, "collectedAt">;
|
||||
|
||||
type TerminalAttentionReason = "finished" | "needs_input";
|
||||
|
||||
@@ -403,6 +409,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
| null;
|
||||
private serverCapabilities: ServerCapabilities | undefined;
|
||||
private readonly runtimeMetrics = new WebSocketRuntimeMetricsWindow();
|
||||
private lastRuntimeMetricsSnapshot: WebSocketRuntimeDiagnosticPayload | null = null;
|
||||
private runtimeMetricsInterval: ReturnType<typeof setInterval> | null = null;
|
||||
private eventLoopDelayMonitor: ReturnType<typeof monitorEventLoopDelay> | null = null;
|
||||
private unsubscribeSpeechReadiness: (() => void) | null = null;
|
||||
@@ -1020,6 +1027,7 @@ export class VoiceAssistantWebSocketServer {
|
||||
serverId: this.serverId,
|
||||
daemonVersion: this.daemonVersion,
|
||||
daemonRuntimeConfig: this.daemonRuntimeConfig,
|
||||
getWebSocketRuntimeMetrics: () => this.lastRuntimeMetricsSnapshot,
|
||||
});
|
||||
|
||||
connection = {
|
||||
@@ -1173,6 +1181,8 @@ export class VoiceAssistantWebSocketServer {
|
||||
providerUsageList: true,
|
||||
// COMPAT(agentDetach): added in v0.1.98, remove gate after 2026-12-19 once daemon floor >= v0.1.98.
|
||||
agentDetach: true,
|
||||
// COMPAT(daemonDiagnostics): added in v0.1.100, remove gate after 2026-12-25 once daemon floor >= v0.1.100.
|
||||
daemonDiagnostics: true,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -1719,36 +1729,38 @@ export class VoiceAssistantWebSocketServer {
|
||||
).length;
|
||||
const sessionMetrics = this.collectSessionRuntimeMetrics();
|
||||
const agentSnapshot = this.agentManager.getMetricsSnapshot();
|
||||
|
||||
this.logger.info(
|
||||
{
|
||||
windowMs: runtimeMetrics.windowMs,
|
||||
final: Boolean(options?.final),
|
||||
sessions: {
|
||||
activeConnections,
|
||||
externalSessionKeys: this.externalSessionsByKey.size,
|
||||
reconnectGraceSessions,
|
||||
},
|
||||
sockets: {
|
||||
activeSockets,
|
||||
pendingConnections,
|
||||
},
|
||||
counters: runtimeMetrics.counters,
|
||||
inboundMessageTypesTop: runtimeMetrics.inboundMessageTypesTop,
|
||||
inboundSessionRequestTypesTop: runtimeMetrics.inboundSessionRequestTypesTop,
|
||||
outboundMessageTypesTop: runtimeMetrics.outboundMessageTypesTop,
|
||||
outboundSessionMessageTypesTop: runtimeMetrics.outboundSessionMessageTypesTop,
|
||||
outboundAgentStreamTypesTop: runtimeMetrics.outboundAgentStreamTypesTop,
|
||||
outboundAgentStreamAgentsTop: runtimeMetrics.outboundAgentStreamAgentsTop,
|
||||
outboundBinaryFrameTypesTop: runtimeMetrics.outboundBinaryFrameTypesTop,
|
||||
bufferedAmount: runtimeMetrics.bufferedAmount,
|
||||
eventLoopDelay: this.snapshotEventLoopDelay(),
|
||||
runtime: sessionMetrics,
|
||||
latency: runtimeMetrics.latency,
|
||||
agents: agentSnapshot,
|
||||
const loggedMetrics = {
|
||||
windowMs: runtimeMetrics.windowMs,
|
||||
final: Boolean(options?.final),
|
||||
sessions: {
|
||||
activeConnections,
|
||||
externalSessionKeys: this.externalSessionsByKey.size,
|
||||
reconnectGraceSessions,
|
||||
},
|
||||
"ws_runtime_metrics",
|
||||
);
|
||||
sockets: {
|
||||
activeSockets,
|
||||
pendingConnections,
|
||||
},
|
||||
counters: runtimeMetrics.counters,
|
||||
inboundMessageTypesTop: runtimeMetrics.inboundMessageTypesTop,
|
||||
inboundSessionRequestTypesTop: runtimeMetrics.inboundSessionRequestTypesTop,
|
||||
outboundMessageTypesTop: runtimeMetrics.outboundMessageTypesTop,
|
||||
outboundSessionMessageTypesTop: runtimeMetrics.outboundSessionMessageTypesTop,
|
||||
outboundAgentStreamTypesTop: runtimeMetrics.outboundAgentStreamTypesTop,
|
||||
outboundAgentStreamAgentsTop: runtimeMetrics.outboundAgentStreamAgentsTop,
|
||||
outboundBinaryFrameTypesTop: runtimeMetrics.outboundBinaryFrameTypesTop,
|
||||
bufferedAmount: runtimeMetrics.bufferedAmount,
|
||||
eventLoopDelay: this.snapshotEventLoopDelay(),
|
||||
runtime: sessionMetrics,
|
||||
latency: runtimeMetrics.latency,
|
||||
agents: agentSnapshot,
|
||||
} satisfies WebSocketRuntimeMetricsLogPayload;
|
||||
|
||||
this.lastRuntimeMetricsSnapshot = {
|
||||
collectedAt: new Date().toISOString(),
|
||||
...loggedMetrics,
|
||||
};
|
||||
this.logger.info(loggedMetrics, "ws_runtime_metrics");
|
||||
}
|
||||
|
||||
private getClientActivityState(session: Session): ClientPresenceState {
|
||||
|
||||
@@ -42,6 +42,30 @@ export interface WebSocketRuntimeMetricsSnapshot {
|
||||
}>;
|
||||
}
|
||||
|
||||
export interface WebSocketRuntimeDiagnosticSnapshot<
|
||||
TRuntime = unknown,
|
||||
TAgents = unknown,
|
||||
> extends WebSocketRuntimeMetricsSnapshot {
|
||||
collectedAt: string;
|
||||
final: boolean;
|
||||
sessions: {
|
||||
activeConnections: number;
|
||||
externalSessionKeys: number;
|
||||
reconnectGraceSessions: number;
|
||||
};
|
||||
sockets: {
|
||||
activeSockets: number;
|
||||
pendingConnections: number;
|
||||
};
|
||||
eventLoopDelay: {
|
||||
p50Ms: number;
|
||||
p99Ms: number;
|
||||
maxMs: number;
|
||||
} | null;
|
||||
runtime: TRuntime;
|
||||
agents: TAgents;
|
||||
}
|
||||
|
||||
type Clock = () => number;
|
||||
|
||||
export class WebSocketRuntimeMetricsWindow {
|
||||
|
||||
Reference in New Issue
Block a user