diff --git a/agent/duet_episode_transport.py b/agent/duet_episode_transport.py index e1b7740ae8..4132ab3264 100644 --- a/agent/duet_episode_transport.py +++ b/agent/duet_episode_transport.py @@ -81,6 +81,10 @@ def from_bound_agent(cls, *, artifacts, owner_duet_id, agent, model_types): from agent.model_call_recovery_policy import resolve_recovery_policy route["recovery"] = resolve_recovery_policy(route) + from agent.refiner_role_routes import configured_role_spec, resolve_role_routes + + spec, providers = configured_role_spec() + role_routes = resolve_role_routes(spec, route, providers) record = { "schema_version": 1, "owner_duet_id": owner_duet_id, @@ -91,6 +95,9 @@ def from_bound_agent(cls, *, artifacts, owner_duet_id, agent, model_types): "auth_mode": runtime.get("auth_mode", ""), "credential_source": "owning_duet_host_memory", } + if role_routes: + # Pinned per-role overrides (agent/refiner_role_routes.py); no secrets. + record["role_routes"] = role_routes reference = put_data(artifacts, owner_duet_id, "duet_model_binding", record) return cls(owner_duet_id, reference, record, runtime.get("api_key") or None) @@ -113,6 +120,29 @@ def validate(self, *, artifacts, reference, owner_duet_id, model_types): "The refiner's required model slots differ from the owning Duet binding." ) + def route_for(self, episode_local_id=None): + """(role, route, key) for one call: a pinned role override or the Duet's route.""" + from agent.refiner_role_routes import route_key, select_route + + role, route = select_route(self.record.get("role_routes") or {}, self.record["route"], episode_local_id) + return role, route, route_key(route, self.api_key) + + def for_role(self, *roles): + """This binding with its route replaced by the first configured override of *roles*.""" + import dataclasses + + from agent.refiner_role_routes import route_key + + overrides = self.record.get("role_routes") or {} + for role in roles: + if role in overrides: + route = overrides[role] + return dataclasses.replace( + self, record={**self.record, "route": route, "refiner_role": role}, + api_key=route_key(route, self.api_key), + ) + return self + def transport(self, *, record_attempt, cancel_event=None): return DuetEpisodeTransport( self, record_attempt=record_attempt, cancel_event=cancel_event @@ -128,9 +158,10 @@ def __init__(self, binding, *, record_attempt, cancel_event=None): async def __call__(self, request): if request.model_type not in self.binding.record["model_types"]: raise ValueError("Model call names a slot outside the frozen Duet binding.") - route = self.binding.record["route"] + role, route, key = self.binding.route_for(request.episode_local_id) receipt = { "binding_ref": self.binding.reference, + **({"refiner_role": role} if role else {}), "owner_duet_id": self.binding.owner_duet_id, "session_id": self.binding.record["session_id"], "call_id": uuid.uuid4().hex, @@ -145,7 +176,7 @@ async def __call__(self, request): self.record_attempt({**receipt, "state": "started"}) try: text, actual_model = await invoke_pinned_route( - route, self.binding.api_key, request, self.cancel, lambda: None, + route, key, request, self.cancel, lambda: None, record_activity=lambda details: self.record_attempt({**receipt, **details}), ) except asyncio.CancelledError: @@ -159,7 +190,7 @@ async def __call__(self, request): self.record_attempt({ **receipt, "state": "failed", - **provider_failure(exc, route, credential=self.binding.api_key, request=request), + **provider_failure(exc, route, credential=key, request=request), "elapsed_seconds": time.monotonic() - started, }) raise ModelCallFailed( diff --git a/agent/refiner_role_routes.py b/agent/refiner_role_routes.py new file mode 100644 index 0000000000..e140289f5a --- /dev/null +++ b/agent/refiner_role_routes.py @@ -0,0 +1,140 @@ +"""Per-role model and reasoning for the refiner's calls on the owning Duet's binding. + +By default every refiner call uses the owning Duet's route (model, provider, +reasoning). Roles differ in what they need: the Designer and the module emitter +benefit from deep reasoning in a single call, while a coding loop pays for that +reasoning on every step. ``openchia.refiner_roles`` in the Hermes config lets an +operator override, per role, the custom provider, the model and the reasoning: + + openchia: + refiner_roles: + designer: {reasoning: default} # omit reasoning_effort: the server default + emitter: {reasoning: default} + coder: {custom_provider: Mango Coder, model: Qwen3-Coder-Next} + implementer: {reasoning: low} + +Role keys are the refiner's Episode local IDs (``launch``/``designer``, +``materialization_implementer``, ``implementer``, ``measure``, ``verify``, +``question``, ``support``, ``*_parts``) plus two call kinds: ``emitter`` (the +Builder emitter transcribing missing modules, #81) and ``coder`` (the Implementer's +coding session). ``designer`` also covers the root ``launch`` Episode. + +Overrides are resolved when the Duet binding is frozen and stored in it without +secrets, so they are pinned, audited and preserved by ``/build continue``. +Credentials are resolved at call time from the provider's ``key_env``. +""" + +from __future__ import annotations + +import os +from contextlib import contextmanager +from contextvars import ContextVar +from typing import Any, Iterator, Mapping + +OVERRIDE_FIELDS = frozenset({"custom_provider", "model", "reasoning"}) +REASONING_LEVELS = frozenset({"minimal", "low", "medium", "high", "xhigh"}) + +_SCOPED_ROLE: ContextVar[str | None] = ContextVar("openchia_refiner_role", default=None) + + +@contextmanager +def role_scope(role: str) -> Iterator[None]: + """Attribute model calls made in this context to *role* (e.g. ``emitter``).""" + token = _SCOPED_ROLE.set(role) + try: + yield + finally: + _SCOPED_ROLE.reset(token) + + +def scoped_role() -> str | None: + return _SCOPED_ROLE.get() + + +def _reasoning(value: Any) -> dict | None: + """``default`` omits the field (server default); ``none`` disables; a level enables.""" + text = str(value).strip().lower() + if text in {"default", "server", "omit"}: + return None + if value is False or text in {"none", "false", "disabled"}: + return {"enabled": False} + if text in REASONING_LEVELS: + return {"enabled": True, "effort": text} + raise ValueError(f"refiner role reasoning must be default, none or one of {sorted(REASONING_LEVELS)}, not {value!r}") + + +def resolve_role_routes(spec: Mapping[str, Any] | None, duet_route: Mapping[str, Any], + custom_providers: list[Mapping[str, Any]] | None) -> dict[str, dict]: + """Concrete, secret-free routes for each configured role, derived from the Duet's route.""" + from agent.model_call_recovery_policy import resolve_recovery_policy + + if not spec: + return {} + if not isinstance(spec, Mapping): + raise ValueError("openchia.refiner_roles must be a mapping of role -> override") + providers = {str(item.get("name")): item for item in (custom_providers or []) if isinstance(item, Mapping)} + routes: dict[str, dict] = {} + for role, override in spec.items(): + if not isinstance(override, Mapping) or not override: + raise ValueError(f"refiner role {role!r} needs a mapping with {sorted(OVERRIDE_FIELDS)}") + unknown = set(override) - OVERRIDE_FIELDS + if unknown: + raise ValueError(f"refiner role {role!r} has unknown fields {sorted(unknown)}") + route = {key: value for key, value in duet_route.items() if key != "recovery"} + if "custom_provider" in override: + name = str(override["custom_provider"]) + provider = providers.get(name) + if provider is None: + raise ValueError(f"refiner role {role!r} names unknown custom provider {name!r}") + route.update(provider="custom", base_url=str(provider["base_url"]).rstrip("/"), + api_mode="chat_completions", model=str(provider.get("model") or route["model"]), + key_env=provider.get("key_env")) + route.pop("reasoning", None) # a different server: inherit no Duet reasoning setting + if "model" in override: + route["model"] = str(override["model"]) + if "reasoning" in override: + route["reasoning"] = _reasoning(override["reasoning"]) + route["recovery"] = resolve_recovery_policy(route) + routes[str(role)] = route + return routes + + +def role_candidates(episode_local_id: str | None) -> list[str]: + """Override keys to try for one call, most specific first.""" + candidates = [] + scoped = scoped_role() + if scoped: + candidates.append(scoped) + if episode_local_id: + candidates.append(episode_local_id) + if episode_local_id == "launch": + candidates.append("designer") + return candidates + + +def select_route(role_routes: Mapping[str, dict], duet_route: dict, episode_local_id: str | None) -> tuple[str | None, dict]: + for role in role_candidates(episode_local_id): + if role in role_routes: + return role, role_routes[role] + return None, duet_route + + +def route_key(route: Mapping[str, Any], duet_key: str | None) -> str | None: + """The credential for *route*: its provider's ``key_env``, or the Duet's own key.""" + if "key_env" in route: + name = route["key_env"] + return os.environ.get(name) if name else None + return duet_key + + +def configured_role_spec() -> tuple[Mapping | None, list]: + from openchia_cli.config import load_config_readonly + + config = load_config_readonly() or {} + return (config.get("openchia") or {}).get("refiner_roles"), list(config.get("custom_providers") or []) + + +__all__ = [ + "configured_role_spec", "resolve_role_routes", "role_candidates", "role_scope", + "route_key", "scoped_role", "select_route", +] diff --git a/iterative_episode_refiner/coding.py b/iterative_episode_refiner/coding.py index 0de894425a..713a54774f 100644 --- a/iterative_episode_refiner/coding.py +++ b/iterative_episode_refiner/coding.py @@ -291,9 +291,15 @@ def produce(): def _run(self, request, prompt, cancel, active, diagnostics): call, candidate, context = self._working_context(request) + # Per-role routes (agent/refiner_role_routes.py): the coding session may + # use its own model/reasoning ("coder", else the invocation's role). + binding = self.binding.for_role("coder", call.assignment.body["role"]) + return self._run_bound(request, prompt, cancel, active, diagnostics, call, candidate, context, binding) + + def _run_bound(self, request, prompt, cancel, active, diagnostics, call, candidate, context, binding): context = _thaw_json(context) context["inputs"]["target_environment"]["coding_diagnostics"] = diagnostics - backend = coding_backend(self.binding) + backend = coding_backend(binding) from .measure_coding import INSTRUCTIONS as MEASURE_INSTRUCTIONS measuring = ROLE_SPECIALIZATION[call.assignment.body["role"]] == "measure" @@ -307,11 +313,11 @@ def _run(self, request, prompt, cancel, active, diagnostics): # one Builder emitter call each instead of an exploratory coding turn (#80). transcriber = None if measuring or resume is not None else self._transcriber(call, workspace.root) runtime_id = backend.runtime_id if transcriber is None else transcriber.runtime_id - route = self.binding.record["route"] + route = binding.record["route"] receipt = { - **identity, "binding_ref": self.binding.reference, - "owner_duet_id": self.binding.owner_duet_id, - "session_id": self.binding.record["session_id"], + **identity, "binding_ref": binding.reference, + "owner_duet_id": binding.owner_duet_id, + "session_id": binding.record["session_id"], "run_id": self.session.registration.run_id.value, "call_id": uuid.uuid4().hex, "task": request.task, "model_type": request.model_type, "episode_local_id": request.episode_local_id, @@ -353,14 +359,14 @@ def observe(note): extra["command_shell"] = shell coder = transcriber if transcriber is not None else backend( **extra, - binding=self.binding, workspace=workspace.root, state_dir=native_home, + binding=binding, workspace=workspace.root, state_dir=native_home, instructions=instructions, resume_thread_id=resume, on_event=observe, ) active.append(coder) thread_id = coder.ensure_started() self.session.put_data("coding_thread", { "invocation_id": call.invocation_id.value, "context_id": context_id, - "thread_id": thread_id, "binding_ref": self.binding.reference, + "thread_id": thread_id, "binding_ref": binding.reference, "coding_runtime": runtime_id, "process": coder.process_identity(), }) @@ -422,7 +428,7 @@ def observe(note): self.record_attempt({**receipt, "state": "cancelled", "elapsed_seconds": time.monotonic() - started}) raise except Exception as exc: - details = provider_failure(exc, route, credential=self.binding.api_key, request=request) + details = provider_failure(exc, route, credential=binding.api_key, request=request) self.record_attempt({**receipt, **details, "state": "failed", "elapsed_seconds": time.monotonic() - started}) raise ModelCallFailed(f"{call.assignment.body['role']} coding call failed; its owning Run must stop.", receipt) from None finally: diff --git a/iterative_episode_refiner/transcription.py b/iterative_episode_refiner/transcription.py index 19ecddd916..9171d194fb 100644 --- a/iterative_episode_refiner/transcription.py +++ b/iterative_episode_refiner/transcription.py @@ -81,7 +81,11 @@ async def emit_modules(*, projection, targets, store, transport, emitter=None, r resolver = resolver or EpisodeReferenceResolver() inherited = inherited_refinement_requests(build_request, store) sources: dict[str, str] = {} - with model_transport_scope(transport): + from agent.refiner_role_routes import role_scope + + # Emitter calls are attributed to the "emitter" role so they can use their own + # model/reasoning (e.g. deep reasoning for one long call), see refiner_role_routes. + with model_transport_scope(transport), role_scope("emitter"): for target in targets: node = node_by_id[target.local_id] frozen = frozen_by_id[target.local_id] diff --git a/tests/agent/test_refiner_role_routes.py b/tests/agent/test_refiner_role_routes.py new file mode 100644 index 0000000000..2caa5b8723 --- /dev/null +++ b/tests/agent/test_refiner_role_routes.py @@ -0,0 +1,141 @@ +"""Per-role model and reasoning for refiner calls (openchia.refiner_roles). + +Motivation (2026-10-09, mango Qwen3.8-27B): the server's best reasoning (xhigh) +is reachable only by omitting reasoning_effort, and is worth it for one long +Designer or emitter call but not for every step of a coding loop. A single Duet +route forced one setting on every refiner role. +""" + +from __future__ import annotations + +import asyncio + +import pytest + +from agent.duet_episode_transport import DuetEpisodeBinding +from agent.refiner_role_routes import resolve_role_routes, role_scope, route_key, select_route + +DUET_ROUTE = {"provider": "custom", "model": "Qwen3.8-27B", "base_url": "http://mango:8008/v1", + "api_mode": "chat_completions", "reasoning": {"enabled": True, "effort": "medium"}, + "recovery": {"mode": "retry_on_healthy_probe"}} +PROVIDERS = [{"name": "Mango Coder", "base_url": "http://mango:8010/v1/", "model": "Qwen3-Coder-Next"}, + {"name": "Argo", "base_url": "https://argo/v1", "model": "Claude Opus 5", "key_env": "ARGO_KEY"}] +SPEC = { + "designer": {"reasoning": "default"}, + "emitter": {"reasoning": "default"}, + "coder": {"custom_provider": "Mango Coder"}, + "implementer": {"reasoning": "low"}, + "measure": {"custom_provider": "Argo", "reasoning": "none"}, +} + + +def _routes(): + return resolve_role_routes(SPEC, DUET_ROUTE, PROVIDERS) + + +def test_overrides_resolve_from_the_duet_route() -> None: + routes = _routes() + assert routes["designer"]["reasoning"] is None # omitted on the wire -> server default + assert routes["designer"]["model"] == "Qwen3.8-27B" and "key_env" not in routes["designer"] + assert routes["implementer"]["reasoning"] == {"enabled": True, "effort": "low"} + coder = routes["coder"] + assert (coder["base_url"], coder["model"], coder["key_env"]) == ("http://mango:8010/v1", "Qwen3-Coder-Next", None) + assert "reasoning" not in coder # another server inherits no Duet reasoning + assert routes["measure"]["reasoning"] == {"enabled": False} and routes["measure"]["key_env"] == "ARGO_KEY" + assert all("recovery" in route for route in routes.values()) + + +@pytest.mark.parametrize("spec, message", [ + ({"coder": {"custom_provider": "Nope"}}, "unknown custom provider"), + ({"coder": {"temperature": 1}}, "unknown fields"), + ({"coder": {"reasoning": "extreme"}}, "reasoning must be"), + ({"coder": {}}, "needs a mapping"), +]) +def test_invalid_overrides_are_rejected(spec, message) -> None: + with pytest.raises(ValueError, match=message): + resolve_role_routes(spec, DUET_ROUTE, PROVIDERS) + + +def test_no_spec_means_no_overrides() -> None: + assert resolve_role_routes(None, DUET_ROUTE, PROVIDERS) == {} + + +def test_selection_order_scoped_role_then_episode_then_designer_alias() -> None: + routes = _routes() + assert select_route(routes, DUET_ROUTE, "launch")[0] == "designer" + assert select_route(routes, DUET_ROUTE, "implementer")[0] == "implementer" + assert select_route(routes, DUET_ROUTE, "verify") == (None, DUET_ROUTE) + with role_scope("emitter"): + assert select_route(routes, DUET_ROUTE, "implementer")[0] == "emitter" + + +def test_keys_come_from_key_env_or_the_duet(monkeypatch) -> None: + routes = _routes() + monkeypatch.setenv("ARGO_KEY", "secret-from-env") + assert route_key(routes["measure"], "duet-key") == "secret-from-env" + assert route_key(routes["coder"], "duet-key") is None # keyless provider + assert route_key(routes["designer"], "duet-key") == "duet-key" + + +def _binding(): + record = {"route": DUET_ROUTE, "role_routes": _routes(), "model_types": ["refinement"], "session_id": "s"} + return DuetEpisodeBinding("owner", {"artifact_id": "b"}, record, "duet-key") + + +def test_binding_for_role_gives_the_coder_its_route() -> None: + coder = _binding().for_role("coder", "implementer") + assert coder.record["route"]["model"] == "Qwen3-Coder-Next" and coder.api_key is None + assert coder.record["refiner_role"] == "coder" + plain = _binding().for_role("verify") + assert plain.record["route"] is DUET_ROUTE + + +def test_transport_dispatches_each_call_on_its_role_route(monkeypatch) -> None: + import agent.duet_episode_transport as transport_module + from llm_call_library.transport import ModelTransportRequest + + seen = [] + + async def fake_invoke(route, key, request, cancel, progress, *, record_activity): + seen.append((route.get("model"), route.get("reasoning", "absent"), key)) + return "ok", route["model"] + + monkeypatch.setattr(transport_module, "invoke_pinned_route", fake_invoke) + records = [] + transport = _binding().transport(record_attempt=records.append) + + def call(local_id): + request = ModelTransportRequest(task="t", model_type="refinement", messages=({"role": "user", "content": "x"},), + temperature=None, max_tokens=8, timeout=5, reasoning_config=None, + main_runtime=None, episode_local_id=local_id) + return asyncio.run(transport(request)) + + call("launch") + call("verify") + + async def emitter_call(): + with role_scope("emitter"): + from llm_call_library.transport import ModelTransportRequest as Request + return await transport(Request(task="t", model_type="refinement", messages=({"role": "user", "content": "x"},), + temperature=None, max_tokens=8, timeout=5, reasoning_config=None, + main_runtime=None, episode_local_id="asm_next_broker_probe")) + + asyncio.run(emitter_call()) + assert seen == [ + ("Qwen3.8-27B", None, "duet-key"), # designer: reasoning omitted + ("Qwen3.8-27B", {"enabled": True, "effort": "medium"}, "duet-key"), # verify: the Duet's route + ("Qwen3.8-27B", None, "duet-key"), # emitter + ] + started = [r for r in records if r["state"] == "started"] + assert [r.get("refiner_role") for r in started] == ["designer", None, "emitter"] + + +def test_default_reasoning_omits_reasoning_effort_on_the_wire() -> None: + # _invoke sends reasoning_effort only when the route's reasoning is not None. + import inspect + + from agent import episode_launch_transport + + source = inspect.getsource(episode_launch_transport._invoke) + assert 'reasoning = route.get("reasoning", request.reasoning_config)' in source + assert "if reasoning is not None:" in source