metrics: expose + persist measured-vs-estimated provenance (#4, #6)

Thread the per-factor source through impact_for() instead of discarding it.
Numeric output (energy_wh, co2e_g) is byte-identical — this is a reporting
change, not a recalculation.

#6 persistence: additive columns tokens_measured, tokens_estimated and
factor_revision (a stable hash of the factor table) on metrics_daily, added via
the existing guarded ALTER TABLE idiom in both app/main.py and
metrics_snapshot.py. Existing rows keep NULL.

#4 API: metrics_series() selects the new columns and derives a per-row
provenance indicator (measured | mixed | estimated | none) from the split; both
/admin/metrics/series and /broker/metrics return it. NULL/unknown splits report
'none', never a confident 'measured'.

docs/energy-metrics.md documents the indicator, the new columns, and the
forward-only-vs-backfill behaviour.

metrics_snapshot.py

@@ -27,6 +27,7 @@ Rules this file follows deliberately:
 """
 
 import datetime as dt
+import hashlib
 import json
 import os
 import sqlite3
@@ -60,10 +61,23 @@ def connect() -> sqlite3.Connection:
                out_tokens INTEGER,
                energy_wh REAL,
                co2e_g REAL,
+               tokens_measured INTEGER,
+               tokens_estimated INTEGER,
+               factor_revision TEXT,
                recorded_at TEXT)"""
     )
     # CREATE IF NOT EXISTS never adds columns to an existing table, so migrate.
-    for col, decl in (("out_tokens", "INTEGER"), ("energy_wh", "REAL"), ("co2e_g", "REAL")):
+    for col, decl in (
+        ("out_tokens", "INTEGER"),
+        ("energy_wh", "REAL"),
+        ("co2e_g", "REAL"),
+        # Measured-vs-estimated provenance (see co-op/support#26 follow-ups).
+        # Additive: existing rows keep NULL, and NULL reports as "unknown", never
+        # as a confident "measured".
+        ("tokens_measured", "INTEGER"),
+        ("tokens_estimated", "INTEGER"),
+        ("factor_revision", "TEXT"),
+    ):
         try:
             conn.execute(f"ALTER TABLE metrics_daily ADD COLUMN {col} {decl}")
         except sqlite3.OperationalError:
@@ -105,6 +119,30 @@ IMPACT_FACTORS = {
 }
 IMPACT_PROXY = {"wh_per_1k_out": 0.5000, "gco2e_per_kwh": 250.0, "source": ESTIMATED}
 
+
+def _factor_revision() -> str:
+    """Short, stable fingerprint of the factor table in force.
+
+    Serialised WITHOUT sort_keys, preserving both the per-factor key order and the
+    table's longest-prefix-first ordering — because reordering IMPACT_FACTORS
+    changes which factor a model matches, so the revision must move with it. A
+    deterministic hash of values rather than a timestamp: two runs against the
+    same table produce the same string, and a chart label can trust it.
+    """
+    payload = json.dumps(
+        {
+            "factors": [[prefix, IMPACT_FACTORS[prefix]] for prefix in IMPACT_FACTORS],
+            "proxy": IMPACT_PROXY,
+        },
+        separators=(",", ":"),
+    )
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:8]
+
+
+# Revision of the CURRENT factor table. Written onto every row computed from it, so
+# a chart can tell which factor set produced a number. See docs/energy-metrics.md.
+FACTOR_REVISION = _factor_revision()
+
 # ---------------------------------------------------------------- sources
 
 def listmonk_joins() -> list[str] | None:
@@ -325,8 +363,38 @@ def tokens_by_day_model(since: dt.date | None = None) -> dict[str, dict[str, int
     return out
 
 
-def impact_for(models: dict[str, int]) -> tuple[float | None, float | None]:
-    """(Wh, gCO2e) for a day's {model: output_tokens}.
+def _no_impact() -> dict:
+    """The impact block for a day with no output tokens: all NULL, fail-closed.
+
+    NULL energy/carbon means "not measured" (see the module docstring). The
+    provenance split is NULL too, so an unknown split cannot be read as a
+    confident "0 measured".
+    """
+    return {
+        "wh": None,
+        "co2e_g": None,
+        "tokens_measured": None,
+        "tokens_estimated": None,
+        "factor_revision": None,
+    }
+
+
+def impact_for(models: dict[str, int]) -> dict:
+    """Impact for a day's {model: output_tokens}, and the provenance split.
+
+    Returns a dict:
+      wh, co2e_g        — totals, unchanged from before (same factors, same
+                          matching, same arithmetic, same rounding).
+      tokens_measured   — output tokens attributed to a factor whose `source` is
+                          MEASURED (a direct provider measurement).
+      tokens_estimated  — the rest: SAME_FAMILY (borrowed from a sibling) and the
+                          flat proxy. A borrowed or proxied factor is an
+                          assumption, so it is NOT counted as measured.
+      factor_revision   — the revision of the factor table used.
+
+    The split is reported, not recomputed: the numeric behaviour is identical to
+    the previous (Wh, gCO2e)-only version. `tokens_measured + tokens_estimated`
+    equals the day's output tokens for any day with tokens.
 
     Factors live in IMPACT_FACTORS. Unknown models fall back to the documented
     proxy row rather than being skipped, so a total is never quietly missing a
@@ -334,15 +402,17 @@ def impact_for(models: dict[str, int]) -> tuple[float | None, float | None]:
     docs say so. Energy: Wh per 1,000 output tokens. Carbon: gCO2e per kWh.
     """
     if not models:
-        return None, None
+        return _no_impact()
     if not sum(t for t in models.values() if t):
         # No measured tokens for this day: report "not measured" (NULL), never a
         # zero. A zero reads as "the co-op used no energy" — a lie that would
         # survive into a chart. The first backfill shipped exactly that, because
         # it summed the wrong column and every model totalled 0.
-        return None, None
+        return _no_impact()
     wh = 0.0
     co2 = 0.0
+    tokens_measured = 0
+    tokens_estimated = 0
     for model, toks in models.items():
         if not toks:
             continue
@@ -352,6 +422,12 @@ def impact_for(models: dict[str, int]) -> tuple[float | None, float | None]:
                 f = factor
                 break
         f = f or IMPACT_PROXY
+        # Provenance is a pure reporting side-channel: only MEASURED is a direct
+        # measurement; SAME_FAMILY (borrowed) and the proxy are not.
+        if f.get("source") == MEASURED:
+            tokens_measured += toks
+        else:
+            tokens_estimated += toks
         wh += toks / 1000.0 * f["wh_per_1k_out"]
         # Wh -> kWh is /1e3, then x gCO2e/kWh. This divided by 1e6 once, which
         # understated carbon exactly 1000x: the energy column stayed plausible
@@ -366,7 +442,13 @@ def impact_for(models: dict[str, int]) -> tuple[float | None, float | None]:
                 f"metrics: WARNING implied grid intensity {implied:.1f} gCO2e/kWh is outside "
                 f"the plausible band — suspect a unit error before trusting these numbers"
             )
-    return round(wh, 6), round(co2, 6)
+    return {
+        "wh": round(wh, 6),
+        "co2e_g": round(co2, 6),
+        "tokens_measured": tokens_measured,
+        "tokens_estimated": tokens_estimated,
+        "factor_revision": FACTOR_REVISION,
+    }
 
 
 # ---------------------------------------------------------------- writes
@@ -378,8 +460,8 @@ def upsert(day: str, row: dict) -> None:
             """INSERT INTO metrics_daily
                    (day, members_active, members_total, members_new, newsletter_subs,
                     matrix_space_members, spend_usd, requests, out_tokens, energy_wh,
-                    co2e_g, recorded_at)
-               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    co2e_g, tokens_measured, tokens_estimated, factor_revision, recorded_at)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                ON CONFLICT(day) DO UPDATE SET
                    members_active=excluded.members_active,
                    members_total=excluded.members_total,
@@ -391,6 +473,9 @@ def upsert(day: str, row: dict) -> None:
                    out_tokens=excluded.out_tokens,
                    energy_wh=excluded.energy_wh,
                    co2e_g=excluded.co2e_g,
+                   tokens_measured=excluded.tokens_measured,
+                   tokens_estimated=excluded.tokens_estimated,
+                   factor_revision=excluded.factor_revision,
                    recorded_at=excluded.recorded_at""",
             (
                 day,
@@ -404,6 +489,9 @@ def upsert(day: str, row: dict) -> None:
                 row.get("out_tokens"),
                 row.get("energy_wh"),
                 row.get("co2e_g"),
+                row.get("tokens_measured"),
+                row.get("tokens_estimated"),
+                row.get("factor_revision"),
                 dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"),
             ),
         )
@@ -424,7 +512,7 @@ def snapshot(day: str | None = None) -> dict:
     day_models = (tokens_by_day_model(since=since).get(day)) or {}
     tot = sum(day_models.values())
     out_tokens = tot or None
-    energy_wh, co2e_g = impact_for(day_models) if tot else (None, None)
+    impact = impact_for(day_models)
     row = {
         "members_active": m["active"],
         "members_total": m["total"],
@@ -434,8 +522,11 @@ def snapshot(day: str | None = None) -> dict:
         "spend_usd": today_spend[0],
         "requests": today_spend[1],
         "out_tokens": out_tokens,
-        "energy_wh": energy_wh,
-        "co2e_g": co2e_g,
+        "energy_wh": impact["wh"],
+        "co2e_g": impact["co2e_g"],
+        "tokens_measured": impact["tokens_measured"],
+        "tokens_estimated": impact["tokens_estimated"],
+        "factor_revision": impact["factor_revision"],
     }
     upsert(day, row)
     print(f"metrics: recorded {day} -> {json.dumps(row)}")
@@ -485,7 +576,12 @@ def backfill() -> None:
         day_models = tokens.get(key) or {}
         if sum(day_models.values()):
             row["out_tokens"] = sum(day_models.values())
-            row["energy_wh"], row["co2e_g"] = impact_for(day_models)
+            impact = impact_for(day_models)
+            row["energy_wh"] = impact["wh"]
+            row["co2e_g"] = impact["co2e_g"]
+            row["tokens_measured"] = impact["tokens_measured"]
+            row["tokens_estimated"] = impact["tokens_estimated"]
+            row["factor_revision"] = impact["factor_revision"]
         if key == today():
             # today's row carries the exact live metrics
             row.update(