fix(intake): Derive evaluator_results name from Harbor reward keys

services/intake/src/nmp/intake/spans/ingest/atif_mapping.py

@@ -497,10 +497,10 @@ def trajectory_to_evaluator_results(
 ) -> list[EvaluatorResult]:
     """Extract evaluator_results rows from an ATIF trajectory's verifier_result block.
 
-    Returns one evaluator_result row targeting the EVALUATOR-kind span that
-    ``trajectory_to_spans`` already produced for the verifier. The span preserves
-    the original tree structure; this row makes the score queryable by name and
-    value without parsing the span payload.
+    Emits one row per Harbor reward key (``verifier_result.rewards``), named by that
+    key, targeting the EVALUATOR-kind span that ``trajectory_to_spans`` produced for
+    the verifier. The span preserves the original tree structure; these rows make each
+    score queryable by name and value without parsing the span payload.
     """
 
     extra = trajectory.extra or {}
@@ -510,30 +510,47 @@ def trajectory_to_evaluator_results(
     evaluator_span = next((span for span in spans if span.kind == SpanKind.EVALUATOR), None)
     if evaluator_span is None:
         return []
-    score = _evaluator_score(verifier_result)
-    if score is None:
-        return []
-    data_type, value, string_value = _coerce_evaluator_value(score)
-    return [
-        EvaluatorResult(
-            evaluator_result_id=stable_id(
-                evaluator_span.external_span_id,
-                "harbor.verifier",
-                prefix="eval",
-            ),
-            span_id=evaluator_span.external_span_id,
-            session_id=trajectory.session_id,
-            workspace=workspace,
-            name="harbor.verifier",
-            value=value,
-            string_value=string_value,
-            data_type=data_type,
-            comment=None,
-            created_by="intake:atif_importer",
-            created_at=ingested_at,
-            ingested_at=ingested_at,
+    results: list[EvaluatorResult] = []
+    for name, raw_value in _evaluator_rewards(verifier_result):
+        data_type, value, string_value = _coerce_evaluator_value(raw_value)
+        results.append(
+            EvaluatorResult(
+                # Per-key id: the reward name keeps each criterion's row distinct on the
+                # same span, and an identical re-ingest hashes to the same id (dedupe).
+                evaluator_result_id=stable_id(evaluator_span.external_span_id, name, prefix="eval"),
+                span_id=evaluator_span.external_span_id,
+                session_id=trajectory.session_id,
+                workspace=workspace,
+                name=name,
+                value=value,
+                string_value=string_value,
+                data_type=data_type,
+                comment=None,
+                created_by="intake:atif_importer",
+                created_at=ingested_at,
+                ingested_at=ingested_at,
+            )
         )
-    ]
+    return results
+
+
+def _evaluator_rewards(verifier_result: dict[str, Any]) -> list[tuple[str, bool | int | float | str]]:
+    """Per-reward ``(name, value)`` pairs from a Harbor ``verifier_result``.
+
+    Harbor writes a ``rewards`` dict (``reward.json``) whose keys are the metric
+    identities — one named reward per ``tests/`` subdirectory, or the 1D
+    ``{"reward": <score>}`` convention. Each key becomes its own evaluator_result, so
+    multi-criterion verifiers keep their per-criterion breakdown and keys (incl.
+    namespaced ones like ``v1/correctness``) pass through verbatim. Falls back to a
+    single ``reward`` row when only a bare top-level ``score`` scalar is present.
+    """
+    rewards = verifier_result.get("rewards")
+    if isinstance(rewards, dict):
+        return [(name, value) for name, value in rewards.items() if isinstance(value, (int, float, str))]
+    score = verifier_result.get("score")
+    if isinstance(score, (int, float, str)):
+        return [("reward", score)]
+    return []
 
 
 def _coerce_evaluator_value(

services/intake/tests/integration/spans/test_evaluator_results_atif.py

@@ -34,13 +34,77 @@ def test_atif_ingest_extracts_verifier_reward_into_evaluator_results(client: Tes
     rows = listed.json()["data"]
     assert len(rows) == 1
     row = rows[0]
-    assert row["name"] == "harbor.verifier"
+    # 1D reward convention -> named by its key, "reward" (not the old hardcoded "harbor.verifier").
+    assert row["name"] == "reward"
     assert row["data_type"] == "NUMERIC"
     assert row["value"] == 0.42
     assert row["session_id"] == "atif-eval-session"
     assert row["created_by"] == "intake:atif_importer"
 
 
+def test_atif_ingest_emits_one_row_per_reward_key(client: TestClient):
+    body = {
+        "schema_version": "ATIF-v1.6",
+        "session_id": "atif-multi-criterion",
+        "extra": {
+            "task_name": "eval-task",
+            "verifier_result": {"rewards": {"correctness": 0.75, "structure": 1.0, "v1/quality": 0.5}},
+        },
+        "agent": {"name": "agent-x", "version": "1.0"},
+        "steps": [],
+    }
+    response = client.post(ATIF_INGEST, json=body)
+    assert response.status_code == 201, response.text
+
+    listed = client.get(EVAL_BASE, params={"page_size": 50})
+    assert listed.status_code == 200, listed.text
+    rows = listed.json()["data"]
+    # One row per reward key; namespaced keys (v1/quality) pass through verbatim.
+    assert {row["name"]: row["value"] for row in rows} == {
+        "correctness": 0.75,
+        "structure": 1.0,
+        "v1/quality": 0.5,
+    }
+
+
+def test_atif_re_ingest_dedupes_per_reward(client: TestClient):
+    body = {
+        "schema_version": "ATIF-v1.6",
+        "session_id": "atif-reingest",
+        "extra": {
+            "task_name": "eval-task",
+            "verifier_result": {"rewards": {"correctness": 0.75, "structure": 1.0}},
+        },
+        "agent": {"name": "agent-x", "version": "1.0"},
+        "steps": [],
+    }
+    assert client.post(ATIF_INGEST, json=body).status_code == 201, "first ingest"
+    assert client.post(ATIF_INGEST, json=body).status_code == 201, "identical re-ingest"
+
+    listed = client.get(EVAL_BASE, params={"page_size": 50})
+    # Deterministic per-(span, key) ids -> identical re-ingest dedupes, no doubling.
+    assert listed.json()["pagination"]["total_results"] == 2
+    assert {row["name"] for row in listed.json()["data"]} == {"correctness", "structure"}
+
+
+def test_atif_ingest_falls_back_to_reward_for_bare_score(client: TestClient):
+    body = {
+        "schema_version": "ATIF-v1.6",
+        "session_id": "atif-bare-score",
+        "extra": {"task_name": "eval-task", "verifier_result": {"score": 0.9}},
+        "agent": {"name": "agent-x", "version": "1.0"},
+        "steps": [],
+    }
+    response = client.post(ATIF_INGEST, json=body)
+    assert response.status_code == 201, response.text
+
+    listed = client.get(EVAL_BASE, params={"page_size": 50})
+    rows = listed.json()["data"]
+    assert len(rows) == 1
+    assert rows[0]["name"] == "reward"
+    assert rows[0]["value"] == 0.9
+
+
 def test_atif_ingest_without_verifier_result_writes_no_evaluator_results(client: TestClient):
     body = {
         "schema_version": "ATIF-v1.6",

services/intake/tests/integration/spans/test_experiment_rollups.py

@@ -67,10 +67,10 @@ def test_experiment_response_hydrates_clickhouse_rollups(client: TestClient) ->
     experiment = fetched.json()
 
     assert experiment["run_count"] == 4
-    assert experiment["evaluator_names"] == ["harbor.verifier"]
+    assert experiment["evaluator_names"] == ["reward"]
     assert experiment["model_names"] == ["provider/sample-model"]
 
-    score = experiment["aggregate_scores"]["harbor.verifier"]
+    score = experiment["aggregate_scores"]["reward"]
     assert score["sum"] == pytest.approx(3.0)
     assert score["mean"] == pytest.approx(0.75)
     assert score["median"] == pytest.approx(0.8)
@@ -100,7 +100,7 @@ def test_experiment_response_hydrates_clickhouse_rollups(client: TestClient) ->
     listed = client.get(EXPERIMENTS)
     assert listed.status_code == 200, listed.text
     listed_experiment = next(item for item in listed.json()["data"] if item["name"] == experiment_id)
-    assert listed_experiment["aggregate_scores"]["harbor.verifier"]["mean"] == pytest.approx(0.75)
+    assert listed_experiment["aggregate_scores"]["reward"]["mean"] == pytest.approx(0.75)
 
 
 def test_atif_ingest_rejects_deleted_experiment(client: TestClient) -> None:
@@ -193,7 +193,7 @@ def test_deprecated_evaluation_context_hydrates_experiment_rollups(client: TestC
     assert fetched.status_code == 200, fetched.text
     experiment = fetched.json()
     assert experiment["run_count"] == 1
-    assert experiment["aggregate_scores"]["harbor.verifier"]["mean"] == pytest.approx(1.0)
+    assert experiment["aggregate_scores"]["reward"]["mean"] == pytest.approx(1.0)
 
 
 def _atif_body(

services/intake/tests/integration/spans/test_experiment_sessions.py

@@ -89,7 +89,7 @@ def test_list_experiment_sessions_returns_joined_session_rows(client: TestClient
     assert case_a["input_tokens"] == 100
     assert case_a["output_tokens"] == 10
     assert case_a["cost_total_usd"] == pytest.approx(0.05)
-    assert case_a["evaluator_scores"] == {"harbor.verifier": pytest.approx(1.0)}
+    assert case_a["evaluator_scores"] == {"reward": pytest.approx(1.0)}
     assert case_a["status"] in {"success", "unknown"}
 
     paged = client.get(f"{EXPERIMENTS}/{experiment_name}/sessions", params={"page": 2, "page_size": 1})
@@ -98,7 +98,7 @@ def test_list_experiment_sessions_returns_joined_session_rows(client: TestClient
     assert paged_body["pagination"]["total_results"] == 3
     assert len(paged_body["data"]) == 1
     assert paged_body["data"][0]["test_case_id"] == "case-b"
-    assert paged_body["data"][0]["evaluator_scores"] == {"harbor.verifier": pytest.approx(0.5)}
+    assert paged_body["data"][0]["evaluator_scores"] == {"reward": pytest.approx(0.5)}
 
 
 def test_list_experiment_sessions_filter_by_test_case(client: TestClient) -> None:

services/intake/tests/test_experiment_rollup_repository.py

@@ -44,7 +44,7 @@ async def test_experiment_rollups_anchor_on_root_session_membership():
                 ["experiment_id", "run_count"],
             ),
             _QueryResult(
-                [("exp-a", "harbor.verifier", 3.0, 0.75, 0.8, 1.0, 1.0, 1.0, 4)],
+                [("exp-a", "reward", 3.0, 0.75, 0.8, 1.0, 1.0, 1.0, 4)],
                 ["experiment_id", "evaluator_name", "sum", "mean", "median", "p90", "p95", "p99", "count"],
             ),
             _QueryResult(
@@ -99,17 +99,17 @@ async def test_experiment_rollups_anchor_on_root_session_membership():
 
     rollup = rollups["exp-a"]
     assert rollup.run_count == 3
-    assert rollup.evaluator_names == ["harbor.verifier"]
+    assert rollup.evaluator_names == ["reward"]
     assert rollup.model_names == ["model-a", "model-b"]
     assert rollup.agent_names == ["agent-a"]
     assert rollup.agent_versions == ["1.0.0", "1.0.1"]
-    assert rollup.evaluator_scores["harbor.verifier"].sum == 3.0
-    assert rollup.evaluator_scores["harbor.verifier"].mean == 0.75
-    assert rollup.evaluator_scores["harbor.verifier"].median == 0.8
-    assert rollup.evaluator_scores["harbor.verifier"].p90 == 1.0
-    assert rollup.evaluator_scores["harbor.verifier"].p95 == 1.0
-    assert rollup.evaluator_scores["harbor.verifier"].p99 == 1.0
-    assert rollup.evaluator_scores["harbor.verifier"].count == 4
+    assert rollup.evaluator_scores["reward"].sum == 3.0
+    assert rollup.evaluator_scores["reward"].mean == 0.75
+    assert rollup.evaluator_scores["reward"].median == 0.8
+    assert rollup.evaluator_scores["reward"].p90 == 1.0
+    assert rollup.evaluator_scores["reward"].p95 == 1.0
+    assert rollup.evaluator_scores["reward"].p99 == 1.0
+    assert rollup.evaluator_scores["reward"].count == 4
     assert rollup.cost_usd is not None
     assert rollup.cost_usd.sum == 0.65
     assert rollup.cost_usd.mean == 0.1625