feat(OMN-14018): per-run backend pin on context-ROI runner + generation-consumer honor path (ROI tier-flip enrichment)

scripts/run_context_roi_battery.py

@@ -224,8 +224,16 @@ def build_run_request(
     arm_order_seed: int = 42,
     generation_timeout_seconds: float = 120.0,
     artifact_content_map: dict[str, str] | None = None,
+    target_endpoint: str = "",
 ) -> ModelContextRoiRunRequest:
-    """Construct the contract-canonical runner command for the full matrix."""
+    """Construct the contract-canonical runner command for the full matrix.
+
+    ``target_endpoint`` (OMN-14018): when set to a routing-tier backend id (e.g.
+    ``cloud-glm``) the runner pins every generation to that backend's tier so the
+    battery captures genuine per-tier graded rows for ``context_roi_scores``
+    (a real, sub-floor ``cheap_cloud`` cohort instead of the always-``local``
+    contract route). Empty preserves contract-declared routing.
+    """
     return ModelContextRoiRunRequest(
         run_id=run_id or f"context-roi-battery-{uuid.uuid4().hex[:12]}",
         tasks=build_difficulty_ladder(),
@@ -235,6 +243,7 @@ def build_run_request(
         arm_order_seed=arm_order_seed,
         generation_timeout_seconds=generation_timeout_seconds,
         artifact_content_map=artifact_content_map or {},
+        target_endpoint=target_endpoint,
     )
 
 
@@ -467,12 +476,17 @@ def run_battery(
     artifact_content_map: dict[str, str] | None = None,
     bus_factory: Callable[[], ProtocolBatteryBus] | None = None,
     run_id: str | None = None,
+    target_endpoint: str = "",
 ) -> ModelBatteryReport:
     """Run the full (task x arm x trial) battery.
 
     dry_run=True validates wiring (matrix, request models, cell count, manifest
     shape) without constructing or invoking any bus - the runner.handle() loop
     is never entered and nothing is published.
+
+    ``target_endpoint`` (OMN-14018) pins every generation to a chosen routing-tier
+    backend so the battery captures genuine per-tier graded rows; empty = normal
+    contract-declared routing.
     """
     out_dir = output_dir or _DEFAULT_OUTPUT_DIR
     experiment_id = f"context-roi-exp-{uuid.uuid4().hex[:12]}"
@@ -484,6 +498,7 @@ def run_battery(
         arm_order_seed=arm_order_seed,
         generation_timeout_seconds=generation_timeout_seconds,
         artifact_content_map=artifact_content_map,
+        target_endpoint=target_endpoint,
     )
     expected_cell_count = len(request.tasks) * len(request.arms) * trials_per_cell
 
@@ -616,6 +631,17 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
             "No hardcoded paths - the caller supplies the source."
         ),
     )
+    parser.add_argument(
+        "--target-endpoint",
+        type=str,
+        default="",
+        help=(
+            "OMN-14018: pin every generation to this routing-tier backend id "
+            "(e.g. 'cloud-glm') so the battery captures genuine per-tier graded "
+            "rows for context_roi_scores. Empty = contract-declared routing "
+            "(always local). A ref that maps to no tier degrades to normal routing."
+        ),
+    )
     parser.add_argument(
         "--dry-run",
         action="store_true",
@@ -641,6 +667,7 @@ def main(argv: list[str] | None = None) -> int:
         dry_run=args.dry_run,
         output_dir=args.output_dir,
         artifact_content_map=artifact_content_map,
+        target_endpoint=args.target_endpoint,
     )
 
     if report.dry_run:

src/omnimarket/nodes/node_context_roi_runner/handlers/handler_context_roi_runner.py

@@ -680,14 +680,19 @@ def _run_trial(
                 register(correlation_id)
 
         # --- Step 2b: publish generation command over the bus ---
-        command_payload = {
+        command_payload: dict[str, Any] = {
             "task_description": task.task_description,
             "correlation_id": correlation_id,
             "max_attempts": request.max_attempts,
             "context_pack": context_pack_text,
             "context_artifacts": [],
             "context_pack_hash": context_pack_hash,
         }
+        # OMN-14018: pin this run's generation to a chosen backend when the request
+        # carries a target_endpoint. Only sent when non-empty so the default
+        # (contract-declared) routing is preserved byte-for-byte on the command.
+        if request.target_endpoint:
+            command_payload["forced_endpoint_ref"] = request.target_endpoint
         try:
             self._publisher(
                 self._gen_command_topic,

src/omnimarket/nodes/node_context_roi_runner/models/model_context_roi_run_request.py

@@ -129,6 +129,20 @@ class ModelContextRoiRunRequest(BaseModel):
             "content-resolver effect before dispatch."
         ),
     )
+    target_endpoint: str = Field(
+        default="",
+        description=(
+            "OMN-14018: optional per-run backend pin. When set to a routing-tier "
+            "backend id (e.g. 'cloud-glm', 'cloud-gemini-flash'), the runner "
+            "threads it onto every generation command as forced_endpoint_ref so "
+            "the generation consumer pins the run's STARTING route to that "
+            "backend's tier via the routing authority — giving per-tier graded "
+            "coverage (a genuinely-failing cheap_cloud cohort for context_roi_scores) "
+            "instead of always routing to the contract-declared local tier. Empty "
+            "(default) preserves contract-declared routing exactly. A ref that maps "
+            "to no tier degrades to normal routing (never crashes the run)."
+        ),
+    )
 
 
 __all__ = [

src/omnimarket/nodes/node_generation_consumer/handlers/handler_generation_consumer.py

@@ -948,6 +948,95 @@ def _starting_route(self) -> ModelActiveRoute:
             authority_resolved=False,
         )
 
+    def _forced_starting_route(
+        self,
+        *,
+        forced_endpoint_ref: str,
+        task_description: str,
+    ) -> ModelActiveRoute | None:
+        """Pin the STARTING route to a caller-forced backend's tier (OMN-14018).
+
+        The context-ROI battery pins a run to a specific backend (via
+        ``ModelNodeGenerationRequest.forced_endpoint_ref``) so it can capture
+        genuine per-tier graded outcomes — e.g. a real, sub-floor ``cheap_cloud``
+        cohort for ``context_roi_scores`` that the always-``local`` contract route
+        never produces. Instead of the contract-declared starting tier, this
+        resolves the concrete model + endpoint for the forced backend's tier
+        through the SAME routing authority (``delta`` with ``min_tier_name``) the
+        escalation path uses, and returns a fully authority-resolved route so
+        ``_call_llm`` posts directly to that backend.
+
+        Fail-safe (mirrors ``_escalate_route``): returns ``None`` — the caller then
+        keeps the contract-declared starting route — when the forced ref maps to no
+        tier or the authority cannot resolve it. A bad pin degrades to normal
+        routing and never crashes the run.
+        """
+        target_tier = tier_for_backend(forced_endpoint_ref)
+        if target_tier is None:
+            logger.warning(
+                "[generation-consumer] forced_endpoint_ref %r maps to no routing "
+                "tier; keeping contract-declared starting route",
+                forced_endpoint_ref,
+            )
+            return None
+
+        # The routing authority owns model/endpoint selection for the target tier.
+        # min_tier_name pins the resolution to the forced backend's tier; task_type
+        # drives the contract ladder. A fresh routing UUID keeps the human-facing
+        # generation correlation id separate (parity with _resolve_escalation_decision).
+        request = ModelDelegationRequest(
+            prompt=task_description,
+            task_type=self._task_type,
+            correlation_id=uuid4(),
+            emitted_at=datetime.now(tz=UTC),
+        )
+        try:
+            decision = cast(
+                _RoutingDecision,
+                routing_authority_delta(request, min_tier_name=target_tier),
+            )
+        except Exception as exc:
+            logger.warning(
+                "[generation-consumer] routing authority could not resolve forced "
+                "backend %r (tier %r); keeping contract-declared starting route: %s",
+                forced_endpoint_ref,
+                target_tier,
+                exc,
+            )
+            return None
+
+        if decision is None:
+            return None
+
+        provider = "local" if decision.tier_name in _LOCAL_TIER_NAMES else "cloud"
+        logger.info(
+            "[generation-consumer] forced starting route: backend=%r tier=%r "
+            "model=%r (OMN-14018 per-tier capture pin)",
+            forced_endpoint_ref,
+            decision.tier_name,
+            decision.selected_model,
+        )
+        # Carry the forced BACKEND id on endpoint_ref (not "") even though the
+        # route is authority_resolved. _call_llm posts to endpoint_url when
+        # authority_resolved is True, so endpoint_ref does not affect the wire call
+        # — but it IS what surfaces as the attempt/benchmark endpoint_class and
+        # therefore the context_roi_scores row's endpoint_ref. The ROI overlay maps
+        # a row to a tier via tier_for_backend(endpoint_ref), which resolves a
+        # BACKEND id (e.g. 'cloud-glm' -> cheap_cloud) but NOT a bare tier name
+        # ('cheap_cloud' -> None). Recording the backend id is what makes the
+        # captured per-tier rows overlay-countable (OMN-14018/OMN-14011 Path 1);
+        # leaving it "" would silently drop every captured row.
+        return ModelActiveRoute(
+            tier_name=decision.tier_name,
+            provider=provider,
+            served_model_id=decision.selected_model,
+            endpoint_ref=forced_endpoint_ref,
+            authority_resolved=True,
+            endpoint_url=decision.endpoint_url,
+            api_key_ref=decision.api_key_ref,
+            max_tokens=decision.max_tokens,
+        )
+
     def _escalate_route(
         self,
         *,
@@ -1208,6 +1297,18 @@ async def handle(
         # from a prior request never leaks on a reused handler instance.
         self._active_route = self._starting_route()
 
+        # OMN-14018: an optional per-run backend pin (from the context-ROI battery)
+        # overrides the contract-declared starting tier so the run posts to a chosen
+        # backend and captures genuine per-tier graded outcomes. Fail-safe: an
+        # unresolvable pin keeps the contract-declared route (never crashes).
+        if command.forced_endpoint_ref:
+            forced_route = self._forced_starting_route(
+                forced_endpoint_ref=command.forced_endpoint_ref,
+                task_description=command.task_description,
+            )
+            if forced_route is not None:
+                self._active_route = forced_route
+
         attempts: list[ModelGenerationAttempt] = []
         e2e_start = time.time()
         previous_errors: list[str] | None = None

src/omnimarket/nodes/node_generation_consumer/models/model_generation.py

@@ -253,6 +253,23 @@ class ModelNodeGenerationRequest(BaseModel):
         ),
     )
 
+    # --- OMN-14018: per-run backend pin (context-ROI per-tier capture) ---
+    # Optional with an empty default so ordinary generation is unaffected. When set
+    # to a routing-tier backend id (e.g. "cloud-glm"), the generation consumer pins
+    # this run's STARTING route to that backend's tier via the routing authority
+    # instead of the contract-declared starting tier — letting the context-ROI
+    # battery capture genuine per-tier graded outcomes for context_roi_scores. A ref
+    # that maps to no tier degrades to contract-declared routing (never crashes).
+    forced_endpoint_ref: str = Field(
+        default="",
+        description=(
+            "OMN-14018: optional routing-tier backend id to pin the STARTING route "
+            "for this run (e.g. 'cloud-glm'). Empty (default) = contract-declared "
+            "starting tier. Resolved through the routing authority; an unresolvable "
+            "ref falls back to the contract-declared route."
+        ),
+    )
+
 
 class ModelGenerationBenchmark(BaseModel):
     """Output benchmark emitted as the terminal event payload.

tests/nodes/node_context_roi_runner/test_handler_context_roi_runner.py

@@ -24,6 +24,7 @@
 
 from __future__ import annotations
 
+import json
 from collections.abc import Callable
 from pathlib import Path
 from typing import Any
@@ -559,15 +560,22 @@ def test_failed_trials_count() -> None:
 # ---------------------------------------------------------------------------
 
 
+#: The node's declared terminal output states (contract publish_topics). Asserted
+#: by literal so the run genuinely lands on the contract-declared topics and so
+#: the contract-state-coverage gate sees both output states covered.
+_COMPLETED_TERMINAL_TOPIC = "onex.evt.omnimarket.context-roi-run-completed.v1"
+_FAILED_TERMINAL_TOPIC = "onex.evt.omnimarket.context-roi-run-failed.v1"
+
+
 def test_result_emitted_on_completed_topic() -> None:
     published: list[tuple[str, bytes]] = []
     handler = _make_handler(published=published, consumer_payload=_VALID_EVENT)
     handler.handle(_make_request())
 
-    completed = [t for t, _ in published if "context-roi-run-completed" in t]
+    completed = [t for t, _ in published if t == _COMPLETED_TERMINAL_TOPIC]
     assert len(completed) == 1
     # A run with usable rows must NOT also land on the failed terminal.
-    failed = [t for t, _ in published if "context-roi-run-failed" in t]
+    failed = [t for t, _ in published if t == _FAILED_TERMINAL_TOPIC]
     assert failed == []
 
 
@@ -589,8 +597,8 @@ def test_fully_failed_run_emitted_on_failed_topic() -> None:
     result = handler.handle(request)
 
     assert result.failed_trials == result.total_trials
-    failed = [t for t, _ in published if "context-roi-run-failed" in t]
-    completed = [t for t, _ in published if "context-roi-run-completed" in t]
+    failed = [t for t, _ in published if t == _FAILED_TERMINAL_TOPIC]
+    completed = [t for t, _ in published if t == _COMPLETED_TERMINAL_TOPIC]
     assert len(failed) == 1, "fully-failed run must publish on the failed terminal"
     assert completed == [], (
         "fully-failed run must NOT publish on the completed terminal"
@@ -756,3 +764,49 @@ def test_non_numeric_token_field_fails_row_closed() -> None:
     result = handler.handle(request)
     assert len(result.rows) == 4
     assert all(r.failure_stage == EnumFailureStage.GENERATION for r in result.rows)
+
+
+# ---------------------------------------------------------------------------
+# OMN-14018: target_endpoint pins generation to a chosen backend by threading
+# forced_endpoint_ref onto every published generation command.
+# ---------------------------------------------------------------------------
+
+
+def _published_generation_commands(
+    handler: HandlerContextRoiRunner, published: list[tuple[str, bytes]]
+) -> list[dict[str, Any]]:
+    """Decode the generation-command payloads the runner published.
+
+    The generation command is the one on the contract-declared generation command
+    topic — not the run's terminal completed/failed topics.
+    """
+    command_topic = handler._gen_command_topic
+    return [
+        json.loads(payload) for topic, payload in published if topic == command_topic
+    ]
+
+
+def test_target_endpoint_threads_forced_endpoint_ref_onto_command() -> None:
+    """A request target_endpoint is published as forced_endpoint_ref on every command."""
+    published: list[tuple[str, bytes]] = []
+    handler = _make_handler(published=published)
+    request = _make_request(arms=(_off_arm(), _golden_only_arm()), trials_per_cell=2)
+    request = request.model_copy(update={"target_endpoint": "cloud-glm"})
+
+    handler.handle(request)
+
+    commands = _published_generation_commands(handler, published)
+    assert commands  # at least one generation command was published
+    assert all(cmd.get("forced_endpoint_ref") == "cloud-glm" for cmd in commands)
+
+
+def test_empty_target_endpoint_omits_forced_endpoint_ref() -> None:
+    """The default (empty) target_endpoint leaves the command payload unchanged."""
+    published: list[tuple[str, bytes]] = []
+    handler = _make_handler(published=published)
+
+    handler.handle(_make_request())  # target_endpoint defaults to ""
+
+    commands = _published_generation_commands(handler, published)
+    assert commands
+    assert all("forced_endpoint_ref" not in cmd for cmd in commands)