diff --git a/internal/cloud/backend-performance.test.ts b/internal/cloud/backend-performance.test.ts index 32fbd533..26cba64f 100644 --- a/internal/cloud/backend-performance.test.ts +++ b/internal/cloud/backend-performance.test.ts @@ -221,6 +221,34 @@ test("backend performance store records Workbench projection and AgentRun result assert.doesNotMatch(text, /trc_secret|0123456789abcdef|run_secret|cmd_secret|ses_secret|cnv_secret/u); }); +test("backend performance reports zero projection lag for caught-up terminal traces", () => { + 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: 501, + lastProjectedSeq: 501, + sourceLatestEventAt: "2026-06-19T03:00:00.000Z", + projectedLatestEventAt: "2026-06-19T02:00:00.000Z", + terminalSourceAt: "2026-06-19T02:00:00.000Z", + terminalProjectedAt: "2026-06-19T02:00:00.020Z", + status: "completed", + reason: "none", + projectionStatus: "caught-up", + source: "agentrun-v01", + }); + + const text = store.metricsText(); + const lagSumLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_sum") && line.includes('projection_status="caught_up"')); + const lagBucketLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_bucket") && line.includes('projection_status="caught_up"') && line.includes('le="0.05"')); + const stuckLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_stuck_traces") && line.includes('projection_status="caught_up"')); + const terminalDelayBucketLine = text.split("\n").find((line) => line.startsWith("hwlab_workbench_terminal_projection_delay_seconds_bucket") && line.includes('projection_status="caught_up"') && line.includes('le="0.05"')); + assert.ok(lagSumLine); + assert.match(lagSumLine, / 0\.000000$/u); + assert.ok(lagBucketLine?.endsWith(" 1")); + assert.ok(stuckLine?.endsWith(" 0")); + assert.ok(terminalDelayBucketLine?.endsWith(" 1")); +}); + 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) => { diff --git a/internal/cloud/backend-performance.ts b/internal/cloud/backend-performance.ts index ebb347ad..b2c23282 100644 --- a/internal/cloud/backend-performance.ts +++ b/internal/cloud/backend-performance.ts @@ -246,7 +246,7 @@ export function createBackendPerformanceStore(options: BackendPerformanceStoreOp const sourceAt = unixMs(input.sourceLatestEventAt); const projectedAt = unixMs(input.projectedLatestEventAt); if (sourceAt !== null && projectedAt !== null) { - const lagSeconds = Math.max(0, (sourceAt - projectedAt) / 1000); + const lagSeconds = isTerminalProjectionStatus(labels.projection_status) ? 0 : 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); diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 1f6cca0e..7aec058b 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -5,6 +5,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import { createCloudApiServer } from "./server.ts"; +import { createBackendPerformanceStore } from "./backend-performance.ts"; import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts"; import { createWorkbenchTurnProjection, projectionDiagnostics } from "./workbench-turn-projection.ts"; @@ -475,7 +476,8 @@ test("workbench read model projects completed current turn for idle session summ async ensureBootstrap() {}, async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } }; - const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results }); + const backendPerformanceStore = createBackendPerformanceStore({ env: { POD_NAMESPACE: "hwlab-v03", HWLAB_GITOPS_TARGET: "v03", HWLAB_RUNTIME_LANE: "v03", HWLAB_CODE_AGENT_AGENTRUN_PROVIDER_ID: "D601" } }); + const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, backendPerformanceStore }); await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); try { @@ -501,6 +503,11 @@ test("workbench read model projects completed current turn for idle session summ assert.equal(turn.body.projectionStatus, "caught-up"); assert.equal(turn.body.sourceRunId, "run_workbench_idle_completed"); assert.equal(turn.body.sourceCommandId, "cmd_workbench_idle_completed"); + const metricsText = backendPerformanceStore.metricsText(); + const lagCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_projection_lag_seconds_count") && line.includes('projection_status="caught_up"')); + const turnGetCountLine = metricsText.split("\n").find((line) => line.startsWith("hwlab_workbench_turn_get_duration_seconds_count") && line.includes('route="/v1/workbench/turns/:traceId"')); + assert.ok(lagCountLine?.endsWith(" 1")); + assert.ok(turnGetCountLine?.endsWith(" 1")); const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`); assert.equal(trace.status, 200); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 59384046..c24ced08 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -201,6 +201,76 @@ function recordWorkbenchTurnReadMetric(options = {}, url, statusCode, body = {}, }); } +function recordWorkbenchProjectionMetric(options = {}, { result = null, trace = null, projection = null, diagnostic = null } = {}) { + const performanceStore = options.backendPerformanceStore ?? options.backendPerformance; + if (!performanceStore?.recordWorkbenchProjection || !projection) return; + const sourceLatestSeq = workbenchProjectionSourceLatestSeq(result, trace, projection); + const lastProjectedSeq = Number(diagnostic?.lastProjectedSeq ?? 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: workbenchProjectionSourceLatestAt(result, trace, projection), + projectedLatestEventAt: diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null, + terminalSourceAt: workbenchTerminalSourceAt(result, projection), + terminalProjectedAt: diagnostic?.projectionStatus === "caught-up" ? diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null : null, + status: projection?.status ?? result?.status ?? trace?.status ?? "unknown", + reason: workbenchProjectionReason(diagnostic), + projectionStatus: diagnostic?.projectionStatus ?? "unknown", + source: result?.agentRun ? "agentrun_v01" : "workbench_read_model" + }); +} + +function workbenchProjectionSourceLatestSeq(result = null, trace = null, projection = 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, + trace?.lastSeq, + trace?.lastEvent?.sourceSeq, + trace?.lastEvent?.seq, + trace?.eventCount, + projection?.lastProjectedSeq, + projection?.eventCount + ]; + for (const value of candidates) { + const seq = Number(value); + if (Number.isFinite(seq) && seq >= 0) return Math.floor(seq); + } + return NaN; +} + +function workbenchProjectionSourceLatestAt(result = null, trace = null, projection = null) { + return result?.traceSummary?.updatedAt + ?? result?.agentRun?.updatedAt + ?? result?.providerTrace?.updatedAt + ?? result?.runnerTrace?.updatedAt + ?? trace?.updatedAt + ?? result?.updatedAt + ?? projection?.updatedAt + ?? null; +} + +function workbenchTerminalSourceAt(result = null, projection = null) { + if (projection?.terminal !== true) return null; + return result?.traceSummary?.updatedAt + ?? result?.agentRun?.updatedAt + ?? result?.updatedAt + ?? projection?.updatedAt + ?? null; +} + +function workbenchProjectionReason(diagnostic = null) { + if (diagnostic?.blocker) return diagnostic.blocker.code ?? "projection_blocked"; + if (diagnostic?.projectionStatus === "projecting") return "projecting"; + if (diagnostic?.projectionStatus === "caught-up") return "none"; + return diagnostic?.projectionStatus ?? "unknown"; +} + function workbenchTurnDegradedReason(body = {}) { if (body?.projection?.blocker?.code) return body.projection.blocker.code; if (body?.blocker?.code) return body.blocker.code; @@ -342,6 +412,7 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur return sendJson(response, 404, body); } const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection }); + recordWorkbenchProjectionMetric(options, { result, trace, projection: turnProjection, diagnostic: projection }); const body = { ok: true, status, @@ -379,6 +450,7 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT const page = traceEventPage(trace, tracePageOptions(url)); const turnProjection = createWorkbenchTurnProjection({ traceId, result, session, trace }); const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection }); + recordWorkbenchProjectionMetric(options, { result, trace, projection: turnProjection, diagnostic: projection }); sendJson(response, 200, { ok: true, status: "succeeded",