diff --git a/internal/cloud/backend-performance.test.ts b/internal/cloud/backend-performance.test.ts index ac01b813..32fbd533 100644 --- a/internal/cloud/backend-performance.test.ts +++ b/internal/cloud/backend-performance.test.ts @@ -6,7 +6,7 @@ import { createServer as createHttpServer } from "node:http"; import { test } from "bun:test"; import { createBackendPerformanceStore, backendPerformanceRouteTemplate, withBackendPerformanceContext } from "./backend-performance.ts"; -import { submitAgentRunChatTurn } from "./code-agent-agentrun-adapter.ts"; +import { submitAgentRunChatTurn, syncAgentRunChatResult } from "./code-agent-agentrun-adapter.ts"; import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { createCloudApiServer } from "./server.ts"; import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts"; @@ -189,3 +189,103 @@ test("AgentRun adapter records upstream timing through backend performance conte await new Promise((resolve, reject) => agentRunServer.close((error) => error ? reject(error) : resolve())); } }); + +test("backend performance store records Workbench projection and AgentRun result metrics without identifiers", () => { + const store = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } }); + + store.recordWorkbenchProjection({ + sourceLatestSeq: 2626, + lastProjectedSeq: 137, + sourceLatestEventAt: "2026-06-19T02:00:30.000Z", + projectedLatestEventAt: "2026-06-19T02:00:00.000Z", + status: "running", + reason: "projecting", + projectionStatus: "projecting", + source: "agentrun-v01", + }); + store.recordWorkbenchTurnRead({ route: "/v1/workbench/turns/trc_secret_0123456789abcdef", status: "running", degradedReason: "projecting", durationMs: 3364, responseBytes: 4096 }); + store.recordAgentRunResult({ durationMs: 9800, eventCount: 2626, eventsScanned: 2626, pagesScanned: 6, status: "timeout", budget: 2500, timedOut: true }); + store.recordWorkbenchProjectorBatch({ phase: "agentrun_events_fetch", status: "ok", durationMs: 42 }); + store.recordWorkbenchProjectorCandidate({ status: "seen", reason: "read_side_refresh" }); + store.recordWorkbenchProjectorEvents({ eventKind: "assistant_message", status: "ok", count: 2 }); + store.recordWorkbenchProjectorError({ reason: "agentrun_timeout" }); + + const text = store.metricsText(); + assert.match(text, /hwlab_workbench_projection_lag_events_bucket\{[^}]*projection_status="projecting"[^}]*\}/u); + assert.match(text, /hwlab_workbench_projection_stuck_traces\{[^}]*reason="projecting"[^}]*\} 1/u); + assert.match(text, /hwlab_workbench_turn_get_duration_seconds_count\{[^}]*route="\/v1\/workbench\/turns\/:traceId"[^}]*degraded_reason="projecting"[^}]*\} 1/u); + assert.match(text, /hwlab_agentrun_result_duration_seconds_count\{[^}]*event_count_bucket="2001_5000"[^}]*status="timeout"[^}]*\} 1/u); + assert.match(text, /hwlab_agentrun_result_timeout_total\{[^}]*budget="turn_2_5s"[^}]*status="timeout"[^}]*\} 1/u); + assert.match(text, /hwlab_workbench_projector_batch_duration_seconds_count\{[^}]*phase="agentrun_events_fetch"[^}]*status="ok"[^}]*\} 1/u); + assert.match(text, /hwlab_workbench_projector_last_success_unixtime\{[^}]*node="D601"[^}]*\} /u); + assert.doesNotMatch(text, /trc_secret|0123456789abcdef|run_secret|cmd_secret|ses_secret|cnv_secret/u); +}); + +test("AgentRun read-side sync advances event cursor without forcing result aggregation before terminal evidence", async () => { + const calls: string[] = []; + const agentRunServer = createHttpServer(async (request, response) => { + const url = new URL(request.url || "/", "http://127.0.0.1"); + calls.push(`${request.method} ${url.pathname}`); + const send = (data: Record) => { + response.writeHead(200, { "content-type": "application/json" }); + response.end(`${JSON.stringify({ ok: true, data })}\n`); + }; + if (request.method === "GET" && url.pathname === "/api/v1/runs/run_read_side/events") { + return send({ + items: [ + { seq: 11, type: "assistant_message", runId: "run_read_side", payload: { commandId: "cmd_read_side", text: "working" }, createdAt: "2026-06-19T02:00:01.000Z" }, + { seq: 12, type: "tool_call", runId: "run_read_side", payload: { commandId: "cmd_read_side", toolName: "shell" }, createdAt: "2026-06-19T02:00:02.000Z" }, + ], + }); + } + if (request.method === "GET" && url.pathname.endsWith("/result")) return send({ terminalStatus: "completed", text: "should not be fetched" }); + response.writeHead(404, { "content-type": "application/json" }); + response.end(`${JSON.stringify({ ok: false, message: `unexpected ${request.method} ${url.pathname}` })}\n`); + }); + await new Promise((resolve) => agentRunServer.listen(0, "127.0.0.1", resolve)); + try { + const address = agentRunServer.address(); + assert.ok(address && typeof address === "object"); + const store = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } }); + const traceStore = createCodeAgentTraceStore(); + const synced = await syncAgentRunChatResult({ + traceId: "trc_read_side_cursor", + currentResult: { + status: "running", + traceId: "trc_read_side_cursor", + agentRun: { + adapter: "agentrun-v01", + runId: "run_read_side", + commandId: "cmd_read_side", + traceId: "trc_read_side_cursor", + providerTrace: { traceId: "trc_read_side_cursor" }, + lastSeq: 10, + managerUrl: `http://127.0.0.1:${address.port}`, + backendProfile: "codex", + }, + }, + options: { + backendPerformanceStore: store, + env: { + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601", + HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS: "2500", + }, + }, + traceStore, + forceResultSync: false, + }); + + assert.equal(synced.resultSynced, false); + assert.equal(synced.eventsRefreshed, true); + assert.equal(synced.result.agentRun.lastSeq, 12); + assert.ok(calls.some((call) => call === "GET /api/v1/runs/run_read_side/events")); + assert.ok(!calls.some((call) => call.endsWith("/result"))); + const text = store.metricsText(); + assert.match(text, /hwlab_workbench_projection_cursor_advance_total\{[^}]*status="advanced"[^}]*\} 2/u); + assert.match(text, /hwlab_workbench_projector_events_processed_total\{[^}]*event_kind="assistant_message"[^}]*\} 1/u); + assert.match(text, /hwlab_workbench_projector_batch_duration_seconds_count\{[^}]*phase="agentrun_events_fetch"[^}]*status="ok"[^}]*\} 1/u); + } finally { + await new Promise((resolve, reject) => agentRunServer.close((error) => error ? reject(error) : resolve())); + } +}); diff --git a/internal/cloud/backend-performance.ts b/internal/cloud/backend-performance.ts index 1140277a..ebb347ad 100644 --- a/internal/cloud/backend-performance.ts +++ b/internal/cloud/backend-performance.ts @@ -6,6 +6,14 @@ import { AsyncLocalStorage } from "node:async_hooks"; const REQUEST_BUCKETS_SECONDS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]; const PHASE_BUCKETS_SECONDS = [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30]; const UPSTREAM_BUCKETS_SECONDS = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120]; +const PROJECTION_LAG_EVENT_BUCKETS = [0, 1, 5, 10, 50, 100, 500, 1000, 2000, 5000, 10000]; +const PROJECTION_LAG_SECONDS_BUCKETS = [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 900, 1800, 3600]; +const WORKBENCH_TURN_DURATION_BUCKETS_SECONDS = [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]; +const WORKBENCH_TURN_RESPONSE_BYTES_BUCKETS = [512, 1024, 4096, 16384, 65536, 262144, 1048576, 4194304]; +const AGENTRUN_RESULT_BUCKETS_SECONDS = [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120]; +const AGENTRUN_RESULT_EVENTS_BUCKETS = [0, 50, 100, 500, 1000, 2000, 5000, 10000]; +const AGENTRUN_RESULT_PAGES_BUCKETS = [0, 1, 2, 5, 10, 20, 50, 100]; +const PROJECTOR_BATCH_BUCKETS_SECONDS = [0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]; const MAX_SERVER_TIMING_ENTRIES = 8; const backendPerformanceStorage = new AsyncLocalStorage(); @@ -22,6 +30,11 @@ interface CounterSeries { value: number; } +interface GaugeSeries { + labels: Record; + value: number; +} + interface BackendPerformanceStoreOptions { env?: Record; maxSeries?: number; @@ -48,6 +61,51 @@ interface BackendUpstreamInput { durationMs: number; } +interface WorkbenchProjectionInput { + sourceLatestSeq?: unknown; + lastProjectedSeq?: unknown; + sourceLatestEventAt?: unknown; + projectedLatestEventAt?: unknown; + terminalSourceAt?: unknown; + terminalProjectedAt?: unknown; + status?: unknown; + reason?: unknown; + projectionStatus?: unknown; + source?: unknown; +} + +interface WorkbenchTurnReadInput { + route: string; + status?: unknown; + degradedReason?: unknown; + durationMs: number; + responseBytes?: unknown; + timedOut?: boolean; +} + +interface AgentRunResultInput { + durationMs?: unknown; + eventsScanned?: unknown; + pagesScanned?: unknown; + eventCount?: unknown; + status?: unknown; + budget?: unknown; + timedOut?: boolean; +} + +interface WorkbenchProjectorBatchInput { + phase: string; + status?: unknown; + durationMs: number; +} + +interface WorkbenchProjectorCounterInput { + status?: unknown; + reason?: unknown; + eventKind?: unknown; + count?: unknown; +} + export function createBackendPerformanceStore(options: BackendPerformanceStoreOptions = {}) { const env = options.env ?? process.env; const maxSeries = parsePositiveInteger(env.HWLAB_BACKEND_PERFORMANCE_MAX_SERIES, options.maxSeries ?? 320); @@ -56,11 +114,33 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp const requestPhaseSeries = new Map(); const upstreamSeries = new Map(); const droppedSeries = new Map(); + const projectionLagEventsSeries = new Map(); + const projectionLagSecondsSeries = new Map(); + const projectionStuckGaugeSeries = new Map(); + const projectionCursorAdvanceSeries = new Map(); + const terminalProjectionDelaySeries = new Map(); + const turnGetDurationSeries = new Map(); + const turnGetTimeoutSeries = new Map(); + const turnGetResponseBytesSeries = new Map(); + const agentRunResultDurationSeries = new Map(); + const agentRunResultPagesSeries = new Map(); + const agentRunResultEventsSeries = new Map(); + const agentRunResultTimeoutSeries = new Map(); + const projectorBatchDurationSeries = new Map(); + const projectorCandidatesSeries = new Map(); + const projectorEventsProcessedSeries = new Map(); + const projectorErrorsSeries = new Map(); + const projectorLastSuccessSeries = new Map(); const baseLabels = { service: "hwlab-cloud-api", namespace: sanitizeLabelValue(env.POD_NAMESPACE ?? env.HWLAB_METRICS_NAMESPACE ?? "hwlab-v02"), gitops_target: sanitizeLabelValue(env.HWLAB_GITOPS_TARGET ?? env.HWLAB_GITOPS_PROFILE ?? env.HWLAB_RUNTIME_LANE ?? "v02") }; + const workbenchBaseLabels = { + ...baseLabels, + node: sanitizeLabelValue(env.HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID ?? env.AGENTRUN_PROVIDER_ID ?? env.HWLAB_NODE_ID ?? "unknown"), + lane: sanitizeLabelValue(env.HWLAB_RUNTIME_LANE ?? env.HWLAB_GITOPS_TARGET ?? env.HWLAB_GITOPS_PROFILE ?? "unknown") + }; function beginHttpRequest(request: any, response: any) { const url = safeUrl(request?.url || "/"); @@ -148,6 +228,130 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp return { route, method, startedAt, recordPhase, measure, recordUpstream, setStatus, applyServerTimingHeader, finish, get statusCode() { return statusCode; } }; } + function recordWorkbenchProjection(input: WorkbenchProjectionInput = {}) { + const labels = { + ...workbenchBaseLabels, + status: normalizeStatusLabel(input.status ?? "unknown"), + reason: normalizeReason(input.reason ?? "none"), + projection_status: normalizeProjectionStatus(input.projectionStatus ?? "unknown"), + source: normalizeSourceLabel(input.source ?? "agentrun_v01") + }; + const sourceSeq = finiteNonNegativeNumber(input.sourceLatestSeq); + const projectedSeq = finiteNonNegativeNumber(input.lastProjectedSeq); + if (sourceSeq !== null && projectedSeq !== null) { + const lagEvents = Math.max(0, Math.floor(sourceSeq - projectedSeq)); + if (!recordHistogram(projectionLagEventsSeries, labels, lagEvents, PROJECTION_LAG_EVENT_BUCKETS, maxSeries)) incrementDropped("projection_lag_events_series_limit"); + setGauge(projectionStuckGaugeSeries, labels, lagEvents > 0 && labels.projection_status !== "caught_up" ? 1 : 0, maxSeries) || incrementDropped("projection_stuck_series_limit"); + } + const sourceAt = unixMs(input.sourceLatestEventAt); + const projectedAt = unixMs(input.projectedLatestEventAt); + if (sourceAt !== null && projectedAt !== null) { + const lagSeconds = Math.max(0, (sourceAt - projectedAt) / 1000); + if (!recordHistogram(projectionLagSecondsSeries, labels, lagSeconds, PROJECTION_LAG_SECONDS_BUCKETS, maxSeries)) incrementDropped("projection_lag_seconds_series_limit"); + } + const terminalSourceAt = unixMs(input.terminalSourceAt); + const terminalProjectedAt = unixMs(input.terminalProjectedAt) ?? (isTerminalProjectionStatus(labels.projection_status) ? projectedAt : null); + if (terminalSourceAt !== null) { + const delaySeconds = Math.max(0, ((terminalProjectedAt ?? Date.now()) - terminalSourceAt) / 1000); + if (!recordHistogram(terminalProjectionDelaySeries, labels, delaySeconds, PROJECTION_LAG_SECONDS_BUCKETS, maxSeries)) incrementDropped("terminal_projection_delay_series_limit"); + } + } + + function recordWorkbenchProjectionCursorAdvance(input: WorkbenchProjectorCounterInput = {}) { + const labels = { + ...workbenchBaseLabels, + status: normalizeStatusLabel(input.status ?? "advanced"), + reason: normalizeReason(input.reason ?? "sync"), + projection_status: normalizeProjectionStatus((input as any).projectionStatus ?? "projecting"), + source: normalizeSourceLabel((input as any).source ?? "agentrun_v01") + }; + incrementCounterBy(projectionCursorAdvanceSeries, labels, boundedCounterValue(input.count), maxSeries) || incrementDropped("projection_cursor_advance_series_limit"); + } + + function recordWorkbenchTurnRead(input: WorkbenchTurnReadInput) { + const labels = { + ...workbenchBaseLabels, + route: workbenchTurnRouteTemplate(input.route), + status: normalizeStatusLabel(input.status ?? "unknown"), + degraded_reason: normalizeReason(input.degradedReason ?? "none") + }; + if (!recordHistogram(turnGetDurationSeries, labels, boundedDurationMs(input.durationMs, 120_000) / 1000, WORKBENCH_TURN_DURATION_BUCKETS_SECONDS, maxSeries)) incrementDropped("turn_get_duration_series_limit"); + const bytes = finiteNonNegativeNumber(input.responseBytes); + if (bytes !== null) { + const byteLabels = { ...workbenchBaseLabels, route: labels.route, status: labels.status }; + if (!recordHistogram(turnGetResponseBytesSeries, byteLabels, bytes, WORKBENCH_TURN_RESPONSE_BYTES_BUCKETS, maxSeries)) incrementDropped("turn_get_response_bytes_series_limit"); + } + if (input.timedOut === true || labels.status === "timeout") { + const timeoutLabels = { ...workbenchBaseLabels, route: labels.route, degraded_reason: labels.degraded_reason }; + incrementCounter(turnGetTimeoutSeries, timeoutLabels, maxSeries) || incrementDropped("turn_get_timeout_series_limit"); + } + } + + function recordAgentRunResult(input: AgentRunResultInput = {}) { + const eventCount = finiteNonNegativeNumber(input.eventCount ?? input.eventsScanned) ?? 0; + const labels = { + ...workbenchBaseLabels, + event_count_bucket: agentRunEventCountBucket(eventCount), + status: normalizeStatusLabel(input.status ?? (input.timedOut ? "timeout" : "unknown")) + }; + const durationMs = finiteNonNegativeNumber(input.durationMs); + if (durationMs !== null) { + if (!recordHistogram(agentRunResultDurationSeries, labels, boundedDurationMs(durationMs, 120_000) / 1000, AGENTRUN_RESULT_BUCKETS_SECONDS, maxSeries)) incrementDropped("agentrun_result_duration_series_limit"); + } + const pagesScanned = finiteNonNegativeNumber(input.pagesScanned); + if (pagesScanned !== null) { + if (!recordHistogram(agentRunResultPagesSeries, workbenchBaseLabels, pagesScanned, AGENTRUN_RESULT_PAGES_BUCKETS, maxSeries)) incrementDropped("agentrun_result_pages_series_limit"); + } + const eventsScanned = finiteNonNegativeNumber(input.eventsScanned); + if (eventsScanned !== null) { + if (!recordHistogram(agentRunResultEventsSeries, workbenchBaseLabels, eventsScanned, AGENTRUN_RESULT_EVENTS_BUCKETS, maxSeries)) incrementDropped("agentrun_result_events_series_limit"); + } + if (input.timedOut === true || labels.status === "timeout") { + const timeoutLabels = { + ...workbenchBaseLabels, + budget: normalizeBudgetLabel(input.budget), + status: labels.status + }; + incrementCounter(agentRunResultTimeoutSeries, timeoutLabels, maxSeries) || incrementDropped("agentrun_result_timeout_series_limit"); + } + } + + function recordWorkbenchProjectorBatch(input: WorkbenchProjectorBatchInput) { + const labels = { + ...workbenchBaseLabels, + phase: normalizeProjectorPhase(input.phase), + status: normalizeStatusLabel(input.status ?? "ok") + }; + if (!recordHistogram(projectorBatchDurationSeries, labels, boundedDurationMs(input.durationMs, 120_000) / 1000, PROJECTOR_BATCH_BUCKETS_SECONDS, maxSeries)) incrementDropped("projector_batch_series_limit"); + if (labels.status === "ok") setGauge(projectorLastSuccessSeries, workbenchBaseLabels, Math.floor(Date.now() / 1000), maxSeries) || incrementDropped("projector_last_success_series_limit"); + } + + function recordWorkbenchProjectorCandidate(input: WorkbenchProjectorCounterInput = {}) { + const labels = { + ...workbenchBaseLabels, + status: normalizeStatusLabel(input.status ?? "seen"), + reason: normalizeReason(input.reason ?? "unknown") + }; + incrementCounterBy(projectorCandidatesSeries, labels, boundedCounterValue(input.count), maxSeries) || incrementDropped("projector_candidates_series_limit"); + } + + function recordWorkbenchProjectorEvents(input: WorkbenchProjectorCounterInput = {}) { + const labels = { + ...workbenchBaseLabels, + event_kind: normalizeReason(input.eventKind ?? "unknown"), + status: normalizeStatusLabel(input.status ?? "ok") + }; + incrementCounterBy(projectorEventsProcessedSeries, labels, boundedCounterValue(input.count), maxSeries) || incrementDropped("projector_events_series_limit"); + } + + function recordWorkbenchProjectorError(input: WorkbenchProjectorCounterInput = {}) { + const labels = { + ...workbenchBaseLabels, + reason: normalizeReason(input.reason ?? "unknown") + }; + incrementCounterBy(projectorErrorsSeries, labels, boundedCounterValue(input.count), maxSeries) || incrementDropped("projector_errors_series_limit"); + } + function incrementDropped(reason: string) { incrementCounter(droppedSeries, { ...baseLabels, reason }, Math.max(maxSeries, 16)); } @@ -169,6 +373,57 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp "# HELP hwlab_cloud_api_performance_dropped_total Backend performance samples dropped before Prometheus export.", "# TYPE hwlab_cloud_api_performance_dropped_total counter", ...renderCounters("hwlab_cloud_api_performance_dropped_total", droppedSeries), + "# HELP hwlab_workbench_projection_lag_events Workbench projection source seq lag in AgentRun events.", + "# TYPE hwlab_workbench_projection_lag_events histogram", + ...renderHistogram("hwlab_workbench_projection_lag_events", projectionLagEventsSeries, PROJECTION_LAG_EVENT_BUCKETS), + "# HELP hwlab_workbench_projection_lag_seconds Workbench projection source time lag in seconds.", + "# TYPE hwlab_workbench_projection_lag_seconds histogram", + ...renderHistogram("hwlab_workbench_projection_lag_seconds", projectionLagSecondsSeries, PROJECTION_LAG_SECONDS_BUCKETS), + "# HELP hwlab_workbench_projection_stuck_traces Low-cardinality gauge set to 1 when the observed projection is behind its source.", + "# TYPE hwlab_workbench_projection_stuck_traces gauge", + ...renderGauges("hwlab_workbench_projection_stuck_traces", projectionStuckGaugeSeries), + "# HELP hwlab_workbench_projection_cursor_advance_total Workbench projection cursor advance events.", + "# TYPE hwlab_workbench_projection_cursor_advance_total counter", + ...renderCounters("hwlab_workbench_projection_cursor_advance_total", projectionCursorAdvanceSeries), + "# HELP hwlab_workbench_terminal_projection_delay_seconds Delay from AgentRun terminal source to Workbench terminal projection.", + "# TYPE hwlab_workbench_terminal_projection_delay_seconds histogram", + ...renderHistogram("hwlab_workbench_terminal_projection_delay_seconds", terminalProjectionDelaySeries, PROJECTION_LAG_SECONDS_BUCKETS), + "# HELP hwlab_workbench_turn_get_duration_seconds Workbench turn/read API duration in seconds.", + "# TYPE hwlab_workbench_turn_get_duration_seconds histogram", + ...renderHistogram("hwlab_workbench_turn_get_duration_seconds", turnGetDurationSeries, WORKBENCH_TURN_DURATION_BUCKETS_SECONDS), + "# HELP hwlab_workbench_turn_get_timeout_total Workbench turn/read API timeout count.", + "# TYPE hwlab_workbench_turn_get_timeout_total counter", + ...renderCounters("hwlab_workbench_turn_get_timeout_total", turnGetTimeoutSeries), + "# HELP hwlab_workbench_turn_get_response_bytes Workbench turn/read API response bytes.", + "# TYPE hwlab_workbench_turn_get_response_bytes histogram", + ...renderHistogram("hwlab_workbench_turn_get_response_bytes", turnGetResponseBytesSeries, WORKBENCH_TURN_RESPONSE_BYTES_BUCKETS), + "# HELP hwlab_agentrun_result_duration_seconds AgentRun command result aggregation duration in seconds.", + "# TYPE hwlab_agentrun_result_duration_seconds histogram", + ...renderHistogram("hwlab_agentrun_result_duration_seconds", agentRunResultDurationSeries, AGENTRUN_RESULT_BUCKETS_SECONDS), + "# HELP hwlab_agentrun_result_pages_scanned AgentRun result estimated pages scanned.", + "# TYPE hwlab_agentrun_result_pages_scanned histogram", + ...renderHistogram("hwlab_agentrun_result_pages_scanned", agentRunResultPagesSeries, AGENTRUN_RESULT_PAGES_BUCKETS), + "# HELP hwlab_agentrun_result_events_scanned AgentRun result estimated events scanned.", + "# TYPE hwlab_agentrun_result_events_scanned histogram", + ...renderHistogram("hwlab_agentrun_result_events_scanned", agentRunResultEventsSeries, AGENTRUN_RESULT_EVENTS_BUCKETS), + "# HELP hwlab_agentrun_result_timeout_total AgentRun result timeout count by refresh budget.", + "# TYPE hwlab_agentrun_result_timeout_total counter", + ...renderCounters("hwlab_agentrun_result_timeout_total", agentRunResultTimeoutSeries), + "# HELP hwlab_workbench_projector_batch_duration_seconds Workbench projector/finalizer batch phase duration in seconds.", + "# TYPE hwlab_workbench_projector_batch_duration_seconds histogram", + ...renderHistogram("hwlab_workbench_projector_batch_duration_seconds", projectorBatchDurationSeries, PROJECTOR_BATCH_BUCKETS_SECONDS), + "# HELP hwlab_workbench_projector_candidates_total Workbench projector/finalizer candidate count.", + "# TYPE hwlab_workbench_projector_candidates_total counter", + ...renderCounters("hwlab_workbench_projector_candidates_total", projectorCandidatesSeries), + "# HELP hwlab_workbench_projector_events_processed_total Workbench projector/finalizer processed event count.", + "# TYPE hwlab_workbench_projector_events_processed_total counter", + ...renderCounters("hwlab_workbench_projector_events_processed_total", projectorEventsProcessedSeries), + "# HELP hwlab_workbench_projector_errors_total Workbench projector/finalizer error count.", + "# TYPE hwlab_workbench_projector_errors_total counter", + ...renderCounters("hwlab_workbench_projector_errors_total", projectorErrorsSeries), + "# HELP hwlab_workbench_projector_last_success_unixtime Last successful Workbench projector/finalizer sync unix time.", + "# TYPE hwlab_workbench_projector_last_success_unixtime gauge", + ...renderGauges("hwlab_workbench_projector_last_success_unixtime", projectorLastSuccessSeries), "" ].join("\n"); } @@ -181,11 +436,28 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp requestPhaseCount: histogramSampleCount(requestPhaseSeries), upstreamSeries: upstreamSeries.size, upstreamCount: histogramSampleCount(upstreamSeries), + projectionLagSeries: projectionLagEventsSeries.size + projectionLagSecondsSeries.size, + turnReadSeries: turnGetDurationSeries.size, + agentRunResultSeries: agentRunResultDurationSeries.size, + projectorSeries: projectorBatchDurationSeries.size + projectorCandidatesSeries.size + projectorEventsProcessedSeries.size + projectorErrorsSeries.size, droppedSeries: droppedSeries.size }; } - return { beginHttpRequest, createManualContext, metricsText, snapshot }; + return { + beginHttpRequest, + createManualContext, + recordWorkbenchProjection, + recordWorkbenchProjectionCursorAdvance, + recordWorkbenchTurnRead, + recordAgentRunResult, + recordWorkbenchProjectorBatch, + recordWorkbenchProjectorCandidate, + recordWorkbenchProjectorEvents, + recordWorkbenchProjectorError, + metricsText, + snapshot + }; } export function withBackendPerformanceContext(context: unknown, callback: () => T): T { @@ -210,6 +482,13 @@ export function backendPerformanceRouteTemplate(input: unknown): string { return path.length > 140 ? `${path.slice(0, 137)}...` : path; } +function workbenchTurnRouteTemplate(input: unknown): string { + const route = backendPerformanceRouteTemplate(input); + if (route === "/v1/workbench/turns/:id") return "/v1/workbench/turns/:traceId"; + if (route === "/v1/agent/turns/:id") return "/v1/agent/turns/:traceId"; + return route; +} + function serverTimingHeader(totalMs: number, phases: Array<{ name: string; durationMs: number }>) { const entries = phases.slice(-MAX_SERVER_TIMING_ENTRIES).map((phase) => `${serverTimingToken(phase.name)};dur=${roundDurationMs(phase.durationMs)}`); entries.push(`total;dur=${roundDurationMs(totalMs)}`); @@ -244,10 +523,38 @@ function incrementCounter(series: Map, labels: Record, labels: Record, value: number, maxSeries: number) { + const key = stableLabelKey(labels); + let entry = series.get(key); + if (!entry) { + if (series.size >= maxSeries) return false; + entry = { labels, value: 0 }; + series.set(key, entry); + } + entry.value += value; + return true; +} + +function setGauge(series: Map, labels: Record, value: number, maxSeries: number) { + const key = stableLabelKey(labels); + let entry = series.get(key); + if (!entry) { + if (series.size >= maxSeries) return false; + entry = { labels, value: 0 }; + series.set(key, entry); + } + entry.value = Number.isFinite(value) ? value : 0; + return true; +} + function renderCounters(metricName: string, series: Map) { return [...series.values()].map((entry) => `${metricName}{${renderLabels(entry.labels)}} ${entry.value}`); } +function renderGauges(metricName: string, series: Map) { + return [...series.values()].map((entry) => `${metricName}{${renderLabels(entry.labels)}} ${formatPrometheusValue(entry.value)}`); +} + function renderHistogram(metricName: string, series: Map, buckets: number[]) { const lines: string[] = []; for (const entry of series.values()) { @@ -311,11 +618,61 @@ function normalizeStatusClass(value: unknown) { return /^(?:1xx|2xx|3xx|4xx|5xx|network|timeout|unknown)$/u.test(text) ? text : "unknown"; } +function normalizeStatusLabel(value: unknown) { + const text = sanitizeMetricName(value, "unknown"); + return /^(?:ok|error|timeout|denied|network|cancelled|canceled|unknown|running|pending|completed|failed|blocked|degraded|advanced|unchanged|scheduled|seen|skipped|caught_up|projecting|stale|missing)$/u.test(text) ? text : "unknown"; +} + function normalizeOutcome(value: unknown) { const text = sanitizeMetricName(value, "unknown"); return /^(?:ok|error|timeout|denied|network|cancelled|canceled|unknown)$/u.test(text) ? text : "unknown"; } +function normalizeProjectionStatus(value: unknown) { + const text = sanitizeMetricName(value, "unknown"); + if (text === "caught_up" || text === "caughtup") return "caught_up"; + return /^(?:projecting|caught_up|blocked|unknown|missing|stale|degraded|terminal|running)$/u.test(text) ? text : "unknown"; +} + +function isTerminalProjectionStatus(value: unknown) { + const status = normalizeProjectionStatus(value); + return status === "caught_up" || status === "terminal"; +} + +function normalizeReason(value: unknown) { + return sanitizeMetricName(value, "unknown").slice(0, 64) || "unknown"; +} + +function normalizeSourceLabel(value: unknown) { + const text = sanitizeMetricName(value, "unknown"); + if (text === "agentrun_v0_1" || text === "agentrun_v01") return "agentrun_v01"; + return text.slice(0, 64) || "unknown"; +} + +function normalizeProjectorPhase(value: unknown) { + const phase = sanitizeMetricName(value, "unknown"); + return /^(?:candidate_scan|agentrun_events_fetch|event_map|trace_append|turn_finalize|session_write|status_effects|unknown)$/u.test(phase) ? phase : "unknown"; +} + +function normalizeBudgetLabel(value: unknown) { + const parsed = Number(value); + if (Number.isFinite(parsed) && parsed > 0) { + if (parsed <= 2500) return "turn_2_5s"; + if (parsed <= 10000) return "short_10s"; + if (parsed <= 30000) return "medium_30s"; + if (parsed <= 120000) return "long_120s"; + return "long_over_120s"; + } + return normalizeReason(value ?? "unknown"); +} + +function agentRunEventCountBucket(value: number) { + if (value <= 500) return "0_500"; + if (value <= 2000) return "501_2000"; + if (value <= 5000) return "2001_5000"; + return "5000_plus"; +} + function sanitizeMetricName(value: unknown, fallback = "unknown") { const text = String(value ?? "").trim().toLowerCase().replace(/[^a-z0-9_:-]+/gu, "_").replace(/[-:]+/gu, "_").replace(/^_+|_+$/gu, ""); return text || fallback; @@ -342,6 +699,33 @@ function boundedDurationMs(value: unknown, maxMs: number) { return Math.min(parsed, maxMs); } +function boundedCounterValue(value: unknown) { + const parsed = Number(value ?? 1); + if (!Number.isFinite(parsed) || parsed <= 0) return 1; + return Math.min(Math.floor(parsed), 1_000_000); +} + +function finiteNonNegativeNumber(value: unknown): number | null { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : null; +} + +function unixMs(value: unknown): number | null { + if (value instanceof Date) { + const time = value.getTime(); + return Number.isFinite(time) ? time : null; + } + const numeric = Number(value); + if (Number.isFinite(numeric) && numeric > 0) return numeric > 10_000_000_000 ? numeric : numeric * 1000; + const parsed = Date.parse(String(value ?? "")); + return Number.isFinite(parsed) ? parsed : null; +} + +function formatPrometheusValue(value: number) { + if (!Number.isFinite(value)) return "0"; + return Number.isInteger(value) ? String(value) : value.toFixed(6); +} + function roundDurationMs(value: number) { return Math.max(0, value).toFixed(1); } diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 33eff649..f93c16ab 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -441,6 +441,7 @@ function isAgentRunCommandAlreadyClaimed(error) { export async function syncAgentRunChatResult({ traceId, currentResult = null, options = {}, traceStore = defaultCodeAgentTraceStore, appendResultEvent = true, refreshEvents = true, forceResultSync = false }) { const initial = currentResult ?? await loadPersistedAgentRunResult(traceId, options); const mapped = initial ? await resolveAgentRunTraceCommandMapping({ traceId, mapped: initial, options }) : initial; + const performanceStore = options.backendPerformanceStore ?? currentBackendPerformanceContext(); if (!forceResultSync && options.skipAgentRunTerminalRefreshIfComplete === true && agentRunTerminalResultRefreshSatisfied(mapped, traceId)) { const runnerTrace = mapped.runnerTrace ?? traceStore.snapshot(traceId); return { result: mapped, runnerTrace, found: true, eventsRefreshed: false, resultSynced: false, terminalRefreshSkipped: true }; @@ -450,12 +451,15 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op const fetchImpl = options.fetchImpl ?? globalThis.fetch; const managerUrl = resolveAgentRunManagerUrl(env, mapped.agentRun.managerUrl); const timeoutMs = parsePositiveInteger(env.HWLAB_CODE_AGENT_AGENTRUN_HTTP_TIMEOUT_MS, 20_000); - const eventsResponse = refreshEvents ? await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } }) : null; + const previousLastSeq = Number(mapped.agentRun.lastSeq ?? 0); + const eventsResponse = refreshEvents ? await measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping: { ...mapped.agentRun, traceSummary: mapped.traceSummary } }) : null; if (eventsResponse) appendAgentRunEventsToTrace(traceStore, traceId, eventsResponse.events, mapped.agentRun); + const nextLastSeq = eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq; + recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse, previousLastSeq, nextLastSeq); if (!forceResultSync && !agentRunCommandResultSyncRequired(mapped, eventsResponse)) { const nextMapping = withAgentRunRunnerJobCount({ ...mapped.agentRun, - lastSeq: eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq, + lastSeq: nextLastSeq, status: mapped.agentRun.status ?? "running", runStatus: mapped.agentRun.runStatus ?? null, commandState: mapped.agentRun.commandState ?? null, @@ -467,15 +471,11 @@ export async function syncAgentRunChatResult({ traceId, currentResult = null, op options.codeAgentChatResults?.set?.(traceId, payload); return { result: payload, runnerTrace: payload.runnerTrace ?? traceStore.snapshot(traceId), found: true, eventsRefreshed: Boolean(eventsResponse), resultSynced: false }; } - const result = await agentRunJson(fetchImpl, managerUrl, `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`, { - method: "GET", - timeoutMs, - env - }); + const result = await measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped, eventsResponse }); const nextMapping = withAgentRunRunnerJobCount({ ...mapped.agentRun, ...agentRunResultRefs(result), - lastSeq: eventsResponse ? agentRunTraceCursorSeq(eventsResponse, mapped.agentRun.lastSeq) : mapped.agentRun.lastSeq, + lastSeq: nextLastSeq, status: result?.status ?? mapped.agentRun.status ?? "running", runStatus: result?.runStatus ?? mapped.agentRun.runStatus ?? null, commandState: result?.commandState ?? mapped.agentRun.commandState ?? null, @@ -1878,12 +1878,109 @@ async function fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, e events, afterSeq, endSeq, + rawEventCount: rawEvents.length, commandFiltered: Boolean(currentCommandId), maxSeq: Math.max(afterSeq, ...rawEvents.map((event) => Number(event?.seq ?? 0))), traceLastSeq: Math.max(afterSeq, ...events.map((event) => Number(event?.seq ?? 0)).filter(Number.isFinite)) }; } +async function measuredAgentRunEventsFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapping = {} }) { + const startedAt = nowMs(); + try { + const response = await fetchAgentRunEventsForTrace({ fetchImpl, managerUrl, timeoutMs, env, mapping }); + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: "ok", durationMs: nowMs() - startedAt }); + return response; + } catch (error) { + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "agentrun_events_fetch", status: agentRunMetricStatusForError(error), durationMs: nowMs() - startedAt }); + performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_events_fetch_failed" }); + throw error; + } +} + +async function measuredAgentRunResultFetch({ performanceStore, fetchImpl, managerUrl, timeoutMs, env, mapped = {}, eventsResponse = null }) { + const startedAt = nowMs(); + const path = `/api/v1/runs/${encodeURIComponent(mapped.agentRun.runId)}/commands/${encodeURIComponent(mapped.agentRun.commandId)}/result`; + try { + const result = await agentRunJson(fetchImpl, managerUrl, path, { method: "GET", timeoutMs, env }); + const eventCount = agentRunResultEventCount(result, eventsResponse); + performanceStore?.recordAgentRunResult?.({ + durationMs: nowMs() - startedAt, + eventCount, + eventsScanned: eventCount, + pagesScanned: estimateAgentRunResultPages(eventCount), + status: result?.terminalStatus ?? result?.status ?? "ok", + budget: timeoutMs + }); + return result; + } catch (error) { + const status = agentRunMetricStatusForError(error); + const eventCount = agentRunResultEventCount(null, eventsResponse); + performanceStore?.recordAgentRunResult?.({ + durationMs: nowMs() - startedAt, + eventCount, + eventsScanned: eventCount, + pagesScanned: estimateAgentRunResultPages(eventCount), + status, + budget: timeoutMs, + timedOut: status === "timeout" + }); + performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_result_failed" }); + throw error; + } +} + +function recordAgentRunEventsProjectionMetrics(performanceStore, eventsResponse = null, previousLastSeq = 0, nextLastSeq = 0) { + if (!performanceStore || !eventsResponse) return; + const cursorAdvance = Math.max(0, Number(nextLastSeq ?? 0) - Number(previousLastSeq ?? 0)); + if (cursorAdvance > 0) { + performanceStore.recordWorkbenchProjectionCursorAdvance?.({ + status: "advanced", + reason: eventsResponse.commandFiltered ? "command_filtered_events" : "run_events", + projectionStatus: "projecting", + source: "agentrun_v01", + count: cursorAdvance + }); + } + const counts = new Map(); + for (const event of eventsResponse.events ?? []) { + const kind = String(event?.type ?? event?.payload?.phase ?? "unknown").trim() || "unknown"; + counts.set(kind, (counts.get(kind) ?? 0) + 1); + } + if (counts.size === 0 && Number(eventsResponse.rawEventCount ?? 0) > 0) counts.set("filtered", Number(eventsResponse.rawEventCount)); + for (const [eventKind, count] of counts.entries()) { + performanceStore.recordWorkbenchProjectorEvents?.({ eventKind, status: "ok", count }); + } +} + +function agentRunResultEventCount(result = null, eventsResponse = null) { + const candidates = [ + result?.traceSummary?.sourceEventCount, + result?.traceSummary?.eventCount, + result?.sourceEventCount, + result?.eventCount, + result?.providerTrace?.eventCount, + eventsResponse?.maxSeq, + eventsResponse?.rawEventCount, + eventsResponse?.events?.length + ]; + for (const value of candidates) { + const count = Number(value); + if (Number.isFinite(count) && count >= 0) return Math.floor(count); + } + return 0; +} + +function estimateAgentRunResultPages(eventCount) { + const count = Number(eventCount); + if (!Number.isFinite(count) || count <= 0) return 0; + return Math.max(1, Math.ceil(count / 500)); +} + +function agentRunMetricStatusForError(error) { + return error?.code === "agentrun_timeout" || error?.name === "AbortError" || Number(error?.statusCode) === 504 ? "timeout" : "error"; +} + function agentRunTraceCursorSeq(eventsResponse = {}, previousLastSeq = 0) { const afterSeq = Number(eventsResponse.afterSeq ?? 0); const endSeq = Number(eventsResponse.endSeq ?? 0); diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index b29472cc..139bd4d7 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -855,6 +855,7 @@ function scheduleAgentRunProjectionSync({ traceId, params = {}, options = {}, tr isTerminal: isTraceCommandTerminalStatus, activitySignature: agentRunProjectionActivitySignature, sleep: sleepAgentRunProjectionSync, + performanceStore: options.backendPerformanceStore, syncResult: ({ result }) => syncAgentRunChatResult({ traceId, currentResult: result, @@ -1181,10 +1182,11 @@ export async function handleCodeAgentChatResultHttp(request, response, url, opti } export async function handleCodeAgentTurnHttp(request, response, url, options) { + const startedAt = nowMs(); const parts = url.pathname.split("/").filter(Boolean); const traceId = decodeURIComponent(parts[3] ?? ""); if (!safeTraceId(traceId)) { - sendJson(response, 400, { + const body = { ok: false, status: "unknown", running: false, @@ -1193,16 +1195,20 @@ export async function handleCodeAgentTurnHttp(request, response, url, options) { code: "invalid_trace_id", message: "traceId must start with trc_ and contain only safe identifier characters" } - }); + }; + recordCodeAgentTurnReadMetric(options, url, 400, body, startedAt); + sendJson(response, 400, body); return; } const requestOptions = codeAgentRequestScopedOptions(options); const projectMismatch = await measureCodeAgentHttpPhase(requestOptions, "turn_project_check", () => traceProjectMismatchSnapshot(traceId, url, requestOptions, "turn")); if (projectMismatch) { + recordCodeAgentTurnReadMetric(options, url, projectMismatch.statusCode, projectMismatch.body, startedAt); sendJson(response, projectMismatch.statusCode, projectMismatch.body); return; } const resolved = await measureCodeAgentHttpPhase(requestOptions, "turn_resolve_status", () => resolveCodeAgentTurnStatusSnapshot(traceId, requestOptions)); + recordCodeAgentTurnReadMetric(options, url, resolved.statusCode, resolved.body, startedAt); sendJson(response, resolved.statusCode, resolved.body); } @@ -1240,13 +1246,16 @@ async function refreshAgentRunProjectionForCompatRead({ traceId, result = null, if (!safeTraceId(traceId) || !shouldRefreshAgentRunProjectionOnRead(result)) return { result, runnerTrace: trace, refreshError: null }; const traceStore = options.traceStore ?? defaultCodeAgentTraceStore; const refreshOptions = codeAgentTurnStatusRefreshOptions(options); + const performanceStore = options.backendPerformanceStore ?? options.backendPerformance; + const startedAt = nowMs(); + performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "seen", reason: "read_side_refresh" }); try { const synced = await syncAgentRunChatResult({ traceId, currentResult: result, options: refreshOptions, traceStore, - forceResultSync: true + forceResultSync: agentRunProjectionObservedTerminal(result, trace) }); const payload = synced.result ?? result; if (payload && isTraceCommandTerminalStatus(payload.status)) { @@ -1256,8 +1265,11 @@ async function refreshAgentRunProjectionForCompatRead({ traceId, result = null, options: refreshOptions }); } + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "turn_finalize", status: "ok", durationMs: nowMs() - startedAt }); return { result: payload, runnerTrace: synced.runnerTrace ?? traceStore.snapshot(traceId), refreshError: null }; } catch (error) { + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "turn_finalize", status: error?.code === "agentrun_timeout" ? "timeout" : "error", durationMs: nowMs() - startedAt }); + performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "agentrun_read_side_sync_failed" }); traceStore.append(traceId, { type: "projection_refresh", status: "degraded", @@ -1336,6 +1348,7 @@ async function readCodeAgentCompatProjection(traceId, options = {}) { trace = traceSnapshotWithTerminalEvidence(refreshed.runnerTrace ?? trace, result, traceId, refreshed.refreshError ?? null); } const projection = readModel.projectionDiagnostics({ traceId, result, trace, refreshError: refreshed.refreshError ?? null }); + recordCodeAgentProjectionMetrics(options, { result, trace, projection, refreshError: refreshed.refreshError ?? null }); const found = Boolean(result || session || (trace && trace.status !== "missing") || trace?.persisted === true); return { result, session, trace, projection, found, refreshError: refreshed.refreshError ?? null }; } @@ -3028,8 +3041,105 @@ function createCodeAgentM3HwlabApiRequestJson(options = {}) { }; }; } +function recordCodeAgentProjectionMetrics(options = {}, { result = null, trace = null, projection = null, refreshError = null } = {}) { + const performanceStore = options.backendPerformanceStore ?? options.backendPerformance; + if (!performanceStore?.recordWorkbenchProjection || !projection) return; + const sourceLatestSeq = sourceLatestSeqFromResult(result); + const lastProjectedSeq = Number(projection.lastProjectedSeq ?? 0); + if (!Number.isFinite(sourceLatestSeq) && !Number.isFinite(lastProjectedSeq)) return; + performanceStore.recordWorkbenchProjection({ + sourceLatestSeq: Number.isFinite(sourceLatestSeq) ? sourceLatestSeq : lastProjectedSeq, + lastProjectedSeq: Number.isFinite(lastProjectedSeq) ? lastProjectedSeq : 0, + sourceLatestEventAt: sourceLatestAtFromResult(result, trace), + projectedLatestEventAt: projection.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null, + terminalSourceAt: terminalSourceAtFromResult(result), + terminalProjectedAt: projection.projectionStatus === "caught-up" ? projection.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null : null, + status: result?.status ?? trace?.status ?? "unknown", + reason: projectionReason(projection, refreshError), + projectionStatus: projection.projectionStatus ?? "unknown", + source: result?.agentRun ? "agentrun_v01" : "workbench_read_model" + }); +} +function recordCodeAgentTurnReadMetric(options = {}, url, statusCode, body = {}, startedAt = nowMs()) { + const performanceStore = options.backendPerformanceStore ?? options.backendPerformance; + if (!performanceStore?.recordWorkbenchTurnRead) return; + performanceStore.recordWorkbenchTurnRead({ + route: url?.pathname ?? "/v1/agent/turns/:id", + status: body?.status ?? statusClassLabel(statusCode), + degradedReason: turnReadDegradedReason(body), + durationMs: nowMs() - startedAt, + responseBytes: responseBodyBytes(body), + timedOut: statusCode === 504 || body?.error?.code === "timeout" || body?.refreshError?.code === "agentrun_timeout" + }); +} +function turnReadDegradedReason(body = {}) { + if (body?.refreshError?.code) return body.refreshError.code; + if (body?.projection?.blocker?.code) return body.projection.blocker.code; + if (body?.blocker?.code) return body.blocker.code; + if (body?.error?.code) return body.error.code; + if (body?.projectionStatus === "projecting") return "projecting"; + if (body?.projectionStatus === "blocked") return "projection_blocked"; + return "none"; +} + +function responseBodyBytes(body = {}) { + try { return Buffer.byteLength(JSON.stringify(body)); } catch { return 0; } +} + +function statusClassLabel(value) { + const status = Number(value); + if (!Number.isFinite(status) || status <= 0) return "unknown"; + return `${Math.floor(status / 100)}xx`; +} + +function sourceLatestSeqFromResult(result = null) { + const candidates = [ + result?.agentRun?.lastSeq, + result?.traceSummary?.agentRun?.lastSeq, + result?.traceSummary?.lastSeq, + result?.providerTrace?.lastSeq, + result?.runnerTrace?.lastEvent?.sourceSeq, + result?.runnerTrace?.lastEvent?.seq, + result?.runnerTrace?.eventCount + ]; + for (const value of candidates) { + const seq = Number(value); + if (Number.isFinite(seq) && seq >= 0) return Math.floor(seq); + } + return NaN; +} + +function sourceLatestAtFromResult(result = null, trace = null) { + return result?.traceSummary?.updatedAt + ?? result?.agentRun?.updatedAt + ?? result?.providerTrace?.updatedAt + ?? result?.runnerTrace?.updatedAt + ?? trace?.updatedAt + ?? result?.updatedAt + ?? null; +} + +function terminalSourceAtFromResult(result = null) { + if (!result) return null; + const terminalStatus = result?.agentRun?.terminalStatus ?? result?.traceSummary?.terminalStatus ?? result?.status; + if (!isTraceCommandTerminalStatus(terminalStatus)) return null; + return result?.traceSummary?.updatedAt ?? result?.agentRun?.updatedAt ?? result?.updatedAt ?? null; +} + +function projectionReason(projection = null, refreshError = null) { + if (refreshError) return refreshError.code ?? "refresh_error"; + if (projection?.blocker) return projection.blocker.code ?? "projection_blocked"; + if (projection?.projectionStatus === "projecting") return "projecting"; + if (projection?.projectionStatus === "caught-up") return "none"; + return projection?.projectionStatus ?? "unknown"; +} + +function nowMs() { + if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now(); + return Date.now(); +} function uniqueStrings(values) { return [...new Set(values.map((value) => String(value ?? "").trim()).filter(Boolean))]; diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 42342c6a..ed077c05 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -188,6 +188,38 @@ function nowMs() { return Date.now(); } +function recordWorkbenchTurnReadMetric(options = {}, url, statusCode, body = {}, startedAt = nowMs()) { + const performanceStore = options.backendPerformanceStore ?? options.backendPerformance; + if (!performanceStore?.recordWorkbenchTurnRead) return; + performanceStore.recordWorkbenchTurnRead({ + route: url?.pathname ?? "/v1/workbench/turns/:id", + status: body?.status ?? statusClassLabel(statusCode), + degradedReason: workbenchTurnDegradedReason(body), + durationMs: nowMs() - startedAt, + responseBytes: responseBodyBytes(body), + timedOut: statusCode === 504 || body?.error?.code === "timeout" + }); +} + +function workbenchTurnDegradedReason(body = {}) { + if (body?.projection?.blocker?.code) return body.projection.blocker.code; + if (body?.blocker?.code) return body.blocker.code; + if (body?.error?.code) return body.error.code; + if (body?.projectionStatus === "projecting") return "projecting"; + if (body?.projectionStatus === "blocked") return "projection_blocked"; + return "none"; +} + +function responseBodyBytes(body = {}) { + try { return Buffer.byteLength(JSON.stringify(body)); } catch { return 0; } +} + +function statusClassLabel(value) { + const status = Number(value); + if (!Number.isFinite(status) || status <= 0) return "unknown"; + return `${Math.floor(status / 100)}xx`; +} + async function authenticateWorkbenchRead(request, response, options) { const access = options.accessController; if (!access?.authenticate) { @@ -282,24 +314,35 @@ async function handleWorkbenchMessagePage(response, url, options, actor, session } async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTurnId) { + const startedAt = nowMs(); const queryTraceId = safeTraceId(url.searchParams.get("traceId")); const traceId = safeTraceId(rawTurnId) ?? queryTraceId; const turnId = safeTurnId(rawTurnId) || traceId || rawTurnId; - if (!traceId) return sendJson(response, 400, workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId })); + if (!traceId) { + const body = workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId }); + recordWorkbenchTurnReadMetric(options, url, 400, body, startedAt); + return sendJson(response, 400, body); + } const readModel = createWorkbenchReadModel(options, actor); const result = readModel.resultForTrace(traceId); const session = await readModel.getSessionByTraceId(traceId); if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) { - return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId })); + const body = workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId }); + recordWorkbenchTurnReadMetric(options, url, 403, body, startedAt); + return sendJson(response, 403, body); } const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); const trace = await readModel.traceSnapshot(traceId); const turnProjection = createWorkbenchTurnProjection({ turnId, traceId, result, session, trace }); const status = turnProjection.status; const found = Boolean(resultVisible || session); - if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId })); + if (!found) { + const body = workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId }); + recordWorkbenchTurnReadMetric(options, url, 404, body, startedAt); + return sendJson(response, 404, body); + } const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection }); - sendJson(response, 200, { + const body = { ok: true, status, contractVersion: "workbench-turn-snapshot-v1", @@ -314,7 +357,9 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur blocker: projection.blocker, valuesRedacted: true, secretMaterialStored: false - }); + }; + recordWorkbenchTurnReadMetric(options, url, 200, body, startedAt); + sendJson(response, 200, body); } async function handleWorkbenchTraceEventPage(response, url, options, actor, rawTraceId) { diff --git a/internal/cloud/workbench-projection-finalizer.ts b/internal/cloud/workbench-projection-finalizer.ts index 2bcd6d28..3c7c6ebd 100644 --- a/internal/cloud/workbench-projection-finalizer.ts +++ b/internal/cloud/workbench-projection-finalizer.ts @@ -4,11 +4,12 @@ */ import { appendProjectionDiagnostic } from "./workbench-projection-writer.ts"; -export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = null, traceStore, getCachedResult, isCanceled, isTerminal, syncResult, onTerminal, activitySignature, noResponseTimeoutMs = null, pollIntervalMs = 1000, sleep = defaultSleep } = {}) { +export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = null, traceStore, getCachedResult, isCanceled, isTerminal, syncResult, onTerminal, activitySignature, noResponseTimeoutMs = null, pollIntervalMs = 1000, sleep = defaultSleep, performanceStore = null } = {}) { if (!traceId || !currentResult?.agentRun?.runId || !currentResult?.agentRun?.commandId || typeof syncResult !== "function") return null; let lastActivityAt = Date.now(); let lastActivitySignature = activitySignature?.(currentResult, traceStore?.snapshot?.(traceId)) ?? null; let idleDegraded = false; + performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "scheduled", reason: "finalizer" }); setImmediate(() => { void (async () => { @@ -18,13 +19,17 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = const cached = getCachedResult?.(traceId); if (isCanceled?.(cached)) return; if (isTerminal?.(cached?.status)) { + performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "completed", reason: "cached_terminal" }); onTerminal?.(cached, { preserveLastTraceId: true }); return; } + const syncStartedAt = Date.now(); try { const synced = await syncResult({ result }); + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "candidate_scan", status: "ok", durationMs: Date.now() - syncStartedAt }); result = synced?.result ?? result; if (result && isTerminal?.(result.status)) { + performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "completed", reason: "synced_terminal" }); onTerminal?.(result, { preserveLastTraceId: true }); return; } @@ -36,10 +41,14 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = } lastError = null; } catch (error) { + performanceStore?.recordWorkbenchProjectorBatch?.({ phase: "candidate_scan", status: error?.code === "agentrun_timeout" ? "timeout" : "error", durationMs: Date.now() - syncStartedAt }); + performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "finalizer_sync_failed" }); lastError = error; } const idleMs = Date.now() - lastActivityAt; if (noResponseTimeoutMs && idleMs >= noResponseTimeoutMs && !idleDegraded) { + performanceStore?.recordWorkbenchProjectorCandidate?.({ status: "degraded", reason: "no_response_idle_timeout" }); + performanceStore?.recordWorkbenchProjectorError?.({ reason: lastError?.code ?? "no_response_idle_timeout" }); appendProjectionDiagnostic(traceStore, traceId, { type: "turn-status", status: "degraded", @@ -60,6 +69,7 @@ export function scheduleWorkbenchProjectionFinalizer({ traceId, currentResult = await sleep(pollIntervalMs); } })().catch((error) => { + performanceStore?.recordWorkbenchProjectorError?.({ reason: error?.code ?? "workbench_projection_finalizer_crashed" }); appendProjectionDiagnostic(traceStore, traceId, { type: "turn-status", status: "degraded",