diff --git a/scripts/run_context_roi_battery.py b/scripts/run_context_roi_battery.py index c58d7ce5e7..47fc0f8641 100644 --- a/scripts/run_context_roi_battery.py +++ b/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: diff --git a/src/omnimarket/nodes/node_context_roi_runner/handlers/handler_context_roi_runner.py b/src/omnimarket/nodes/node_context_roi_runner/handlers/handler_context_roi_runner.py index b05fdb57f3..e3e0baadf7 100644 --- a/src/omnimarket/nodes/node_context_roi_runner/handlers/handler_context_roi_runner.py +++ b/src/omnimarket/nodes/node_context_roi_runner/handlers/handler_context_roi_runner.py @@ -680,7 +680,7 @@ 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, @@ -688,6 +688,11 @@ def _run_trial( "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, diff --git a/src/omnimarket/nodes/node_context_roi_runner/models/model_context_roi_run_request.py b/src/omnimarket/nodes/node_context_roi_runner/models/model_context_roi_run_request.py index f6c47f3519..ffb9d0ffbc 100644 --- a/src/omnimarket/nodes/node_context_roi_runner/models/model_context_roi_run_request.py +++ b/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__ = [ diff --git a/src/omnimarket/nodes/node_generation_consumer/handlers/handler_generation_consumer.py b/src/omnimarket/nodes/node_generation_consumer/handlers/handler_generation_consumer.py index 571c06eb3a..6919501e7d 100644 --- a/src/omnimarket/nodes/node_generation_consumer/handlers/handler_generation_consumer.py +++ b/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 diff --git a/src/omnimarket/nodes/node_generation_consumer/models/model_generation.py b/src/omnimarket/nodes/node_generation_consumer/models/model_generation.py index ef098cd7f0..b1ca9b154c 100644 --- a/src/omnimarket/nodes/node_generation_consumer/models/model_generation.py +++ b/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. diff --git a/tests/nodes/node_context_roi_runner/test_handler_context_roi_runner.py b/tests/nodes/node_context_roi_runner/test_handler_context_roi_runner.py index f7986a027f..ebdaae42b4 100644 --- a/tests/nodes/node_context_roi_runner/test_handler_context_roi_runner.py +++ b/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) diff --git a/tests/unit/nodes/node_generation_consumer/test_generation_rides_ladder_omn13359.py b/tests/unit/nodes/node_generation_consumer/test_generation_rides_ladder_omn13359.py index 566fb6f347..6dab26dd2d 100644 --- a/tests/unit/nodes/node_generation_consumer/test_generation_rides_ladder_omn13359.py +++ b/tests/unit/nodes/node_generation_consumer/test_generation_rides_ladder_omn13359.py @@ -479,3 +479,100 @@ async def test_emits_escalation_proof_matching_active_route(tmp_path: Path) -> N assert event["tier"] == "cheap_cloud" assert event["model"] == effect.calls[1].model == _CLOUD_MODEL assert event["endpoint"] == effect.calls[1].endpoint_url == _CLOUD_ENDPOINT + + +# --------------------------------------------------------------------------- +# OMN-14018: a per-run forced_endpoint_ref pins the STARTING route to a chosen +# backend's tier (context-ROI per-tier capture), overriding the contract tier. +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +@pytest.mark.usefixtures("hermetic_routing") +@pytest.mark.asyncio +async def test_forced_endpoint_ref_pins_first_attempt_to_that_tier( + tmp_path: Path, +) -> None: + """forced_endpoint_ref=cloud-gemini-flash → attempt #1 posts to the CLOUD tier. + + The contract declares local-coder (local); without a pin attempt #1 is local + (proven by ``test_first_attempt_routes_to_local_tier``). With the pin the + routing authority resolves the cheap_cloud tier's model + endpoint for the + STARTING route, so the first (and only) call rides cloud. This is the runner- + driven per-tier capture: the battery pins a run to a backend to record genuine + graded outcomes for that tier in context_roi_scores. + """ + handler, effect = _make_real_path_handler(tmp_path, [_VALID_LLM_RESPONSE]) + + result = await handler.handle( + ModelNodeGenerationRequest( + task_description="Build a stub node", + correlation_id="corr-forced-cloud-1", + max_attempts=3, + forced_endpoint_ref="cloud-gemini-flash", + ) + ) + + assert result.contract_passed is True + assert result.attempt_count == 1 + assert len(effect.calls) == 1 + # The pin overrode the contract-declared local start — first call rode cloud. + assert effect.calls[0].model == _CLOUD_MODEL + assert effect.calls[0].endpoint_url == _CLOUD_ENDPOINT + assert result.attempts[0].provider == "cloud" + assert result.attempts[0].model_id == _CLOUD_MODEL + # OMN-14018 capture-attribution: the attempt AND the run-level benchmark must + # record the forced BACKEND id (not the bare tier name) as endpoint_class, so + # the context_roi_scores row's endpoint_ref resolves through + # tier_for_backend(endpoint_ref) -> cheap_cloud and the captured per-tier rows + # are overlay-countable. A bare tier name ("cheap_cloud") resolves to None and + # the row would be silently dropped. + assert result.attempts[0].endpoint_class == "cloud-gemini-flash" + assert result.endpoint_class == "cloud-gemini-flash" + + +@pytest.mark.unit +@pytest.mark.usefixtures("hermetic_routing") +@pytest.mark.asyncio +async def test_empty_forced_endpoint_ref_uses_contract_starting_tier( + tmp_path: Path, +) -> None: + """An empty forced_endpoint_ref (default) preserves contract-declared routing.""" + handler, effect = _make_real_path_handler(tmp_path, [_VALID_LLM_RESPONSE]) + + result = await handler.handle( + ModelNodeGenerationRequest( + task_description="Build a stub node", + correlation_id="corr-forced-empty-1", + max_attempts=3, + forced_endpoint_ref="", + ) + ) + + assert result.attempt_count == 1 + assert effect.calls[0].model == _LOCAL_MODEL + assert effect.calls[0].endpoint_url == _LOCAL_ENDPOINT + + +@pytest.mark.unit +@pytest.mark.usefixtures("hermetic_routing") +@pytest.mark.asyncio +async def test_unresolvable_forced_endpoint_ref_falls_back_to_contract_route( + tmp_path: Path, +) -> None: + """A forced ref that maps to no tier degrades to the contract route (fail-safe).""" + handler, effect = _make_real_path_handler(tmp_path, [_VALID_LLM_RESPONSE]) + + result = await handler.handle( + ModelNodeGenerationRequest( + task_description="Build a stub node", + correlation_id="corr-forced-bad-1", + max_attempts=3, + forced_endpoint_ref="no-such-backend", + ) + ) + + # The bad pin did not crash the run; it kept the contract-declared local route. + assert result.contract_passed is True + assert effect.calls[0].model == _LOCAL_MODEL + assert effect.calls[0].endpoint_url == _LOCAL_ENDPOINT