From a4a582d0363c1f290a6412e7be632b0bac793aaa Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 6 Jul 2026 12:25:24 +0530 Subject: [PATCH 1/4] feat(harness): persist builder cost/context telemetry into metrics.json (#83, #135) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - src/core/harness/telemetry.js: shared, pure extraction of the builder contract's structured cost/context/execution fields (legacy bare-number cost tolerated, non-numeric values ignored — the contract gate is what fails validation on them), pinned cost_recorded / context_recorded event payloads, and the updateRunMetrics payload with per-stage shares split evenly across the task's canonical stages (mirrors deriveStageElapsed). - updateRunMetrics gains an atomic in-lock 'increment' path: cumulative cost/tokens/tool-calls add instead of overwrite, per-stage stage_cost_usd / stage_tokens maps accumulate per key. cumulative_tokens doubles as the 'persisted totals are authoritative' marker, so unrelated updates never stamp it onto legacy runs. - sdlc_validate now records telemetry on every validation (retries cost money too) and fixes the v2 bug where an object builder.cost was written into the cost_recorded 'usd' field. - pipeline-state rollup: cost_context.cumulative_tokens totals plus per-stage cost_usd/tokens on each stage entry (additive, schema_version stays 1); pipeline status text shows the token total when present. Co-Authored-By: Claude Fable 5 --- src/commands/pipeline.js | 4 + src/core/harness/pipeline-state.js | 20 +++ src/core/harness/run-state.js | 95 ++++++++++- src/core/harness/telemetry.js | 193 +++++++++++++++++++++ src/index.js | 1 + src/integrations/pi/rstack-sdlc.ts | 39 +++-- tests/extension-cost-context.test.js | 133 +++++++++++++++ tests/harness-telemetry.test.js | 242 +++++++++++++++++++++++++++ tests/pipeline-state.test.js | 12 ++ 9 files changed, 722 insertions(+), 17 deletions(-) create mode 100644 src/core/harness/telemetry.js create mode 100644 tests/extension-cost-context.test.js create mode 100644 tests/harness-telemetry.test.js diff --git a/src/commands/pipeline.js b/src/commands/pipeline.js index 7a23321..8910bad 100644 --- a/src/commands/pipeline.js +++ b/src/commands/pipeline.js @@ -130,6 +130,10 @@ export function formatPipelineStatus(state) { `cost $${Number(cost.cumulative_cost_usd ?? 0).toFixed(2)}`, `tool calls ${cost.cumulative_tool_calls ?? 0}`, ]; + const tokenTotal = Number(cost.cumulative_tokens?.total); + if (Number.isFinite(tokenTotal) && tokenTotal > 0) { + totals.push(`tokens ${tokenTotal}`); + } if (cost.context_tokens_used != null) { totals.push(`context ${cost.context_tokens_used}${cost.context_tokens_available != null ? `/${cost.context_tokens_available}` : ''} tokens`); } diff --git a/src/core/harness/pipeline-state.js b/src/core/harness/pipeline-state.js index 5cf530f..eefa9b7 100644 --- a/src/core/harness/pipeline-state.js +++ b/src/core/harness/pipeline-state.js @@ -213,11 +213,28 @@ function buildPipelineStatus(manifest, stages) { }; } +function normalizeTokenTotals(value) { + const source = value && typeof value === 'object' && !Array.isArray(value) ? value : {}; + const input = Number(source.input); + const output = Number(source.output); + const total = Number(source.total); + const safeInput = Number.isFinite(input) ? input : 0; + const safeOutput = Number.isFinite(output) ? output : 0; + return { + input: safeInput, + output: safeOutput, + total: Number.isFinite(total) ? total : safeInput + safeOutput, + }; +} + function buildCostContext(metrics) { return { cumulative_duration_ms: metrics.cumulative_duration_ms ?? 0, cumulative_cost_usd: metrics.cumulative_cost_usd ?? 0, cumulative_tool_calls: metrics.cumulative_tool_calls ?? 0, + // Persisted token totals (#83/#135): accumulated incrementally at validate + // time in metrics.json — zeros for legacy runs that never recorded them. + cumulative_tokens: normalizeTokenTotals(metrics.cumulative_tokens), context_tokens_used: metrics.context_tokens_used ?? null, context_tokens_available: metrics.context_tokens_available ?? null, }; @@ -343,6 +360,9 @@ export async function buildPipelineState(projectRoot, runId, { generatedAt = new task_ids: stageTaskIds, validation_status: deriveValidationStatus(stage.id, evidenceEvents), elapsed_ms: metrics.stage_elapsed_ms?.[stage.id] ?? null, + // Per-stage cost/token telemetry (#83/#135) — null when never recorded. + cost_usd: metrics.stage_cost_usd?.[stage.id] ?? null, + tokens: metrics.stage_tokens?.[stage.id] ?? null, evidence_paths: [...new Set([...artifactPaths, ...evidencePaths])], }); } diff --git a/src/core/harness/run-state.js b/src/core/harness/run-state.js index f953f4a..9fb3824 100644 --- a/src/core/harness/run-state.js +++ b/src/core/harness/run-state.js @@ -36,6 +36,74 @@ export async function prepareRunState(runDir) { return runDir; } +// Cumulative token totals shape (#83). Tolerates malformed/legacy values — +// anything non-numeric collapses to 0, a missing total is derived. +function coerceTokenCounts(value) { + const source = value && typeof value === 'object' && !Array.isArray(value) ? value : {}; + const input = Number(source.input); + const output = Number(source.output); + const total = Number(source.total); + const safeInput = Number.isFinite(input) ? input : 0; + const safeOutput = Number.isFinite(output) ? output : 0; + return { + input: safeInput, + output: safeOutput, + total: Number.isFinite(total) ? total : safeInput + safeOutput, + }; +} + +function addTokenCounts(base, delta) { + const a = coerceTokenCounts(base); + const b = coerceTokenCounts(delta); + return { input: a.input + b.input, output: a.output + b.output, total: a.total + b.total }; +} + +// Strip float dust from accumulated USD without losing sub-cent telemetry. +function roundUsd(value) { + return Math.round(value * 1e6) / 1e6; +} + +// Apply an incremental delta (#83): unlike the top-level cumulative_* fields +// (which overwrite), `increment` ADDS to the running totals — the whole +// read-modify-write already runs inside the file lock, so concurrent +// increments from parallel validations both land. +function applyMetricsIncrement(merged, increment) { + if (!increment || typeof increment !== 'object') return; + + const cost = Number(increment.cost_usd); + if (Number.isFinite(cost)) { + merged.cumulative_cost_usd = roundUsd((Number(merged.cumulative_cost_usd) || 0) + cost); + } + + const toolCalls = Number(increment.tool_calls); + if (Number.isFinite(toolCalls)) { + merged.cumulative_tool_calls = (Number(merged.cumulative_tool_calls) || 0) + toolCalls; + } + + if (increment.tokens && typeof increment.tokens === 'object') { + merged.cumulative_tokens = addTokenCounts(merged.cumulative_tokens, increment.tokens); + } + + if (increment.stage_cost_usd && typeof increment.stage_cost_usd === 'object') { + const stageCost = { ...(merged.stage_cost_usd && typeof merged.stage_cost_usd === 'object' ? merged.stage_cost_usd : {}) }; + for (const [stageId, usd] of Object.entries(increment.stage_cost_usd)) { + const delta = Number(usd); + if (!Number.isFinite(delta)) continue; + stageCost[stageId] = roundUsd((Number(stageCost[stageId]) || 0) + delta); + } + merged.stage_cost_usd = stageCost; + } + + if (increment.stage_tokens && typeof increment.stage_tokens === 'object') { + const stageTokens = { ...(merged.stage_tokens && typeof merged.stage_tokens === 'object' ? merged.stage_tokens : {}) }; + for (const [stageId, tokens] of Object.entries(increment.stage_tokens)) { + if (!tokens || typeof tokens !== 'object') continue; + stageTokens[stageId] = addTokenCounts(stageTokens[stageId], tokens); + } + merged.stage_tokens = stageTokens; + } +} + export async function updateRunMetrics(runDir, metricsUpdate = {}) { const path = join(runDir, 'metrics.json'); // Lock the whole read-modify-write: concurrent stage updates (parallel @@ -56,16 +124,31 @@ export async function updateRunMetrics(runDir, metricsUpdate = {}) { } catch { current = {}; } } + const { increment, ...update } = metricsUpdate; + const merged = { ...current, - ...metricsUpdate, - stage_elapsed_ms: { ...current.stage_elapsed_ms, ...(metricsUpdate.stage_elapsed_ms || {}) }, - stage_status: { ...current.stage_status, ...(metricsUpdate.stage_status || {}) }, + ...update, + stage_elapsed_ms: { ...current.stage_elapsed_ms, ...(update.stage_elapsed_ms || {}) }, + stage_status: { ...current.stage_status, ...(update.stage_status || {}) }, }; + // Per-stage cost/token maps (#83) merge per key like the other stage maps, + // but are only materialized when data exists — `cumulative_tokens` doubles + // as the "incremental totals are authoritative" marker for readers + // (derive.js), so unrelated updates must never stamp it onto legacy runs. + if (current.stage_cost_usd || update.stage_cost_usd) { + merged.stage_cost_usd = { ...(current.stage_cost_usd || {}), ...(update.stage_cost_usd || {}) }; + } + if (current.stage_tokens || update.stage_tokens) { + merged.stage_tokens = { ...(current.stage_tokens || {}), ...(update.stage_tokens || {}) }; + } + if (update.cumulative_tokens !== undefined) merged.cumulative_tokens = coerceTokenCounts(update.cumulative_tokens); + + if (update.cumulative_duration_ms !== undefined) merged.cumulative_duration_ms = update.cumulative_duration_ms; + if (update.cumulative_cost_usd !== undefined) merged.cumulative_cost_usd = update.cumulative_cost_usd; + if (update.cumulative_tool_calls !== undefined) merged.cumulative_tool_calls = update.cumulative_tool_calls; - if (metricsUpdate.cumulative_duration_ms !== undefined) merged.cumulative_duration_ms = metricsUpdate.cumulative_duration_ms; - if (metricsUpdate.cumulative_cost_usd !== undefined) merged.cumulative_cost_usd = metricsUpdate.cumulative_cost_usd; - if (metricsUpdate.cumulative_tool_calls !== undefined) merged.cumulative_tool_calls = metricsUpdate.cumulative_tool_calls; + applyMetricsIncrement(merged, increment); await writeJsonAtomic(path, merged); return merged; diff --git a/src/core/harness/telemetry.js b/src/core/harness/telemetry.js new file mode 100644 index 0000000..8905818 --- /dev/null +++ b/src/core/harness/telemetry.js @@ -0,0 +1,193 @@ +// owner: RStack developed by Richardson Gunde +// +// Builder-contract cost/context telemetry (#83, #135). +// +// The builder contract carries structured `cost`, `context`, and `execution` +// fields; until now they were optional decoration — tokens lived only inside +// cost_recorded events and were re-derived O(events) on every dashboard poll. +// This module is the shared extraction point: hosts (the Pi extension's +// sdlc_validate, or any framework) call extractBuilderTelemetry at validate +// time, append the pinned events from builderTelemetryEvents, and persist the +// totals incrementally via updateRunMetrics(runDir, telemetryMetricsUpdate(...)). +// +// Everything here is pure — no filesystem, no model calls. Invalid values are +// ignored here; the contract gate (validateBuilderContract's +// builder_v2_cost_values_are_numeric check) is what fails validation on +// non-numeric cost telemetry. + +function finiteNumber(value) { + if (value === null || value === undefined || value === '') return null; + const num = Number(value); + return Number.isFinite(num) ? num : null; +} + +function nonEmptyString(value) { + if (typeof value !== 'string') return null; + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : null; +} + +function plainObject(value) { + return value && typeof value === 'object' && !Array.isArray(value) ? value : null; +} + +/** + * Extract cost/context/execution telemetry from a builder contract. + * + * Returns { cost, tokens, tools_used_count, context } where every section is + * null when the contract carries no usable data for it: + * - cost: { estimated_usd, actual_usd, usd, currency } — `usd` is the + * effective spend (actual wins over estimate). Legacy contracts where + * `builder.cost` is a bare number are treated as actual spend. + * - tokens: { input, output, total } from optional cost.input_tokens / + * cost.output_tokens / cost.total_tokens (cost.tokens accepted as total). + * - tools_used_count: length of execution.tools_used (distinct tools, not + * call counts — execution.tool_calls stays the guardrail-budget signal). + * - context: { profile, workflow, injected_source_count, tokens_used, + * tokens_available } — the token gauges are the context-pressure hook (#136). + */ +export function extractBuilderTelemetry(builder) { + const telemetry = { cost: null, tokens: null, tools_used_count: null, context: null }; + if (!plainObject(builder)) return telemetry; + + const rawCost = builder.cost; + if (typeof rawCost === 'number' && Number.isFinite(rawCost)) { + // Legacy v1 shape: builder.cost is a bare USD number. + telemetry.cost = { estimated_usd: null, actual_usd: rawCost, usd: rawCost, currency: 'USD' }; + } else if (plainObject(rawCost)) { + const estimated = finiteNumber(rawCost.estimated_usd); + const actual = finiteNumber(rawCost.actual_usd); + if (estimated !== null || actual !== null) { + telemetry.cost = { + estimated_usd: estimated, + actual_usd: actual, + usd: actual ?? estimated, + currency: nonEmptyString(rawCost.currency) ?? 'USD', + }; + } + const input = finiteNumber(rawCost.input_tokens); + const output = finiteNumber(rawCost.output_tokens); + const total = finiteNumber(rawCost.total_tokens) ?? finiteNumber(rawCost.tokens); + if (input !== null || output !== null || total !== null) { + telemetry.tokens = { + input: input ?? 0, + output: output ?? 0, + total: total ?? (input ?? 0) + (output ?? 0), + }; + } + } + + if (Array.isArray(builder.execution?.tools_used)) { + telemetry.tools_used_count = builder.execution.tools_used.length; + } + + const rawContext = plainObject(builder.context); + if (rawContext) { + const profile = nonEmptyString(rawContext.profile); + const workflow = nonEmptyString(rawContext.workflow); + const injected = Array.isArray(rawContext.injected_sources) + ? rawContext.injected_sources.length + : finiteNumber(rawContext.injected_source_count); + const tokensUsed = finiteNumber(rawContext.tokens_used); + const tokensAvailable = finiteNumber(rawContext.tokens_available); + if (profile !== null || workflow !== null || injected !== null || tokensUsed !== null || tokensAvailable !== null) { + telemetry.context = { + profile, + workflow, + injected_source_count: injected, + tokens_used: tokensUsed, + tokens_available: tokensAvailable, + }; + } + } + + return telemetry; +} + +/** + * Pinned event payloads for the extracted telemetry (contract style follows + * retry_decision — downstream consumers key on these exact shapes; change + * only with a schema migration): + * + * cost_recorded: { type, task_id, usd, cost (legacy alias), estimated_usd, + * actual_usd, currency, tokens, input_tokens, output_tokens, source } + * context_recorded: { type, task_id, profile, workflow, injected_sources, + * tokens_used, tokens_available, source } + * + * Returns [] when the contract carried no cost and no context data. + */ +export function builderTelemetryEvents(taskId, telemetry) { + const events = []; + if (telemetry?.cost || telemetry?.tokens) { + const usd = telemetry.cost?.usd ?? 0; + events.push({ + type: 'cost_recorded', + task_id: taskId ?? null, + usd, + cost: usd, + estimated_usd: telemetry.cost?.estimated_usd ?? null, + actual_usd: telemetry.cost?.actual_usd ?? null, + currency: telemetry.cost?.currency ?? 'USD', + tokens: telemetry.tokens?.total ?? 0, + input_tokens: telemetry.tokens?.input ?? 0, + output_tokens: telemetry.tokens?.output ?? 0, + source: 'builder_contract', + }); + } + if (telemetry?.context) { + events.push({ + type: 'context_recorded', + task_id: taskId ?? null, + profile: telemetry.context.profile, + workflow: telemetry.context.workflow, + injected_sources: telemetry.context.injected_source_count ?? 0, + tokens_used: telemetry.context.tokens_used, + tokens_available: telemetry.context.tokens_available, + source: 'builder_contract', + }); + } + return events; +} + +/** + * updateRunMetrics payload for the extracted telemetry, or null when there is + * nothing to persist. Cost and tokens land as an `increment` (added to the + * running totals atomically in-lock); the per-stage share is split evenly + * across the task's canonical stages, mirroring deriveStageElapsed's + * multi-stage normalization so nothing is double-counted. Context token + * gauges are point-in-time values, not counters, so they overwrite. + */ +export function telemetryMetricsUpdate(telemetry, stageIds = []) { + const ids = (Array.isArray(stageIds) ? stageIds : []).filter((id) => typeof id === 'string' && id); + const increment = {}; + + const usd = telemetry?.cost?.usd ?? null; + if (usd !== null) { + increment.cost_usd = usd; + if (ids.length > 0) { + increment.stage_cost_usd = Object.fromEntries(ids.map((id) => [id, usd / ids.length])); + } + } + + if (telemetry?.tokens) { + increment.tokens = telemetry.tokens; + if (ids.length > 0) { + const share = (value) => Math.round((Number(value) || 0) / ids.length); + increment.stage_tokens = Object.fromEntries(ids.map((id) => [id, { + input: share(telemetry.tokens.input), + output: share(telemetry.tokens.output), + total: share(telemetry.tokens.total), + }])); + } + } + + const update = {}; + if (Object.keys(increment).length > 0) update.increment = increment; + if (telemetry?.context?.tokens_used !== null && telemetry?.context?.tokens_used !== undefined) { + update.context_tokens_used = telemetry.context.tokens_used; + } + if (telemetry?.context?.tokens_available !== null && telemetry?.context?.tokens_available !== undefined) { + update.context_tokens_available = telemetry.context.tokens_available; + } + return Object.keys(update).length > 0 ? update : null; +} diff --git a/src/index.js b/src/index.js index cf25837..bcbad28 100644 --- a/src/index.js +++ b/src/index.js @@ -18,6 +18,7 @@ export { validateBuilderContract, validateValidatorContract } from './core/harne export { appendEvidenceEvent, validateEvidenceEvent } from './core/harness/evidence.js'; export { DEFAULT_HARNESS_GUARDRAILS, guardrailSummary } from './core/harness/guardrails.js'; export { updateRunMetrics, createStageCheckpoint, rollbackStage, prepareRunState } from './core/harness/run-state.js'; +export { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate } from './core/harness/telemetry.js'; export { buildPipelineState, readPipelineState, summarizePipelineState, writePipelineState } from './core/harness/pipeline-state.js'; export { addDecision, decide, readDecisions, summarizeDecisions, writeDecisions } from './core/harness/decisions.js'; export { assertReadyForStage, dorCheck, readinessModeForProfile } from './core/harness/readiness.js'; diff --git a/src/integrations/pi/rstack-sdlc.ts b/src/integrations/pi/rstack-sdlc.ts index c5aba65..af7c705 100644 --- a/src/integrations/pi/rstack-sdlc.ts +++ b/src/integrations/pi/rstack-sdlc.ts @@ -15,6 +15,7 @@ import { MANIFEST_SCHEMA_VERSION, migrateManifest } from "../../core/harness/mig import { appendEvidenceEvent } from "../../core/harness/evidence.js"; import { DEFAULT_HARNESS_GUARDRAILS, guardrailSummary, loadProjectGuardrails, evaluateTaskClaim, evaluateBuilderTelemetry, guardrailEvent, isDestructiveTask } from "../../core/harness/guardrails.js"; import { classifyRetryDecision } from "../../core/harness/retry-policy.js"; +import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate } from "../../core/harness/telemetry.js"; import { VALIDATOR_CONTEXT_ENV, VALIDATOR_RUN_ID_ENV, VALIDATOR_READ_ONLY_TOOLS, evaluateValidatorAction, isValidatorContext, isValidatorRole, isValidatorSandboxDebug } from "../../core/harness/validator-sandbox.js"; import { loadValidatorRegistry, resolveValidatorProfile } from "../../core/harness/validator-registry.js"; import { budgetEnvelopeForTask, loadBudgetPolicy, loadProjectProfile } from "../../core/profiles.js"; @@ -1736,26 +1737,42 @@ export default function (pi: ExtensionAPI) { const passChecks = checks.filter((c: any) => c.status === "PASS").length; const qualityScore = checks.length > 0 ? Math.round((passChecks / checks.length) * 100) / 100 : (status === "PASS" ? 0.9 : 0.25); + // Attribute telemetry and completion to the task's canonical stage(s). + // Task ids (e.g. "007-documentation") are plan ids, not canonical stage + // ids — consumers (reporter stage aggregation, alerts, stage matrix, + // per-stage cost maps) key by canonical stage. + const canonicalStageIds = [...new Set( + (task.stage_artifacts ?? []) + .map((artifact: any) => artifact?.stage_id) + .filter((id: any) => typeof id === "string" && getCanonicalStage(id)), + )]; + if (canonicalStageIds.length === 0 && getCanonicalStage(task.id)) canonicalStageIds.push(task.id); + await appendEvent(projectRoot, manifest.run_id, { type: "task_validated", task_id: task.id, status }); for (const violation of telemetryViolations) { await appendEvent(projectRoot, manifest.run_id, guardrailEvent(task.id, violation)); } if (builderContract) { - if (builderContract.cost) { - await appendEvent(projectRoot, manifest.run_id, { type: "cost_recorded", task_id: task.id, usd: builderContract.cost, cost: builderContract.cost }); + // Cost/context telemetry (#83/#135): shared extraction from the builder + // contract's structured cost/context fields, pinned cost_recorded / + // context_recorded events, and incremental metrics.json accumulation — + // recorded on every validation (retries cost money too), so totals stop + // being re-derived O(events) on each dashboard poll. + const telemetry = extractBuilderTelemetry(builderContract); + for (const telemetryEvent of builderTelemetryEvents(task.id, telemetry)) { + await appendEvent(projectRoot, manifest.run_id, telemetryEvent); + } + const metricsUpdate = telemetryMetricsUpdate(telemetry, canonicalStageIds); + if (metricsUpdate) { + try { + await updateRunMetrics(join(runsDir(projectRoot), manifest.run_id), metricsUpdate); + } catch (metricsError) { + console.error("Failed to persist cost/context metrics:", metricsError); + } } } await appendEvent(projectRoot, manifest.run_id, { type: "quality_score_recorded", task_id: task.id, score: qualityScore, pass_checks: passChecks, total_checks: checks.length }); if (status === "PASS") { - // Attribute completion to the task's canonical stage(s). Task ids (e.g. - // "007-documentation") are plan ids, not canonical stage ids — consumers - // (reporter stage aggregation, alerts, stage matrix) key by canonical stage. - const canonicalStageIds = [...new Set( - (task.stage_artifacts ?? []) - .map((artifact: any) => artifact?.stage_id) - .filter((id: any) => typeof id === "string" && getCanonicalStage(id)), - )]; - if (canonicalStageIds.length === 0 && getCanonicalStage(task.id)) canonicalStageIds.push(task.id); if (canonicalStageIds.length === 0) { // No canonical mapping — keep the timing signal but never invent a stage id. await appendEvent(projectRoot, manifest.run_id, { type: "stage_completed", stage_id: null, task_id: task.id, elapsed_ms: elapsedMs }); diff --git a/tests/extension-cost-context.test.js b/tests/extension-cost-context.test.js new file mode 100644 index 0000000..a1ba0e2 --- /dev/null +++ b/tests/extension-cost-context.test.js @@ -0,0 +1,133 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import extension from '../extensions/rstack-sdlc.ts'; + +// owner: RStack developed by Richardson Gunde +// +// Integration coverage for #83/#135: sdlc_validate extracts cost/context +// telemetry from the builder contract, appends the pinned cost_recorded / +// context_recorded events, and accumulates the totals in metrics.json. + +function createMockPi() { + return { + tools: {}, + commands: {}, + on: () => {}, + registerTool(tool) { + this.tools[tool.name] = tool; + }, + registerCommand(cmd, opts) { + this.commands[cmd] = opts; + }, + }; +} + +function builderContract(taskId, stageIds) { + return { + task_id: taskId, + agent: 'builder', + status: 'PASS', + summary: 'Implemented telemetry smoke test with cost and context data.', + files_modified: [], + tests_run: ['npm test (telemetry smoke)'], + risks: [], + next_steps: [], + execution: { tools_used: ['read_file', 'patch'], events: [], artifacts_written: [] }, + cost: { currency: 'USD', estimated_usd: 0.5, actual_usd: 0.42, input_tokens: 12000, output_tokens: 3000 }, + context: { profile: 'business-flex', workflow: 'feature', injected_sources: ['requirements'], tokens_used: 42000, tokens_available: 200000 }, + memory_summary: { + work_done: 'Recorded telemetry smoke-test context for cost/context capture.', + decisions: ['Report cost telemetry in the builder contract.'], + evidence: ['builder.json'], + context_to_keep: [], + context_to_drop: [], + next_agent_hints: [], + }, + stage_summaries: stageIds.map((stageId) => ({ + stage_id: stageId, + agent_id: `agent.${stageId}`, + work_done: `Completed ${stageId} for the telemetry smoke test.`, + evidence: ['builder.json'], + context_to_keep: [], + context_to_drop: [], + })), + }; +} + +test('sdlc_validate persists builder cost/context telemetry into events and metrics.json', async () => { + const projectRoot = mkdtempSync(join(tmpdir(), 'rstack-telemetry-')); + const memoryRoot = mkdtempSync(join(tmpdir(), 'rstack-telemetry-mem-')); + const previousProjectRoot = process.env.RSTACK_PROJECT_ROOT; + const previousMemoryDir = process.env.RSTACK_MEMORY_DIR; + process.env.RSTACK_PROJECT_ROOT = projectRoot; + process.env.RSTACK_MEMORY_DIR = memoryRoot; + + try { + const pi = createMockPi(); + extension(pi); + const start = await pi.tools.sdlc_start.execute('start', { goal: 'Build cost telemetry smoke test', mode: 'express' }); + const runId = start.details.run_id; + await pi.tools.sdlc_plan.execute('plan', { run_id: runId }); + const build = await pi.tools.sdlc_build_next.execute('build', { run_id: runId }); + const taskId = build.details.task.id; + const stageIds = [...new Set( + (build.details.task.stage_artifacts ?? []).map((artifact) => artifact?.stage_id).filter(Boolean), + )]; + + const runDir = join(projectRoot, '.rstack', 'runs', runId); + writeFileSync( + join(runDir, 'tasks', taskId, 'builder.json'), + JSON.stringify(builderContract(taskId, stageIds), null, 2), + ); + + await pi.tools.sdlc_validate.execute('validate', { run_id: runId, task_id: taskId }); + + const events = readFileSync(join(runDir, 'events.jsonl'), 'utf8') + .split('\n').filter(Boolean).map((line) => JSON.parse(line)); + + const costEvent = events.find((event) => event.type === 'cost_recorded'); + assert.ok(costEvent, 'cost_recorded event should be appended'); + assert.equal(costEvent.task_id, taskId); + assert.equal(costEvent.usd, 0.42); + assert.equal(costEvent.cost, 0.42); + assert.equal(costEvent.estimated_usd, 0.5); + assert.equal(costEvent.actual_usd, 0.42); + assert.equal(costEvent.tokens, 15000); + assert.equal(costEvent.input_tokens, 12000); + assert.equal(costEvent.output_tokens, 3000); + assert.equal(costEvent.source, 'builder_contract'); + + const contextEvent = events.find((event) => event.type === 'context_recorded'); + assert.ok(contextEvent, 'context_recorded event should be appended'); + assert.equal(contextEvent.task_id, taskId); + assert.equal(contextEvent.profile, 'business-flex'); + assert.equal(contextEvent.workflow, 'feature'); + assert.equal(contextEvent.injected_sources, 1); + assert.equal(contextEvent.tokens_used, 42000); + assert.equal(contextEvent.tokens_available, 200000); + + const metrics = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + assert.equal(metrics.cumulative_cost_usd, 0.42); + assert.deepEqual(metrics.cumulative_tokens, { input: 12000, output: 3000, total: 15000 }); + assert.equal(metrics.context_tokens_used, 42000); + assert.equal(metrics.context_tokens_available, 200000); + if (stageIds.length > 0) { + const stageCostTotal = Object.values(metrics.stage_cost_usd ?? {}).reduce((sum, usd) => sum + usd, 0); + assert.ok(Math.abs(stageCostTotal - 0.42) < 1e-9, 'per-stage cost shares should sum to the task cost'); + const stageTokenTotal = Object.values(metrics.stage_tokens ?? {}).reduce((sum, tokens) => sum + tokens.total, 0); + // Per-stage token shares are rounded, so the sum may drift by at most + // one token per stage. + assert.ok(Math.abs(stageTokenTotal - 15000) <= stageIds.length, `stage token shares (${stageTokenTotal}) should sum to ~15000`); + } + } finally { + if (previousProjectRoot === undefined) delete process.env.RSTACK_PROJECT_ROOT; + else process.env.RSTACK_PROJECT_ROOT = previousProjectRoot; + if (previousMemoryDir === undefined) delete process.env.RSTACK_MEMORY_DIR; + else process.env.RSTACK_MEMORY_DIR = previousMemoryDir; + rmSync(projectRoot, { recursive: true, force: true }); + rmSync(memoryRoot, { recursive: true, force: true }); + } +}); diff --git a/tests/harness-telemetry.test.js b/tests/harness-telemetry.test.js new file mode 100644 index 0000000..3dd4a65 --- /dev/null +++ b/tests/harness-telemetry.test.js @@ -0,0 +1,242 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate } from '../src/core/harness/telemetry.js'; +import { updateRunMetrics } from '../src/core/harness/run-state.js'; + +// owner: RStack developed by Richardson Gunde + +// ── extractBuilderTelemetry ────────────────────────────────────────────────── + +test('extractBuilderTelemetry reads v2 cost/context/execution fields', () => { + const telemetry = extractBuilderTelemetry({ + task_id: '004-implementation', + status: 'PASS', + cost: { currency: 'USD', estimated_usd: 1.5, actual_usd: 1.25, input_tokens: 12000, output_tokens: 3000 }, + context: { profile: 'business-flex', workflow: 'feature', injected_sources: ['requirements', 'memory'], tokens_used: 42000, tokens_available: 200000 }, + execution: { tools_used: ['read_file', 'patch', 'bash'] }, + }); + + assert.deepEqual(telemetry.cost, { estimated_usd: 1.5, actual_usd: 1.25, usd: 1.25, currency: 'USD' }); + assert.deepEqual(telemetry.tokens, { input: 12000, output: 3000, total: 15000 }); + assert.equal(telemetry.tools_used_count, 3); + assert.deepEqual(telemetry.context, { + profile: 'business-flex', + workflow: 'feature', + injected_source_count: 2, + tokens_used: 42000, + tokens_available: 200000, + }); +}); + +test('extractBuilderTelemetry falls back to the estimate when actual spend is missing', () => { + const telemetry = extractBuilderTelemetry({ cost: { estimated_usd: 0.8 } }); + assert.equal(telemetry.cost.usd, 0.8); + assert.equal(telemetry.cost.actual_usd, null); +}); + +test('extractBuilderTelemetry accepts the legacy bare-number cost shape', () => { + const telemetry = extractBuilderTelemetry({ cost: 0.37 }); + assert.deepEqual(telemetry.cost, { estimated_usd: null, actual_usd: 0.37, usd: 0.37, currency: 'USD' }); + assert.equal(telemetry.tokens, null); +}); + +test('extractBuilderTelemetry ignores non-numeric cost and token values', () => { + const telemetry = extractBuilderTelemetry({ + cost: { estimated_usd: 'high', actual_usd: 'unknown', input_tokens: 'many' }, + context: { profile: '', workflow: ' ' }, + }); + assert.equal(telemetry.cost, null); + assert.equal(telemetry.tokens, null); + // empty-string profile/workflow carry no data — context stays null + assert.equal(telemetry.context, null); +}); + +test('extractBuilderTelemetry returns nulls for missing or malformed contracts', () => { + assert.deepEqual(extractBuilderTelemetry(null), { cost: null, tokens: null, tools_used_count: null, context: null }); + assert.deepEqual(extractBuilderTelemetry({}), { cost: null, tokens: null, tools_used_count: null, context: null }); + assert.deepEqual(extractBuilderTelemetry({ cost: [], context: 'nope' }).cost, null); +}); + +test('extractBuilderTelemetry derives token total when only splits are present', () => { + const telemetry = extractBuilderTelemetry({ cost: { tokens: 9000 } }); + assert.deepEqual(telemetry.tokens, { input: 0, output: 0, total: 9000 }); +}); + +// ── builderTelemetryEvents ─────────────────────────────────────────────────── + +test('builderTelemetryEvents emits pinned cost_recorded and context_recorded payloads', () => { + const telemetry = extractBuilderTelemetry({ + cost: { currency: 'USD', estimated_usd: 1.5, actual_usd: 1.25, input_tokens: 100, output_tokens: 50 }, + context: { profile: 'business-flex', workflow: 'feature', injected_sources: ['memory'] }, + }); + const events = builderTelemetryEvents('004-implementation', telemetry); + assert.equal(events.length, 2); + + assert.deepEqual(events[0], { + type: 'cost_recorded', + task_id: '004-implementation', + usd: 1.25, + cost: 1.25, + estimated_usd: 1.5, + actual_usd: 1.25, + currency: 'USD', + tokens: 150, + input_tokens: 100, + output_tokens: 50, + source: 'builder_contract', + }); + assert.deepEqual(events[1], { + type: 'context_recorded', + task_id: '004-implementation', + profile: 'business-flex', + workflow: 'feature', + injected_sources: 1, + tokens_used: null, + tokens_available: null, + source: 'builder_contract', + }); +}); + +test('builderTelemetryEvents emits nothing when the contract carries no telemetry', () => { + assert.deepEqual(builderTelemetryEvents('task', extractBuilderTelemetry({})), []); + assert.deepEqual(builderTelemetryEvents('task', null), []); +}); + +// ── telemetryMetricsUpdate ─────────────────────────────────────────────────── + +test('telemetryMetricsUpdate splits cost and tokens evenly across canonical stages', () => { + const telemetry = extractBuilderTelemetry({ + cost: { actual_usd: 0.5, input_tokens: 1000, output_tokens: 500 }, + context: { tokens_used: 42000, tokens_available: 200000 }, + }); + const update = telemetryMetricsUpdate(telemetry, ['07-code', '08-testing']); + + assert.equal(update.increment.cost_usd, 0.5); + assert.deepEqual(update.increment.stage_cost_usd, { '07-code': 0.25, '08-testing': 0.25 }); + assert.deepEqual(update.increment.tokens, { input: 1000, output: 500, total: 1500 }); + assert.deepEqual(update.increment.stage_tokens, { + '07-code': { input: 500, output: 250, total: 750 }, + '08-testing': { input: 500, output: 250, total: 750 }, + }); + assert.equal(update.context_tokens_used, 42000); + assert.equal(update.context_tokens_available, 200000); +}); + +test('telemetryMetricsUpdate returns null when there is nothing to persist', () => { + assert.equal(telemetryMetricsUpdate(extractBuilderTelemetry({}), ['07-code']), null); + assert.equal(telemetryMetricsUpdate(null, []), null); +}); + +test('telemetryMetricsUpdate omits stage maps when the task has no canonical stages', () => { + const update = telemetryMetricsUpdate(extractBuilderTelemetry({ cost: { actual_usd: 0.2 } }), []); + assert.equal(update.increment.cost_usd, 0.2); + assert.equal(update.increment.stage_cost_usd, undefined); +}); + +// ── updateRunMetrics increments (#83) ──────────────────────────────────────── + +test('updateRunMetrics accumulates cost/token increments across calls', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-incr-')); + try { + await updateRunMetrics(runDir, { + increment: { + cost_usd: 0.25, + tokens: { input: 1000, output: 200, total: 1200 }, + stage_cost_usd: { '07-code': 0.25 }, + stage_tokens: { '07-code': { input: 1000, output: 200, total: 1200 } }, + }, + }); + const second = await updateRunMetrics(runDir, { + increment: { + cost_usd: 0.15, + tokens: { input: 500, output: 100, total: 600 }, + stage_cost_usd: { '07-code': 0.05, '08-testing': 0.1 }, + stage_tokens: { '08-testing': { input: 500, output: 100, total: 600 } }, + }, + }); + + assert.equal(second.cumulative_cost_usd, 0.4); + assert.deepEqual(second.cumulative_tokens, { input: 1500, output: 300, total: 1800 }); + assert.deepEqual(second.stage_cost_usd, { '07-code': 0.3, '08-testing': 0.1 }); + assert.deepEqual(second.stage_tokens, { + '07-code': { input: 1000, output: 200, total: 1200 }, + '08-testing': { input: 500, output: 100, total: 600 }, + }); + + // Increments land on disk, not just in the returned object. + const persisted = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + assert.equal(persisted.cumulative_cost_usd, 0.4); + assert.deepEqual(persisted.cumulative_tokens, { input: 1500, output: 300, total: 1800 }); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('concurrent metric increments both land under the file lock', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-race-')); + try { + await Promise.all([ + updateRunMetrics(runDir, { increment: { cost_usd: 0.1, tokens: { input: 100, output: 10, total: 110 } } }), + updateRunMetrics(runDir, { increment: { cost_usd: 0.2, tokens: { input: 200, output: 20, total: 220 } } }), + ]); + const persisted = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + assert.equal(persisted.cumulative_cost_usd, 0.3); + assert.deepEqual(persisted.cumulative_tokens, { input: 300, output: 30, total: 330 }); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('updateRunMetrics keeps overwrite semantics for cumulative fields and never stamps token fields onto legacy runs', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-legacy-')); + try { + // Legacy-style metrics.json: no cumulative_tokens, no stage maps. + writeFileSync(join(runDir, 'metrics.json'), JSON.stringify({ + cumulative_duration_ms: 1000, + cumulative_cost_usd: 0.02, + cumulative_tool_calls: 5, + stage_elapsed_ms: {}, + stage_status: { '00-environment': 'PASS' }, + })); + + // An unrelated stage-status update (goal-loop reset, harvest) must not + // materialize cumulative_tokens — its presence is the "persisted totals + // are authoritative" marker for derive.js. + const updated = await updateRunMetrics(runDir, { stage_status: { '01-transcript': 'PASS' } }); + assert.equal('cumulative_tokens' in updated, false); + assert.equal('stage_cost_usd' in updated, false); + assert.equal('stage_tokens' in updated, false); + assert.equal(updated.stage_status['00-environment'], 'PASS'); + assert.equal(updated.stage_status['01-transcript'], 'PASS'); + + // Explicit cumulative_* values still overwrite (pre-#83 behavior). + const overwritten = await updateRunMetrics(runDir, { cumulative_cost_usd: 0.5, cumulative_tokens: { input: 10, output: 5 } }); + assert.equal(overwritten.cumulative_cost_usd, 0.5); + assert.deepEqual(overwritten.cumulative_tokens, { input: 10, output: 5, total: 15 }); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('updateRunMetrics tolerates malformed increments and token shapes', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-malformed-')); + try { + const result = await updateRunMetrics(runDir, { + increment: { + cost_usd: 'not-a-number', + tokens: { input: 'garbage', output: 7 }, + stage_cost_usd: { '07-code': 'NaN' }, + stage_tokens: { '07-code': null }, + }, + }); + assert.equal(result.cumulative_cost_usd, 0); + assert.deepEqual(result.cumulative_tokens, { input: 0, output: 7, total: 7 }); + assert.deepEqual(result.stage_cost_usd ?? {}, {}); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); diff --git a/tests/pipeline-state.test.js b/tests/pipeline-state.test.js index b3699b5..f2398e7 100644 --- a/tests/pipeline-state.test.js +++ b/tests/pipeline-state.test.js @@ -63,6 +63,7 @@ test('buildPipelineState derives status from canonical run files', async () => { cumulative_duration_ms: 1200, cumulative_cost_usd: 0.42, cumulative_tool_calls: 7, + cumulative_tokens: { input: 12000, output: 3000, total: 15000 }, context_tokens_used: 1000, context_tokens_available: 3000, stage_status: { @@ -72,6 +73,12 @@ test('buildPipelineState derives status from canonical run files', async () => { stage_elapsed_ms: { '07-code': 900, }, + stage_cost_usd: { + '07-code': 0.3, + }, + stage_tokens: { + '07-code': { input: 8000, output: 2000, total: 10000 }, + }, }); await writeJson(path.join(dir, 'approvals.json'), [ { artifact: 'plan.md', status: 'PENDING', stage_id: '04-planning' }, @@ -116,9 +123,14 @@ test('buildPipelineState derives status from canonical run files', async () => { cumulative_duration_ms: 1200, cumulative_cost_usd: 0.42, cumulative_tool_calls: 7, + cumulative_tokens: { input: 12000, output: 3000, total: 15000 }, context_tokens_used: 1000, context_tokens_available: 3000, }); + assert.equal(codeStage.cost_usd, 0.3); + assert.deepEqual(codeStage.tokens, { input: 8000, output: 2000, total: 10000 }); + assert.equal(state.stages.find((stage) => stage.id === '08-testing').cost_usd, null); + assert.equal(state.stages.find((stage) => stage.id === '08-testing').tokens, null); assert.deepEqual(summarizePipelineState(state), { run_id: runId, From 821dc860e9b31ff664d50d929bed139db66f4fdc Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 6 Jul 2026 12:28:10 +0530 Subject: [PATCH 2/4] feat(observability): prefer persisted cumulative metrics on the read path (#83) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - derive.js: persistedTokenTotals reads the incremental metrics shape (cumulative_tokens object = authoritative marker); resolveRunTotals overlays persisted cost/token totals on the event-derived dimensions and falls back to full event recompute for legacy runs — so dashboard polls stop paying O(events) for numbers that are already on disk, and totals survive event rotation/archival. - dashboard state: runs.js resolves totals through metrics.json; the rollup index no longer collapses object-shaped cumulative_tokens to 0 and carries the per-stage cost/token maps into index-served lite runs; client state surfaces stageCost/stageTokens for Run Analytics (data flow now, UI rendering later); context_recorded excluded from notable events like its cost_recorded twin. - reporter: buildRunReport cost summary prefers persisted metrics (tagged source: metrics|events); context_recorded added to known event types. - alerts engine: plain-language feed line for context_recorded. Co-Authored-By: Claude Fable 5 --- src/observability/alerts/engine.js | 10 ++ src/observability/collectors/reporter.js | 26 +++- .../dashboard/state/client-state.js | 4 + src/observability/dashboard/state/index.js | 10 +- .../dashboard/state/rollup-index.js | 17 ++- src/observability/dashboard/state/runs.js | 6 +- src/observability/metrics/derive.js | 39 +++++ tests/observability-cost-metrics.test.js | 134 ++++++++++++++++++ 8 files changed, 231 insertions(+), 15 deletions(-) create mode 100644 tests/observability-cost-metrics.test.js diff --git a/src/observability/alerts/engine.js b/src/observability/alerts/engine.js index b821320..a4d7238 100644 --- a/src/observability/alerts/engine.js +++ b/src/observability/alerts/engine.js @@ -140,6 +140,16 @@ export function plainLanguageSummary(event) { const cost = Number(rawCost); return `Cost recorded: $${Number.isFinite(cost) ? cost.toFixed(4) : '0.0000'}`; } + case 'context_recorded': { + const parts = []; + if (event.profile) parts.push(`profile ${event.profile}`); + if (event.workflow) parts.push(`workflow ${event.workflow}`); + parts.push(`${Number(event.injected_sources) || 0} injected source(s)`); + if (event.tokens_used != null) { + parts.push(`${event.tokens_used}${event.tokens_available != null ? `/${event.tokens_available}` : ''} context tokens`); + } + return `Context recorded — ${parts.join(', ')}`; + } case 'guardrail_triggered': return `🛡 Guardrail blocked ${event.task_id ?? 'task'} — ${event.reason ?? event.limit_name ?? event.limit ?? 'budget exceeded'}`; case 'guardrail_overridden': diff --git a/src/observability/collectors/reporter.js b/src/observability/collectors/reporter.js index aa393fa..03d723f 100644 --- a/src/observability/collectors/reporter.js +++ b/src/observability/collectors/reporter.js @@ -1,6 +1,7 @@ import { readFile } from 'node:fs/promises'; import { existsSync } from 'node:fs'; import { join } from 'node:path'; +import { persistedTokenTotals } from '../metrics/derive.js'; // owner: RStack developed by Richardson Gunde @@ -8,7 +9,7 @@ import { join } from 'node:path'; const KNOWN_EVENT_TYPES = [ 'run_started', 'task_started', 'builder_task_prepared', 'task_validated', 'stage_completed', 'approval_gate', 'approval_gate_blocked', 'guardrail_triggered', - 'tool_call', 'tool_result', 'cost_recorded', 'quality_score_recorded', + 'tool_call', 'tool_result', 'cost_recorded', 'context_recorded', 'quality_score_recorded', 'memory_recalled', 'memory_pruned', 'episode_memory_written', 'episode_memory_write_failed', 'session_shutdown', 'clarification_requested', 'clarification_answers_added', 'plan_created', 'validation_failed', @@ -153,13 +154,24 @@ export async function buildRunReport(runDir) { guardrailSummary[k] = (guardrailSummary[k] ?? 0) + 1; } - // Cost summary + // Cost summary — persisted cumulative metrics win (#83): O(1), and they + // survive event rotation. Legacy runs recompute from cost_recorded events. + const metrics = await readJson(join(runDir, 'metrics.json'), {}); const costEvents = events.filter((ev) => ev.type === 'cost_recorded'); - const costSummary = { - total_usd: costEvents.reduce((s, ev) => s + (Number(ev.usd ?? ev.cost ?? 0) || 0), 0), - total_tokens: costEvents.reduce((s, ev) => s + (Number(ev.tokens) || 0), 0), - entries: costEvents, - }; + const persistedTokens = persistedTokenTotals(metrics); + const costSummary = persistedTokens + ? { + total_usd: Number(metrics.cumulative_cost_usd) || 0, + total_tokens: persistedTokens.total, + source: 'metrics', + entries: costEvents, + } + : { + total_usd: costEvents.reduce((s, ev) => s + (Number(ev.usd ?? ev.cost ?? 0) || 0), 0), + total_tokens: costEvents.reduce((s, ev) => s + (Number(ev.tokens) || 0), 0), + source: 'events', + entries: costEvents, + }; // Duration from first/last event timestamps let durationMs = 0; diff --git a/src/observability/dashboard/state/client-state.js b/src/observability/dashboard/state/client-state.js index fb908d8..1432466 100644 --- a/src/observability/dashboard/state/client-state.js +++ b/src/observability/dashboard/state/client-state.js @@ -18,6 +18,10 @@ export function toClientState(state) { timeline: (run.timeline ?? []).slice(0, 120), totals: run.totals ?? null, stageElapsed: run.stageElapsed ?? {}, + // Per-stage cost/token telemetry (#83/#135): the data flows to Run + // Analytics now; dedicated UI rendering is follow-up work. + stageCost: run.metrics?.stage_cost_usd ?? {}, + stageTokens: run.metrics?.stage_tokens ?? {}, approvals: (run.approvals ?? []).slice(0, 40).map((approval) => ({ artifact: approval.artifact, status: approval.status, diff --git a/src/observability/dashboard/state/index.js b/src/observability/dashboard/state/index.js index 4600b35..7d10fc8 100644 --- a/src/observability/dashboard/state/index.js +++ b/src/observability/dashboard/state/index.js @@ -9,7 +9,7 @@ import { buildAgentGroups, buildAgentWork } from './agent-work.js'; import { buildTraceMap } from './traceability.js'; import { buildProjectSummaries } from './projects.js'; import { buildDiagnostics, buildLayerSummaries } from './layers.js'; -import { buildStageTrends } from '../../metrics/derive.js'; +import { buildStageTrends, persistedTokenTotals } from '../../metrics/derive.js'; import { buildPeople, buildPresence } from './people.js'; import { buildBusinessFlexState } from './business-flex.js'; import { buildDecisionState } from './decisions.js'; @@ -33,9 +33,13 @@ export async function buildFullState(projectRoot, options = {}) { getAllApprovals(roots), ]); - // Prefer event-derived totals (single source of truth); fall back to metrics.json. + // run.totals already prefers persisted cumulative metrics (#83) and falls + // back to event recompute for legacy runs; the metrics.json chain here is + // the last resort for index entries without totals. const totalCost = runs.reduce((sum, run) => sum + (run.totals?.cost_usd || run.metrics?.cumulative_cost_usd || 0), 0); - const tokenTotal = runs.reduce((sum, run) => sum + (run.totals?.tokens || Number(run.metrics?.cumulative_tokens ?? run.metrics?.total_tokens ?? 0)), 0); + const tokenTotal = runs.reduce((sum, run) => sum + (run.totals?.tokens + || persistedTokenTotals(run.metrics)?.total + || Number(run.metrics?.cumulative_tokens ?? run.metrics?.total_tokens ?? 0) || 0), 0); const activeRuns = runs.filter((run) => run.derivedStatus === 'active'); const today = new Date().toISOString().slice(0, 10); const todayRuns = runs.filter((run) => run.manifest?.created_at?.startsWith(today)); diff --git a/src/observability/dashboard/state/rollup-index.js b/src/observability/dashboard/state/rollup-index.js index e7fbd77..f6ca16b 100644 --- a/src/observability/dashboard/state/rollup-index.js +++ b/src/observability/dashboard/state/rollup-index.js @@ -28,13 +28,14 @@ import { join } from 'node:path'; import { mkdir, readFile, readdir, rename, stat, writeFile } from 'node:fs/promises'; import { getRunsForRoot } from './runs.js'; import { safeJson } from './files.js'; +import { persistedTokenTotals } from '../../metrics/derive.js'; export const INDEX_VERSION = 1; export const DEFAULT_RETENTION_DAYS = 90; const STALL_MS = 30 * 60 * 1000; // High-volume event types excluded from the per-run notable_events rollup. -const NOTABLE_EXCLUDE = new Set(['tool_call', 'tool_result', 'cost_recorded']); +const NOTABLE_EXCLUDE = new Set(['tool_call', 'tool_result', 'cost_recorded', 'context_recorded']); const NOTABLE_CAP = 200; function defaultIo() { @@ -94,7 +95,10 @@ export function entryFromRun(run, sig = null) { if (stageId) stageStatuses[stageId] = task.status ?? 'READY'; } const metricCost = run.metrics?.cumulative_cost_usd ?? 0; - const metricTokens = Number(run.metrics?.cumulative_tokens ?? run.metrics?.total_tokens ?? 0) || 0; + // cumulative_tokens is an { input, output, total } object on runs written by + // the incremental telemetry path (#83); tolerate the legacy bare number too. + const metricTokens = persistedTokenTotals(run.metrics)?.total + ?? (Number(run.metrics?.cumulative_tokens ?? run.metrics?.total_tokens ?? 0) || 0); return { runId: run.runId, status: run.derivedStatus ?? 'idle', @@ -108,7 +112,14 @@ export function entryFromRun(run, sig = null) { started_by: run.manifest?.started_by ?? null, has_plan: run.hasPlan ?? false, brief: (run.brief ?? '').slice(0, 300), - metrics: { cumulative_cost_usd: metricCost, cumulative_tokens: metricTokens }, + metrics: { + cumulative_cost_usd: metricCost, + cumulative_tokens: run.metrics?.cumulative_tokens ?? metricTokens, + // Per-stage telemetry maps (#83/#135) survive into index-served lite + // runs so Run Analytics keeps its data without a full re-parse. + stage_cost_usd: run.metrics?.stage_cost_usd ?? {}, + stage_tokens: run.metrics?.stage_tokens ?? {}, + }, totals: run.totals ?? null, cost_usd: run.totals?.cost_usd || metricCost || 0, tokens: run.totals?.tokens || metricTokens || 0, diff --git a/src/observability/dashboard/state/runs.js b/src/observability/dashboard/state/runs.js index 151fb5e..71f8bea 100644 --- a/src/observability/dashboard/state/runs.js +++ b/src/observability/dashboard/state/runs.js @@ -3,7 +3,7 @@ import { readFile, readdir, stat } from 'node:fs/promises'; import { join } from 'node:path'; import { CANONICAL_SDLC_STAGES } from '../../../core/harness/stages.js'; import { cleanOrphanedTmpFiles } from '../../../core/harness/safe-write.js'; -import { deriveRunTimeline, deriveRunTotals, deriveStageElapsed } from '../../metrics/derive.js'; +import { deriveRunTimeline, deriveStageElapsed, resolveRunTotals } from '../../metrics/derive.js'; import { stageReportIndex } from './stage-reports.js'; import { readJson, readJsonTracked, readJsonlTracked } from './files.js'; @@ -189,7 +189,9 @@ export async function getRunsForRoot(projectRoot, options = {}) { stageReports: await stageReportIndex(runDir), activityTimeline: buildActivityTimeline(events), timeline: deriveRunTimeline(events, rawTasks), - totals: deriveRunTotals(events), + // Persisted cumulative metrics win over event recompute (#83); legacy + // runs without them still derive totals from events. + totals: resolveRunTotals(events, metrics), stageElapsed: deriveStageElapsed(events, rawTasks), derivedStatus: deriveRunStatus(manifest, events), integrity, diff --git a/src/observability/metrics/derive.js b/src/observability/metrics/derive.js index 5dd7566..2b5196d 100644 --- a/src/observability/metrics/derive.js +++ b/src/observability/metrics/derive.js @@ -136,6 +136,45 @@ export function deriveRunTimeline(events, tasks) { return segments.sort((a, b) => (parseTs(a.started_at) ?? Infinity) - (parseTs(b.started_at) ?? Infinity)); } +/** + * Persisted token totals from metrics.json (#83), or null for legacy runs. + * The presence of a `cumulative_tokens` object is the marker that the run was + * written by the incremental telemetry path — only that path (or an explicit + * caller) ever materializes it, so legacy runs keep falling back to event + * recompute even after unrelated metrics updates touch the file. + */ +export function persistedTokenTotals(metrics) { + const tokens = metrics?.cumulative_tokens; + if (!tokens || typeof tokens !== 'object' || Array.isArray(tokens)) return null; + const input = Number(tokens.input); + const output = Number(tokens.output); + const total = Number(tokens.total); + const safeInput = Number.isFinite(input) ? input : 0; + const safeOutput = Number.isFinite(output) ? output : 0; + return { + input: safeInput, + output: safeOutput, + total: Number.isFinite(total) ? total : safeInput + safeOutput, + }; +} + +/** + * Run totals, preferring persisted cumulative metrics (#83): when metrics.json + * carries incremental cost/token totals they are authoritative — O(1) instead + * of re-parsing the event stream, and they survive event rotation/archival. + * Legacy runs (no persisted totals) recompute from events exactly as before. + * Duration, task outcomes, guardrails, and quality always come from events. + */ +export function resolveRunTotals(events, metrics = {}) { + const totals = deriveRunTotals(events); + const persisted = persistedTokenTotals(metrics); + if (persisted) { + totals.tokens = persisted.total; + totals.cost_usd = Math.round((Number(metrics.cumulative_cost_usd) || 0) * 10000) / 10000; + } + return totals; +} + /** Whole-run totals from the event stream. */ export function deriveRunTotals(events) { const totals = { diff --git a/tests/observability-cost-metrics.test.js b/tests/observability-cost-metrics.test.js new file mode 100644 index 0000000..ea05147 --- /dev/null +++ b/tests/observability-cost-metrics.test.js @@ -0,0 +1,134 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { deriveRunTotals, persistedTokenTotals, resolveRunTotals } from '../src/observability/metrics/derive.js'; +import { entryFromRun } from '../src/observability/dashboard/state/rollup-index.js'; +import { toClientState } from '../src/observability/dashboard/state/client-state.js'; +import { buildRunReport } from '../src/observability/collectors/reporter.js'; + +// owner: RStack developed by Richardson Gunde +// +// Read-path coverage for #83: persisted cumulative metrics are preferred, +// legacy runs without token fields still render from event recompute. + +const COST_EVENTS = [ + { type: 'run_started', ts: '2026-07-01T05:00:00.000Z' }, + { type: 'cost_recorded', usd: 0.25, tokens: 12000, ts: '2026-07-01T05:01:00.000Z' }, + { type: 'cost_recorded', usd: 0.15, tokens: 8000, ts: '2026-07-01T05:02:00.000Z' }, +]; + +test('persistedTokenTotals reads the incremental metrics shape and rejects legacy values', () => { + assert.deepEqual( + persistedTokenTotals({ cumulative_tokens: { input: 100, output: 50, total: 150 } }), + { input: 100, output: 50, total: 150 }, + ); + // total derived when missing; malformed members collapse to 0 + assert.deepEqual( + persistedTokenTotals({ cumulative_tokens: { input: 100, output: 'junk' } }), + { input: 100, output: 0, total: 100 }, + ); + assert.equal(persistedTokenTotals({}), null); + assert.equal(persistedTokenTotals({ cumulative_tokens: 20000 }), null); + assert.equal(persistedTokenTotals(null), null); +}); + +test('resolveRunTotals prefers persisted cumulative metrics over event recompute', () => { + const totals = resolveRunTotals(COST_EVENTS, { + cumulative_cost_usd: 0.9, + cumulative_tokens: { input: 40000, output: 10000, total: 50000 }, + }); + assert.equal(totals.cost_usd, 0.9); + assert.equal(totals.tokens, 50000); + // Event-derived dimensions are untouched. + assert.equal(totals.duration_ms, 2 * 60 * 1000); +}); + +test('resolveRunTotals falls back to event recompute for legacy runs', () => { + const legacyMetrics = { cumulative_cost_usd: 0, stage_status: { '07-code': 'PASS' } }; + const totals = resolveRunTotals(COST_EVENTS, legacyMetrics); + assert.equal(totals.cost_usd, 0.4); + assert.equal(totals.tokens, 20000); + assert.deepEqual(totals, deriveRunTotals(COST_EVENTS)); +}); + +test('rollup index entry carries object token totals and per-stage telemetry maps', () => { + const run = { + runId: 'run-telemetry', + manifest: { created_at: '2026-07-01T05:00:00.000Z' }, + derivedStatus: 'active', + metrics: { + cumulative_cost_usd: 0.42, + cumulative_tokens: { input: 12000, output: 3000, total: 15000 }, + stage_cost_usd: { '07-code': 0.42 }, + stage_tokens: { '07-code': { input: 12000, output: 3000, total: 15000 } }, + }, + totals: null, + tasks: [], + events: [{ type: 'context_recorded', ts: '2026-07-01T05:01:00.000Z' }], + }; + const entry = entryFromRun(run); + assert.equal(entry.tokens, 15000, 'object-shaped cumulative_tokens must not collapse to 0'); + assert.equal(entry.cost_usd, 0.42); + assert.deepEqual(entry.metrics.stage_cost_usd, { '07-code': 0.42 }); + assert.deepEqual(entry.metrics.stage_tokens['07-code'], { input: 12000, output: 3000, total: 15000 }); + // context_recorded is high-volume telemetry — excluded from notable events. + assert.deepEqual(entry.notable_events, []); +}); + +test('client state surfaces per-stage cost/tokens for Run Analytics', () => { + const state = toClientState({ + runs: [{ + runId: 'run-telemetry', + metrics: { + stage_cost_usd: { '07-code': 0.42 }, + stage_tokens: { '07-code': { input: 12000, output: 3000, total: 15000 } }, + }, + events: [], + evidence: [], + tasks: [], + }], + }); + assert.deepEqual(state.runs[0].stageCost, { '07-code': 0.42 }); + assert.deepEqual(state.runs[0].stageTokens, { '07-code': { input: 12000, output: 3000, total: 15000 } }); + // Legacy runs without the maps degrade to empty objects, not crashes. + const legacy = toClientState({ runs: [{ runId: 'legacy', metrics: {}, events: [], evidence: [], tasks: [] }] }); + assert.deepEqual(legacy.runs[0].stageCost, {}); + assert.deepEqual(legacy.runs[0].stageTokens, {}); +}); + +test('buildRunReport cost summary prefers persisted metrics and falls back to events', async () => { + const baseDir = mkdtempSync(join(tmpdir(), 'rstack-report-metrics-')); + try { + const eventsJsonl = `${COST_EVENTS.map((event) => JSON.stringify(event)).join('\n')}\n`; + + const persistedRun = join(baseDir, 'run-persisted'); + mkdirSync(persistedRun, { recursive: true }); + writeFileSync(join(persistedRun, 'manifest.json'), JSON.stringify({ run_id: 'run-persisted' })); + writeFileSync(join(persistedRun, 'events.jsonl'), eventsJsonl); + writeFileSync(join(persistedRun, 'tasks.json'), JSON.stringify({ tasks: [] })); + writeFileSync(join(persistedRun, 'metrics.json'), JSON.stringify({ + cumulative_cost_usd: 0.9, + cumulative_tokens: { input: 40000, output: 10000, total: 50000 }, + })); + const persistedReport = await buildRunReport(persistedRun); + assert.equal(persistedReport.cost_summary.total_usd, 0.9); + assert.equal(persistedReport.cost_summary.total_tokens, 50000); + assert.equal(persistedReport.cost_summary.source, 'metrics'); + assert.equal(persistedReport.cost_summary.entries.length, 2); + + const legacyRun = join(baseDir, 'run-legacy'); + mkdirSync(legacyRun, { recursive: true }); + writeFileSync(join(legacyRun, 'manifest.json'), JSON.stringify({ run_id: 'run-legacy' })); + writeFileSync(join(legacyRun, 'events.jsonl'), eventsJsonl); + writeFileSync(join(legacyRun, 'tasks.json'), JSON.stringify({ tasks: [] })); + const legacyReport = await buildRunReport(legacyRun); + assert.equal(legacyReport.cost_summary.total_usd, 0.4); + assert.equal(legacyReport.cost_summary.total_tokens, 20000); + assert.equal(legacyReport.cost_summary.source, 'events'); + } finally { + rmSync(baseDir, { recursive: true, force: true }); + } +}); From 7b5115955f451043269151ff444aad122bf9d6ff Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 6 Jul 2026 12:28:41 +0530 Subject: [PATCH 3/4] docs(harness): document the full metrics.json schema and telemetry event contracts (#83, #135) New 'Run metrics' section only: schema, overwrite-vs-increment write semantics, the cumulative_tokens authoritative-marker rule for readers, per-stage splitting, and the pinned cost_recorded / context_recorded event shapes. Co-Authored-By: Claude Fable 5 --- docs/HARNESS.md | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/docs/HARNESS.md b/docs/HARNESS.md index ced3b33..00801b3 100644 --- a/docs/HARNESS.md +++ b/docs/HARNESS.md @@ -92,6 +92,42 @@ Raw runtime events are appended to `events.jsonl`. Validator-grounded task evide `src/core/harness/evidence.js` rejects missing `task_id`, `kind`, `status`, or `evidence` fields. +## Run metrics (metrics.json) + +`/metrics.json` is the persisted cost/duration/token rollup for a run (#83, #135). It is written by `updateRunMetrics` (`src/core/harness/run-state.js`) under a file lock with atomic tmp+rename, so concurrent writers both land. Full schema: + +```json +{ + "cumulative_duration_ms": 0, + "cumulative_cost_usd": 0, + "cumulative_tool_calls": 0, + "cumulative_tokens": { "input": 0, "output": 0, "total": 0 }, + "stage_elapsed_ms": { "07-code": 900 }, + "stage_status": { "07-code": "PASS" }, + "stage_cost_usd": { "07-code": 0.42 }, + "stage_tokens": { "07-code": { "input": 12000, "output": 3000, "total": 15000 } }, + "context_tokens_used": null, + "context_tokens_available": null +} +``` + +All fields are additive-tolerant — readers default anything missing, so legacy files need no migration and the file carries no `schema_version`. `cumulative_tokens` doubles as the marker that the run was written by the incremental telemetry path: readers (`resolveRunTotals` in `src/observability/metrics/derive.js`, the reporter's cost summary) treat persisted totals as authoritative when the object is present and recompute from `cost_recorded` events otherwise, so legacy runs still render. Unrelated metrics updates never materialize the marker on legacy files. + +Write semantics: + +- Top-level `cumulative_*` values and the stage maps passed directly to `updateRunMetrics` **overwrite/merge-per-key** (pre-#83 behavior, unchanged). +- An `increment` block **adds** deltas atomically in-lock: `cost_usd`, `tool_calls`, `tokens {input, output, total}`, and per-stage `stage_cost_usd` / `stage_tokens` maps. This is how `cost_recorded` telemetry updates totals incrementally instead of being re-derived O(events) per dashboard poll. +- `context_tokens_used` / `context_tokens_available` are point-in-time gauges (the context-pressure hook for BLE-6.2), so they overwrite. + +Telemetry source: at validate time `sdlc_validate` extracts the builder contract's structured `cost` / `context` / `execution` fields via `extractBuilderTelemetry` (`src/core/harness/telemetry.js`), on **every** validation — retries cost money too. Non-numeric cost values are ignored by extraction; the contract gate's `builder_v2_cost_values_are_numeric` check is what fails validation on them. Cost and tokens are split evenly across the task's canonical stages (the same normalization as stage elapsed), so multi-stage tasks are never double-counted. + +Events (pinned contract, same rules as `retry_decision` — downstream consumers key on these exact shapes): + +- `cost_recorded` — `task_id`, `usd` (effective spend: `actual_usd` wins over `estimated_usd`; `cost` kept as a legacy alias), `estimated_usd`, `actual_usd`, `currency`, `tokens`, `input_tokens`, `output_tokens`, `source: "builder_contract"`. +- `context_recorded` — `task_id`, `profile`, `workflow`, `injected_sources` (count), `tokens_used`, `tokens_available`, `source: "builder_contract"`. + +Rollups: the pipeline-state `cost_context` block carries `cumulative_tokens` alongside the existing cost/duration/tool-call totals, and each stage entry carries its `cost_usd` / `tokens` share (`null` when never recorded). `rstack-agents pipeline status` prints the token total; `--json` exposes the full structure. + ## Agent episodic memory Validator-approved tasks are written to an agent/stage scoped episodic memory store by `src/memory/index.js`. From 0f67d437851ca4ab8cf017963ac3261451ce1f52 Mon Sep 17 00:00:00 2001 From: Richardson Gunde Date: Mon, 6 Jul 2026 14:36:52 +0530 Subject: [PATCH 4/4] =?UTF-8?q?fix(telemetry):=20apply=20PR=20#199=20revie?= =?UTF-8?q?w=20findings=20=E2=80=94=20idempotency,=20drift=20visibility,?= =?UTF-8?q?=20history,=20tool=20calls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit F1 (BLOCKING) double-count: increments now carry an idempotency key = the builder contract's canonical content hash; a replayed/re-validated identical contract is a no-op, while a genuine retry (new contract content) counts again. goal-loop resets clear the prior attempt's builder.json so a stale contract can't be replayed and re-costed. Keeps the loop budget cap enforcing on real spend, not phantom spend. F2: a failed persisted increment after its cost_recorded event landed now emits a metrics_write_failed event; resolveRunTotals detects the drift and falls back to event recompute instead of reporting a total it knows is stale. F3: the first increment that materializes cumulative_tokens on a run with prior cost_recorded history seeds totals from an event recompute — no dropped history. F4: cumulative_tool_calls is now fed by execution.tool_calls (invocations), distinct from tools_used_count (distinct names); docs match code. Co-Authored-By: Claude Opus 4.8 --- docs/HARNESS.md | 12 +- src/core/harness/goal-loop.js | 10 +- src/core/harness/run-state.js | 42 ++++++- src/core/harness/telemetry.js | 69 ++++++++++- src/integrations/pi/rstack-sdlc.ts | 62 ++++++++-- src/observability/metrics/derive.js | 22 +++- tests/harness-telemetry.test.js | 151 ++++++++++++++++++++++- tests/observability-cost-metrics.test.js | 20 ++- tests/pipeline-loop.test.js | 27 ++++ 9 files changed, 393 insertions(+), 22 deletions(-) diff --git a/docs/HARNESS.md b/docs/HARNESS.md index 0d89db1..eeaf76c 100644 --- a/docs/HARNESS.md +++ b/docs/HARNESS.md @@ -107,7 +107,8 @@ Raw runtime events are appended to `events.jsonl`. Validator-grounded task evide "stage_cost_usd": { "07-code": 0.42 }, "stage_tokens": { "07-code": { "input": 12000, "output": 3000, "total": 15000 } }, "context_tokens_used": null, - "context_tokens_available": null + "context_tokens_available": null, + "applied_telemetry_keys": [""] } ``` @@ -119,12 +120,19 @@ Write semantics: - An `increment` block **adds** deltas atomically in-lock: `cost_usd`, `tool_calls`, `tokens {input, output, total}`, and per-stage `stage_cost_usd` / `stage_tokens` maps. This is how `cost_recorded` telemetry updates totals incrementally instead of being re-derived O(events) per dashboard poll. - `context_tokens_used` / `context_tokens_available` are point-in-time gauges (the context-pressure hook for BLE-6.2), so they overwrite. -Telemetry source: at validate time `sdlc_validate` extracts the builder contract's structured `cost` / `context` / `execution` fields via `extractBuilderTelemetry` (`src/core/harness/telemetry.js`), on **every** validation — retries cost money too. Non-numeric cost values are ignored by extraction; the contract gate's `builder_v2_cost_values_are_numeric` check is what fails validation on them. Cost and tokens are split evenly across the task's canonical stages (the same normalization as stage elapsed), so multi-stage tasks are never double-counted. +Idempotency (double-count guard): an `increment` may carry an `idempotency_key` (a SHA-256 of the canonical builder-contract content, from `builderContractKey`). The whole increment is applied **at most once** per key — consumed keys are recorded in `applied_telemetry_keys` and checked/appended inside the same lock as the totals. This is what stops one real builder execution being persisted 2–3× through the automated retry path (#123) and the goal loop's stage resets (#129): re-validating the *same* `builder.json` (identical content → identical key) is a no-op, while a genuine retry that actually re-runs the builder writes a *new* contract (different content → new key) and correctly counts again. The guard is content-based, not timestamp-based, precisely so a real re-run is never mistaken for a replay; two attempts with byte-identical contracts represent the same spend and are collapsed on purpose. Belt-and-braces, the stale `builder.json` is also removed at re-claim (`sdlc_build_next`) and at goal-loop reset (`resetStagesForRetry`), so a reset stage cannot replay the prior attempt's contract at all — it must produce a fresh one to be validated. + +Mid-run upgrade seeding: the first increment to materialize the `cumulative_tokens` marker on a run that already has pre-upgrade `cost_recorded` history seeds the persisted totals from an event recompute (passed as `seed` to `updateRunMetrics`, applied once and only when the marker is being created). Without this, an in-flight run upgrading to the persisted-metrics format mid-run would drop all prior history (e.g. `$5.00` of legacy events + one `$0.10` new validation would report `$0.10`). + +`cumulative_tool_calls`: fed by `increment.tool_calls`, sourced from the builder contract's `execution.tool_calls` (total tool **invocations** in the attempt — the guardrail-budget signal). This is distinct from `tools_used_count` (`execution.tools_used.length`, the count of distinct tool **names**); a contract with no `execution.tool_calls` contributes nothing to the counter. + +Telemetry source: at validate time `sdlc_validate` extracts the builder contract's structured `cost` / `context` / `execution` fields via `extractBuilderTelemetry` (`src/core/harness/telemetry.js`), on **every** validation — retries cost money too, and the idempotency key (above) is what prevents re-validation of the same contract from double-counting. Non-numeric cost values are ignored by extraction; the contract gate's `builder_v2_cost_values_are_numeric` check is what fails validation on them. Cost and tokens are split evenly across the task's canonical stages (the same normalization as stage elapsed), so multi-stage tasks are never double-counted. Events (pinned contract, same rules as `retry_decision` — downstream consumers key on these exact shapes): - `cost_recorded` — `task_id`, `usd` (effective spend: `actual_usd` wins over `estimated_usd`; `cost` kept as a legacy alias), `estimated_usd`, `actual_usd`, `currency`, `tokens`, `input_tokens`, `output_tokens`, `source: "builder_contract"`. - `context_recorded` — `task_id`, `profile`, `workflow`, `injected_sources` (count), `tokens_used`, `tokens_available`, `source: "builder_contract"`. +- `metrics_write_failed` — `task_id`, `operation` (e.g. `"telemetry_increment"`), `error`. Emitted when a `cost_recorded` event landed but its matching persisted increment failed to write. It marks the persisted totals as behind the events; `resolveRunTotals` detects the event (via `hasMetricsWriteDrift`) and falls back to event recompute rather than reporting a total it knows is stale. Rollups: the pipeline-state `cost_context` block carries `cumulative_tokens` alongside the existing cost/duration/tool-call totals, and each stage entry carries its `cost_usd` / `tokens` share (`null` when never recorded). `rstack-agents pipeline status` prints the token total; `--json` exposes the full structure. diff --git a/src/core/harness/goal-loop.js b/src/core/harness/goal-loop.js index 26a4c33..8b6ee04 100644 --- a/src/core/harness/goal-loop.js +++ b/src/core/harness/goal-loop.js @@ -10,7 +10,7 @@ // goal_evaluated, loop_iteration_retrying_stages, loop_completed, // loop_blocked) so trace/status/feed render the loop without reading source. -import { appendFile, mkdir, readFile } from 'node:fs/promises'; +import { appendFile, mkdir, readFile, rm } from 'node:fs/promises'; import { existsSync } from 'node:fs'; import { dirname, join } from 'node:path'; @@ -252,6 +252,14 @@ export async function resetStagesForRetry(projectRoot, runId, stageIds) { const matched = taskStageIds(task).filter((stageId) => stageSet.has(stageId)); if (!matched.length) continue; task.status = 'PENDING'; + // Clear the prior attempt's builder.json so the next iteration cannot + // replay — and re-cost — a stale contract through validation (#83). The + // telemetry increment is also idempotency-keyed on contract content, so + // this is defence in depth: the reset stage starts with no contract and + // must produce a fresh one to be validated. + if (typeof task.output_dir === 'string' && task.output_dir) { + await rm(join(projectRoot, task.output_dir, 'builder.json'), { force: true }).catch(() => {}); + } resetTaskIds.push(task.id); for (const stageId of matched) resetStageIds.add(stageId); } diff --git a/src/core/harness/run-state.js b/src/core/harness/run-state.js index 9fb3824..27a9307 100644 --- a/src/core/harness/run-state.js +++ b/src/core/harness/run-state.js @@ -67,8 +67,26 @@ function roundUsd(value) { // (which overwrite), `increment` ADDS to the running totals — the whole // read-modify-write already runs inside the file lock, so concurrent // increments from parallel validations both land. +// +// Idempotency (#83 double-count fix): when the increment carries an +// `idempotency_key` (the builder-contract content hash), the whole increment +// is applied at most once. The consumed keys are recorded in +// `applied_telemetry_keys` on the metrics object, checked and appended inside +// the same lock as the totals — so a retry or goal-loop reset that re-validates +// the SAME builder.json (identical content → identical key) never counts twice, +// while a genuine re-run of the builder (new content → new key) still counts. +// Returns true when the increment was applied, false when skipped as a replay. function applyMetricsIncrement(merged, increment) { - if (!increment || typeof increment !== 'object') return; + if (!increment || typeof increment !== 'object') return false; + + const key = typeof increment.idempotency_key === 'string' && increment.idempotency_key + ? increment.idempotency_key + : null; + if (key) { + const applied = Array.isArray(merged.applied_telemetry_keys) ? merged.applied_telemetry_keys : []; + if (applied.includes(key)) return false; // already counted — no-op replay + merged.applied_telemetry_keys = [...applied, key]; + } const cost = Number(increment.cost_usd); if (Number.isFinite(cost)) { @@ -102,6 +120,8 @@ function applyMetricsIncrement(merged, increment) { } merged.stage_tokens = stageTokens; } + + return true; } export async function updateRunMetrics(runDir, metricsUpdate = {}) { @@ -124,7 +144,25 @@ export async function updateRunMetrics(runDir, metricsUpdate = {}) { } catch { current = {}; } } - const { increment, ...update } = metricsUpdate; + const { increment, seed, ...update } = metricsUpdate; + + // Mid-run upgrade seeding (#83): the first increment to materialize the + // `cumulative_tokens` marker on a run that already has pre-upgrade cost + // history (legacy cost_recorded events) must fold that history in, or a + // $5.00 run that logs one $0.10 new validation would report $0.10. The + // caller (which can see events) passes `seed` = event-recomputed totals; + // we apply it exactly once — only when the marker isn't present yet and we + // are about to create it — so subsequent increments don't re-seed. + const willMaterializeMarker = + current.cumulative_tokens === undefined && update.cumulative_tokens === undefined && + increment && typeof increment === 'object'; + if (willMaterializeMarker && seed && typeof seed === 'object') { + const seedCost = Number(seed.cost_usd); + if (Number.isFinite(seedCost)) current.cumulative_cost_usd = roundUsd((Number(current.cumulative_cost_usd) || 0) + seedCost); + if (seed.tokens && typeof seed.tokens === 'object') { + current.cumulative_tokens = addTokenCounts(current.cumulative_tokens, seed.tokens); + } + } const merged = { ...current, diff --git a/src/core/harness/telemetry.js b/src/core/harness/telemetry.js index 8905818..8250315 100644 --- a/src/core/harness/telemetry.js +++ b/src/core/harness/telemetry.js @@ -15,6 +15,8 @@ // builder_v2_cost_values_are_numeric check) is what fails validation on // non-numeric cost telemetry. +import { createHash } from 'node:crypto'; + function finiteNumber(value) { if (value === null || value === undefined || value === '') return null; const num = Number(value); @@ -41,13 +43,15 @@ function plainObject(value) { * `builder.cost` is a bare number are treated as actual spend. * - tokens: { input, output, total } from optional cost.input_tokens / * cost.output_tokens / cost.total_tokens (cost.tokens accepted as total). - * - tools_used_count: length of execution.tools_used (distinct tools, not - * call counts — execution.tool_calls stays the guardrail-budget signal). + * - tools_used_count: length of execution.tools_used (distinct tool NAMES, not + * call counts). + * - tool_calls: execution.tool_calls (total tool INVOCATIONS) — the + * guardrail-budget signal and what feeds cumulative_tool_calls. * - context: { profile, workflow, injected_source_count, tokens_used, * tokens_available } — the token gauges are the context-pressure hook (#136). */ export function extractBuilderTelemetry(builder) { - const telemetry = { cost: null, tokens: null, tools_used_count: null, context: null }; + const telemetry = { cost: null, tokens: null, tools_used_count: null, tool_calls: null, context: null }; if (!plainObject(builder)) return telemetry; const rawCost = builder.cost; @@ -80,6 +84,12 @@ export function extractBuilderTelemetry(builder) { if (Array.isArray(builder.execution?.tools_used)) { telemetry.tools_used_count = builder.execution.tools_used.length; } + // execution.tool_calls is the guardrail-budget signal: the total number of + // tool INVOCATIONS in the attempt (distinct from tools_used_count, which is + // the count of distinct tool NAMES). This is what feeds cumulative_tool_calls + // — a real call count, not a name count. + const toolCalls = finiteNumber(builder.execution?.tool_calls); + if (toolCalls !== null) telemetry.tool_calls = toolCalls; const rawContext = plainObject(builder.context); if (rawContext) { @@ -104,6 +114,42 @@ export function extractBuilderTelemetry(builder) { return telemetry; } +/** + * Stable idempotency key for a builder contract (#83 double-count fix). + * + * The key is a SHA-256 of the CANONICAL contract content — object keys sorted + * recursively so semantically-identical JSON always hashes the same. This is + * what makes the cost increment idempotent: the SAME builder.json validated + * twice (an automated retry that re-runs validation without re-running the + * builder, or a goal-loop stage reset that replays a stale contract) produces + * the SAME key and must count only once. + * + * A LEGITIMATE re-spend — a genuine retry that actually re-runs the builder — + * writes a NEW builder.json (different summary, files_modified, tests_run, + * cost, or any other field), which hashes to a NEW key and correctly counts + * again. We key on content, not on a timestamp, precisely so that a builder + * that re-does real work is never mistaken for a replay; two attempts that + * produced byte-identical contracts genuinely represent the same spend and are + * collapsed on purpose. + * + * Returns null for a non-object contract (nothing to key on). + */ +export function builderContractKey(builder) { + if (!plainObject(builder)) return null; + return createHash('sha256').update(canonicalJson(builder)).digest('hex'); +} + +// Deterministic JSON: object keys sorted at every level so key ordering never +// changes the hash. Arrays keep their order (order is meaningful there). +function canonicalJson(value) { + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]`; + if (value && typeof value === 'object') { + const keys = Object.keys(value).sort(); + return `{${keys.map((k) => `${JSON.stringify(k)}:${canonicalJson(value[k])}`).join(',')}}`; + } + return JSON.stringify(value === undefined ? null : value); +} + /** * Pinned event payloads for the extracted telemetry (contract style follows * retry_decision — downstream consumers key on these exact shapes; change @@ -156,8 +202,14 @@ export function builderTelemetryEvents(taskId, telemetry) { * across the task's canonical stages, mirroring deriveStageElapsed's * multi-stage normalization so nothing is double-counted. Context token * gauges are point-in-time values, not counters, so they overwrite. + * + * When `idempotencyKey` is supplied (the builder-contract hash from + * builderContractKey), it is stamped on the increment so updateRunMetrics can + * make the whole increment a no-op if that key was already applied — this is + * the double-count guard for retries and loop resets that re-validate the same + * contract (#83). */ -export function telemetryMetricsUpdate(telemetry, stageIds = []) { +export function telemetryMetricsUpdate(telemetry, stageIds = [], idempotencyKey = null) { const ids = (Array.isArray(stageIds) ? stageIds : []).filter((id) => typeof id === 'string' && id); const increment = {}; @@ -181,8 +233,15 @@ export function telemetryMetricsUpdate(telemetry, stageIds = []) { } } + if (telemetry?.tool_calls !== null && telemetry?.tool_calls !== undefined) { + increment.tool_calls = telemetry.tool_calls; + } + const update = {}; - if (Object.keys(increment).length > 0) update.increment = increment; + if (Object.keys(increment).length > 0) { + if (typeof idempotencyKey === 'string' && idempotencyKey) increment.idempotency_key = idempotencyKey; + update.increment = increment; + } if (telemetry?.context?.tokens_used !== null && telemetry?.context?.tokens_used !== undefined) { update.context_tokens_used = telemetry.context.tokens_used; } diff --git a/src/integrations/pi/rstack-sdlc.ts b/src/integrations/pi/rstack-sdlc.ts index 0ca9b9c..bf119bf 100644 --- a/src/integrations/pi/rstack-sdlc.ts +++ b/src/integrations/pi/rstack-sdlc.ts @@ -5,7 +5,7 @@ import { spawn } from "node:child_process"; import { request as httpRequest } from "node:http"; import { createConnection } from "node:net"; import { existsSync, openSync } from "node:fs"; -import { mkdir, readFile, readdir, writeFile, appendFile } from "node:fs/promises"; +import { mkdir, readFile, readdir, writeFile, appendFile, rm } from "node:fs/promises"; import { basename, dirname, join, relative, resolve } from "node:path"; import { homedir } from "node:os"; import { fileURLToPath } from "node:url"; @@ -17,7 +17,8 @@ import { MANIFEST_SCHEMA_VERSION, migrateManifest } from "../../core/harness/mig import { appendEvidenceEvent } from "../../core/harness/evidence.js"; import { DEFAULT_HARNESS_GUARDRAILS, guardrailSummary, loadProjectGuardrails, evaluateTaskClaim, evaluateBuilderTelemetry, guardrailEvent, isDestructiveTask } from "../../core/harness/guardrails.js"; import { classifyRetryDecision } from "../../core/harness/retry-policy.js"; -import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate } from "../../core/harness/telemetry.js"; +import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate, builderContractKey } from "../../core/harness/telemetry.js"; +import { deriveRunTotals } from "../../observability/metrics/derive.js"; import { VALIDATOR_CONTEXT_ENV, VALIDATOR_RUN_ID_ENV, VALIDATOR_READ_ONLY_TOOLS, evaluateValidatorAction, isValidatorContext, isValidatorRole, isValidatorSandboxDebug } from "../../core/harness/validator-sandbox.js"; import { loadValidatorRegistry, resolveValidatorProfile } from "../../core/harness/validator-registry.js"; import { budgetEnvelopeForTask, loadBudgetPolicy, loadProjectProfile } from "../../core/profiles.js"; @@ -1525,6 +1526,16 @@ export default function (pi: ExtensionAPI) { await writeJsonAtomic(approvalsFile, approvals); }); } + // Clear any stale builder.json from a prior attempt before granting the + // new one (#83 replay guard). If a re-claimed FAIL/BLOCKED task starts a + // fresh attempt but the builder crashes before rewriting builder.json, + // sdlc_validate would otherwise re-validate — and re-cost — the previous + // attempt's contract. Removing it in-lock at the transition means the + // next validation sees no contract (FAIL: builder_contract_exists) + // instead of silently replaying the old one's spend. + try { + await rm(join(projectRoot, task.output_dir, "builder.json"), { force: true }); + } catch { /* best-effort; a missing file is the expected case */ } task.status = "IN_PROGRESS"; task._started_at = Date.now(); // Real attribution: stamp the routed pipeline agent so builder.json and @@ -1770,19 +1781,56 @@ export default function (pi: ExtensionAPI) { if (builderContract) { // Cost/context telemetry (#83/#135): shared extraction from the builder // contract's structured cost/context fields, pinned cost_recorded / - // context_recorded events, and incremental metrics.json accumulation — - // recorded on every validation (retries cost money too), so totals stop - // being re-derived O(events) on each dashboard poll. + // context_recorded events, and incremental metrics.json accumulation. + // Recorded on every validation (retries cost money too), but the + // metrics increment is keyed on the builder-contract content hash so + // re-validating the SAME contract (a retry that didn't re-run the + // builder, or a goal-loop reset replaying a stale builder.json) never + // double-counts — while a genuine re-run (new contract content → new + // key) still counts. + const runDir = join(runsDir(projectRoot), manifest.run_id); const telemetry = extractBuilderTelemetry(builderContract); + const idempotencyKey = builderContractKey(builderContract); + // Seed from pre-existing events BEFORE we append this validation's + // cost_recorded event, so a mid-run upgrade (legacy cost_recorded + // history + first new-style validation) folds the prior history into + // the persisted totals instead of dropping it. Only used when the + // marker (cumulative_tokens) isn't present yet; updateRunMetrics guards + // that in-lock. + let seed: { cost_usd: number; tokens: { input: number; output: number; total: number } } | undefined; + try { + const existingMetricsPath = join(runDir, "metrics.json"); + const hasMarker = existsSync(existingMetricsPath) + && !!JSON.parse(await readFile(existingMetricsPath, "utf8"))?.cumulative_tokens; + if (!hasMarker) { + const priorTotals = deriveRunTotals(await readJsonl(join(runDir, "events.jsonl"))); + if (priorTotals.cost_usd > 0 || priorTotals.tokens > 0) { + seed = { cost_usd: priorTotals.cost_usd, tokens: { input: 0, output: 0, total: priorTotals.tokens } }; + } + } + } catch { + // Best-effort seeding; a read failure just skips the fold-in. + } for (const telemetryEvent of builderTelemetryEvents(task.id, telemetry)) { await appendEvent(projectRoot, manifest.run_id, telemetryEvent); } - const metricsUpdate = telemetryMetricsUpdate(telemetry, canonicalStageIds); + const metricsUpdate = telemetryMetricsUpdate(telemetry, canonicalStageIds, idempotencyKey); if (metricsUpdate) { + if (seed) (metricsUpdate as any).seed = seed; try { - await updateRunMetrics(join(runsDir(projectRoot), manifest.run_id), metricsUpdate); + await updateRunMetrics(runDir, metricsUpdate); } catch (metricsError) { + // F2: a swallowed write failure permanently diverges the persisted + // totals from the events that recorded the cost. Emit a pinned + // metrics_write_failed event so the drift is visible and readers + // can reconcile (derive.js falls back to event recompute). console.error("Failed to persist cost/context metrics:", metricsError); + await appendEvent(projectRoot, manifest.run_id, { + type: "metrics_write_failed", + task_id: task.id, + operation: "telemetry_increment", + error: String((metricsError as any)?.message ?? metricsError), + }).catch(() => {}); } } } diff --git a/src/observability/metrics/derive.js b/src/observability/metrics/derive.js index 2b5196d..29f9faf 100644 --- a/src/observability/metrics/derive.js +++ b/src/observability/metrics/derive.js @@ -158,17 +158,37 @@ export function persistedTokenTotals(metrics) { }; } +/** + * True when the event stream records a metrics write failure (F2). A + * `metrics_write_failed` event means a cost_recorded event landed but its + * matching persisted increment did NOT, so the persisted cumulative totals are + * known to be behind the events. When this has happened, event recompute is the + * safe source of truth for cost/tokens (cost_recorded is always appended before + * the increment is attempted), so resolveRunTotals ignores the stale persisted + * marker and recomputes. + */ +export function hasMetricsWriteDrift(events) { + for (const ev of events ?? []) { + if (ev?.type === 'metrics_write_failed') return true; + } + return false; +} + /** * Run totals, preferring persisted cumulative metrics (#83): when metrics.json * carries incremental cost/token totals they are authoritative — O(1) instead * of re-parsing the event stream, and they survive event rotation/archival. * Legacy runs (no persisted totals) recompute from events exactly as before. * Duration, task outcomes, guardrails, and quality always come from events. + * + * Exception (F2): if a `metrics_write_failed` event is present the persisted + * totals are known to lag the events, so we fall back to the event recompute + * rather than report a total we know is stale. */ export function resolveRunTotals(events, metrics = {}) { const totals = deriveRunTotals(events); const persisted = persistedTokenTotals(metrics); - if (persisted) { + if (persisted && !hasMetricsWriteDrift(events)) { totals.tokens = persisted.total; totals.cost_usd = Math.round((Number(metrics.cumulative_cost_usd) || 0) * 10000) / 10000; } diff --git a/tests/harness-telemetry.test.js b/tests/harness-telemetry.test.js index 3dd4a65..76aeac3 100644 --- a/tests/harness-telemetry.test.js +++ b/tests/harness-telemetry.test.js @@ -4,7 +4,7 @@ import { mkdtempSync, rmSync, readFileSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate } from '../src/core/harness/telemetry.js'; +import { extractBuilderTelemetry, builderTelemetryEvents, telemetryMetricsUpdate, builderContractKey } from '../src/core/harness/telemetry.js'; import { updateRunMetrics } from '../src/core/harness/run-state.js'; // owner: RStack developed by Richardson Gunde @@ -56,8 +56,8 @@ test('extractBuilderTelemetry ignores non-numeric cost and token values', () => }); test('extractBuilderTelemetry returns nulls for missing or malformed contracts', () => { - assert.deepEqual(extractBuilderTelemetry(null), { cost: null, tokens: null, tools_used_count: null, context: null }); - assert.deepEqual(extractBuilderTelemetry({}), { cost: null, tokens: null, tools_used_count: null, context: null }); + assert.deepEqual(extractBuilderTelemetry(null), { cost: null, tokens: null, tools_used_count: null, tool_calls: null, context: null }); + assert.deepEqual(extractBuilderTelemetry({}), { cost: null, tokens: null, tools_used_count: null, tool_calls: null, context: null }); assert.deepEqual(extractBuilderTelemetry({ cost: [], context: 'nope' }).cost, null); }); @@ -222,6 +222,151 @@ test('updateRunMetrics keeps overwrite semantics for cumulative fields and never } }); +// ── extractBuilderTelemetry tool_calls (F4) ────────────────────────────────── + +test('extractBuilderTelemetry reads execution.tool_calls as the invocation count', () => { + const telemetry = extractBuilderTelemetry({ + execution: { tools_used: ['read_file', 'patch'], tool_calls: 17 }, + }); + // tools_used_count is distinct tool NAMES; tool_calls is total invocations. + assert.equal(telemetry.tools_used_count, 2); + assert.equal(telemetry.tool_calls, 17); +}); + +test('extractBuilderTelemetry leaves tool_calls null when execution has no count', () => { + assert.equal(extractBuilderTelemetry({ execution: { tools_used: ['a'] } }).tool_calls, null); + assert.equal(extractBuilderTelemetry({}).tool_calls, null); +}); + +test('telemetryMetricsUpdate carries tool_calls into the increment', () => { + const update = telemetryMetricsUpdate(extractBuilderTelemetry({ execution: { tool_calls: 9 } }), []); + assert.equal(update.increment.tool_calls, 9); +}); + +// ── builderContractKey (F1 idempotency) ────────────────────────────────────── + +test('builderContractKey is stable across key ordering and differs on content', () => { + const a = builderContractKey({ task_id: 't', status: 'PASS', cost: { actual_usd: 1 } }); + const b = builderContractKey({ cost: { actual_usd: 1 }, status: 'PASS', task_id: 't' }); + assert.equal(a, b, 'reordered keys hash the same'); + const c = builderContractKey({ task_id: 't', status: 'PASS', cost: { actual_usd: 2 } }); + assert.notEqual(a, c, 'different content hashes differently'); + assert.equal(builderContractKey(null), null); +}); + +test('telemetryMetricsUpdate stamps the idempotency key when supplied', () => { + const update = telemetryMetricsUpdate(extractBuilderTelemetry({ cost: { actual_usd: 0.3 } }), [], 'abc123'); + assert.equal(update.increment.idempotency_key, 'abc123'); + const noKey = telemetryMetricsUpdate(extractBuilderTelemetry({ cost: { actual_usd: 0.3 } }), []); + assert.equal('idempotency_key' in noKey.increment, false); +}); + +// ── updateRunMetrics idempotency (F1) ──────────────────────────────────────── + +test('the same contract validated 2×/3× is counted once', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-idem-')); + try { + const builder = { task_id: '07-code', status: 'PASS', cost: { actual_usd: 1.0, input_tokens: 1000, output_tokens: 200 } }; + const key = builderContractKey(builder); + const telemetry = extractBuilderTelemetry(builder); + const update = telemetryMetricsUpdate(telemetry, ['07-code'], key); + + for (let i = 0; i < 3; i++) { + await updateRunMetrics(runDir, update); + } + const persisted = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + assert.equal(persisted.cumulative_cost_usd, 1.0, 'a 3× loop over a $1.00 stage persists $1.00, not $3.00'); + assert.deepEqual(persisted.cumulative_tokens, { input: 1000, output: 200, total: 1200 }); + assert.deepEqual(persisted.applied_telemetry_keys, [key]); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('a new contract after a genuine retry counts again', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-rerun-')); + try { + const first = { task_id: '07-code', status: 'FAIL', summary: 'attempt 1', cost: { actual_usd: 1.0 } }; + const second = { task_id: '07-code', status: 'PASS', summary: 'attempt 2 re-ran the builder', cost: { actual_usd: 1.0 } }; + await updateRunMetrics(runDir, telemetryMetricsUpdate(extractBuilderTelemetry(first), ['07-code'], builderContractKey(first))); + await updateRunMetrics(runDir, telemetryMetricsUpdate(extractBuilderTelemetry(second), ['07-code'], builderContractKey(second))); + const persisted = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + // Two genuinely different attempts (different content → different keys) = real re-spend. + assert.equal(persisted.cumulative_cost_usd, 2.0); + assert.equal(persisted.applied_telemetry_keys.length, 2); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('idempotency guard holds under concurrent replays of the same key', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-idem-race-')); + try { + const builder = { task_id: '07-code', cost: { actual_usd: 0.5 } }; + const update = telemetryMetricsUpdate(extractBuilderTelemetry(builder), ['07-code'], builderContractKey(builder)); + await Promise.all([ + updateRunMetrics(runDir, update), + updateRunMetrics(runDir, update), + updateRunMetrics(runDir, update), + ]); + const persisted = JSON.parse(readFileSync(join(runDir, 'metrics.json'), 'utf8')); + assert.equal(persisted.cumulative_cost_usd, 0.5); + assert.equal(persisted.applied_telemetry_keys.length, 1); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +// ── mid-run upgrade seeding (F3) ───────────────────────────────────────────── + +test('first new-style increment on a run with prior events seeds from history', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-seed-')); + try { + // Realistic legacy metrics.json: pre-#83, cost/tokens lived ONLY in + // cost_recorded events — metrics.json carried no cumulative_cost_usd + // accrual and no cumulative_tokens marker. The read path recomputed both. + writeFileSync(join(runDir, 'metrics.json'), JSON.stringify({ + cumulative_duration_ms: 0, + cumulative_cost_usd: 0, + cumulative_tool_calls: 0, + stage_elapsed_ms: {}, + stage_status: {}, + })); + const builder = { task_id: '07-code', cost: { actual_usd: 0.1, input_tokens: 100, output_tokens: 20 } }; + const update = telemetryMetricsUpdate(extractBuilderTelemetry(builder), ['07-code'], builderContractKey(builder)); + // Caller recomputes the pre-upgrade history from events ($5.00, 900 tokens) + // and passes it as the seed so no history is dropped. + update.seed = { cost_usd: 5.0, tokens: { input: 0, output: 0, total: 900 } }; + const after = await updateRunMetrics(runDir, update); + // Cost: seeded $5.00 + this validation's $0.10 = $5.10 (not $0.10 — F3 bug). + assert.equal(after.cumulative_cost_usd, 5.1); + // Tokens: seeded 900 + this validation's 120. + assert.equal(after.cumulative_tokens.total, 1020); + // A second increment must not re-seed (marker now present). + const again = await updateRunMetrics(runDir, { increment: { tokens: { input: 0, output: 0, total: 5 }, idempotency_key: 'other' } }); + assert.equal(again.cumulative_cost_usd, 5.1); + assert.equal(again.cumulative_tokens.total, 1025); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + +test('seeding does not fire once the cumulative_tokens marker exists', async () => { + const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-noseed-')); + try { + await updateRunMetrics(runDir, { increment: { tokens: { input: 10, output: 5, total: 15 }, idempotency_key: 'k1' } }); + // Marker now present; a later increment carrying a seed must ignore it. + const after = await updateRunMetrics(runDir, { + increment: { cost_usd: 0.1, idempotency_key: 'k2' }, + seed: { cost_usd: 99, tokens: { input: 0, output: 0, total: 99999 } }, + }); + assert.equal(after.cumulative_tokens.total, 15, 'seed ignored — marker already present'); + assert.equal(after.cumulative_cost_usd, 0.1); + } finally { + rmSync(runDir, { recursive: true, force: true }); + } +}); + test('updateRunMetrics tolerates malformed increments and token shapes', async () => { const runDir = mkdtempSync(join(tmpdir(), 'rstack-metrics-malformed-')); try { diff --git a/tests/observability-cost-metrics.test.js b/tests/observability-cost-metrics.test.js index ea05147..8f81e8b 100644 --- a/tests/observability-cost-metrics.test.js +++ b/tests/observability-cost-metrics.test.js @@ -4,7 +4,7 @@ import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { deriveRunTotals, persistedTokenTotals, resolveRunTotals } from '../src/observability/metrics/derive.js'; +import { deriveRunTotals, persistedTokenTotals, resolveRunTotals, hasMetricsWriteDrift } from '../src/observability/metrics/derive.js'; import { entryFromRun } from '../src/observability/dashboard/state/rollup-index.js'; import { toClientState } from '../src/observability/dashboard/state/client-state.js'; import { buildRunReport } from '../src/observability/collectors/reporter.js'; @@ -46,6 +46,24 @@ test('resolveRunTotals prefers persisted cumulative metrics over event recompute assert.equal(totals.duration_ms, 2 * 60 * 1000); }); +test('resolveRunTotals falls back to events when a metrics_write_failed drift event exists (F2)', () => { + // Persisted totals say $0.90 but a write failed after the last cost_recorded + // landed — the persisted number is known-stale, so we recompute from events. + const eventsWithDrift = [ + ...COST_EVENTS, + { type: 'metrics_write_failed', operation: 'telemetry_increment', ts: '2026-07-01T05:03:00.000Z' }, + ]; + assert.equal(hasMetricsWriteDrift(eventsWithDrift), true); + assert.equal(hasMetricsWriteDrift(COST_EVENTS), false); + const totals = resolveRunTotals(eventsWithDrift, { + cumulative_cost_usd: 0.9, + cumulative_tokens: { input: 40000, output: 10000, total: 50000 }, + }); + // Ignores the stale persisted marker; recomputes 0.25 + 0.15 = 0.40 from events. + assert.equal(totals.cost_usd, 0.4); + assert.equal(totals.tokens, 20000); +}); + test('resolveRunTotals falls back to event recompute for legacy runs', () => { const legacyMetrics = { cumulative_cost_usd: 0, stage_status: { '07-code': 'PASS' } }; const totals = resolveRunTotals(COST_EVENTS, legacyMetrics); diff --git a/tests/pipeline-loop.test.js b/tests/pipeline-loop.test.js index 3008e3a..886b000 100644 --- a/tests/pipeline-loop.test.js +++ b/tests/pipeline-loop.test.js @@ -141,6 +141,33 @@ test('resetStagesForRetry only overrides stage_status for stages where a task wa assert.equal(metrics.stage_status['06-architecture'], 'FAILED', 'a stage with only gated tasks keeps its real status'); }); +test('resetStagesForRetry clears the reset task\'s stale builder.json so it cannot be replayed (#83)', async () => { + const projectRoot = mkdtempSync(path.join(os.tmpdir(), 'rstack-loop-')); + const outputDir = path.join('runs', 'run-a', 'tasks', '002'); + const absOutputDir = path.join(projectRoot, outputDir); + mkdirSync(absOutputDir, { recursive: true }); + const builderPath = path.join(absOutputDir, 'builder.json'); + // A prior attempt's contract with real cost sitting in the task dir. + writeFileSync(builderPath, JSON.stringify({ task_id: '002', status: 'FAIL', cost: { actual_usd: 1.0 } })); + + const runDir = seedRun(projectRoot, 'run-a', { + tasks: [ + { id: '001', title: '001', status: 'PASS', stage_artifacts: [{ stage_id: '06-architecture' }] }, + { id: '002', title: '002', status: 'FAIL', stage_artifacts: [{ stage_id: '07-code' }], output_dir: outputDir }, + ], + metrics: null, + }); + assert.equal(existsSync(builderPath), true, 'precondition: stale contract present'); + + const reset = await resetStagesForRetry(projectRoot, 'run-a', ['07-code']); + assert.deepEqual(reset, ['002']); + assert.equal(existsSync(builderPath), false, 'stale builder.json removed at reset so validation cannot re-cost it'); + // The un-reset task in another stage keeps its dir untouched (nothing to prove + // beyond: reset only cleans what it resets). + const tasks = JSON.parse(readFileSync(path.join(runDir, 'tasks.json'), 'utf8')).tasks; + assert.equal(tasks.find((item) => item.id === '002').status, 'PENDING'); +}); + // ── The loop ───────────────────────────────────────────────────────────────── test('retry then pass: loop resets the recommended stage, reruns it, and completes with the full event trail', async () => {