"""Delegation config knobs (delegation.* keys) and child credential/provider resolution.""" from __future__ import annotations import logging import os from typing import Any, Dict, List, Optional from utils import base_url_hostname, is_truthy_value from hermes_cli.fallback_config import scoped_fallback_chain logger = logging.getLogger("tools.delegate_tool") # log-record parity with the origin module # Runtime-provider sentinel for providers that are not natively known; must # match hermes_cli.runtime_provider.RUNTIME_PROVIDER_TYPE_CUSTOM. _RUNTIME_PROVIDER_CUSTOM = "custom" _DEFAULT_MAX_CONCURRENT_CHILDREN = 10 # One-shot guard: _get_max_concurrent_children() runs on every get_definitions() # schema rebuild, so the >10 cost advisory would otherwise log on every turn. _HIGH_CONCURRENCY_WARNED = False MAX_DEPTH = 1 # flat by default: parent (0) -> child (1); deeper needs max_spawn_depth _MIN_SPAWN_DEPTH = 1 # floor for the configurable cap; MAX_DEPTH stays the default _LEGACY_MAX_ASYNC_WARNED = False # No default wall-clock cap on children: legitimate heavy work (deep reviews, research fan-outs, slow reasoning # models) was being killed mid-task. Stuck-child detection is the heartbeat staleness monitor; # delegation.child_timeout_seconds opts back in. DEFAULT_CHILD_TIMEOUT: Optional[float] = None def _cfg() -> dict: """The ``delegation`` section, read through the origin so tests can patch it.""" from tools.delegate_tool import _load_config return _load_config() # ── Subagent approval callbacks ───────────────────────────────────────────── # Subagent worker threads don't inherit the CLI's threading.local approval # callback, so prompt_dangerous_approval() would fall back to input() and # deadlock the parent's prompt_toolkit TUI. Every worker gets a non-interactive # callback via ThreadPoolExecutor(initializer=...): deny by default, approve when # delegation.subagent_auto_approve is true. Both warn for audit. Gateway sessions # are unaffected (they resolve approvals via tools/approval.py's per-session queue). def _subagent_auto_deny(command: str, description: str, **kwargs) -> str: """Auto-deny (safe default): returns 'deny' so the child sees a recoverable refusal.""" logger.warning( "Subagent auto-denied dangerous command: %s (%s). Set delegation.subagent_auto_approve: true to allow.", command, description, ) return "deny" def _subagent_auto_approve(command: str, description: str, **kwargs) -> str: """Auto-approve (opt-in YOLO via delegation.subagent_auto_approve): returns 'once'.""" logger.warning("Subagent auto-approved dangerous command: %s (%s)", command, description) return "once" def _get_subagent_approval_callback(): """Callback for subagent worker threads per delegation.subagent_auto_approve (default False).""" if is_truthy_value(_cfg().get("subagent_auto_approve", False)): return _subagent_auto_approve return _subagent_auto_deny def _knob(key: str, env_var: Optional[str], parse, default, invalid_msg: str): """delegation. > > default. A config value that fails ``parse`` logs ``invalid_msg`` (``%r`` = the value) and yields the default; an env value that fails is silently ignored.""" val = _cfg().get(key) if val is not None: try: return parse(val) except (TypeError, ValueError): logger.warning(invalid_msg, val) return default env_val = os.getenv(env_var) if env_var else None if env_val: try: return parse(env_val) except (TypeError, ValueError): pass return default def _warn_once(flag_name: str, message: str, *args: Any) -> None: """Module-global one-shot warning (``flag_name`` is the ``_*_WARNED`` global).""" if not globals()[flag_name]: globals()[flag_name] = True logger.warning(message, *args) def _get_oneshot_max_children() -> int: """delegation.oneshot_max_children (total children per finite one-shot session; 0 = unlimited).""" return _knob( "oneshot_max_children", None, lambda v: max(0, int(v)), 2, "delegation.oneshot_max_children=%r is not a valid integer; using default 2", ) def _get_max_concurrent_children() -> int: """delegation.max_concurrent_children > DELEGATION_MAX_CONCURRENT_CHILDREN env > 10. Floor of 1 is the only bound enforced; there is no ceiling. """ result = _knob( "max_concurrent_children", "DELEGATION_MAX_CONCURRENT_CHILDREN", lambda v: max(1, int(v)), _DEFAULT_MAX_CONCURRENT_CHILDREN, f"delegation.max_concurrent_children=%r is not a valid integer; using default {_DEFAULT_MAX_CONCURRENT_CHILDREN}", ) if result > 10 and _cfg().get("max_concurrent_children") is not None: _warn_once( "_HIGH_CONCURRENCY_WARNED", "delegation.max_concurrent_children=%d: each child consumes API tokens " "independently. High values multiply cost linearly.", result, ) return result def _get_independent_completions() -> bool: """delegation.independent_completions (bool, default False): split a background call into per-task / per-group completion messages that land as each finishes. Off = one consolidated message when the whole call is done.""" return is_truthy_value(_cfg().get("independent_completions", False)) def _get_worktree_isolation() -> bool: """delegation.worktree_isolation (bool, default False): each child gets its own git worktree off the parent's HEAD so parallel children never contend for one working copy. Git-only and local-backend-only; otherwise silently ignored.""" return bool(_cfg().get("worktree_isolation", False)) def _get_max_async_children() -> int: """Concurrency cap for background delegations == delegation.max_concurrent_children. At capacity a new async dispatch is REJECTED (not queued) so a runaway model can't pile up unbounded background work; the caller then runs synchronously. A leftover ``delegation.max_async_children`` key is ignored with a one-time warning.""" from tools.delegate_tool import _get_max_concurrent_children if _cfg().get("max_async_children") is not None: _warn_once( "_LEGACY_MAX_ASYNC_WARNED", "delegation.max_async_children is deprecated and ignored; " "delegation.max_concurrent_children now caps background " "delegations too. Remove the stale key from config.yaml.", ) return _get_max_concurrent_children() def _parse_timeout(raw: Any) -> Optional[float]: """Seconds → None (<= 0 disables) or max(30, value). Raises on non-numeric.""" parsed = float(raw) return None if parsed <= 0 else max(30.0, parsed) def _get_child_timeout() -> Optional[float]: """Inactivity cap for one child (seconds of NO progress), or None (default: no cap). Failures should come from what the child does (API/tool errors, iteration budget), not a stopwatch: the cap restarts on every sign of progress — a completed call, a tool change, an activity-clock tick — so a slow provider serving multi-minute completions never loses a live child, and a child frozen for the whole window is still caught. A configured value pre-empts nothing the heartbeat staleness monitor would not also catch. delegation.child_timeout_seconds > 0 opts in (floor 30 s); 0 or negative disables. Env fallback: DELEGATION_CHILD_TIMEOUT_SECONDS.""" return _knob( "child_timeout_seconds", "DELEGATION_CHILD_TIMEOUT_SECONDS", _parse_timeout, DEFAULT_CHILD_TIMEOUT, "delegation.child_timeout_seconds=%r is not a valid number; using default (no timeout)", ) def _get_max_spawn_depth() -> int: """delegation.max_spawn_depth floored at 1 (no ceiling). Depth 0 is the parent; agents at depths 0..N-1 may spawn, depth N is the leaf floor. Default 1 is flat. Each extra level multiplies API cost.""" def _floored(v): ival = int(v) if ival < _MIN_SPAWN_DEPTH: logger.warning("delegation.max_spawn_depth=%d below floor %d; using %d", ival, _MIN_SPAWN_DEPTH, _MIN_SPAWN_DEPTH) return max(_MIN_SPAWN_DEPTH, ival) return _knob( "max_spawn_depth", None, _floored, MAX_DEPTH, f"delegation.max_spawn_depth=%r is not a valid integer; using default {MAX_DEPTH}", ) def _get_orchestrator_enabled() -> bool: """delegation.orchestrator_enabled kill switch (default True): False forces every child to leaf.""" val = _cfg().get("orchestrator_enabled", True) if isinstance(val, bool): return val # Accept "true"/"false" strings from YAML that doesn't auto-coerce. if isinstance(val, str): return val.strip().lower() in {"true", "1", "yes", "on"} return True def _get_inherit_mcp_toolsets() -> bool: """Whether narrowed child toolsets should keep the parent's MCP toolsets.""" return is_truthy_value(_cfg().get("inherit_mcp_toolsets"), default=True) def _normalized_runtime_url(value: Any) -> str: return str(value or "").strip().rstrip("/") def _inherit_parent_capabilities(parent_agent, override_provider, override_base_url) -> Optional[dict]: """Parent's endpoint-trust capability map for a child, or None. ``agent.capabilities`` is a trust decision scoped to one provider+endpoint: inherited ONLY when the child runs the parent's exact route; any provider or base_url override stays DEFAULT-DENY (matches the /model switch posture). See #94036, #97292. """ if override_provider or override_base_url: return None parent_caps = getattr(parent_agent, "capabilities", None) if not isinstance(parent_caps, dict): return None return {key: value for key, value in parent_caps.items() if isinstance(key, str) and isinstance(value, bool)} def _inherit_parent_endpoint(parent_agent, surface_base_url: Optional[str], surface_api_key: Any) -> tuple: """``(base_url, api_key)`` the parent is actually calling, taken from ONE source. ``parent_agent.base_url`` / ``api_key`` can lag the live client (old OpenRouter URL vs local Ollama; a fallback runtime the surface attributes have not caught up with), and pairing the live endpoint with the surface key hands the child a (base_url, key) pair that was never valid anywhere — an instant, non-retryable 401 (#90009). The live client's key travels with its URL; the surface attributes are used only when there is no live OpenAI-wire client (native Anthropic/Bedrock runtimes keep ``client=None``).""" client_kwargs = getattr(parent_agent, "_client_kwargs", None) client = getattr(parent_agent, "client", None) live_candidates = ( (client_kwargs.get("base_url"), client_kwargs.get("api_key")) if isinstance(client_kwargs, dict) else (None, None), # OpenAI SDK exposes base_url as httpx.URL — coerce before comparing. (getattr(client, "base_url", ""), getattr(client, "api_key", None)) if client is not None else (None, None), ) for raw_url, live_key in live_candidates: url = _normalized_runtime_url(raw_url) if url and url.startswith(("http://", "https://")): return url, (live_key or surface_api_key) return (surface_base_url or None), surface_api_key def _loaded_pool(key: Any): """``load_pool(key)`` when it holds credentials, else None.""" from agent.credential_pool import load_pool pool = load_pool(key) return pool if pool is not None and pool.has_credentials() else None def _pool_serves_endpoint(pool: Any, provider: Optional[str], base_url: Optional[str]) -> bool: """Provider identity AND at least one entry for the child's endpoint; pools without entry metadata pass.""" from agent.credential_pool import ( credential_pool_entry_serves_endpoint as _entry_serves_endpoint, credential_pool_matches_provider, ) if not credential_pool_matches_provider(pool, provider, base_url=base_url): return False entries_fn = getattr(pool, "entries", None) if not callable(entries_fn): return True entries = entries_fn() return not isinstance(entries, list) or any(_entry_serves_endpoint(entry, base_url) for entry in entries) def _resolve_child_credential_pool( effective_provider: Optional[str], parent_agent, effective_base_url: Optional[str] = None, effective_requested_provider: Optional[str] = None, ): """Credential pool for the child: parent's pool (same provider), that provider's own pool, or None (child keeps its fixed credential). Custom endpoints all collapse to ``provider="custom"``, so they are matched by endpoint identity (the ``custom:`` pool key) — sharing the parent's pool across different custom endpoints would overwrite the child's delegated base_url on lease; an unregistered custom endpoint (no custom_providers entry) keeps the child's fixed credential rather than inherit the parent's. Custom endpoints are a special case: every direct ``delegation.base_url`` runtime collapses to ``provider="custom"``, so bare provider equality would treat two *different* custom endpoints as interchangeable and let the child inherit the parent's pool. We therefore resolve custom runtimes by endpoint identity (the ``custom:`` pool key derived from the base_url) and only share the parent's pool when both resolve to the *same* custom endpoint. See #7833. Named custom providers may share one gateway URL with different credentials, so the inherited ``requested_provider`` identity takes precedence over URL-only matching (#45763): the child must not lease the first pool registered for the shared endpoint. """ parent_pool = getattr(parent_agent, "_credential_pool", None) if not effective_provider: return parent_pool parent_provider = getattr(parent_agent, "provider", None) or "" try: if effective_provider == "custom": from agent.credential_pool import get_custom_provider_pool_key child_key = get_custom_provider_pool_key(effective_base_url, provider_name=effective_requested_provider) if child_key is None: return None parent_key = get_custom_provider_pool_key( getattr(parent_agent, "base_url", None), provider_name=getattr(parent_agent, "requested_provider", None), ) if parent_pool is not None and parent_provider == "custom" and parent_key is not None and parent_key == child_key: return parent_pool return _loaded_pool(child_key) if parent_pool is not None and effective_provider == parent_provider: if not effective_base_url or _pool_serves_endpoint(parent_pool, effective_provider, effective_base_url): return parent_pool logger.debug("Parent %s pool has no entry for child endpoint %s; not sharing it", effective_provider, effective_base_url) pool = _loaded_pool(effective_provider) if pool is not None and effective_base_url and not _pool_serves_endpoint(pool, effective_provider, effective_base_url): return None # child keeps its fixed credential return pool except Exception as exc: if effective_provider == "custom": logger.debug("Could not resolve custom credential pool for child endpoint '%s': %s", effective_base_url, exc) else: logger.debug("Could not load credential pool for child provider '%s': %s", effective_provider, exc) return None def _merge_request_overrides(runtime_overrides, explicit_overrides): """Merge explicit ``delegation.request_overrides`` OVER runtime-derived ones. Explicit top-level keys win; ``extra_body`` is deep-merged ONE level so provider personality (e.g. ``thinking: {type: disabled}``) survives unless the explicit dict redefines that exact key. Both sides are deep-copied so transport-side mutation can't leak into the config/runtime cache. None when both are empty.""" import copy as _copy runtime_overrides = runtime_overrides if isinstance(runtime_overrides, dict) else None explicit_overrides = explicit_overrides if isinstance(explicit_overrides, dict) else None if not runtime_overrides and not explicit_overrides: return None merged = _copy.deepcopy(runtime_overrides) if runtime_overrides else {} explicit = _copy.deepcopy(explicit_overrides) if explicit_overrides else {} runtime_extra = merged.get("extra_body") explicit_extra = explicit.pop("extra_body", None) merged.update(explicit) if isinstance(runtime_extra, dict) and isinstance(explicit_extra, dict): runtime_extra.update(explicit_extra) merged["extra_body"] = runtime_extra elif explicit_extra is not None: merged["extra_body"] = explicit_extra return merged or None # Native-SDK providers speak their own wire protocol and can't be reached via chat_completions against a base_url: # always take the runtime-provider path (a configured base_url still flows through it, e.g. a Bedrock region). _NATIVE_SDK_PROVIDERS = frozenset({"bedrock", "vertex", "google", "google-genai"}) _EXPLICIT_API_MODES = frozenset({"chat_completions", "codex_responses", "anthropic_messages"}) def _require_pinned_command(command: Optional[str], message: str) -> None: """A pinned ACP transport command must exist on PATH — refuse loudly rather than let the child silently fall back to another transport.""" import shutil as _shutil if command and not _shutil.which(command): raise ValueError(message) def _credential_bundle(model, provider, base_url, api_key, api_mode, request_overrides, **extra) -> dict: """The child credential dict every branch of ``_resolve_delegation_credentials`` returns.""" return { "model": model, "provider": provider, "base_url": base_url, "api_key": api_key, "api_mode": api_mode, "request_overrides": request_overrides, **extra, } def _direct_endpoint_credentials(v: dict, explicit_request_overrides) -> dict: """``delegation.base_url`` branch: provider/api_mode from URL heuristics.""" # Shared URL-based api_mode detector so Anthropic-compatible direct endpoints (/anthropic suffix: Azure AI # Foundry, MiniMax, Zhipu, LiteLLM) get the Messages transport instead of 404ing on chat_completions. # Without this, subagents would default to chat_completions and hit 404s on endpoints that only speak # the Anthropic Messages protocol. Fixes #10213. from hermes_cli.runtime_provider import _detect_api_mode_for_url base_lower = v["base_url"].lower() host = base_url_hostname(v["base_url"]) provider = "custom" api_mode = _detect_api_mode_for_url(v["base_url"]) or "chat_completions" if host == "chatgpt.com" and "/backend-api/codex" in base_lower: provider, api_mode = "openai-codex", "codex_responses" elif host == "api.anthropic.com": provider, api_mode = "anthropic", "anthropic_messages" elif "api.kimi.com/coding" in base_lower: api_mode = "anthropic_messages" # Explicit delegation.api_mode always wins over the URL heuristic; a provider plugin's # registered dialect counts as explicit. from agent.transports import registered_api_modes if v["api_mode"] in _EXPLICIT_API_MODES or (v["api_mode"] and v["api_mode"] in registered_api_modes()): api_mode = v["api_mode"] # Preserve the configured provider's request personality on an explicit endpoint. request_overrides = None if v["provider"]: try: from hermes_cli.runtime_provider import resolve_runtime_provider runtime = resolve_runtime_provider(requested=v["provider"], target_model=v["model"]) request_overrides = dict(runtime.get("request_overrides") or {}) or None except Exception as exc: logger.debug( "delegation.base_url: runtime resolution for provider '%s' failed; proceeding without request_overrides: %s", v["provider"], exc, ) # api_key None → inherited from parent in _build_child_agent return _credential_bundle( v["model"], provider, v["base_url"], v["api_key"], api_mode, _merge_request_overrides(request_overrides, explicit_request_overrides), ) def _runtime_provider_credentials(v: dict, explicit_request_overrides) -> dict: """``delegation.provider`` branch: full bundle via the runtime provider system.""" configured_provider = v["provider"] try: from hermes_cli.runtime_provider import resolve_runtime_provider runtime = resolve_runtime_provider(requested=configured_provider, target_model=v["model"]) except Exception as exc: raise ValueError( f"Cannot resolve delegation provider '{configured_provider}': {exc}. " f"Check that the provider is configured (API key set, valid provider name), " f"or set delegation.base_url/delegation.api_key for a direct endpoint. " f"Available providers: openrouter, nous, zai, kimi-coding, minimax." ) from exc api_key = runtime.get("api_key", "") if not api_key: raise ValueError( f"Delegation provider '{configured_provider}' resolved but has no API key. " f"Set the appropriate environment variable or run 'hermes auth'." ) # A pinned ACP transport command must exist — refuse the spawn loudly rather than letting the child # silently fall back to another transport (#80450). pinned_command = runtime.get("command") _require_pinned_command( pinned_command, f"Delegation provider '{configured_provider}' is pinned to the " f"'{pinned_command}' command, which was not found on PATH. " f"Install it or choose a different delegation provider.", ) # A provider override is assembled into the child as one bundle (see _resolve_child_runtime), so it must # carry its own endpoint. ACP transports are addressed by command and native-SDK providers by their SDK, # so neither needs a URL; anything else without one would otherwise be built with no endpoint at all. if (not runtime.get("base_url") and not pinned_command and configured_provider.strip().lower() not in _NATIVE_SDK_PROVIDERS): raise ValueError( f"Delegation provider '{configured_provider}' resolved without a base_url. " f"Refusing to build a subagent with an incomplete credential bundle — check the provider's " f"configuration / auth, or set delegation.base_url for a direct endpoint." ) return _credential_bundle( v["model"] or runtime.get("model") or None, configured_provider if runtime.get("provider") == _RUNTIME_PROVIDER_CUSTOM else runtime.get("provider"), runtime.get("base_url"), api_key, runtime.get("api_mode"), _merge_request_overrides(runtime.get("request_overrides"), explicit_request_overrides) or {}, command=pinned_command, args=list(runtime.get("args") or []), ) def _resolve_delegation_credentials(cfg: dict, parent_agent) -> dict: """Child credential bundle from the ``delegation`` config section. Three branches: ``base_url`` set → direct endpoint (``api_key`` None means inherit the parent's key, so providers keyed outside OPENAI_API_KEY work); ``provider`` set → full bundle via the runtime provider system (same path as CLI/gateway startup); neither → None values, child inherits everything. ``request_overrides`` is honored on every branch. Raises ValueError with a user-facing message.""" values = {k: str(cfg.get(k) or "").strip() or None for k in ("model", "provider", "base_url", "api_key")} values["api_mode"] = str(cfg.get("api_mode") or "").strip().lower() or None explicit_request_overrides = cfg.get("request_overrides") if isinstance(cfg.get("request_overrides"), dict) else None is_native_sdk_provider = (values["provider"] or "").strip().lower() in _NATIVE_SDK_PROVIDERS if values["base_url"] and not is_native_sdk_provider: return _direct_endpoint_credentials(values, explicit_request_overrides) if not values["provider"]: # Pure inherit; explicit request_overrides still merge OVER the parent's. return _credential_bundle( values["model"], None, None, None, None, _merge_request_overrides(getattr(parent_agent, "request_overrides", None), explicit_request_overrides), ) return _runtime_provider_credentials(values, explicit_request_overrides) def _load_config() -> dict: """The ``delegation`` config section (read-only — do NOT mutate). Prefers the shared ``load_config_readonly()`` (follows HERMES_HOME/profile; no deepcopy, since this runs on every get_definitions() rebuild) over the legacy ``cli.CLI_CONFIG``, which can hide user-set keys — except that ``HERMES_IGNORE_USER_CONFIG=1`` is only honored by the legacy loader, so it stays authoritative when that flag is set.""" if os.environ.get("HERMES_IGNORE_USER_CONFIG") != "1": try: from hermes_cli.config import load_config_readonly cfg = load_config_readonly().get("delegation") or {} if isinstance(cfg, dict): return cfg except Exception: pass try: from cli import CLI_CONFIG cfg = CLI_CONFIG.get("delegation") or {} return cfg if isinstance(cfg, dict) else {} except Exception: return {} # OpenRouter routing filters: inherited from the parent, but reset to these defaults under a pinned provider — parent # filters (e.g. only=["Anthropic"]) would silently force the child back onto the parent's provider. # openrouter_min_coding_score stays inherited: model-gated, no-op elsewhere. _ROUTING_FILTER_DEFAULTS = ( ("providers_allowed", None), ("providers_ignored", None), ("providers_order", None), ("provider_sort", None), ("provider_require_parameters", False), ("provider_data_collection", ""), ) _NOUS_PROVIDERS = frozenset({"nous", "nous-portal", "nousresearch"}) def _resolve_child_fallback_chain(parent_agent, routing_cfg: Any, pinned: bool) -> Optional[List[Dict[str, Any]]]: """Fallback chain for a child, owned by the same config block as its route. Pinned children (provider, endpoint or model override) never borrow the parent chain; unpinned children inherit it when ``fallback_providers`` is absent/null. An explicit ``[]`` disables fallback either way. Malformed entries are dropped by the canonical normalizer. Same rule as a pinned cron job (``cron/scheduler.py::_job_fallback_chain``). """ return scoped_fallback_chain( getattr(parent_agent, "_fallback_chain", None), routing_cfg.get("fallback_providers") if isinstance(routing_cfg, dict) else None, pinned=pinned, owner="delegation") def _resolve_child_runtime( parent_agent, delegation_cfg: dict, parent_api_key: Any, *, model: Optional[str], override_provider: Optional[str], override_base_url: Optional[str], override_api_key: Optional[str], override_api_mode: Optional[str], override_acp_command: Optional[str], override_acp_args: Optional[List[str]], routing_cfg: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Child credentials, transport and routing (config override > parent inherit) as ``AIAgent`` kwargs. Rules that are easy to break: api_mode is re-derived (not inherited) when the child's provider differs from the parent's or is Nous Portal (dual-wire); a pinned ``delegation.command`` must exist on PATH or the spawn fails loudly; ``override_provider`` clears the parent's ACP transport, fallback chain and OpenRouter routing filters so the pinned provider is actually honoured.""" effective_model = model or parent_agent.model # provider/base_url are one bundle: all from the override, or all from the parent. Per-field fallback built # children like an override provider pointed at the PARENT's endpoint (e.g. copilot credentials on the # parent's Codex URL), which 404s on every request and can't be rescued by the fallback chain, whose dedup # matches provider+model and so skips the entry as a self-loop. _inherit_parent_endpoint recovers the # parent's live endpoint (with its key), which is meaningless for a different provider. if override_provider: effective_provider = override_provider effective_base_url = override_base_url elif override_base_url: effective_provider = getattr(parent_agent, "provider", None) effective_base_url = override_base_url else: effective_provider = getattr(parent_agent, "provider", None) effective_base_url, parent_api_key = _inherit_parent_endpoint(parent_agent, parent_agent.base_url, parent_api_key) # api_mode: each provider has its own wire, so a different provider re-derives (None) instead of inheriting (404s # otherwise). Nous Portal is dual-wire within one provider (anthropic/* → Messages, else chat_completions), so # same-provider inheritance would pin the child on the wrong wire — re-derive. # Bug #20558 / PR #20563: api_mode must NOT be inherited when the child uses a different provider than # the parent — each provider has its own API surface (e.g. MiniMax uses anthropic_messages, DeepSeek # uses chat_completions). Inheriting the parent's mode causes 404 errors when the child routes to the # wrong endpoint. Derive the mode from the target provider when it differs. Same-provider inheritance # would pin a child Hermes/Qwen subagent onto the parent's Claude Messages wire (or the reverse). # agent_init honors an explicit api_mode above its nous branch, so re-derive here before construction. _parent_provider = getattr(parent_agent, "provider", None) or "" if override_api_mode is not None: effective_api_mode = override_api_mode elif (effective_provider or "").strip().lower() in _NOUS_PROVIDERS: from hermes_cli.providers import nous_api_mode effective_api_mode = nous_api_mode(effective_model) elif effective_provider != _parent_provider: effective_api_mode = None # force re-derivation from provider's defaults else: effective_api_mode = getattr(parent_agent, "api_mode", None) # A pinned transport that cannot run must fail the spawn loudly, never fall # back silently (delegate_task pre-validates; this covers direct callers). _require_pinned_command( override_acp_command, f"Pinned delegation command '{override_acp_command}' was not " f"found on PATH. Install it or remove delegation.command from config.yaml.", ) effective_acp_command = override_acp_command or getattr(parent_agent, "acp_command", None) effective_acp_args = list( override_acp_args if override_acp_args is not None else (getattr(parent_agent, "acp_args", []) or []) ) # A pinned provider must use direct API calls; inheriting the parent's ACP # transport would bypass the override credentials entirely. # Inheriting acp_command unconditionally causes run_agent.py to initialize CopilotACPClient, bypassing # override credentials entirely (issue #16816). if override_provider and not override_acp_command: effective_acp_command, effective_acp_args = None, [] # Defensive: validate trusted delegation.command exists on PATH before honoring it. An explicitly pinned # transport that cannot run must fail the spawn loudly (#80450) — silently falling back to the default # transport would run the child somewhere the user explicitly routed it away from. Normally unreachable # via delegate_task, which pre-validates the command in _resolve_delegation_credentials. if override_acp_command: from providers import get_provider_profile profile = get_provider_profile(effective_provider or "") # A generic process command does not imply the legacy ACP protocol. if profile is None or profile.auth_type != "external_process": effective_provider, effective_api_mode = "copilot-acp", "chat_completions" # A named provider identity is endpoint-scoped. Preserve it only when the # child inherits the exact parent route; an override owns its final identity. effective_requested_provider = effective_provider if not override_provider and not override_base_url and not override_acp_command: effective_requested_provider = ( getattr(parent_agent, "requested_provider", None) or effective_provider ) # Reasoning: delegation.reasoning_effort > parent. Keep the raw value — a # YAML ``false`` must disable thinking, not coerce to "" and inherit. child_reasoning = getattr(parent_agent, "reasoning_config", None) try: delegation_effort = delegation_cfg.get("reasoning_effort") if delegation_effort or delegation_effort is False: from hermes_constants import parse_reasoning_effort parsed = parse_reasoning_effort(delegation_effort) if parsed is None: logger.warning("Unknown delegation.reasoning_effort '%s', inheriting parent level", delegation_effort) else: child_reasoning = parsed except Exception as exc: logger.debug("Could not load delegation reasoning_effort: %s", exc) kwargs: Dict[str, Any] = { "base_url": effective_base_url, "api_key": override_api_key or parent_api_key, "model": effective_model, "provider": effective_provider, "requested_provider": effective_requested_provider, "capabilities": _inherit_parent_capabilities(parent_agent, override_provider, override_base_url), "api_mode": effective_api_mode, "acp_command": effective_acp_command, "acp_args": effective_acp_args, "reasoning_config": child_reasoning, # Resolve routing and recovery policy from the same configuration owner. A pinned provider, endpoint, or # model never borrows the parent's chain; an explicitly declared child chain still remains available. "fallback_model": _resolve_child_fallback_chain( parent_agent, delegation_cfg if routing_cfg is None else routing_cfg, pinned=bool(override_provider or override_base_url or model)), "openrouter_min_coding_score": getattr(parent_agent, "openrouter_min_coding_score", None), # Routing filters reset to their defaults under a pinned provider (see _ROUTING_FILTER_DEFAULTS). **{a: d if override_provider else getattr(parent_agent, a, d) for a, d in _ROUTING_FILTER_DEFAULTS}, } if not override_provider: kwargs["provider_data_collection"] = kwargs["provider_data_collection"] or "" child_max_tokens = getattr(parent_agent, "max_tokens", None) if isinstance(child_max_tokens, int): kwargs["max_tokens"] = child_max_tokens return kwargs