diff --git a/internal/cloud/backend-performance.test.ts b/internal/cloud/backend-performance.test.ts new file mode 100644 index 00000000..b94442f1 --- /dev/null +++ b/internal/cloud/backend-performance.test.ts @@ -0,0 +1,111 @@ +// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 +// Verifies Cloud API backend RED metrics, Server-Timing, and AgentRun upstream timing privacy. + +import assert from "node:assert/strict"; +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 { createCodeAgentTraceStore } from "./code-agent-trace-store.ts"; +import { createCloudApiServer } from "./server.ts"; + +test("backend performance store templates routes and appends Cloud API metrics to loopback metrics", async () => { + assert.equal(backendPerformanceRouteTemplate("/v1/agent/traces/trc_secret_0123456789abcdef?full=1"), "/v1/agent/traces/:id"); + assert.equal(backendPerformanceRouteTemplate("/api/v1/runs/run_0123456789abcdef/commands/cmd_0123456789abcdef/result"), "/api/v1/runs/:id/commands/:id/result"); + assert.equal(backendPerformanceRouteTemplate("/v1/admin/billing/users/12345/credits/adjust"), "/v1/admin/billing/users/:id/credits/adjust"); + + const backendPerformanceStore = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03" } }); + const server = createCloudApiServer({ backendPerformanceStore, env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03" } }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + try { + const address = server.address(); + assert.ok(address && typeof address === "object"); + const baseUrl = `http://127.0.0.1:${address.port}`; + const summary = await fetch(`${baseUrl}/v1/web-performance/summary`); + assert.equal(summary.status, 200); + assert.match(summary.headers.get("server-timing") ?? "", /total;dur=/u); + + const metrics = await fetch(`${baseUrl}/v1/web-performance/metrics`); + const text = await metrics.text(); + assert.equal(metrics.status, 200); + assert.match(text, /hwlab_cloud_api_requests_total\{[^}]*route="\/v1\/web-performance\/summary"[^}]*status_class="2xx"[^}]*\} 1/u); + assert.match(text, /hwlab_cloud_api_request_duration_seconds_count\{[^}]*route="\/v1\/web-performance\/summary"/u); + assert.doesNotMatch(text, /trc_secret|0123456789abcdef/u); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + +test("AgentRun adapter records upstream timing through backend performance context", async () => { + const calls: Array<{ method?: string; path: string; body?: any }> = []; + const agentRunServer = createHttpServer(async (request, response) => { + const url = new URL(request.url || "/", "http://127.0.0.1"); + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + const body = chunks.length ? JSON.parse(Buffer.concat(chunks).toString("utf8")) : null; + calls.push({ method: request.method, path: url.pathname, body }); + const send = (data: Record) => { + response.writeHead(200, { "content-type": "application/json" }); + response.end(`${JSON.stringify({ ok: true, data, traceId: "trc_backend_perf" })}\n`); + }; + if (request.method === "POST" && url.pathname === "/api/v1/sessions") return send({ ok: true }); + if (request.method === "POST" && url.pathname === "/api/v1/runs") return send({ id: "run_backend_perf", status: "pending", backendProfile: "codex", sessionRef: body.sessionRef, resourceBundleRef: body.resourceBundleRef }); + if (request.method === "POST" && url.pathname === "/api/v1/runs/run_backend_perf/commands") return send({ id: "cmd_backend_perf", runId: "run_backend_perf", state: "pending", type: "turn", seq: 1 }); + if (request.method === "POST" && url.pathname === "/api/v1/runs/run_backend_perf/runner-jobs") return send({ action: "create-kubernetes-job", runId: "run_backend_perf", commandId: body.commandId, attemptId: "attempt_backend_perf", runnerId: "runner_backend_perf", namespace: "agentrun-v01", jobName: "agentrun-v01-runner-backend-perf" }); + 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" } }); + const context = store.createManualContext({ route: "/v1/agent/chat", method: "POST" }); + const traceStore = createCodeAgentTraceStore(); + + const result = await withBackendPerformanceContext(context, () => submitAgentRunChatTurn({ + traceId: "trc_backend_perf", + traceStore, + params: { + message: "backend perf smoke", + ownerUserId: "usr_backend_perf", + projectId: "prj_hwpod_workbench", + conversationId: "cnv_backend_perf", + sessionId: "ses_backend_perf", + providerProfile: "codex-api" + }, + options: { + env: { + HWLAB_CODE_AGENT_ADAPTER: "agentrun-v01", + AGENTRUN_MGR_URL: `http://127.0.0.1:${address.port}`, + HWLAB_CODE_AGENT_AGENTRUN_ALLOW_NON_K3S_URL: "1", + HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601", + HWLAB_CODE_AGENT_AGENTRUN_SOURCE_COMMIT: "0123456789abcdef0123456789abcdef01234567", + HWLAB_CODE_AGENT_DEFAULT_PROVIDER_PROFILE: "codex-api", + HWLAB_RUNTIME_API_URL: "http://hwlab-cloud-api.hwlab-v03.svc.cluster.local:6667", + HWLAB_RUNTIME_WEB_URL: "http://hwlab-cloud-web.hwlab-v03.svc.cluster.local:8080", + HWLAB_RUNTIME_NAMESPACE: "hwlab-v03", + HWLAB_RUNTIME_LANE: "v03", + HWLAB_RUNTIME_ENDPOINT_LOCKED: "1", + HWLAB_CODE_AGENT_ASSEMBLED_RUNTIME: "1", + UNIDESK_MAIN_SERVER_IP: "http://74.48.78.17:18081" + }, + accessController: { + async codeAgentToolCapabilitiesForOwner() { return { contractVersion: "admin-access-v1", tools: {} }; }, + store: { async findActiveDefaultApiKeyForUser() { return { displaySecret: "hwl_live_owner_secret" }; } } + } + } + })); + + assert.equal(result.status, "running"); + assert.ok(calls.some((call) => call.path === "/api/v1/runs/run_backend_perf/runner-jobs")); + const text = store.metricsText(); + assert.match(text, /hwlab_cloud_api_upstream_duration_seconds_count\{[^}]*component="agentrun"[^}]*route="\/api\/v1\/runs"[^}]*\} 1/u); + assert.match(text, /hwlab_cloud_api_upstream_duration_seconds_count\{[^}]*component="agentrun"[^}]*route="\/api\/v1\/runs\/:id\/commands"[^}]*\} 1/u); + assert.match(text, /hwlab_cloud_api_upstream_duration_seconds_count\{[^}]*component="agentrun"[^}]*route="\/api\/v1\/runs\/:id\/runner-jobs"[^}]*\} 1/u); + assert.doesNotMatch(text, /run_backend_perf|cmd_backend_perf|ses_backend_perf|cnv_backend_perf/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 new file mode 100644 index 00000000..1140277a --- /dev/null +++ b/internal/cloud/backend-performance.ts @@ -0,0 +1,357 @@ +// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 +// Records low-cardinality Cloud API backend performance metrics and Server-Timing phases. + +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 MAX_SERVER_TIMING_ENTRIES = 8; + +const backendPerformanceStorage = new AsyncLocalStorage(); + +interface HistogramSeries { + labels: Record; + buckets: number[]; + sum: number; + count: number; +} + +interface CounterSeries { + labels: Record; + value: number; +} + +interface BackendPerformanceStoreOptions { + env?: Record; + maxSeries?: number; +} + +interface BackendPerformanceContextInput { + route: string; + method: string; +} + +interface BackendPhaseInput { + phase: string; + durationMs: number; + outcome?: string; +} + +interface BackendUpstreamInput { + component: string; + operation?: string; + route: string; + method: string; + statusClass?: string; + outcome?: string; + durationMs: number; +} + +export function createBackendPerformanceStore(options: BackendPerformanceStoreOptions = {}) { + const env = options.env ?? process.env; + const maxSeries = parsePositiveInteger(env.HWLAB_BACKEND_PERFORMANCE_MAX_SERIES, options.maxSeries ?? 320); + const requestDurationSeries = new Map(); + const requestTotalSeries = new Map(); + const requestPhaseSeries = new Map(); + const upstreamSeries = new Map(); + const droppedSeries = 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") + }; + + function beginHttpRequest(request: any, response: any) { + const url = safeUrl(request?.url || "/"); + const context = createContext({ route: backendPerformanceRouteTemplate(url.pathname), method: normalizeMethod(request?.method) }); + const originalWriteHead = typeof response?.writeHead === "function" ? response.writeHead : null; + if (originalWriteHead) { + response.writeHead = function writeHeadWithServerTiming(statusCode: number, ...args: unknown[]) { + context.setStatus(statusCode); + context.applyServerTimingHeader(response); + return originalWriteHead.call(this, statusCode, ...args); + }; + } + if (typeof response?.once === "function") { + response.once("finish", () => context.finish(Number(response.statusCode ?? context.statusCode ?? 0))); + } + return context; + } + + function createManualContext(input: BackendPerformanceContextInput) { + return createContext({ route: backendPerformanceRouteTemplate(input.route), method: normalizeMethod(input.method) }); + } + + function createContext(input: BackendPerformanceContextInput) { + const startedAt = nowMs(); + const phases: Array<{ name: string; durationMs: number; outcome: string }> = []; + let statusCode = 200; + let finished = false; + const route = input.route; + const method = input.method; + + function recordPhase(inputPhase: BackendPhaseInput) { + const durationMs = boundedDurationMs(inputPhase.durationMs, 30_000); + const phase = sanitizeMetricName(inputPhase.phase, "unknown"); + const outcome = normalizeOutcome(inputPhase.outcome ?? "ok"); + phases.push({ name: phase, durationMs, outcome }); + const labels = { ...baseLabels, route, method, phase, outcome }; + if (!recordHistogram(requestPhaseSeries, labels, durationMs / 1000, PHASE_BUCKETS_SECONDS, maxSeries)) incrementDropped("phase_series_limit"); + } + + async function measure(phase: string, fn: () => Promise | T): Promise { + const phaseStartedAt = nowMs(); + try { + const result = await fn(); + recordPhase({ phase, durationMs: nowMs() - phaseStartedAt, outcome: "ok" }); + return result; + } catch (error) { + recordPhase({ phase, durationMs: nowMs() - phaseStartedAt, outcome: "error" }); + throw error; + } + } + + function recordUpstream(inputUpstream: BackendUpstreamInput) { + const labels = { + ...baseLabels, + component: sanitizeMetricName(inputUpstream.component, "unknown"), + operation: sanitizeMetricName(inputUpstream.operation ?? inputUpstream.route, "unknown"), + route: backendPerformanceRouteTemplate(inputUpstream.route), + method: normalizeMethod(inputUpstream.method), + status_class: normalizeStatusClass(inputUpstream.statusClass ?? "unknown"), + outcome: normalizeOutcome(inputUpstream.outcome ?? "ok") + }; + if (!recordHistogram(upstreamSeries, labels, boundedDurationMs(inputUpstream.durationMs, 120_000) / 1000, UPSTREAM_BUCKETS_SECONDS, maxSeries)) incrementDropped("upstream_series_limit"); + } + + function setStatus(nextStatusCode: number) { + if (Number.isInteger(nextStatusCode) && nextStatusCode > 0) statusCode = nextStatusCode; + } + + function applyServerTimingHeader(response: any) { + if (!response || typeof response.setHeader !== "function" || response.headersSent) return; + const header = serverTimingHeader(nowMs() - startedAt, phases); + if (header) response.setHeader("server-timing", header); + } + + function finish(finalStatusCode = statusCode) { + if (finished) return; + finished = true; + const statusClassValue = statusClass(finalStatusCode); + const outcome = statusClassValue === "5xx" ? "error" : statusClassValue === "4xx" ? "denied" : "ok"; + const labels = { ...baseLabels, route, method, status_class: statusClassValue, outcome }; + if (!recordHistogram(requestDurationSeries, labels, boundedDurationMs(nowMs() - startedAt, 120_000) / 1000, REQUEST_BUCKETS_SECONDS, maxSeries)) incrementDropped("request_series_limit"); + incrementCounter(requestTotalSeries, labels, maxSeries) || incrementDropped("request_counter_series_limit"); + } + + return { route, method, startedAt, recordPhase, measure, recordUpstream, setStatus, applyServerTimingHeader, finish, get statusCode() { return statusCode; } }; + } + + function incrementDropped(reason: string) { + incrementCounter(droppedSeries, { ...baseLabels, reason }, Math.max(maxSeries, 16)); + } + + function metricsText() { + return [ + "# HELP hwlab_cloud_api_requests_total Cloud API HTTP requests by low-cardinality route template.", + "# TYPE hwlab_cloud_api_requests_total counter", + ...renderCounters("hwlab_cloud_api_requests_total", requestTotalSeries), + "# HELP hwlab_cloud_api_request_duration_seconds Cloud API HTTP request duration in seconds.", + "# TYPE hwlab_cloud_api_request_duration_seconds histogram", + ...renderHistogram("hwlab_cloud_api_request_duration_seconds", requestDurationSeries, REQUEST_BUCKETS_SECONDS), + "# HELP hwlab_cloud_api_request_phase_duration_seconds Cloud API handler phase duration in seconds.", + "# TYPE hwlab_cloud_api_request_phase_duration_seconds histogram", + ...renderHistogram("hwlab_cloud_api_request_phase_duration_seconds", requestPhaseSeries, PHASE_BUCKETS_SECONDS), + "# HELP hwlab_cloud_api_upstream_duration_seconds Cloud API upstream dependency call duration in seconds.", + "# TYPE hwlab_cloud_api_upstream_duration_seconds histogram", + ...renderHistogram("hwlab_cloud_api_upstream_duration_seconds", upstreamSeries, UPSTREAM_BUCKETS_SECONDS), + "# 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), + "" + ].join("\n"); + } + + function snapshot() { + return { + requestSeries: requestDurationSeries.size, + requestCount: [...requestTotalSeries.values()].reduce((sum, entry) => sum + entry.value, 0), + requestPhaseSeries: requestPhaseSeries.size, + requestPhaseCount: histogramSampleCount(requestPhaseSeries), + upstreamSeries: upstreamSeries.size, + upstreamCount: histogramSampleCount(upstreamSeries), + droppedSeries: droppedSeries.size + }; + } + + return { beginHttpRequest, createManualContext, metricsText, snapshot }; +} + +export function withBackendPerformanceContext(context: unknown, callback: () => T): T { + return backendPerformanceStorage.run(context, callback); +} + +export function currentBackendPerformanceContext(): any { + return backendPerformanceStorage.getStore() ?? null; +} + +export function backendPerformanceRouteTemplate(input: unknown): string { + const raw = String(input ?? "").trim() || "/"; + const pathOnly = raw.split(/[?#]/u, 1)[0] || "/"; + let pathname = pathOnly; + try { + pathname = new URL(raw, "http://hwlab.local").pathname || pathOnly; + } catch { + // Keep pathOnly. + } + const parts = pathname.split("/").filter(Boolean).map((part) => routeSegmentTemplate(safeDecode(part))); + const path = `/${parts.join("/")}`; + return path.length > 140 ? `${path.slice(0, 137)}...` : path; +} + +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)}`); + return entries.join(", "); +} + +function recordHistogram(series: Map, labels: Record, value: number, buckets: number[], maxSeries: number) { + const key = stableLabelKey(labels); + let entry = series.get(key); + if (!entry) { + if (series.size >= maxSeries) return false; + entry = { labels, buckets: buckets.map(() => 0), sum: 0, count: 0 }; + series.set(key, entry); + } + for (let index = 0; index < buckets.length; index += 1) { + if (value <= buckets[index]) entry.buckets[index] += 1; + } + entry.sum += value; + entry.count += 1; + return true; +} + +function incrementCounter(series: Map, labels: Record, 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 += 1; + return true; +} + +function renderCounters(metricName: string, series: Map) { + return [...series.values()].map((entry) => `${metricName}{${renderLabels(entry.labels)}} ${entry.value}`); +} + +function renderHistogram(metricName: string, series: Map, buckets: number[]) { + const lines: string[] = []; + for (const entry of series.values()) { + for (let index = 0; index < buckets.length; index += 1) { + lines.push(`${metricName}_bucket{${renderLabels({ ...entry.labels, le: String(buckets[index]) })}} ${entry.buckets[index]}`); + } + lines.push(`${metricName}_bucket{${renderLabels({ ...entry.labels, le: "+Inf" })}} ${entry.count}`); + lines.push(`${metricName}_sum{${renderLabels(entry.labels)}} ${entry.sum.toFixed(6)}`); + lines.push(`${metricName}_count{${renderLabels(entry.labels)}} ${entry.count}`); + } + return lines; +} + +function histogramSampleCount(series: Map) { + return [...series.values()].reduce((sum, entry) => sum + entry.count, 0); +} + +function renderLabels(labels: Record) { + return Object.entries(labels).map(([key, value]) => `${key}="${escapeLabelValue(value)}"`).join(","); +} + +function stableLabelKey(labels: Record) { + return Object.keys(labels).sort().map((key) => `${key}=${labels[key]}`).join("\u0000"); +} + +function routeSegmentTemplate(segment: string) { + if (/^\d+$/u.test(segment)) return ":id"; + if (/^[a-z][a-z0-9]{1,12}_[A-Za-z0-9_.:-]+$/iu.test(segment)) return ":id"; + if (/^[0-9a-f]{12,}$/iu.test(segment)) return ":id"; + if (/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/iu.test(segment)) return ":id"; + if (segment.length > 48) return ":id"; + return sanitizePathSegment(segment); +} + +function safeUrl(value: string) { + try { return new URL(value, "http://hwlab-cloud-api.local"); } catch { return new URL("/", "http://hwlab-cloud-api.local"); } +} + +function safeDecode(value: string) { + try { return decodeURIComponent(value); } catch { return value; } +} + +function sanitizePathSegment(value: string) { + const sanitized = value.replace(/[^A-Za-z0-9_.:-]/gu, "_"); + return sanitized.slice(0, 64) || "_"; +} + +function normalizeMethod(value: unknown) { + const method = String(value ?? "GET").trim().toUpperCase(); + return /^(GET|POST|PUT|PATCH|DELETE|OPTIONS|HEAD)$/u.test(method) ? method : "OTHER"; +} + +function statusClass(value: unknown) { + const status = Number(value); + if (!Number.isFinite(status) || status <= 0) return "unknown"; + return `${Math.floor(status / 100)}xx`; +} + +function normalizeStatusClass(value: unknown) { + const text = String(value ?? "").trim().toLowerCase(); + return /^(?:1xx|2xx|3xx|4xx|5xx|network|timeout|unknown)$/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 sanitizeMetricName(value: unknown, fallback = "unknown") { + const text = String(value ?? "").trim().toLowerCase().replace(/[^a-z0-9_:-]+/gu, "_").replace(/[-:]+/gu, "_").replace(/^_+|_+$/gu, ""); + return text || fallback; +} + +function sanitizeLabelValue(value: unknown, fallback = "unknown") { + const text = String(value ?? "").trim(); + if (!text) return fallback; + return text.replace(/[\n\r\t]/gu, " ").slice(0, 120); +} + +function escapeLabelValue(value: string) { + return String(value).replace(/\\/gu, "\\\\").replace(/"/gu, "\\\"").replace(/\n/gu, "\\n"); +} + +function parsePositiveInteger(value: unknown, fallback: number) { + const parsed = Number.parseInt(String(value ?? ""), 10); + return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback; +} + +function boundedDurationMs(value: unknown, maxMs: number) { + const parsed = Number(value); + if (!Number.isFinite(parsed) || parsed < 0) return 0; + return Math.min(parsed, maxMs); +} + +function roundDurationMs(value: number) { + return Math.max(0, value).toFixed(1); +} + +function serverTimingToken(value: string) { + const token = sanitizeMetricName(value, "phase").replace(/_/gu, "-"); + return token || "phase"; +} + +function nowMs() { + if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now(); + return Date.now(); +} diff --git a/internal/cloud/code-agent-agentrun-adapter.ts b/internal/cloud/code-agent-agentrun-adapter.ts index 8f6a48f7..ee76e28c 100644 --- a/internal/cloud/code-agent-agentrun-adapter.ts +++ b/internal/cloud/code-agent-agentrun-adapter.ts @@ -1,3 +1,6 @@ +// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. +// Responsibility: AgentRun v0.1 adapter and upstream timing projection for Cloud API observability. + import { createHash, randomUUID } from "node:crypto"; import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; @@ -11,6 +14,7 @@ import { text, truthyFlag } from "./server-http-utils.ts"; +import { backendPerformanceRouteTemplate, currentBackendPerformanceContext } from "./backend-performance.ts"; const ADAPTER_ID = "agentrun-v01"; const DEFAULT_AGENTRUN_MGR_URL = "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080"; @@ -1968,6 +1972,10 @@ function agentRunCommandExecutionEvent(base, payload = {}) { async function agentRunJson(fetchImpl, managerUrl, path, { method = "GET", body, timeoutMs = 20_000, env = process.env } = {}) { const controller = new AbortController(); const timeout = setTimeout(() => controller.abort(), timeoutMs); + const performanceContext = currentBackendPerformanceContext(); + const startedAt = nowMs(); + let statusCode = 0; + let outcome = "ok"; try { const headers: Record = {}; const apiKey = resolveAgentRunApiKey(env); @@ -1979,9 +1987,11 @@ async function agentRunJson(fetchImpl, managerUrl, path, { method = "GET", body, body: body === undefined ? undefined : JSON.stringify(body), signal: controller.signal }); + statusCode = Number(response.status ?? 0); const text = await response.text(); const parsed = text ? JSON.parse(text) : {}; if (!response.ok || parsed?.ok === false) { + outcome = "error"; throw Object.assign(new Error(parsed?.message ?? `AgentRun ${method} ${path} failed with HTTP ${response.status}`), { code: parsed?.failureKind ?? "agentrun_request_failed", statusCode: response.status, @@ -1991,14 +2001,42 @@ async function agentRunJson(fetchImpl, managerUrl, path, { method = "GET", body, return parsed?.ok === true && Object.hasOwn(parsed, "data") ? parsed.data : parsed; } catch (error) { if (error?.name === "AbortError") { + outcome = "timeout"; + statusCode = 504; throw Object.assign(new Error(`AgentRun ${method} ${path} timed out after ${timeoutMs}ms`), agentRunHttpErrorContext({ managerUrl, path, method, timeoutMs, code: "agentrun_timeout", statusCode: 504, category: "upstream-timeout" })); } + if (outcome === "ok") outcome = "error"; throw error; } finally { clearTimeout(timeout); + performanceContext?.recordUpstream?.({ + component: "agentrun", + operation: agentRunOperationName(method, path), + route: backendPerformanceRouteTemplate(path), + method, + statusClass: statusClass(statusCode), + outcome, + durationMs: nowMs() - startedAt + }); } } +function agentRunOperationName(method, path) { + const route = backendPerformanceRouteTemplate(path).replace(/^\/api\/v1\//u, "").replace(/\/:id/gu, ""); + return `${String(method ?? "GET").toLowerCase()}_${route || "root"}`.replace(/[^a-z0-9]+/giu, "_").replace(/^_+|_+$/gu, "").toLowerCase(); +} + +function statusClass(value) { + const status = Number(value); + if (!Number.isFinite(status) || status <= 0) return "unknown"; + return `${Math.floor(status / 100)}xx`; +} + +function nowMs() { + if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now(); + return Date.now(); +} + function agentRunHttpErrorContext({ managerUrl, path, method, timeoutMs, code, statusCode, category }) { return { code, diff --git a/internal/cloud/server-code-agent-http.ts b/internal/cloud/server-code-agent-http.ts index bfd6897b..bec7d471 100644 --- a/internal/cloud/server-code-agent-http.ts +++ b/internal/cloud/server-code-agent-http.ts @@ -1,4 +1,4 @@ -// SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0. +// SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-0102 Agent编排 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. // Responsibility: Code Agent HTTP turn admission, lifecycle compatibility wrappers, and session evidence projection. import { createHash, randomUUID } from "node:crypto"; @@ -52,7 +52,8 @@ const DEFAULT_CODE_AGENT_DEEPSEEK_MODEL = "deepseek-chat"; const DEFAULT_CODE_AGENT_MINIMAX_M3_MODEL = "MiniMax-M3"; export async function handleCodeAgentChatHttp(request, response, options) { - const body = await readBody(request, options.bodyLimitBytes); + const perf = options.backendPerformance; + const body = perf ? await perf.measure("body_read", () => readBody(request, options.bodyLimitBytes)) : await readBody(request, options.bodyLimitBytes); let params = {}; try { @@ -95,15 +96,16 @@ export async function handleCodeAgentChatHttp(request, response, options) { ownerUserId: options.actor?.id ?? params.ownerUserId, ownerRole: options.actor?.role ?? params.ownerRole }; - const manualSession = await prepareManualCodeAgentSession({ params: chatParams, options, traceId, response }); + const manualSession = perf ? await perf.measure("manual_session", () => prepareManualCodeAgentSession({ params: chatParams, options, traceId, response })) : await prepareManualCodeAgentSession({ params: chatParams, options, traceId, response }); if (manualSession?.blocked) return; - const workspaceClaim = await claimWorkbenchWorkspaceTurn({ params: chatParams, options, traceId, response }); + const workspaceClaim = perf ? await perf.measure("workspace_claim", () => claimWorkbenchWorkspaceTurn({ params: chatParams, options, traceId, response })) : await claimWorkbenchWorkspaceTurn({ params: chatParams, options, traceId, response }); if (workspaceClaim?.blocked) return; - const billingPreflight = await preflightCodeAgentBilling({ params: chatParams, options, traceId, response }); + const billingPreflight = perf ? await perf.measure("billing_preflight", () => preflightCodeAgentBilling({ params: chatParams, options, traceId, response })) : await preflightCodeAgentBilling({ params: chatParams, options, traceId, response }); if (billingPreflight?.blocked) return; const nativeSessionChatParams = stripSyntheticConversationContext({ ...chatParams, userBillingReservation: billingPreflight?.reservation ?? null }, traceId, options); if (codeAgentChatShortConnectionRequested(request, params, options)) { + if (perf) perf.recordPhase({ phase: "short_connection_accept", durationMs: 0, outcome: "ok" }); submitCodeAgentChatTurn({ params: nativeSessionChatParams, options, @@ -136,9 +138,9 @@ export async function handleCodeAgentChatHttp(request, response, options) { return; } - const payload = await runCodeAgentChat(nativeSessionChatParams, options); - await finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options }); - await recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: payload.status === "completed" ? "active" : payload.status }); + const payload = perf ? await perf.measure("code_agent_run", () => runCodeAgentChat(nativeSessionChatParams, options)) : await runCodeAgentChat(nativeSessionChatParams, options); + await (perf ? perf.measure("billing_finalize", () => finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options })) : finalizeCodeAgentBillingUsage({ payload, params: nativeSessionChatParams, options })); + await (perf ? perf.measure("session_owner_record", () => recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: payload.status === "completed" ? "active" : payload.status })) : recordCodeAgentSessionOwner({ payload, params: nativeSessionChatParams, options, status: payload.status === "completed" ? "active" : payload.status })); const responsePayload = annotateOwner(payload, nativeSessionChatParams); sendJson(response, responsePayload.status === "failed" && responsePayload.error?.code === "invalid_params" ? 400 : 200, responsePayload); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 62b4c7f3..1dfcb324 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0 + * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。 */ import { createHash } from "node:crypto"; @@ -22,40 +22,41 @@ const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]); export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) { - const auth = await authenticateWorkbenchRead(request, response, options); + const perf = options.backendPerformance; + const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options); if (!auth) return; if (url.pathname === "/v1/workbench/sessions") { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await handleWorkbenchSessionList(response, url, options, auth.actor); + await (perf ? perf.measure("workbench_session_list", () => handleWorkbenchSessionList(response, url, options, auth.actor)) : handleWorkbenchSessionList(response, url, options, auth.actor)); return; } const sessionMessagesMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)\/messages$/u); if (sessionMessagesMatch) { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await handleWorkbenchMessagePage(response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1])); + await (perf ? perf.measure("workbench_message_page", () => handleWorkbenchMessagePage(response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1]))) : handleWorkbenchMessagePage(response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1]))); return; } const sessionMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)$/u); if (sessionMatch) { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await handleWorkbenchSessionDetail(response, options, auth.actor, decodeURIComponent(sessionMatch[1])); + await (perf ? perf.measure("workbench_session_detail", () => handleWorkbenchSessionDetail(response, options, auth.actor, decodeURIComponent(sessionMatch[1]))) : handleWorkbenchSessionDetail(response, options, auth.actor, decodeURIComponent(sessionMatch[1]))); return; } const turnMatch = url.pathname.match(/^\/v1\/workbench\/turns\/([^/]+)$/u); if (turnMatch) { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await handleWorkbenchTurnSnapshot(response, url, options, auth.actor, decodeURIComponent(turnMatch[1])); + await (perf ? perf.measure("workbench_turn_snapshot", () => handleWorkbenchTurnSnapshot(response, url, options, auth.actor, decodeURIComponent(turnMatch[1]))) : handleWorkbenchTurnSnapshot(response, url, options, auth.actor, decodeURIComponent(turnMatch[1]))); return; } const traceEventsMatch = url.pathname.match(/^\/v1\/workbench\/traces\/([^/]+)\/events$/u); if (traceEventsMatch) { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - await handleWorkbenchTraceEventPage(response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1])); + await (perf ? perf.measure("workbench_trace_events", () => handleWorkbenchTraceEventPage(response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1]))) : handleWorkbenchTraceEventPage(response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1]))); return; } @@ -64,7 +65,8 @@ export async function handleWorkbenchReadModelHttp(request, response, url, optio export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) { if (request.method !== "GET") return methodNotAllowed(response, "GET"); - const auth = await authenticateWorkbenchRead(request, response, options); + const perf = options.backendPerformance; + const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options); if (!auth) return; const projectId = textValue(url.searchParams.get("projectId")) || DEFAULT_PROJECT_ID; @@ -86,8 +88,12 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option const writeEvent = (name, payload = {}) => { if (closed || response.destroyed) return; + const startedAt = nowMs(); + const eventCreatedAt = realtimeEventCreatedAt(payload); + const traceSeq = realtimeTraceSeq(payload); response.write(`event: ${name}\n`); - response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", projectId, ...payload })}\n\n`); + response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", projectId, serverSentAt: new Date().toISOString(), eventCreatedAt, traceSeq, ...payload })}\n\n`); + perf?.recordPhase({ phase: "sse_write", durationMs: nowMs() - startedAt, outcome: "ok" }); }; writeEvent("workbench.connected", { @@ -100,7 +106,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option } }); - const initialWorkspace = await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId }); + const initialWorkspace = perf ? await perf.measure("workbench_initial_workspace", () => visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId })) : await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId }); if (initialWorkspace) { latestWorkspaceRevision = Number(initialWorkspace.revision ?? 0) || 0; writeEvent("workbench.workspace.snapshot", { @@ -113,7 +119,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option const activeTraceId = requestedTraceId ?? safeTraceId(initialWorkspace?.activeTraceId ?? initialWorkspace?.workspace?.activeTraceId ?? initialWorkspace?.workspace?.lastTraceId); if (activeTraceId) { - await writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }); + await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })); const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => { if (event) { writeEvent("workbench.trace.event", { @@ -174,6 +180,25 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option }); } +function realtimeEventCreatedAt(payload) { + const candidates = [payload?.event?.createdAt, payload?.snapshot?.lastEvent?.createdAt, payload?.turn?.updatedAt, payload?.workspace?.updatedAt]; + for (const value of candidates) { + const text = textValue(value); + if (text) return text; + } + return null; +} + +function realtimeTraceSeq(payload) { + const seq = Number(payload?.cursor?.traceSeq ?? payload?.event?.seq ?? payload?.snapshot?.lastEvent?.seq); + return Number.isFinite(seq) && seq >= 0 ? seq : null; +} + +function nowMs() { + if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now(); + return Date.now(); +} + async function authenticateWorkbenchRead(request, response, options) { const access = options.accessController; if (!access?.authenticate) { diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 31a44b35..5c0e8b67 100644 --- a/internal/cloud/server.ts +++ b/internal/cloud/server.ts @@ -1,5 +1,5 @@ /* - * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0 + * SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0 * 职责: Cloud API REST/SSE route dispatcher。Workbench 新资源路由应委托 read model / compat wrapper，不在 dispatcher 内写业务事实。 */ import { createServer } from "node:http"; @@ -55,6 +55,7 @@ import { handleWebPerformanceMetricsHttp, handleWebPerformanceSummaryHttp } from "./web-performance.ts"; +import { createBackendPerformanceStore, withBackendPerformanceContext } from "./backend-performance.ts"; import { createCodeAgentChatResultStore, handleCodeAgentCancelHttp, @@ -104,6 +105,7 @@ export function createCloudApiServer(options = {}) { maxResults: parsePositiveInteger(env.HWLAB_CODE_AGENT_CHAT_RESULT_CACHE_SIZE, 256) }); const webPerformanceStore = options.webPerformanceStore || createWebPerformanceStore({ env }); + const backendPerformanceStore = options.backendPerformanceStore || createBackendPerformanceStore({ env }); const gatewayRegistry = options.gatewayRegistry || createGatewayDemoRegistry({ staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000), dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 120000) @@ -114,18 +116,22 @@ export function createCloudApiServer(options = {}) { accessController.configureCodeAgentWorkspaceContext({ env, fetchImpl: options.fetchImpl, traceStore, codeAgentChatResults }); } return createServer(async (request, response) => { - try { - await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, hwpodNodeWsRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults, webPerformanceStore }); - } catch (error) { - if (error?.alreadySent) return; - sendJson(response, 500, { - error: { - code: "internal_error", - message: "hwlab-cloud-api request handling failed", - reason: error.message - } - }); - } + const backendPerformance = backendPerformanceStore.beginHttpRequest(request, response); + await withBackendPerformanceContext(backendPerformance, async () => { + try { + await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, hwpodNodeWsRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults, webPerformanceStore, backendPerformanceStore, backendPerformance }); + } catch (error) { + if (error?.alreadySent) return; + backendPerformance.recordPhase({ phase: "route_exception", durationMs: 0, outcome: "error" }); + sendJson(response, 500, { + error: { + code: "internal_error", + message: "hwlab-cloud-api request handling failed", + reason: error.message + } + }); + } + }); }); } @@ -315,7 +321,7 @@ async function handleRestAdapter(request, response, url, options) { } if (url.pathname === "/v1/web-performance/metrics" && request.method === "GET") { - handleWebPerformanceMetricsHttp(request, response, { store: options.webPerformanceStore }); + handleWebPerformanceMetricsHttp(request, response, { store: options.webPerformanceStore, extraMetricsText: () => options.backendPerformanceStore.metricsText() }); return; } diff --git a/internal/cloud/web-performance.ts b/internal/cloud/web-performance.ts index 0193101c..68ad7635 100644 --- a/internal/cloud/web-performance.ts +++ b/internal/cloud/web-performance.ts @@ -383,7 +383,11 @@ export function handleWebPerformanceMetricsHttp(request, response, options = {}) sendJson(response, 404, { error: { code: "not_found", message: "route is not public" } }); return; } - const payload = options.store.metricsText(); + const primary = String(options.store.metricsText() ?? "").trimEnd(); + const extra = typeof options.extraMetricsText === "function" + ? String(options.extraMetricsText() ?? "").trimEnd() + : String(options.extraMetricsText ?? "").trimEnd(); + const payload = [primary, extra].filter(Boolean).join("\n") + "\n"; response.writeHead(200, { "content-type": "text/plain; version=0.0.4; charset=utf-8", "content-length": Buffer.byteLength(payload)