fix(power): withhold invalid and ambiguous measured energy / 屏蔽无效与语义不明的实测能耗

packages/app/src/components/inference/types.ts

@@ -137,6 +137,8 @@ export interface AggDataEntry {
   'p99.9_e2el': number;
   // Measured GPU telemetry (emitted by runner's aggregate_power.py).
   // Optional because historical runs predate the fields.
+  power_valid?: number;
+  power_metric_schema_version?: number;
   avg_power_w?: number;
   joules_per_output_token?: number;
   joules_per_total_token?: number;
@@ -145,13 +147,13 @@ export interface AggDataEntry {
   // and decode workers (single-node disagg or multinode disagg). Single-node
   // aggregated configs leave these undefined.
   // - prefill_avg_power_w / decode_avg_power_w: mean per-GPU draw (W) within each role
-  // - joules_per_input_token: prefill_energy / total_input_tokens (prefill GPUs only)
-  // The disagg decode-only J/output is carried by joules_per_output_token above
-  // (the runner overrides it to decode_energy / total_output_tokens on disagg) —
-  // there is no separate _decode field.
+  // Unprefixed joules fields are whole-deployment metrics in schema version 2.
+  // Explicit prefill/decode keys retain role-local energy breakdowns.
   prefill_avg_power_w?: number;
   decode_avg_power_w?: number;
   joules_per_input_token?: number;
+  prefill_joules_per_input_token?: number;
+  decode_joules_per_output_token?: number;
   // Cluster-wide GPU telemetry beyond power (temperature, utilization, memory).
   // Emitted by aggregate_power.py when the perfmon CSVs include the matching
   // sample columns. Optional because older runs (and runs without the relevant

packages/app/src/lib/benchmark-transform.test.ts

@@ -240,20 +240,105 @@ describe('rowToAggDataEntry', () => {
   it('passes through multinode / disagg role-split power scalars when present', () => {
     const entry = rowToAggDataEntry(
       makeRow({
+        disagg: true,
         metrics: {
           tput_per_gpu: 100,
+          power_valid: 1,
+          power_metric_schema_version: 2,
           prefill_avg_power_w: 612.3,
           decode_avg_power_w: 701.5,
           joules_per_input_token: 1.2,
-          // disagg: joules_per_output_token IS the per-stage decode value.
           joules_per_output_token: 9.7,
+          joules_per_total_token: 0.8,
+          prefill_joules_per_input_token: 0.4,
+          decode_joules_per_output_token: 5.1,
         },
       }),
     );
     expect(entry.prefill_avg_power_w).toBe(612.3);
     expect(entry.decode_avg_power_w).toBe(701.5);
     expect(entry.joules_per_input_token).toBe(1.2);
     expect(entry.joules_per_output_token).toBe(9.7);
+    expect(entry.joules_per_total_token).toBe(0.8);
+    expect(entry.prefill_joules_per_input_token).toBe(0.4);
+    expect(entry.decode_joules_per_output_token).toBe(5.1);
+  });
+
+  it('withholds ambiguous unversioned disagg joules while preserving role watts', () => {
+    const entry = rowToAggDataEntry(
+      makeRow({
+        disagg: true,
+        metrics: {
+          avg_power_w: 650,
+          prefill_avg_power_w: 612.3,
+          decode_avg_power_w: 701.5,
+          joules_per_input_token: 1.2,
+          joules_per_output_token: 9.7,
+          joules_per_total_token: 0.8,
+        },
+      }),
+    );
+
+    expect(entry.avg_power_w).toBe(650);
+    expect(entry.prefill_avg_power_w).toBe(612.3);
+    expect(entry.decode_avg_power_w).toBe(701.5);
+    expect(entry.joules_per_input_token).toBeUndefined();
+    expect(entry.joules_per_output_token).toBeUndefined();
+    expect(entry.joules_per_total_token).toBeUndefined();
+  });
+
+  it('withholds disagg joules stamped with an unrecognized future schema version', () => {
+    const entry = rowToAggDataEntry(
+      makeRow({
+        disagg: true,
+        metrics: {
+          power_valid: 1,
+          power_metric_schema_version: 3,
+          avg_power_w: 650,
+          prefill_avg_power_w: 612.3,
+          joules_per_input_token: 1.2,
+          joules_per_output_token: 9.7,
+          joules_per_total_token: 0.8,
+        },
+      }),
+    );
+
+    expect(entry.avg_power_w).toBe(650);
+    expect(entry.prefill_avg_power_w).toBe(612.3);
+    expect(entry.joules_per_input_token).toBeUndefined();
+    expect(entry.joules_per_output_token).toBeUndefined();
+    expect(entry.joules_per_total_token).toBeUndefined();
+  });
+
+  it('treats explicit power_valid=0 as authoritative and scrubs measured power', () => {
+    const workers = [{ role: 'agg', worker_idx: 0, num_gpus: 8, avg_power_w: 685.5 }];
+    const entry = rowToAggDataEntry(
+      makeRow({
+        metrics: {
+          power_valid: 0,
+          power_metric_schema_version: 2,
+          avg_power_w: 685.5,
+          joules_per_input_token: 1.2,
+          joules_per_output_token: 8.4,
+          joules_per_total_token: 0.8,
+          prefill_avg_power_w: 612.3,
+          decode_avg_power_w: 701.5,
+          prefill_joules_per_input_token: 0.4,
+          decode_joules_per_output_token: 5.1,
+        },
+        workers,
+      }),
+    );
+
+    expect(entry.avg_power_w).toBeUndefined();
+    expect(entry.joules_per_input_token).toBeUndefined();
+    expect(entry.joules_per_output_token).toBeUndefined();
+    expect(entry.joules_per_total_token).toBeUndefined();
+    expect(entry.prefill_avg_power_w).toBeUndefined();
+    expect(entry.decode_avg_power_w).toBeUndefined();
+    expect(entry.prefill_joules_per_input_token).toBeUndefined();
+    expect(entry.decode_joules_per_output_token).toBeUndefined();
+    expect(entry.workers).toBeUndefined();
   });
 
   it('passes through per-worker measured power array intact', () => {

packages/app/src/lib/benchmark-transform.ts

@@ -16,6 +16,14 @@ import { getHardwareConfig } from '@/lib/constants';
 import { isPersistedBenchmarkId } from '@/lib/benchmark-id';
 import type { BenchmarkRow } from '@/lib/api';
 
+/**
+ * Producer schema version whose unprefixed `joules_per_*` fields are documented
+ * as whole-deployment energy. Mirrors `POWER_METRIC_SCHEMA_VERSION` in the
+ * runner's `utils/aggregate_power.py`; bump both together when the semantics of
+ * those fields change again.
+ */
+const WHOLE_DEPLOYMENT_ENERGY_SCHEMA_VERSION = 2;
+
 /**
  * Agentic trace-replay runs (`benchmark_type === 'agentic_traces'`) emit ttft/ttlt/itl
  * but not the intvty/e2el/tpot keys the chart pipeline expects. Bridge them here:
@@ -95,6 +103,16 @@ export function rowToAggDataEntry(row: BenchmarkRow): AggDataEntry {
   const aggregateDpAttention = aggregateUsesPrefill
     ? row.prefill_dp_attention
     : row.decode_dp_attention;
+  // An explicit invalid verdict is authoritative. Legacy rows without the
+  // verdict remain eligible so historical single-node measurements do not
+  // disappear. Unversioned disaggregated joules are withheld because their
+  // unprefixed fields changed from role-local to whole-deployment semantics.
+  const measuredPowerValid = m.power_valid !== 0;
+  // Match the version exactly rather than `>= 2`: an open bound would silently
+  // admit a future schema whose semantics changed again, which is the exact
+  // failure that made versioning necessary in the first place.
+  const hasWholeDeploymentEnergySemantics =
+    !row.disagg || m.power_metric_schema_version === WHOLE_DEPLOYMENT_ENERGY_SCHEMA_VERSION;
   // Prefer the dedicated column (added in migration 004); fall back to the
   // legacy stash inside `metrics` for any rows ingested before that column
   // existed.
@@ -168,15 +186,31 @@ export function rowToAggDataEntry(row: BenchmarkRow): AggDataEntry {
     // Measured GPU telemetry (runner's aggregate_power.py). Left undefined for
     // rows predating the field so downstream chart code can distinguish
     // "no measurement" from "0 W" via createChartDataPoint's typeof guard.
-    avg_power_w: m.avg_power_w,
-    joules_per_output_token: m.joules_per_output_token,
-    joules_per_total_token: m.joules_per_total_token,
-    // Multinode / disagg-only role splits — same undefined-for-legacy pattern.
-    // (disagg's decode-only J/output is carried by joules_per_output_token above,
-    // which the runner overrides to the per-stage value — no separate _decode key.)
-    prefill_avg_power_w: m.prefill_avg_power_w,
-    decode_avg_power_w: m.decode_avg_power_w,
-    joules_per_input_token: m.joules_per_input_token,
+    power_valid: m.power_valid,
+    power_metric_schema_version: m.power_metric_schema_version,
+    avg_power_w: measuredPowerValid ? m.avg_power_w : undefined,
+    joules_per_output_token:
+      measuredPowerValid && hasWholeDeploymentEnergySemantics
+        ? m.joules_per_output_token
+        : undefined,
+    joules_per_total_token:
+      measuredPowerValid && hasWholeDeploymentEnergySemantics
+        ? m.joules_per_total_token
+        : undefined,
+    // Role power remains unambiguous across schema versions. Version 2 also
+    // publishes explicit role energy alongside whole-deployment joules.
+    prefill_avg_power_w: measuredPowerValid ? m.prefill_avg_power_w : undefined,
+    decode_avg_power_w: measuredPowerValid ? m.decode_avg_power_w : undefined,
+    joules_per_input_token:
+      measuredPowerValid && hasWholeDeploymentEnergySemantics
+        ? m.joules_per_input_token
+        : undefined,
+    prefill_joules_per_input_token: measuredPowerValid
+      ? m.prefill_joules_per_input_token
+      : undefined,
+    decode_joules_per_output_token: measuredPowerValid
+      ? m.decode_joules_per_output_token
+      : undefined,
     // Cluster-wide GPU telemetry beyond power. Emitted when the perfmon CSVs
     // include the corresponding sample columns; left undefined otherwise so
     // the chart layer can distinguish "no measurement" from a real zero.
@@ -187,7 +221,7 @@ export function rowToAggDataEntry(row: BenchmarkRow): AggDataEntry {
     // Per-worker measured power. Surfaced on BenchmarkRow as a sibling of the
     // scalar `metrics` dict (see api.ts). Narrow defensively so a malformed
     // payload can't poison downstream consumers.
-    workers: Array.isArray(row.workers) ? row.workers : undefined,
+    workers: measuredPowerValid && Array.isArray(row.workers) ? row.workers : undefined,
     disagg: row.disagg,
     num_prefill_gpu: row.num_prefill_gpu,
     num_decode_gpu: row.num_decode_gpu,

packages/constants/src/metric-keys.ts

@@ -131,26 +131,30 @@ export const METRIC_KEYS = new Set([
   // server_metrics.kv_cache.gpu_usage_pct in v3)
   'gpu_kv_cache_usage_pct',
   // measured power / energy (emitted by runner's aggregate_power.py)
+  // power_valid: numeric 1/0 publication verdict; explicit 0 withholds power
+  // power_metric_schema_version: version 2 defines every unprefixed
+  //                              joules_per_* field as whole-deployment energy
   // avg_power_w:             mean per-GPU draw (W) during the load window
   // joules_per_output_token: energy / total_output_tokens. CLUSTER-WIDE on
-  //                          single-node / non-disagg (total_system_energy);
-  //                          PER-STAGE decode_energy on disagg (decode GPUs only),
-  //                          symmetric with joules_per_input_token below.
+  //                          schema-version-2 rows, including disaggregated runs.
   // joules_per_total_token:  total_system_energy / (total_input + total_output)
   //                          — cluster-wide; workload-shape-fair view that
   //                          doesn't treat prompt as free.
+  'power_valid',
+  'power_metric_schema_version',
   'avg_power_w',
   'joules_per_output_token',
   'joules_per_total_token',
   // multinode / disagg role splits (emitted only when the deployment has
   // distinct prefill / decode workers)
   // prefill_avg_power_w / decode_avg_power_w:  mean per-GPU draw within each role
-  // joules_per_input_token:  prefill_energy / total_input_tokens (prefill GPUs only).
-  //   The disagg output counterpart is joules_per_output_token above (decode GPUs
-  //   only) — there is no separate _decode key.
+  // Explicit role-local energy remains separate from the version-2 unprefixed
+  // whole-deployment fields.
   'prefill_avg_power_w',
   'decode_avg_power_w',
   'joules_per_input_token',
+  'prefill_joules_per_input_token',
+  'decode_joules_per_output_token',
   // cluster-wide GPU telemetry beyond power (emitted by aggregate_power.py when
   // the perfmon CSVs include temperature, utilization, or memory samples).
   // avg_temp_c:        mean per-GPU temperature (Celsius) during load window

packages/db/src/etl/benchmark-mapper.test.ts

@@ -127,6 +127,108 @@ describe('mapBenchmarkRow', () => {
       expect(result!.metrics).not.toHaveProperty('ep');
     });
 
+    it.each([
+      {
+        name: 'boolean verdict and junk-suffixed schema',
+        input: { power_valid: true, power_metric_schema_version: '2garbage' },
+        expectedVerdict: 0,
+        expectedSchema: undefined,
+      },
+      {
+        name: 'garbage verdict and valid numeric schema',
+        input: { power_valid: 'garbage', power_metric_schema_version: 2 },
+        expectedVerdict: 0,
+        expectedSchema: 2,
+      },
+      {
+        name: 'canonical numeric strings',
+        input: { power_valid: '1', power_metric_schema_version: '2' },
+        expectedVerdict: 1,
+        expectedSchema: 2,
+      },
+      {
+        name: 'explicit invalid verdict',
+        input: { power_valid: 0, power_metric_schema_version: 2 },
+        expectedVerdict: 0,
+        expectedSchema: 2,
+      },
+      {
+        name: 'legacy row without power contract fields',
+        input: {},
+        expectedVerdict: undefined,
+        expectedSchema: undefined,
+      },
+    ])(
+      'normalizes power contract discriminators: $name',
+      ({ input, expectedVerdict, expectedSchema }) => {
+        const tracker = createSkipTracker();
+        const result = mapBenchmarkRow(makeV1Row(input), tracker);
+
+        if (expectedVerdict === undefined) {
+          expect(result!.metrics).not.toHaveProperty('power_valid');
+        } else {
+          expect(result!.metrics.power_valid).toBe(expectedVerdict);
+        }
+        if (expectedSchema === undefined) {
+          expect(result!.metrics).not.toHaveProperty('power_metric_schema_version');
+        } else {
+          expect(result!.metrics.power_metric_schema_version).toBe(expectedSchema);
+        }
+      },
+    );
+
+    it.each([true, false, null, 0.5, 2, Number.NaN, Number.POSITIVE_INFINITY])(
+      'fails closed for explicitly malformed power_valid=%j',
+      (powerValid) => {
+        const tracker = createSkipTracker();
+        const result = mapBenchmarkRow(makeV1Row({ power_valid: powerValid }), tracker);
+
+        expect(result!.metrics.power_valid).toBe(0);
+      },
+    );
+
+    it.each([
+      null,
+      true,
+      false,
+      0,
+      -1,
+      1.5,
+      Number.NaN,
+      Number.POSITIVE_INFINITY,
+      Number.MAX_SAFE_INTEGER + 1,
+      '0',
+      '-1',
+      '02',
+      '2.0',
+      ' 2',
+      '2 ',
+      '2garbage',
+      '1e2',
+      String(Number.MAX_SAFE_INTEGER + 1),
+    ])('omits malformed power_metric_schema_version=%j', (schemaVersion) => {
+      const tracker = createSkipTracker();
+      const result = mapBenchmarkRow(
+        makeV1Row({ power_metric_schema_version: schemaVersion }),
+        tracker,
+      );
+
+      expect(result!.metrics).not.toHaveProperty('power_metric_schema_version');
+    });
+
+    it.each([1, 2, Number.MAX_SAFE_INTEGER, '1', '2', String(Number.MAX_SAFE_INTEGER)])(
+      'accepts safe positive integer power_metric_schema_version=%j',
+      (schemaVersion) => {
+        const tracker = createSkipTracker();
+        const result = mapBenchmarkRow(
+          makeV1Row({ power_metric_schema_version: schemaVersion }),
+          tracker,
+        );
+
+        expect(result!.metrics.power_metric_schema_version).toBe(Number(schemaVersion));
+      },
+    );
+
     it('preserves the KV transfer engine for fixed-sequence multinode rows', () => {
       const tracker = createSkipTracker();
       const result = mapBenchmarkRow(