Persist builder cost/context telemetry into metrics.json + persisted-first read path (#83, #135)

docs/HARNESS.md

@@ -92,6 +92,50 @@ 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)
+
+`<run_dir>/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,
+  "applied_telemetry_keys": ["<sha256 of the builder contract that was counted>"]
+}
+```
+
+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.
+
+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.
+
 ## Agent episodic memory
 
 Validator-approved tasks are written to an agent/stage scoped episodic memory store by `src/memory/index.js`.

src/commands/pipeline.js

@@ -143,6 +143,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`);
   }

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);
     }

src/core/harness/pipeline-state.js

@@ -214,11 +214,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,
   };
@@ -362,6 +379,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])],
       // Verified on disk, never inferred from events: true only when the
       // checkpoint directory actually exists right now (#132).

src/core/harness/run-state.js

@@ -36,6 +36,94 @@ 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.
+//
+// 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 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)) {
+    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;
+  }
+
+  return true;
+}
+
 export async function updateRunMetrics(runDir, metricsUpdate = {}) {
   const path = join(runDir, 'metrics.json');
   // Lock the whole read-modify-write: concurrent stage updates (parallel
@@ -56,16 +144,49 @@ export async function updateRunMetrics(runDir, metricsUpdate = {}) {
       } catch { current = {}; }
     }
 
+    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,
-      ...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;