Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 34 additions & 3 deletions agent/duet_episode_transport.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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)

Expand All @@ -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
Expand All @@ -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,
Expand All @@ -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:
Expand All @@ -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(
Expand Down
140 changes: 140 additions & 0 deletions agent/refiner_role_routes.py
Original file line number Diff line number Diff line change
@@ -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",
]
22 changes: 14 additions & 8 deletions iterative_episode_refiner/coding.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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,
Expand Down Expand Up @@ -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(),
})
Expand Down Expand Up @@ -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:
Expand Down
6 changes: 5 additions & 1 deletion iterative_episode_refiner/transcription.py
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand Down
Loading
Loading