feat: add workbench projection performance metrics (#1592)
This commit is contained in:
Lyon
@@ -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<void>((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<string, unknown>) => {
|
||||
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<void>((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<void>((resolve, reject) => agentRunServer.close((error) => error ? reject(error) : resolve()));
|
||||
}
|
||||
});
|
||||
|
||||
@@ -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<string, string>;
|
||||
value: number;
|
||||
}
|
||||
|
||||
interface BackendPerformanceStoreOptions {
|
||||
env?: Record<string, unknown>;
|
||||
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<string, HistogramSeries>();
|
||||
const upstreamSeries = new Map<string, HistogramSeries>();
|
||||
const droppedSeries = new Map<string, CounterSeries>();
|
||||
const projectionLagEventsSeries = new Map<string, HistogramSeries>();
|
||||
const projectionLagSecondsSeries = new Map<string, HistogramSeries>();
|
||||
const projectionStuckGaugeSeries = new Map<string, GaugeSeries>();
|
||||
const projectionCursorAdvanceSeries = new Map<string, CounterSeries>();
|
||||
const terminalProjectionDelaySeries = new Map<string, HistogramSeries>();
|
||||
const turnGetDurationSeries = new Map<string, HistogramSeries>();
|
||||
const turnGetTimeoutSeries = new Map<string, CounterSeries>();
|
||||
const turnGetResponseBytesSeries = new Map<string, HistogramSeries>();
|
||||
const agentRunResultDurationSeries = new Map<string, HistogramSeries>();
|
||||
const agentRunResultPagesSeries = new Map<string, HistogramSeries>();
|
||||
const agentRunResultEventsSeries = new Map<string, HistogramSeries>();
|
||||
const agentRunResultTimeoutSeries = new Map<string, CounterSeries>();
|
||||
const projectorBatchDurationSeries = new Map<string, HistogramSeries>();
|
||||
const projectorCandidatesSeries = new Map<string, CounterSeries>();
|
||||
const projectorEventsProcessedSeries = new Map<string, CounterSeries>();
|
||||
const projectorErrorsSeries = new Map<string, CounterSeries>();
|
||||
const projectorLastSuccessSeries = new Map<string, GaugeSeries>();
|
||||
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<T>(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<string, CounterSeries>, labels: Record<str
|
||||
return true;
|
||||
}
|
||||
|
||||
function incrementCounterBy(series: Map<string, CounterSeries>, labels: Record<string, string>, 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<string, GaugeSeries>, labels: Record<string, string>, 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<string, CounterSeries>) {
|
||||
return [...series.values()].map((entry) => `${metricName}{${renderLabels(entry.labels)}} ${entry.value}`);
|
||||
}
|
||||
|
||||
function renderGauges(metricName: string, series: Map<string, GaugeSeries>) {
|
||||
return [...series.values()].map((entry) => `${metricName}{${renderLabels(entry.labels)}} ${formatPrometheusValue(entry.value)}`);
|
||||
}
|
||||
|
||||
function renderHistogram(metricName: string, series: Map<string, HistogramSeries>, 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);
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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))];
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user