diff --git a/packages/server/src/client/daemon-client-runtime-metrics.ts b/packages/server/src/client/daemon-client-runtime-metrics.ts new file mode 100644 index 000000000..23053b31d --- /dev/null +++ b/packages/server/src/client/daemon-client-runtime-metrics.ts @@ -0,0 +1,258 @@ +import type { SessionOutboundMessage } from "../shared/messages.js"; + +type RuntimeMetricsLogger = { + info(obj: object, msg?: string): void; +}; + +type RuntimeMetricsHandlerTiming = { + count: number; + totalMs: number; + maxMs: number; +}; + +type RuntimeMetricsBucket = { + inboundMessageCounts: Map; + inboundMessageBytes: Map; + inboundMessageHandlerMs: Map; + inboundAgentStreamCounts: Map; + inboundAgentStreamByAgentCounts: Map; + inboundBinaryFrameCounts: Map; + endedAt: number; +}; + +type RuntimeMetricsContext = { + connectionPath: "direct" | "relay"; + serverId: string | null; + getConnectionStatus: () => string; +}; + +type RuntimeMetricsOptions = { + windowMs?: number; +}; + +const DEFAULT_ROLLING_WINDOW_MS = 60_000; + +export class DaemonClientRuntimeMetrics { + private readonly startedAt = Date.now(); + private readonly windowMs: number; + private readonly buckets: RuntimeMetricsBucket[] = []; + private readonly inboundMessageCounts = new Map(); + private readonly inboundMessageBytes = new Map(); + private readonly inboundMessageHandlerMs = new Map(); + private readonly inboundAgentStreamCounts = new Map(); + private readonly inboundAgentStreamByAgentCounts = new Map(); + private readonly inboundBinaryFrameCounts = new Map(); + + constructor( + private readonly logger: RuntimeMetricsLogger, + private readonly context: RuntimeMetricsContext, + options?: RuntimeMetricsOptions, + ) { + this.windowMs = + typeof options?.windowMs === "number" && options.windowMs > 0 + ? options.windowMs + : DEFAULT_ROLLING_WINDOW_MS; + } + + recordMessage(type: string, bytes: number, handlerMs: number): void { + incrementCount(this.inboundMessageCounts, type, 1); + incrementCount(this.inboundMessageBytes, type, bytes); + incrementHandlerTiming(this.inboundMessageHandlerMs, type, handlerMs); + } + + recordAgentStream( + payload: Extract["payload"], + ): void { + const { agentId, event } = payload; + const eventType = event.type === "timeline" ? `timeline:${event.item.type}` : event.type; + incrementCount(this.inboundAgentStreamCounts, eventType, 1); + incrementCount(this.inboundAgentStreamByAgentCounts, agentId, 1); + } + + recordBinaryFrame(kind: string, bytes: number, handlerMs: number): void { + incrementCount(this.inboundBinaryFrameCounts, kind, 1); + incrementCount(this.inboundMessageBytes, `binary:${kind}`, bytes); + incrementHandlerTiming(this.inboundMessageHandlerMs, `binary:${kind}`, handlerMs); + } + + flush(options?: { final?: boolean }): void { + const now = Date.now(); + const bucket = this.consumeCurrentBucket(now); + if (bucket) { + this.buckets.push(bucket); + } + this.pruneBuckets(now); + + const aggregate = this.aggregateBuckets(); + const hasActivity = + aggregate.inboundMessageCounts.size > 0 || aggregate.inboundBinaryFrameCounts.size > 0; + if (!hasActivity && !options?.final) { + return; + } + + this.logger.info( + { + windowMs: Math.min(this.windowMs, Math.max(0, now - this.startedAt)), + rollingWindowMs: this.windowMs, + bucketCount: this.buckets.length, + final: Boolean(options?.final), + connectionPath: this.context.connectionPath, + serverId: this.context.serverId, + connectionStatus: this.context.getConnectionStatus(), + inboundMessageTypesTop: getTopCounts(aggregate.inboundMessageCounts, 20), + inboundMessageBytesTop: getTopCounts(aggregate.inboundMessageBytes, 20), + inboundAgentStreamTypesTop: getTopCounts(aggregate.inboundAgentStreamCounts, 20), + inboundAgentStreamAgentsTop: getTopCounts(aggregate.inboundAgentStreamByAgentCounts, 20), + inboundBinaryFrameTypesTop: getTopCounts(aggregate.inboundBinaryFrameCounts, 12), + handlerTimingTop: getTopHandlerTimings(aggregate.inboundMessageHandlerMs, 20), + }, + "ws_runtime_metrics_client", + ); + } + + private consumeCurrentBucket(now: number): RuntimeMetricsBucket | null { + const hasActivity = + this.inboundMessageCounts.size > 0 || this.inboundBinaryFrameCounts.size > 0; + if (!hasActivity) { + return null; + } + + const bucket = { + inboundMessageCounts: new Map(this.inboundMessageCounts), + inboundMessageBytes: new Map(this.inboundMessageBytes), + inboundMessageHandlerMs: cloneHandlerTimingMap(this.inboundMessageHandlerMs), + inboundAgentStreamCounts: new Map(this.inboundAgentStreamCounts), + inboundAgentStreamByAgentCounts: new Map(this.inboundAgentStreamByAgentCounts), + inboundBinaryFrameCounts: new Map(this.inboundBinaryFrameCounts), + endedAt: now, + }; + + this.inboundMessageCounts.clear(); + this.inboundMessageBytes.clear(); + this.inboundMessageHandlerMs.clear(); + this.inboundAgentStreamCounts.clear(); + this.inboundAgentStreamByAgentCounts.clear(); + this.inboundBinaryFrameCounts.clear(); + return bucket; + } + + private pruneBuckets(now: number): void { + const cutoff = now - this.windowMs; + while (this.buckets.length > 0 && this.buckets[0]!.endedAt < cutoff) { + this.buckets.shift(); + } + } + + private aggregateBuckets(): RuntimeMetricsBucket { + const aggregate = createEmptyBucket(Date.now()); + for (const bucket of this.buckets) { + mergeCountMap(aggregate.inboundMessageCounts, bucket.inboundMessageCounts); + mergeCountMap(aggregate.inboundMessageBytes, bucket.inboundMessageBytes); + mergeHandlerTimingMap(aggregate.inboundMessageHandlerMs, bucket.inboundMessageHandlerMs); + mergeCountMap(aggregate.inboundAgentStreamCounts, bucket.inboundAgentStreamCounts); + mergeCountMap( + aggregate.inboundAgentStreamByAgentCounts, + bucket.inboundAgentStreamByAgentCounts, + ); + mergeCountMap(aggregate.inboundBinaryFrameCounts, bucket.inboundBinaryFrameCounts); + } + return aggregate; + } +} + +function createEmptyBucket(endedAt: number): RuntimeMetricsBucket { + return { + inboundMessageCounts: new Map(), + inboundMessageBytes: new Map(), + inboundMessageHandlerMs: new Map(), + inboundAgentStreamCounts: new Map(), + inboundAgentStreamByAgentCounts: new Map(), + inboundBinaryFrameCounts: new Map(), + endedAt, + }; +} + +function incrementCount(map: Map, key: string, amount: number): void { + map.set(key, (map.get(key) ?? 0) + amount); +} + +function incrementHandlerTiming( + map: Map, + key: string, + handlerMs: number, +): void { + const existing = map.get(key); + if (existing) { + existing.count += 1; + existing.totalMs += handlerMs; + existing.maxMs = Math.max(existing.maxMs, handlerMs); + return; + } + map.set(key, { + count: 1, + totalMs: handlerMs, + maxMs: handlerMs, + }); +} + +function cloneHandlerTimingMap( + map: Map, +): Map { + return new Map( + [...map.entries()].map(([key, value]) => [ + key, + { count: value.count, totalMs: value.totalMs, maxMs: value.maxMs }, + ]), + ); +} + +function mergeCountMap(target: Map, source: Map): void { + for (const [key, value] of source) { + incrementCount(target, key, value); + } +} + +function mergeHandlerTimingMap( + target: Map, + source: Map, +): void { + for (const [key, value] of source) { + const existing = target.get(key); + if (existing) { + existing.count += value.count; + existing.totalMs += value.totalMs; + existing.maxMs = Math.max(existing.maxMs, value.maxMs); + continue; + } + target.set(key, { + count: value.count, + totalMs: value.totalMs, + maxMs: value.maxMs, + }); + } +} + +function getTopCounts(map: Map, limit: number): Array<[string, number]> { + return [...map.entries()].sort((a, b) => b[1] - a[1]).slice(0, limit); +} + +function getTopHandlerTimings( + map: Map, + limit: number, +): Array<{ + type: string; + count: number; + totalMs: number; + avgMs: number; + maxMs: number; +}> { + const rows = [...map.entries()].map(([type, value]) => ({ + type, + count: value.count, + totalMs: Math.round(value.totalMs), + avgMs: Math.round((value.totalMs / value.count) * 100) / 100, + maxMs: Math.round(value.maxMs * 100) / 100, + })); + rows.sort((a, b) => b.totalMs - a.totalMs); + return rows.slice(0, limit); +} diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index f4196be99..feccdaa7c 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -99,6 +99,7 @@ import { type DaemonTransportFactory, type WebSocketFactory, } from "./daemon-client-transport.js"; +import { DaemonClientRuntimeMetrics } from "./daemon-client-runtime-metrics.js"; export interface Logger { debug(obj: object, msg?: string): void; @@ -109,11 +110,16 @@ export interface Logger { const consoleLogger: Logger = { debug: () => {}, - info: (obj, msg) => console.info(msg, obj), + info: (obj, msg) => console.log(msg, obj), warn: (obj, msg) => console.warn(msg, obj), error: (obj, msg) => console.error(msg, obj), }; +const perfNow: () => number = + typeof performance !== "undefined" && typeof performance.now === "function" + ? () => performance.now() + : () => Date.now(); + export type { DaemonTransport, DaemonTransportFactory, @@ -198,6 +204,8 @@ export type DaemonClientConfig = { baseDelayMs?: number; maxDelayMs?: number; }; + runtimeMetricsIntervalMs?: number; + runtimeMetricsWindowMs?: number; }; export type SendMessageOptions = { @@ -650,6 +658,8 @@ export class DaemonClient { private readonly logClientIdHash: string; private readonly logGeneration: number | null; private lastServerInfoMessage: ServerInfoStatusPayload | null = null; + private runtimeMetricsInterval: ReturnType | null = null; + private runtimeMetrics: DaemonClientRuntimeMetrics | null = null; constructor(private config: DaemonClientConfig) { this.logger = config.logger ?? consoleLogger; @@ -673,6 +683,28 @@ export class DaemonClient { Number.isFinite(this.config.runtimeGeneration) ? this.config.runtimeGeneration : null; + const runtimeMetricsIntervalMs = + typeof config.runtimeMetricsIntervalMs === "number" && config.runtimeMetricsIntervalMs > 0 + ? config.runtimeMetricsIntervalMs + : 0; + if (runtimeMetricsIntervalMs > 0) { + const runtimeMetricsWindowMs = + typeof config.runtimeMetricsWindowMs === "number" && config.runtimeMetricsWindowMs > 0 + ? Math.max(config.runtimeMetricsWindowMs, runtimeMetricsIntervalMs) + : undefined; + this.runtimeMetrics = new DaemonClientRuntimeMetrics( + this.logger, + { + connectionPath: this.logConnectionPath, + serverId: this.logServerId, + getConnectionStatus: () => this.connectionState.status, + }, + runtimeMetricsWindowMs ? { windowMs: runtimeMetricsWindowMs } : undefined, + ); + this.runtimeMetricsInterval = setInterval(() => { + this.runtimeMetrics?.flush(); + }, runtimeMetricsIntervalMs); + } } // ============================================================================ @@ -883,6 +915,12 @@ export class DaemonClient { this.rejectPendingSendQueue(new Error("Daemon client closed")); this.clearTerminalSlots(); this.lastServerInfoMessage = null; + if (this.runtimeMetricsInterval) { + clearInterval(this.runtimeMetricsInterval); + this.runtimeMetricsInterval = null; + this.runtimeMetrics?.flush({ final: true }); + this.runtimeMetrics = null; + } this.updateConnectionState( { status: "disposed" }, { event: "DISPOSE", reason: "Client closed", reasonCode: "disposed" }, @@ -3703,7 +3741,17 @@ export class DaemonClient { if (rawBytes) { const frame = decodeTerminalStreamFrame(rawBytes); if (frame) { + const binaryStartMs = perfNow(); this.handleBinaryFrame(frame); + this.runtimeMetrics?.recordBinaryFrame( + frame.opcode === TerminalStreamOpcode.Output + ? "output" + : frame.opcode === TerminalStreamOpcode.Snapshot + ? "snapshot" + : "other", + rawBytes.byteLength, + perfNow() - binaryStartMs, + ); return; } } @@ -3712,6 +3760,8 @@ export class DaemonClient { return; } + const bytes = rawBytes?.byteLength ?? payload.length; + const startMs = perfNow(); let parsedJson: unknown; try { parsedJson = JSON.parse(payload); @@ -3727,10 +3777,16 @@ export class DaemonClient { } if (parsed.data.type === "pong") { + this.runtimeMetrics?.recordMessage("pong", bytes, perfNow() - startMs); return; } this.handleSessionMessage(parsed.data.message); + const msgType = parsed.data.message.type; + this.runtimeMetrics?.recordMessage(msgType, bytes, perfNow() - startMs); + if (parsed.data.message.type === "agent_stream") { + this.runtimeMetrics?.recordAgentStream(parsed.data.message.payload); + } } private handleBinaryFrame(frame: TerminalStreamFrame): void {