feat(delegation): require a helper tier when tiers are configured (RIG-8770)

agent/subagent_lifecycle.py

@@ -274,7 +274,7 @@ class SubagentLifecycleService:
     def __init__(self, parent_agent_resolver: Callable[[], Any]) -> None:
         self._parent_agent_resolver = parent_agent_resolver
 
-    def launch(self, request: SubagentLaunchRequest | Mapping[str, Any]) -> SubagentHandle:
+    def launch(self, request: SubagentLaunchRequest | Mapping[str, Any], *, tier: Optional[str] = None) -> SubagentHandle:
         parent = self._parent_agent_resolver()
         if parent is None:
             raise SubagentLifecycleError("No active Hermes parent session is available.")
@@ -290,12 +290,21 @@ def launch(self, request: SubagentLaunchRequest | Mapping[str, Any]) -> Subagent
                 raise SubagentLifecycleError("Duplicate correlation_id for this parent session.")
         # Lazy: delegate construction stays internal, plugins never import private delegation helpers.
         from tools.delegate_tool import (
-            _build_child_preserving_parent_tools, _credential_overrides, _load_config, _resolve_delegation_credentials,
-            DEFAULT_MAX_ITERATIONS,
+            _build_child_preserving_parent_tools, _credential_overrides, _load_config,
+            _resolve_delegation_credentials, _resolve_delegation_tier_routing_config, DEFAULT_MAX_ITERATIONS,
         )
-        # Routing precedence: request.provider > delegation config > parent.
-        routing_cfg = {"provider": request.provider, "model": request.model or ""} if request.provider is not None else _load_config()
+        config = _load_config()
+        routing_cfg = config
+        if config.get("tiers") is not None and tier is not None and (
+            request.provider is not None or request.model is not None
+        ):
+            raise SubagentLifecycleError("tier cannot be combined with a per-launch model or provider override.")
         try:
+            routing_cfg = _resolve_delegation_tier_routing_config(
+                routing_cfg, tier, tier_configs=config.get("tiers"), enforce_tier=tier is not None,
+            )
+            if request.provider is not None:
+                routing_cfg = {"provider": request.provider, "model": request.model or ""}
             creds = _resolve_delegation_credentials(routing_cfg, parent)
         except ValueError as exc:
             raise SubagentLifecycleError(str(exc)) from exc

cli-config.yaml.example

@@ -1852,6 +1852,14 @@ delegation:
   #                                           # delegation.model to an inexpensive one — children carry the
   #                                           # vast majority of tokens, so this is where spend is cut while
   #                                           # planning quality stays with the frontier parent.
+  # Optional per-call model routes. If present, delegate_task requires tier: easy, medium, or hard.
+  # Each tier is a full route spec. Base model/provider/endpoint/credentials/request overrides/fallback providers are cleared first.
+  # A tier with only model inherits the parent runtime provider and endpoint, not delegation.provider.
+  # A supplied tier is ignored when no tiers map is configured.
+  # tiers:
+  #   easy: {model: "google/gemini-3-flash-preview", provider: "openrouter"}
+  #   medium: {model: "anthropic/claude-sonnet-4", provider: "openrouter"}
+  #   hard: {model: "anthropic/claude-opus-4", provider: "openrouter"}
   # fallback_providers:                       # Fallback chain for delegated children (same entry format as the
   #   - provider: "openrouter"                # top-level list). Pinned children use only a declared child chain;
   #     model: "deepseek/deepseek-chat"       # unpinned children inherit when unset. [] disables fallback (#65038).

run_agent.py

@@ -1342,6 +1342,7 @@ def _dispatch_delegate_task(self, function_args: dict) -> str:
             goal=function_args.get("goal"), context=function_args.get("context"),
             tasks=_strip_model_hidden_task_fields(function_args.get("tasks")),
             max_iterations=function_args.get("max_iterations"), role=function_args.get("role"),
+            tier=function_args.get("tier"), enforce_tier=True,
             background=not (getattr(self, "_delegate_depth", 0) > 0), images=function_args.get("images"),
             action=function_args.get("action"),
             subagent_id=function_args.get("subagent_id"), message=function_args.get("message"), parent_agent=self,

tests/agent/test_review_engine.py

@@ -244,6 +244,37 @@ def fake_build(**kw):
     assert evt is not None and evt["type"] == "async_delegation"
     assert evt["results"][0]["summary"] == "REVIEW: looks good"
 
+
+
+def test_start_review_is_exempt_from_configured_delegation_tiers(monkeypatch):
+    import tools.delegate_tool as dt
+
+    captured = {}
+
+    def fake_resolve(config, _parent):
+        captured["routing_cfg"] = config
+        return {
+            "model": "parent-model", "provider": None, "base_url": None,
+            "api_key": None, "api_mode": None, "request_overrides": None,
+            "command": None, "args": None,
+        }
+
+    monkeypatch.setattr(dt, "_load_config", lambda: {
+        "model": "delegation-model",
+        "tiers": {tier: {"model": f"tier-{tier}"} for tier in ("easy", "medium", "hard")},
+    })
+    monkeypatch.setattr(dt, "_resolve_delegation_credentials", fake_resolve)
+    monkeypatch.setattr(dt, "_build_child_preserving_parent_tools", lambda **_kwargs: MagicMock(_delegate_role="leaf"))
+    monkeypatch.setattr(dt, "_run_batch", lambda _batch, _background: '{"status": "dispatched"}')
+    monkeypatch.setattr(re_mod, "_load_review_credentials_cfg", lambda: None)
+
+    parent = _fake_parent()
+    parent._delegate_depth = 0
+    result = start_review(parent, [{"role": "user", "content": "review"}], "")
+
+    assert result["status"] == "dispatched"
+    assert "tiers" not in captured["routing_cfg"]
+    assert captured["routing_cfg"]["model"] == "delegation-model"
 def test_start_review_rejects_empty_conversation():
     with pytest.raises(ValueError, match="empty"):
         start_review(_fake_parent(), [], "")

tests/agent/test_subagent_lifecycle.py

@@ -285,6 +285,67 @@ def test_routing_delegation_provider_bundle(routing_lifecycle):
     assert build.call_args.kwargs["model"] == "request-model"
 
 
+
+def test_lifecycle_launch_without_tier_remains_exempt_when_tiers_configured(routing_lifecycle):
+    service, build, config, resolve = routing_lifecycle
+    config.return_value = {
+        "provider": "openrouter", "model": "delegated-model",
+        "base_url": "https://delegation.example/v1",
+        "tiers": {"hard": {"provider": "openrouter", "model": "tier-model"}},
+    }
+
+    handle = service.launch(SubagentLaunchRequest(goal="route"))
+
+    assert service.wait(handle, timeout_seconds=2).state is SubagentState.SUCCEEDED
+    assert build.call_args.kwargs["model"] == "delegated-model"
+    assert build.call_args.kwargs["routing_cfg"]["model"] == "delegated-model"
+    resolve.assert_called_once_with(requested="openrouter", target_model="delegated-model")
+
+
+def test_lifecycle_launch_resolves_explicit_tier(routing_lifecycle):
+    service, build, config, resolve = routing_lifecycle
+    config.return_value = {
+        "provider": "nous", "model": "delegated-model",
+        "base_url": "https://delegation.example/v1", "api_key": "delegation-fixture-key",
+        "api_mode": "anthropic_messages",
+        "tiers": {"hard": {"provider": "openrouter", "model": "tier-model"}},
+    }
+
+    handle = service.launch(SubagentLaunchRequest(goal="route"), tier="hard")
+
+    assert service.wait(handle, timeout_seconds=2).state is SubagentState.SUCCEEDED
+    routing_cfg = build.call_args.kwargs["routing_cfg"]
+    assert build.call_args.kwargs["model"] == "tier-model"
+    assert routing_cfg["provider"] == "openrouter"
+    assert routing_cfg["model"] == "tier-model"
+    assert not {"base_url", "api_key", "api_mode"} & routing_cfg.keys()
+    resolve.assert_called_once_with(requested="openrouter", target_model="tier-model")
+
+
+def test_lifecycle_ignores_tier_when_tiers_are_unconfigured(routing_lifecycle):
+    service, build, config, resolve = routing_lifecycle
+    config.return_value = {"provider": "openrouter", "model": "delegated-model"}
+
+    handle = service.launch(SubagentLaunchRequest(goal="route"), tier="hard")
+
+    assert service.wait(handle, timeout_seconds=2).state is SubagentState.SUCCEEDED
+    assert build.call_args.kwargs["model"] == "delegated-model"
+    assert build.call_args.kwargs["routing_cfg"] == config.return_value
+    resolve.assert_called_once_with(requested="openrouter", target_model="delegated-model")
+
+
+@pytest.mark.parametrize("override", [{"provider": "openrouter"}, {"model": "request-model"}])
+def test_lifecycle_tier_rejects_per_launch_route_overrides(routing_lifecycle, override):
+    service, build, config, resolve = routing_lifecycle
+    config.return_value = {"tiers": {"hard": {"model": "tier-model"}}}
+
+    with pytest.raises(SubagentLifecycleError, match="tier cannot be combined"):
+        service.launch({"goal": "route", **override}, tier="hard")
+
+    build.assert_not_called()
+    resolve.assert_not_called()
+
+
 @pytest.mark.parametrize("wire", [False, True])
 def test_routing_request_provider_wins_without_mixing_bundles(routing_lifecycle, wire):
     from hermes_cli.plugin_host_wire import decode, encode

tests/tools/test_delegate.py

@@ -1557,6 +1557,25 @@ def fake_delegate_task(**kwargs):
         self.assertNotIn("acp_command", captured["tasks"][0])
         self.assertNotIn("acp_args", captured["tasks"][0])
 
+    def test_tier_is_forwarded_to_delegate_task(self):
+        import run_agent
+
+        captured = {}
+
+        def fake_delegate_task(**kwargs):
+            captured.update(kwargs)
+            return "{}"
+
+        parent = _make_mock_parent(depth=0)
+        with patch("tools.delegate_tool.delegate_task", fake_delegate_task):
+            run_agent.AIAgent._dispatch_delegate_task(
+                parent,
+                {"goal": "test", "tier": "hard"},
+            )
+
+        self.assertEqual(captured["tier"], "hard")
+        self.assertTrue(captured["enforce_tier"])
+
 
 
 class TestConcurrencyDefaults(unittest.TestCase):