test(metrics): migrate e2e coverage to unified metrics

.github/workflows/e2e-bronze-to-api.yml

@@ -24,7 +24,7 @@ name: E2E — Bronze to API
 # The metric-coverage-gate and api-endpoint-coverage-gate jobs then each
 # analyse their own lane's artifact as a pure-Python check (no Docker, no app
 # boot). See
-# src/ingestion/tests/e2e/lib/{metric_coverage,collect_metrics,api_coverage}.py.
+# src/ingestion/tests/e2e/lib/{metric_coverage,collect_metric_definitions,api_coverage}.py.
 #
 # Specs: docs/domain/bronze-to-api-e2e/specs/
 # Code:  src/ingestion/tests/e2e/
@@ -188,7 +188,7 @@ jobs:
       # `./e2e.sh test meta/`.
       #
       # This session ALSO emits BOTH .artifacts/ files while analytics is up:
-      #   • catalog_metrics.json     (lib/collect_metrics.py)
+      #   • metric_definitions.json  (lib/collect_metric_definitions.py)
       #   • observed_endpoints.json  (conftest.pytest_sessionfinish — httpx hook
       #     ledger)
       # but only the lane-relevant one is consumed downstream: this api lane
@@ -294,7 +294,7 @@ jobs:
       # `./e2e.sh test meta/`.
       #
       # This session ALSO emits BOTH .artifacts/ files while analytics is up:
-      #   • catalog_metrics.json     (lib/collect_metrics.py)
+      #   • metric_definitions.json  (lib/collect_metric_definitions.py)
       #   • observed_endpoints.json  (conftest.pytest_sessionfinish — httpx hook
       #     ledger)
       # but only the lane-relevant one is consumed downstream: this metrics
@@ -371,10 +371,10 @@ jobs:
         run: |
           pip install --quiet pyyaml
           python3 src/ingestion/tests/e2e/lib/metric_coverage.py \
-            --universe-file coverage-inputs/catalog_metrics.json | tee metric_coverage.md
+            --universe-file coverage-inputs/metric_definitions.json | tee metric_coverage.md
           rc=${PIPESTATUS[0]}
           if [ -n "${GITHUB_STEP_SUMMARY:-}" ]; then cat metric_coverage.md >> "$GITHUB_STEP_SUMMARY"; fi
-          if [ "$rc" -ne 0 ]; then echo "::error::metric coverage gate failed — a metric_key is value-tested by no test and not in SKIP_TABLES/SKIP_LIST (or a skip is stale). See the report."; fi
+          if [ "$rc" -ne 0 ]; then echo "::error::metric coverage gate failed — builtin metric coverage is incomplete or stale. See the report."; fi
           exit "$rc"
 
   api-endpoint-coverage-gate:

src/ingestion/tests/e2e/README.md

@@ -19,8 +19,8 @@
 Test framework that exercises the full data path:
 
 ```
-metrics/<name>.test.yaml (bronze records)  →  bronze tables  →  dbt staging/silver  →
-ClickHouse migration gold-views  →  analytics HTTP (POST /v1/metrics/queries)  →  expect rules
+metrics/<name>.test.yaml (bronze records)  →  bronze tables  →  dbt staging/silver/gold  →
+analytics HTTP (POST /v1/metric-results)  →  expect rules
 ```
 
 Airbyte / Kestra / Argo are NOT exercised — bronze is seeded by direct INSERT of the
@@ -82,9 +82,9 @@ e2e/
 │   ├── migration_applier.py    # applies src/ingestion/scripts/migrations/*.sql
 │   ├── analytics.py        # builds + spawns the analytics binary
 │   ├── worker.py               # WorkerContext (resolves pytest-xdist worker id)
-│   ├── metric_coverage.py      # metric-coverage gate: SKIP_TABLES + SKIP_LIST (--universe-file)
+│   ├── metric_coverage.py      # builtin metric coverage gate
 │   ├── api_coverage.py         # endpoint-coverage report + httpx recording hook
-│   ├── collect_metrics.py      # script: snapshot the metric catalog → .artifacts/
+│   ├── collect_metric_definitions.py # snapshot enabled builtin definitions
 │   └── config.py               # session config (ports, random creds)
 ├── seed/
 │   └── metrics.yaml            # optional test-specific metric overrides (default: empty)
@@ -94,24 +94,18 @@ e2e/
 
 ## Metric coverage gate
 
-A job (`metric-coverage-gate`) in the **E2E — Bronze to API** workflow, *not* a pytest test. The `e2e-metrics` lane runs the metrics/ suite and, while analytics is up, snapshots the metric catalog (`POST /v1/catalog/get_metrics`) to `.artifacts/catalog_metrics.json` (uploaded as `coverage-inputs-metrics`); the gate job then checks every product `metric_key` the catalog exposes is value-asserted by a test or covered by a `SKIP_TABLES`/`SKIP_LIST` entry — pure Python, no Docker, no second app boot.
+A job (`metric-coverage-gate`) in the **E2E — Bronze to API** workflow checks enabled builtin product definitions. The metrics lane snapshots `metric_key`, computation, dimensions, and cohort metadata to `.artifacts/metric_definitions.json`; the gate then verifies every builtin has period, peer, and timeseries assertions, plus breakdown assertions for declared dimensions and histogram assertions for median metrics.
 
 Locally, after a run:
 
 ```bash
-./e2e.sh test metrics/    # runs the metrics suite (emits both .artifacts files; only catalog_metrics.json feeds this gate)
+./e2e.sh test metrics/    # runs the metrics suite and writes metric_definitions.json
 ./e2e.sh gates metrics    # metric gate only, against .artifacts/ (in the runner image; no DB)
 ```
 
 `./e2e.sh gates` with no argument runs both gates (handy after running both suites locally; see [API endpoint coverage gate](#api-endpoint-coverage-gate) below for the api/-only equivalent). `gates api` / `gates metrics` run one gate against one artifact each — that per-lane shape is what mirrors the two independent CI jobs, each of which only ever needs its own lane's artifact.
 
-The verdict per **metric_key** (each individual number) is **binary**:
-
-- **value-tested** — a `metrics/*.test.yaml` asserts it (`find: {metric_key: …}` paired with `equal`/`assert`) → **PASS**
-- **skip-listed** (in the inline `SKIP_LIST` in [`lib/metric_coverage.py`](lib/metric_coverage.py)) → **PASS** (baseline)
-- **neither** → **FAIL** — a number nobody validates must get an assertion or a `SKIP_LIST` entry.
-
-Catalog keys are dotted (`collab_bullet_rows.m365_emails_sent`); a test asserts the bare response key (`m365_emails_sent`). The column suffix is unique across the catalog, so the gate maps bare→dotted by suffix (a future collision raises). `SKIP_LIST` is the accepted baseline and single source of truth (no side-car file — just `(metric_key, reason)`). Kept honest: a **stale** entry (key no longer in the catalog), a **redundant** one (now value-tested), or a test asserting a **non-catalog** key (typo / unseeded → matches 0 rows) all fail. PASS iff no FAILs.
+Missing views, unknown asserted keys, unknown requested keys, and stale fixtures fail. There are no legacy suffix mappings or skip lists.
 
 ```bash
 # ad hoc against a running analytics (instead of the collected artifact):
@@ -133,7 +127,7 @@ Locally, after a run:
 
 Per-status-code coverage is **reported, not enforced**: the report renders an endpoints × registered-status-codes table (`✓` observed · `✗` declared but not yet observed · `·` excluded · blank = not declared) and an overall coverage percentage. A code is *coverable* — and so counts toward the percentage — only if a black-box rig can produce it: `coverable(op) = declared(op) − {codes ≥ 500} − UNIVERSAL_BOILERPLATE{401,429} − BLOCKED[op]`. `BLOCKED` absorbs the committed spec's `.standard_errors` over-declaration (#1669) plus pinned rig/product limits, and a `·` code that becomes observed (or a `BLOCKED` op dropped from the spec) is surfaced as a non-blocking advisory so the list stays honest.
 
-The [`api/`](api/) contract suite covers all 21 spec operations — one module per path group (`test_metrics.py`, `test_metric_thresholds.py`, `test_admin_thresholds.py`, `test_catalog.py`, `test_columns.py`, `test_persons.py`, `test_metric_results.py`), one test per (path, method, status-code) case, from self-cleaning fixtures (`api/conftest.py`) — so `SKIP_LIST` is empty and adding a spec operation without a test fails the gate as MISSING. Spec/product gaps are pinned by **strict xfails** rather than fixed here: #1663 (legacy threshold reads 500 once a row exists), #1664 (duplicate admin create answers 500 instead of the declared 409), and #1670 (off-schema legacy body answers a non-canonical 422, not the intended 400). `POST /v1/metric-results` — the unified-metric compute endpoint added by the `feat/unified-metrics` merge (#1656) — is covered on its deterministic error paths (400 empty/bad-period/unknown-key, 415 wrong content-type); its 200 happy-path needs seeded unified-metric observation data and shows as a reported `✗` gap until that fixture lands.
+The [`api/`](api/) contract suite covers all 21 spec operations — one module per path group (`test_metrics.py`, `test_metric_thresholds.py`, `test_admin_thresholds.py`, `test_catalog.py`, `test_columns.py`, `test_persons.py`, `test_metric_results.py`), one test per (path, method, status-code) case, from self-cleaning fixtures (`api/conftest.py`). The declarative metrics suite covers successful `/v1/metric-results` computation for every builtin metric.
 
 ## Ports (loopback only)
 
@@ -153,16 +147,17 @@ These ports avoid conflict with a local gitops dev cluster (which forwards 8123
 
 ## `cases` / `expect` (declarative YAML rig)
 
-Tests are `metrics/**/*.test.yaml`; each `case` POSTs a batch to `/v1/metrics/queries` and checks an `expect` list of rules. A rule selects with `in` (batch result by `id`) + an exact-equality `find` (`{field: value}`), then asserts via `equal` (subset of fields, exact / `null`) or `assert` (a CEL boolean). Anything richer than equality (inequalities, counts, predicates) goes in a CEL `assert` — the rig deliberately has no second selector language. See the [yaml-rig FEATURE](../../../../docs/domain/bronze-to-api-e2e/specs/feature-yaml-rig/FEATURE.md) and the `/metric` skill.
+Tests are `metrics/**/*.test.yaml`; each case posts to `/v1/metric-results`. Rules select a full builtin `metric` key and `view`, then optionally use `find` for a row and `equal` for exact or null values. Use CEL `assert` for nested timeseries points, dimensions, histogram bins, counts, and predicates.
 
 Variables available in an `assert` (CEL) expression — assembled in `lib/expect_engine.py::evaluate_case` (the `bindings` dict), converted to CEL in `_eval_cel`:
 
 | Binding | Value | Present when |
 |---------|-------|--------------|
 | `it` | the single row matched by `find` | only with `find` (else `null`) |
-| `items` | the selected result's `items` array | a result is selected (`in` or sole query) |
-| `result` | the selected batch result `{id, status, metric_id, items, page_info}` | a result is selected |
-| `results` | the full `results[]` of the batch | always |
+| `items` | values or series from the selected view | a metric view is selected |
+| `view` | the selected view object | a metric view is selected |
+| `metric` | the selected metric result | a metric is selected |
+| `metrics` | all metric results | always |
 | `status` | the batch HTTP status code (int) | always |
 
 CEL is strictly typed and will not compare an `int` to a `double`. Bindings are passed through unchanged, so when a metric value may be integral (e.g. `40`) and you compare against a fractional literal, cast it: `double(it.value) > 39.5`. `status` and `size(...)` are integers — compare them with integer literals. Use `equal` for exact / `null` comparisons (it uses Python `==`).
@@ -171,25 +166,22 @@ CEL is strictly typed and will not compare an `int` to a `double`. Bindings are
 
 `assert` expressions are written in **CEL — the [Common Expression Language](https://github.com/google/cel-spec)** (the same expression language used by Kubernetes admission policies and Envoy). It is a small, side-effect-free language for boolean/value expressions over structured data: no statements, no loops, no I/O — an expression is evaluated against the bindings above and must return a boolean. The rig evaluates it with the [`cel-python`](https://pypi.org/project/cel-python/) library (`celpy`) in `lib/expect_engine.py::_eval_cel`.
 
-Operators: `== != < <= > >=`, `&& || !`, `+ - * / %`, `in`, ternary `cond ? a : b`. Field/index access: `it.value`, `result.status`, `items[0]`. Useful built-ins & macros: `size(x)`, `has(x.field)`, `x.exists(e, <pred>)`, `x.all(e, <pred>)`, `x.filter(e, <pred>)`, `x.map(e, <expr>)`, string `.startsWith()/.endsWith()/.contains()/.matches(re)`.
+Operators: `== != < <= > >=`, `&& || !`, `+ - * / %`, `in`, ternary `cond ? a : b`. Field/index access: `it.value`, `view.values`, `items[0]`. Useful built-ins & macros: `size(x)`, `has(x.field)`, `x.exists(e, <pred>)`, `x.all(e, <pred>)`, `x.filter(e, <pred>)`, `x.map(e, <expr>)`, string `.startsWith()/.endsWith()/.contains()/.matches(re)`.
 
 Examples:
 
 ```yaml
-- assert: "status == 200"                                  # batch HTTP code
-- in: collaboration
-  assert: "result.status == 'ok'"                           # this query's own status
-- in: collaboration
-  assert: "size(items) == 20"                               # row count
-- in: collaboration
-  find: { metric_key: m365_emails_sent }
-  assert: "double(it.value) > 39.5 && double(it.value) < 40.5"   # cast to double for fractional compare
-- in: collaboration
-  find: { metric_key: slack_dm_ratio }
-  assert: "it.value == null"                                # explicit null
-- assert: "results.exists(r, r.status == 'error')"          # any query in the batch failed?
-- in: collaboration
-  assert: "items.all(r, r.range_min <= r.value)"            # invariant across all rows
+- assert: "status == 200"
+- metric: collab.emails_sent
+  view: period
+  find: { entity_id: erin@example.com }
+  equal: { value: 50 }
+- metric: collab.messages_sent
+  view: timeseries
+  assert: "items.exists(s, s.points.exists(p, double(p.value) > 0.0))"
+- metric: git.lines_added
+  view: breakdown
+  assert: "items.exists(v, v.dimensions.exists(d, d.key == 'category' && d.value == 'code'))"
 ```
 
 Prefer `equal` for exact / `null` checks (it uses Python `==`, so `40 == 40.0` and `value: null` work directly); reach for `assert` when you need inequalities, counts, or cross-row predicates.

src/ingestion/tests/e2e/api/test_metric_results.py

@@ -3,22 +3,16 @@
   POST /v1/metric-results   200 compute · 400 (empty/bad-period/unknown-key) · 415 wrong-ct
 
 Added when the `feat/unified-metrics` merge (#1656) introduced this operation to
-the committed spec. It computes results for CATALOG metrics (`metric_key` +
-computation spec over the observation gold-views) — a different system from the
-legacy `query_ref` metrics the `/v1/metrics` CRUD and the scratch fixtures use.
+the committed spec. It computes builtin metrics over unified observation models.
 
 The endpoint validates its request body in `domain::metric_results::validate_request`
 BEFORE touching ClickHouse, so the whole 400 family is deterministic in the rig:
 an empty `metrics`, a malformed/reversed `period`, and an unknown `metric_key`
 (which is a 400 via `unavailable`, NOT a 404) all reject up front. Wrong
 Content-Type is a 415 at the `axum::Json` extractor.
 
-The 200 happy-path is deferred: a deterministic compute needs a seeded catalog
-metric whose observation gold-view is built in the rig, which this branch does
-not yet stand up — the coverage report marks metric-results' 200 as a `✗` gap
-until that fixture lands. The endpoint-coverage gate blocks on an *unexercised*
-operation, not on that per-code gap, so these error-path cases are enough to
-keep the new endpoint covered.
+The declarative metric suite exercises deterministic 200 responses for every
+builtin metric; this module retains the endpoint contract error cases.
 """
 
 from __future__ import annotations

src/ingestion/tests/e2e/conftest.py

@@ -206,37 +206,18 @@ def dbt_runner(ch_migrations_applied: SessionConfig):
     runner.cleanup()
 
 
-def _collect_metrics(proc: AnalyticsProcess) -> None:
-    """Run `lib/collect_metrics.py` (a script — NOT a test) against the
-    live API, primary worker only. Snapshots the metric catalog into `.artifacts/`
-    so the metric-coverage gate analyses a file with no second app boot.
-    Best-effort: a failure just means the gate finds no artifact and fails loudly —
-    never abort the session for it. Must run while the API is up (called from
-    analytics teardown, before proc.stop())."""
+def _collect_metrics(cfg: SessionConfig) -> None:
     if not _IS_PRIMARY:
         return
-    import subprocess
-    import sys
+    from lib.collect_metric_definitions import collect
 
-    script = Path(__file__).parent / "lib" / "collect_metrics.py"
     out_dir = Path(__file__).parent / ".artifacts"
-    result = subprocess.run(
-        [
-            sys.executable,
-            str(script),
-            "--url",
-            proc.base_url,
-            "--out-dir",
-            str(out_dir),
-            "--tenant",
-            str(TEST_TENANT_ID),
-        ],
-        check=False,
-    )
-    if result.returncode != 0:
+    try:
+        collect(cfg, out_dir)
+    except Exception as error:
         LOG.warning(
-            "coverage-artifact collection failed (rc=%d); gate jobs may lack inputs",
-            result.returncode,
+            "coverage-artifact collection failed: %s; gate jobs may lack inputs",
+            error,
         )
 
 
@@ -258,7 +239,12 @@ def identity_stub():
 
 
 @pytest.fixture(scope="session")
-def analytics(ch_migrations_applied: SessionConfig, identity_stub: IdentityStub):
+def analytics(
+    ch_migrations_applied: SessionConfig,
+    dbt_runner: DbtRunner,
+    worker_ctx: WorkerContext,
+    identity_stub: IdentityStub,
+):
     """Spawn the analytics binary baked into the runner image. Its SeaORM
     migrations run on startup; we then upsert test-specific metrics from
     seed/metrics.yaml.
@@ -270,6 +256,7 @@ def analytics(ch_migrations_applied: SessionConfig, identity_stub: IdentityStub)
     bronze→API tests cannot run, so the only honest result is red.
     """
     cfg = ch_migrations_applied
+    dbt_runner.run("tag:gold", worker_ctx=worker_ctx)
     from lib.analytics import ApiSpawnError  # local import to keep top clean
     try:
         binary = locate_binary(cfg)
@@ -280,11 +267,8 @@ def analytics(ch_migrations_applied: SessionConfig, identity_stub: IdentityStub)
     proc.start()
     seed_test_metrics(cfg)
     yield proc
-    # Snapshot the metric catalog while the API is still up (a script, run via
-    # subprocess — see _collect_metrics). Always
-    # stop the process afterward, even if collection raised.
     try:
-        _collect_metrics(proc)
+        _collect_metrics(cfg)
     finally:
         proc.stop()
 

src/ingestion/tests/e2e/e2e.sh

@@ -98,8 +98,8 @@ case "$cmd" in
                 ;;
         esac
         if [ "$which" = all ] || [ "$which" = metrics ]; then
-            if [ ! -f .artifacts/catalog_metrics.json ]; then
-                echo "no .artifacts/catalog_metrics.json — run './e2e.sh test metrics/' first (it collects the metric catalog)" >&2
+            if [ ! -f .artifacts/metric_definitions.json ]; then
+                echo "no .artifacts/metric_definitions.json — run './e2e.sh test metrics/' first" >&2
                 exit 2
             fi
         fi
@@ -114,7 +114,7 @@ case "$cmd" in
         rc=0
         if [ "$which" = all ] || [ "$which" = metrics ]; then
             echo "── metric coverage (gate) ──"
-            "${run[@]}" python3 lib/metric_coverage.py --universe-file .artifacts/catalog_metrics.json || rc=1
+            "${run[@]}" python3 lib/metric_coverage.py --universe-file .artifacts/metric_definitions.json || rc=1
         fi
         if [ "$which" = all ] || [ "$which" = api ]; then
             echo "── api endpoint coverage (gate) ──"

src/ingestion/tests/e2e/lib/analytics.py

@@ -257,8 +257,7 @@ def client(self) -> httpx.Client:
     def call_request(self, request: dict) -> tuple[int, Any]:
         """Execute a `case.request` ({url, method, body}). Return (status_code, json|text).
 
-        Used by the YAML rig; the primary endpoint is the batch
-        `POST /v1/metrics/queries`. The body is sent as JSON.
+        Used by the YAML rig. The body is sent as JSON.
         """
         url = request["url"]
         method = str(request.get("method", "POST")).upper()