diff --git a/tools/apply_layout_tool.py b/tools/apply_layout_tool.py index 1b7cae8dba..bda3960306 100644 --- a/tools/apply_layout_tool.py +++ b/tools/apply_layout_tool.py @@ -12,6 +12,8 @@ model can self-correct without a registry-listing tool. from tools import desktop_ui from tools.registry import registry, tool_error +# Renderer answers via the blocking-prompt bridge; layout apply is synchronous +# there, so the bridge timeout is generous. _TIMEOUT_NOTE = "Layout apply is only available in the Hermes desktop app." diff --git a/tools/cronjob_job_args.py b/tools/cronjob_job_args.py index 094061e8a3..70e029a28f 100644 --- a/tools/cronjob_job_args.py +++ b/tools/cronjob_job_args.py @@ -42,7 +42,7 @@ def _origin_from_env() -> Optional[Dict[str, str]]: "thread_id": thread_id, # Lets an opt-in delivery mirror resolve the exact participant's # session in per-user-isolated group chats (parity with send_message). - "user_id": get_session_env("HERMES_SESSION_USER_ID") or None, + "user_id": get_session_env("HERMES_SESSION_USER_ID") or None, # harmless for DMs # Workspace/server scope (Slack team, Discord guild...). Slack session # keys embed it, so a continuable cron seed built without it would # create a row no scoped reply ever resolves to. @@ -138,7 +138,9 @@ def _split_monitor_arg( """Resolve the single model-facing ``monitor`` field into the stored ``(monitor_script, monitor_url)`` pair. - Shape decides transport: http(s):// is a URL, anything else a script path. + Shape decides transport: http(s):// is a URL, anything else a script path + (a legal script path can never start with a URL scheme). Storage keeps the + two fields separate — interface merge, not a storage migration. Update semantics: None = unchanged, '' = clear; setting one source clears the other so switching transports never trips mutual exclusion. An explicit ``monitor`` wins over the legacy alias fields. @@ -154,8 +156,8 @@ def _split_monitor_arg( def _repeat_display(job: Dict[str, Any]) -> str: - times = (job.get("repeat") or {}).get("times") - completed = (job.get("repeat") or {}).get("completed", 0) + rep = job.get("repeat") or {} + times, completed = rep.get("times"), rep.get("completed", 0) if times is None: return "forever" if times == 1: @@ -174,12 +176,11 @@ def _clean_str_list(items: Any) -> List[str]: def _canonical_skills(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]: if skills is None: - raw_items = [skill] if skill else [] + skills = [skill] if skill else [] elif isinstance(skills, str): - raw_items = [skills] - else: - raw_items = list(skills) - return list(dict.fromkeys(_clean_str_list(item or "" for item in raw_items))) + skills = [skills] + # `item or ""`: a None entry must drop out, not stringify to "None". + return list(dict.fromkeys(_clean_str_list(item or "" for item in skills))) def _normalize_optional_job_value(value: Optional[Any], *, strip_trailing_slash: bool = False) -> Optional[str]: @@ -209,7 +210,8 @@ def _validate_bot_chat_deliver(deliver: Optional[str]) -> Optional[str]: """Validate ``bot-chat[:]`` deliver elements at create time. Bot Chat delivery is machine-local: the profile must exist where the - scheduler fires. Fail loudly here rather than as a per-run delivery error. + scheduler fires (Desktop multi-gateway rosters may show same-named profiles + from other machines). Fail loudly here rather than as a per-run delivery error. Returns an error string or None. """ if not deliver: @@ -245,6 +247,7 @@ def _resolve_cron_context_deliver(deliver: Optional[str]) -> Optional[str]: (or an omitted value) becomes the creating run's ``platform:chat_id[:thread]`` from the HERMES_CRON_AUTO_DELIVER_* contextvars, or ``local`` when the creating run has no concrete target; other elements pass through verbatim. + Without this the scheduler would fall back to guessing a home channel. """ from gateway.session_context import get_session_env from utils import is_truthy_value @@ -301,6 +304,8 @@ def _validate_cron_base_url( return f"Unable to validate base_url override for provider {prov!r}; refused." if prov.lower() == "custom": + # Pure BYOK: key comes from a pool keyed by THIS base_url or host-gated + # env vars, never an arbitrary stored secret. return None if has_named_custom_provider(prov): # A NAMED custom provider carries a STORED key that the runtime still @@ -377,7 +382,8 @@ def _apply_continuity( def _validate_context_from_refs(refs: List[Any]) -> Optional[str]: """Error string if any non-"self" ref names a job that doesn't exist. - ("self" resolves to the job's own id at run time.)""" + ("self" resolves to the job's own id at run time, so it can't be checked + against the store — the job doesn't exist yet at create time.)""" from cron.jobs import get_job as _get_job for ref_id in refs: if isinstance(ref_id, str) and ref_id.strip().lower() == "self": diff --git a/tools/cronjob_prompt_scan.py b/tools/cronjob_prompt_scan.py index 8ae1beff6f..12906ff178 100644 --- a/tools/cronjob_prompt_scan.py +++ b/tools/cronjob_prompt_scan.py @@ -82,15 +82,6 @@ def _zwj_has_emoji_neighbour(text: str, idx: int) -> bool: ) -def _strip_legitimate_emoji_zwj(prompt: str) -> str: - if '\u200d' not in prompt: - return prompt - return ''.join( - ch for idx, ch in enumerate(prompt) - if not (ch == '\u200d' and _zwj_has_emoji_neighbour(prompt, idx)) - ) - - def _strip_cron_safe_constructs(prompt: str) -> str: """Scrub the bundled GitHub skill's `Authorization: token $GITHUB_TOKEN` + api.github.com curl so it doesn't trip the auth-header exfil rule. @@ -112,10 +103,11 @@ def _strip_cron_safe_constructs(prompt: str) -> str: def _check_invisible_unicode(prompt: str) -> str: - """Error string if the prompt holds invisible-unicode markers (emoji ZWJ allowed).""" - prompt_for_invisible_scan = _strip_legitimate_emoji_zwj(prompt) + """Error string if the prompt holds invisible-unicode markers (emoji ZWJ allowed). + Reports the first offender in ``_CRON_INVISIBLE_CHARS`` order.""" + removed = set(_strip_invisible_unicode(prompt)[1]) for char in _CRON_INVISIBLE_CHARS: - if char in prompt_for_invisible_scan: + if f"U+{ord(char):04X}" in removed: return f"Blocked: prompt contains invisible unicode U+{ord(char):04X} (possible injection)." return "" diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 0b9c8ae7c3..4da0f64e8d 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -5,6 +5,7 @@ Expose a single compressed action-oriented tool to avoid schema/context bloat. Compatibility wrappers remain for direct Python callers and legacy tests. """ +import contextlib import json import logging import sys @@ -19,7 +20,8 @@ logger = logging.getLogger(__name__) # Heartbeat cadence that keeps the calling agent's inactivity watchdog at bay # while a manual `cronjob(action="run")` executes synchronously in-process -# (mirrors tools/environments/base.py::touch_activity_if_due). +# (mirrors tools/environments/base.py::touch_activity_if_due; comfortably +# below the default HERMES_AGENT_TIMEOUT). _CRON_RUN_HEARTBEAT_INTERVAL = 10.0 # Hard ceiling on the heartbeat: with HERMES_CRON_TIMEOUT=0 (unlimited) a truly @@ -54,6 +56,7 @@ from tools.cronjob_prompt_scan import ( # noqa: F401 (re-exported) from tools.cronjob_job_args import ( # noqa: F401 (re-exported) _apply_continuity, _canonical_skills, + _clean_str_list, _format_job, _gateway_liveness_notice, _local_delivery_notice, @@ -117,7 +120,47 @@ def _forward_relay_fronted_run( """ if not _relay_fronted_delivery_platforms(job): return None - job_id = job["id"] + url = f"{_api_server_base_url()}/api/jobs/{job['id']}/run" + + from agent.secret_scope import get_secret + + key = get_secret("API_SERVER_KEY", "") or "" + try: + import httpx + + resp = httpx.post( + url, + headers={"Authorization": f"Bearer {key}"}, + json=({"prompt": extra_prompt} if extra_prompt else {}), + timeout=10.0, + ) + except Exception: + resp = None + + if resp is not None and resp.status_code < 300: + return _dumps({ + "success": True, + "forwarded_to_gateway": True, + "note": ( + "This job targets a relay-fronted platform; it was dispatched " + "to the running gateway, whose live relay adapter owns that " + "delivery." + ), + }) + return _dumps({ + "success": False, + "error": ( + "This job targets a relay-fronted platform, which has no " + "standalone sender. Start the gateway — its ticker will " + "deliver the job on schedule via the live relay adapter." + ), + }) + + +def _api_server_base_url() -> str: + """``http://host:port`` of the local api_server, mirroring its bind + resolution (extra.host -> API_SERVER_HOST -> 127.0.0.1); a wildcard bind + listens on loopback too.""" import os port_raw = os.getenv("API_SERVER_PORT", "").strip() @@ -125,9 +168,6 @@ def _forward_relay_fronted_run( port = int(port_raw) if port_raw else 8642 except ValueError: port = 8642 - # Mirror the api_server's bind resolution (extra.host -> API_SERVER_HOST - # -> 127.0.0.1); a wildcard bind listens on loopback too. - host = "" try: from hermes_cli.config import cfg_get, load_config_readonly @@ -146,49 +186,7 @@ def _forward_relay_fronted_run( host = "127.0.0.1" if ":" in host and not host.startswith("["): host = f"[{host}]" # bare IPv6 literal - url = f"http://{host}:{port}/api/jobs/{job_id}/run" - - from agent.secret_scope import get_secret - - key = get_secret("API_SERVER_KEY", "") or "" - - resp = None - try: - import httpx - - resp = httpx.post( - url, - headers={"Authorization": f"Bearer {key}"}, - json=({"prompt": extra_prompt} if extra_prompt else {}), - timeout=10.0, - ) - except Exception: - resp = None - - if resp is not None and resp.status_code < 300: - return json.dumps( - { - "success": True, - "forwarded_to_gateway": True, - "note": ( - "This job targets a relay-fronted platform; it was dispatched " - "to the running gateway, whose live relay adapter owns that " - "delivery." - ), - }, - indent=2, - ) - return json.dumps( - { - "success": False, - "error": ( - "This job targets a relay-fronted platform, which has no " - "standalone sender. Start the gateway — its ticker will " - "deliver the job on schedule via the live relay adapter." - ), - }, - indent=2, - ) + return f"http://{host}:{port}" def _manual_run_delivery_note(deliver: str, refreshed: Dict[str, Any]) -> str: @@ -258,6 +256,68 @@ def _execute_job_now( return _run_claimed_job(claimed_job, extra_prompt=extra_prompt) +def _gateway_adapters_and_loop(): + """Live gateway adapter map + event loop when running inside the gateway + process, else ``(None, None)``. Manual runs from a gateway agent must + deliver on the loop that owns clients such as Matrix/aiohttp (a standalone + asyncio.run() loop breaks them).""" + runner_ref = getattr(sys.modules.get("gateway.run"), "_gateway_runner_ref", None) + runner = runner_ref() if callable(runner_ref) else None + if runner is None: + return None, None + return getattr(runner, "adapters", None), getattr(runner, "_gateway_loop", None) + + +@contextlib.contextmanager +def _run_heartbeat(job_name: str): + """Heartbeat into the caller's activity tracker while a manual run + executes; stops and joins the thread on exit. + + A manual run executes synchronously on the caller's thread and can take + minutes; without tool activity the gateway inactivity watchdog would kill + the parent turn. Best-effort: no callback -> no thread, unchanged behavior. + """ + stop = threading.Event() + thread = None + try: + from tools.environments.base import get_activity_callback + + # Capture on THIS thread: the callback is thread-local (installed by + # the tool executor), so a freshly spawned thread cannot read it. + activity_cb = get_activity_callback() + except Exception: + activity_cb = None + + def _heartbeat_loop() -> None: + started = time.monotonic() + while not stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL): + elapsed = time.monotonic() - started + if elapsed > _CRON_RUN_HEARTBEAT_CEILING: + # A run this long with an unlimited child watchdog is + # likely wedged — stop masking the gateway watchdog. + logger.warning( + "cronjob run heartbeat ceiling reached for job " + "'%s' (%.0fs) — stopping heartbeat; gateway " + "watchdog regains authority", + job_name, elapsed, + ) + return + try: + activity_cb(f"cronjob: running job '{job_name}' ({int(elapsed)}s elapsed)") + except Exception: + continue # one transient callback error must not drop protection + + if activity_cb is not None: + thread = threading.Thread(target=_heartbeat_loop, daemon=True, name="cronjob-run-heartbeat") + thread.start() + try: + yield + finally: + stop.set() + if thread is not None: + thread.join(timeout=_CRON_RUN_HEARTBEAT_INTERVAL + 1) + + def _run_claimed_job( job: Dict[str, Any], extra_prompt: Optional[str] = None ) -> Dict[str, Any]: @@ -287,73 +347,16 @@ def _run_claimed_job( claim = job.get("fire_claim") fire_owner = str(claim.get("by") or "") if isinstance(claim, dict) else None - # A manual run executes synchronously on the caller's thread and can - # take minutes; without tool activity the gateway inactivity watchdog - # would kill the parent turn. Heartbeat into the caller's activity - # tracker while the job runs (best-effort: no callback -> unchanged). + adapters, gateway_loop = _gateway_adapters_and_loop() try: - from tools.environments.base import get_activity_callback - - # Capture on THIS thread: the callback is thread-local. - activity_cb = get_activity_callback() - except Exception: - activity_cb = None - - _heartbeat_stop = threading.Event() - _heartbeat_thread = None - - if activity_cb is not None: - job_name = str(job.get("name") or job_id) - - def _heartbeat_loop() -> None: - started = time.monotonic() - while not _heartbeat_stop.wait(_CRON_RUN_HEARTBEAT_INTERVAL): - elapsed = time.monotonic() - started - if elapsed > _CRON_RUN_HEARTBEAT_CEILING: - logger.warning( - "cronjob run heartbeat ceiling reached for job " - "'%s' (%.0fs) — stopping heartbeat; gateway " - "watchdog regains authority", - job_name, elapsed, - ) - return - try: - activity_cb( - f"cronjob: running job '{job_name}' ({int(elapsed)}s elapsed)" - ) - except Exception: - continue # one transient callback error must not drop protection - - _heartbeat_thread = threading.Thread( - target=_heartbeat_loop, - daemon=True, - name="cronjob-run-heartbeat", - ) - _heartbeat_thread.start() - - # Manual runs from a gateway agent share the process with live platform - # adapters: pass the gateway adapter map + event loop so delivery runs - # on the loop that owns clients such as Matrix/aiohttp (a standalone - # asyncio.run() loop breaks them). - gateway_module = sys.modules.get("gateway.run") - runner_ref = getattr(gateway_module, "_gateway_runner_ref", None) - runner = runner_ref() if callable(runner_ref) else None - adapters = getattr(runner, "adapters", None) if runner is not None else None - gateway_loop = getattr(runner, "_gateway_loop", None) if runner is not None else None - - try: - try: - # run_one_job records last_run_at/last_status via mark_job_run; - # `job` is the owner-bearing claimed snapshot, so terminal writes - # stay fenced by that owner. + # run_one_job records last_run_at/last_status via mark_job_run; + # `job` is the owner-bearing claimed snapshot, so terminal writes + # stay fenced by that owner. + with _run_heartbeat(str(job.get("name") or job_id)): processed = run_one_job( job, adapters=adapters, loop=gateway_loop, extra_prompt=extra_prompt, ) - finally: - _heartbeat_stop.set() - if _heartbeat_thread is not None: - _heartbeat_thread.join(timeout=_CRON_RUN_HEARTBEAT_INTERVAL + 1) finally: _registered = False release_running_job(job_id) @@ -373,28 +376,23 @@ def _run_claimed_job( except Exception as e: logger.error("Failed to execute cron job %s immediately: %s", job_id, e) if _registered: - # Only release registrations WE took — a bare discard could erase - # a ticker-owned entry. + # We raised before the run's own release (e.g. heartbeat setup); + # don't leave the job marked in-flight. Only release registrations + # WE took — a bare discard could erase a ticker-owned entry. try: - from cron.scheduler import release_running_job as _release - - _release(job_id) + release_running_job(job_id) except Exception: pass try: - mark_job_run( - job_id, - False, - str(e), - expected_fire_owner=fire_owner, - ) + mark_job_run(job_id, False, str(e), expected_fire_owner=fire_owner) except Exception: pass return {"claimed": True, "success": False, "error": str(e)} def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[str]: - """Best-effort excerpt of the job's most recent saved output file. Never raises.""" + """Best-effort excerpt of the job's most recent saved output file, for the + background-run completion block (parent sees what the job produced). Never raises.""" try: from cron.jobs import get_cron_output_dir @@ -412,6 +410,72 @@ def _latest_job_output_excerpt(job_id: str, max_chars: int = 2000) -> Optional[s return None +def _reap_stale_executions(job_name: str) -> None: + """Reap execution rows left 'claimed'/'running' by a dead owner process + (e.g. a prior one-shot `hermes cron run` that exited mid-run). The ticker + does this at its own startup; one-shot invocations have no such moment, so + a stale claim would block every later manual run. Only provably-dead + owners are reaped. Best-effort self-heal: must not block dispatch.""" + try: + from cron.executions import recover_interrupted_executions + + _reclaimed = recover_interrupted_executions() + if _reclaimed: + logger.warning( + "Reclaimed %d stale cron execution(s) from dead owner(s) " + "before dispatching job '%s'", + _reclaimed, + job_name, + ) + except Exception as _reap_exc: + logger.debug("Stale execution reclaim failed: %s", _reap_exc) + + +def _background_session_key(session_id: Optional[str]) -> str: + """Routing key for a detached completion, captured on THIS thread + (contextvars don't cross the pool). Empty string = no durable consumer.""" + try: + from tools.approval import get_current_session_key + + session_key = get_current_session_key(default="") + except Exception: + session_key = "" + if not session_key and session_id: + # CLI path: the approval contextvar is only bound during gateway/TUI + # turns; the CLI drain filters completions by the durable session id, + # and an empty key would fail closed (completion never claimable). + session_key = str(session_id) + return session_key + + +def _manual_run_completion( + res: Dict[str, Any], job_id: str, job_name: str, deliver: str, started_at: float +) -> Dict[str, Any]: + """Async-delegation completion block for a finished background manual run.""" + duration = round(time.time() - started_at, 2) + refreshed = get_job(job_id) or {} + lines = [ + f"Cron job '{job_name}' ({job_id}) finished its manual run.", + f"Result: {'ok' if res.get('success') else 'FAILED'}" + + (f" — {res.get('error')}" if res.get("error") else ""), + f"Delivery target: {deliver}" + + _manual_run_delivery_note(deliver, refreshed), + ] + if refreshed.get("next_run_at"): + lines.append(f"Next scheduled run: {refreshed['next_run_at']}") + excerpt = _latest_job_output_excerpt(job_id) + if excerpt: + lines.append("--- JOB OUTPUT ---") + lines.append(excerpt) + return { + "status": "completed" if res.get("success") else "error", + "summary": "\n".join(lines), + "error": res.get("error"), + "api_calls": 0, + "duration_seconds": duration, + } + + def _try_dispatch_background_run( job: Dict[str, Any], session_id: Optional[str] = None, extra_prompt: Optional[str] = None, @@ -431,7 +495,8 @@ def _try_dispatch_background_run( running in the background; ``{"claimed": True, "dispatched": False, ...}`` when the pool was full and the run executed inline (claim already taken). """ - # Finite sessions cannot route a detached result back after the turn ends. + # Finite sessions cannot route a detached result back after the turn ends + # — mirror delegate_task's gate and fall back to sync execution. try: from gateway.session_context import async_delivery_supported @@ -442,42 +507,14 @@ def _try_dispatch_background_run( job_id = job["id"] job_name = str(job.get("name") or job_id) + _reap_stale_executions(job_name) - # Reap execution rows left 'claimed'/'running' by a dead owner process - # (e.g. a prior one-shot `hermes cron run` that exited mid-run). The - # ticker does this at its own startup; one-shot invocations have no such - # moment, so a stale claim would block every later manual run. Only - # provably-dead owners are reaped. - try: - from cron.executions import recover_interrupted_executions - - _reclaimed = recover_interrupted_executions() - if _reclaimed: - logger.warning( - "Reclaimed %d stale cron execution(s) from dead owner(s) " - "before dispatching job '%s'", - _reclaimed, - job_name, - ) - except Exception as _reap_exc: - logger.debug("Stale execution reclaim failed: %s", _reap_exc) - - # Routing capture on THIS thread (contextvars don't cross the pool), and - # BEFORE the claim: with no routable session there is no durable consumer - # for a detached completion, so we must not claim-and-dispatch. - try: - from tools.approval import get_current_session_key - - session_key = get_current_session_key(default="") - except Exception: - session_key = "" - if not session_key and session_id: - # CLI path: the approval contextvar is only bound during gateway/TUI - # turns; the CLI drain filters completions by the durable session id. - session_key = str(session_id) + # Routing capture BEFORE the claim: with no routable session there is no + # durable consumer for a detached completion, so we must not + # claim-and-dispatch. Direct Python callers (`hermes cron run`, tests): + # process exits right after the tool returns, so run synchronously. + session_key = _background_session_key(session_id) if not session_key: - # Direct Python callers (`hermes cron run`, tests): process exits right - # after the tool returns, so run synchronously. return None # Best-effort early dedupe so a mid-run job reports in THIS tool response @@ -537,28 +574,7 @@ def _try_dispatch_background_run( def _runner() -> Dict[str, Any]: res = _run_claimed_job(claimed_job, extra_prompt=extra_prompt) - duration = round(time.time() - started_at, 2) - refreshed = get_job(job_id) or {} - lines = [ - f"Cron job '{job_name}' ({job_id}) finished its manual run.", - f"Result: {'ok' if res.get('success') else 'FAILED'}" - + (f" — {res.get('error')}" if res.get("error") else ""), - f"Delivery target: {deliver}" - + _manual_run_delivery_note(deliver, refreshed), - ] - if refreshed.get("next_run_at"): - lines.append(f"Next scheduled run: {refreshed['next_run_at']}") - excerpt = _latest_job_output_excerpt(job_id) - if excerpt: - lines.append("--- JOB OUTPUT ---") - lines.append(excerpt) - return { - "status": "completed" if res.get("success") else "error", - "summary": "\n".join(lines), - "error": res.get("error"), - "api_calls": 0, - "duration_seconds": duration, - } + return _manual_run_completion(res, job_id, job_name, deliver, started_at) dispatch = dispatch_async_delegation( goal=f"Manual run of cron job '{job_name}' ({job_id})", @@ -605,7 +621,8 @@ def _dumps(payload: Dict[str, Any]) -> str: def _action_create(a: Dict[str, Any]) -> str: - prompt, script, deliver = a["prompt"], a["script"], a["deliver"] + prompt, script = a["prompt"], a["script"] + deliver = _normalize_deliver_param(a["deliver"]) if not a["schedule"]: return tool_error("schedule is required for create", success=False) canonical_skills = _canonical_skills(a["skill"], a["skills"]) @@ -633,7 +650,7 @@ def _action_create(a: Dict[str, Any]) -> str: # credential to an attacker endpoint. or _validate_cron_base_url(a["provider"], a["base_url"]) # bot-chat targets are machine-local: fail the CREATE, not the run. - or _validate_bot_chat_deliver(_normalize_deliver_param(deliver)) + or _validate_bot_chat_deliver(deliver) # failure_deliver shares deliver's grammar and validators (NS-788). or _validate_bot_chat_deliver(_normalize_deliver_param(a["failure_deliver"])) or (a["context_from"] and _validate_context_from_refs( @@ -658,7 +675,7 @@ def _action_create(a: Dict[str, Any]) -> str: schedule=a["schedule"], name=a["name"], repeat=a["repeat"], - deliver=_resolve_cron_context_deliver(_normalize_deliver_param(deliver)), + deliver=_resolve_cron_context_deliver(deliver), origin=_origin_from_env(), skills=canonical_skills, model=_normalize_optional_job_value(a["model"]), @@ -683,11 +700,12 @@ def _action_create(a: Dict[str, Any]) -> str: _partial = exc.to_dict() return tool_error(_partial.pop("error"), success=False, **_partial) _create_message = f"Cron job '{job['name']}' created." - _local_notice = _local_delivery_notice(job, _normalize_deliver_param(deliver)) + _local_notice = _local_delivery_notice(job, deliver) if _local_notice: _create_message = f"{_create_message} {_local_notice}" - # A job created with no gateway running is stored but never fires — tell - # the model, which otherwise reports a clean success. + # The builtin ticker lives in the gateway process: a job created with no + # gateway running is stored but never fires — tell the model (the CLI + # already warns; the agent path otherwise reports a clean success). _result = { "success": True, "job_id": job["id"], @@ -702,10 +720,15 @@ def _action_create(a: Dict[str, Any]) -> str: "message": _create_message, **_gateway_liveness_notice(), } - _notes = _mode_guidance_notes(job, _normalize_deliver_param(deliver)) + return _dumps(_with_guidance(_result, job, deliver)) + + +def _with_guidance(result: Dict[str, Any], job: Dict[str, Any], deliver: Optional[str]) -> Dict[str, Any]: + """Attach mode/delivery guidance (create and update echo the same notes).""" + _notes = _mode_guidance_notes(job, deliver) if _notes: - _result["guidance"] = _notes - return _dumps(_result) + result["guidance"] = _notes + return result def _action_list(a: Dict[str, Any]) -> str: @@ -734,17 +757,23 @@ def _action_remove(job: Dict[str, Any], a: Dict[str, Any]) -> str: def _action_pause(job: Dict[str, Any], a: Dict[str, Any]) -> str: - updated = pause_job(job["id"], reason=a["reason"]) - _notify_provider_jobs_changed_safe() - return _dumps({"success": True, "job": _format_job(updated)}) + return _job_state_result(pause_job(job["id"], reason=a["reason"])) def _action_resume(job: Dict[str, Any], a: Dict[str, Any]) -> str: - updated = resume_job(job["id"]) + return _job_state_result(resume_job(job["id"])) + + +def _job_state_result(updated: Dict[str, Any]) -> str: _notify_provider_jobs_changed_safe() return _dumps({"success": True, "job": _format_job(updated)}) +def _refreshed_job_view(job_id: str) -> Dict[str, Any]: + """Re-read so the response reflects the post-run last_run_at/last_status.""" + return _format_job(get_job(job_id) or {"id": job_id}) + + def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str: job_id = job["id"] # `prompt` on run is transient per-fire context appended to the stored @@ -762,7 +791,7 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str: ) if bg is not None and bg.get("dispatched"): _notify_provider_jobs_changed_safe() - result = _format_job(get_job(job_id) or {"id": job_id}) + result = _refreshed_job_view(job_id) result["executed"] = True result["execution_mode"] = "background" result["delegation_id"] = bg.get("delegation_id") @@ -786,12 +815,12 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str: return forwarded exec_result = _execute_job_now(job, extra_prompt=extra_prompt) # A claimed direct run advances next_run_at and may race an external - # provider's one-shot for the same occurrence — reconcile after the run. + # provider's one-shot for the same occurrence; a lost consumed fire cannot + # re-arm itself, so reconcile after the run has persisted its final state. claimed = exec_result.get("claimed", False) if claimed: _notify_provider_jobs_changed_safe() - # Re-read so the response reflects the post-run last_run_at/last_status. - result = _format_job(get_job(job_id) or {"id": job_id}) + result = _refreshed_job_view(job_id) result["executed"] = claimed result["execution_success"] = exec_result.get("success", False) if not claimed: @@ -803,16 +832,18 @@ def _action_run(job: Dict[str, Any], a: Dict[str, Any]) -> str: return _dumps({"success": True, "job": result}) -def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: - job_id = job["id"] - updates: Dict[str, Any] = {} +def _pick(updates: Dict[str, Any], job: Dict[str, Any], key: str) -> Any: + """Effective value of ``key`` after this update: pending update wins over the stored job.""" + return updates[key] if key in updates else job.get(key) + + +def _update_core_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]: + """prompt / name / deliver / skills / model pins; returns an error string or None.""" prompt, deliver, skill, skills = a["prompt"], a["deliver"], a["skill"], a["skills"] - script, monitor_script, monitor_url = a["script"], a["monitor_script"], a["monitor_url"] - context_from, continuity, no_agent = a["context_from"], a["continuity"], a["no_agent"] if prompt is not None: scan_error = _scan_cron_prompt(prompt) if scan_error: - return tool_error(scan_error, success=False) + return scan_error updates["prompt"] = prompt if a["name"] is not None and a["name"].strip(): # Blank name is a no-op, not a clear: a model re-sending the whole @@ -821,7 +852,7 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: if deliver is not None: bot_chat_error = _validate_bot_chat_deliver(_normalize_deliver_param(deliver)) if bot_chat_error: - return tool_error(bot_chat_error, success=False) + return bot_chat_error updates["deliver"] = _resolve_cron_context_deliver(_normalize_deliver_param(deliver)) if a["failure_deliver"] is not None: # '' clears the override (job falls back to deliver on failures); @@ -832,7 +863,7 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: if _norm_fd: bot_chat_error = _validate_bot_chat_deliver(_norm_fd) if bot_chat_error: - return tool_error(bot_chat_error, success=False) + return bot_chat_error _norm_fd = _resolve_cron_context_deliver(_norm_fd) updates["failure_deliver"] = _norm_fd if skills is not None or skill is not None: @@ -850,50 +881,55 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: updates["reasoning_effort"] = a["reasoning_effort"] # Re-validate the EFFECTIVE provider/base_url on EVERY update: a job # persisted before this guard may already hold an unsafe pair, and editing - # an unrelated field must not leave it schedulable. - base_url_error = _validate_cron_base_url( - updates["provider"] if "provider" in updates else job.get("provider"), - updates["base_url"] if "base_url" in updates else job.get("base_url"), - ) - if base_url_error: - return tool_error(base_url_error, success=False) - # Empty string clears script / monitor fields. - for field, value in (("script", script), ("monitor_script", monitor_script)): + # an unrelated field must not leave it schedulable. Merging this update's + # values over the stored job lets an operator remediate in the same call. + return _validate_cron_base_url(_pick(updates, job, "provider"), _pick(updates, job, "base_url")) + + +def _update_script_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]: + """script / monitor_script / monitor_url (empty string clears); returns an error string or None.""" + monitor_script, monitor_url = a["monitor_script"], a["monitor_url"] + for field, value in (("script", a["script"]), ("monitor_script", monitor_script)): if value is not None: if value: path_error = _validate_cron_script_path(value) if path_error: - return tool_error(path_error, success=False) + return path_error updates[field] = _normalize_optional_job_value(value) if value else None if monitor_url is not None: updates["monitor_url"] = _normalize_optional_job_value(monitor_url) if monitor_url else None - if monitor_script is not None or monitor_url is not None: - eff_mon_script = updates["monitor_script"] if "monitor_script" in updates else job.get("monitor_script") - eff_mon_url = updates["monitor_url"] if "monitor_url" in updates else job.get("monitor_url") - if eff_mon_script and eff_mon_url: - return tool_error( - "monitor_script and monitor_url are mutually exclusive — " - "clear one before setting the other.", - success=False, - ) - if context_from is not None or continuity is not None: - # Empty string / list clears; otherwise every ref must exist. Stored - # as a list (or None) to match create_job(). - if context_from is None: - # continuity-only update: start from the job's stored refs. - existing = job.get("context_from") or [] - refs = [str(j).strip() for j in existing if str(j).strip()] - elif isinstance(context_from, str): - refs = [context_from.strip()] if context_from.strip() else [] - else: - refs = [str(j).strip() for j in context_from if str(j).strip()] - if continuity is not None: - refs = _apply_continuity(refs, continuity) or [] - if refs: - ref_error = _validate_context_from_refs(refs) - if ref_error: - return tool_error(ref_error, success=False) - updates["context_from"] = refs or None + if (monitor_script is not None or monitor_url is not None) and ( + _pick(updates, job, "monitor_script") and _pick(updates, job, "monitor_url") + ): + return ( + "monitor_script and monitor_url are mutually exclusive — " + "clear one before setting the other." + ) + return None + + +def _update_context_from(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]: + """context_from / continuity: empty string / list clears; otherwise every + ref must exist. Stored as a list (or None) to match create_job().""" + context_from, continuity = a["context_from"], a["continuity"] + if context_from is None and continuity is None: + return None + if context_from is None: + # continuity-only update: start from the job's stored refs. + context_from = list(job.get("context_from") or []) + refs = _clean_str_list(context_from) + if continuity is not None: + refs = _apply_continuity(refs, continuity) or [] + if refs: + ref_error = _validate_context_from_refs(refs) + if ref_error: + return ref_error + updates["context_from"] = refs or None + return None + + +def _update_run_fields(job: Dict[str, Any], a: Dict[str, Any], updates: Dict[str, Any]) -> Optional[str]: + """enabled_toolsets / attach_to_session / workdir / no_agent / repeat / schedule.""" if a["enabled_toolsets"] is not None: updates["enabled_toolsets"] = a["enabled_toolsets"] or None if a["attach_to_session"] is not None: @@ -901,22 +937,19 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: if a["workdir"] is not None: # Empty string clears; otherwise update_job() validates/normalizes. updates["workdir"] = _normalize_optional_job_value(a["workdir"]) or None - if no_agent is not None: + if a["no_agent"] is not None: # Flipping to True needs a script on the job or in this same update, # otherwise the next tick would error out. - target_no_agent = bool(no_agent) - if target_no_agent: - effective_script = updates.get("script") if "script" in updates else job.get("script") - if not effective_script: - return tool_error( - "Cannot set no_agent=True on a job without a script. " - "Set `script` in the same update, or on the job first.", - success=False, - ) + target_no_agent = bool(a["no_agent"]) + if target_no_agent and not _pick(updates, job, "script"): + return ( + "Cannot set no_agent=True on a job without a script. " + "Set `script` in the same update, or on the job first." + ) updates["no_agent"] = target_no_agent if a["repeat"] is not None: # Shared chokepoint coerces string forms ('forever'/'once'/'3') and - # 0/negative values. + # 0/negative values (a bare `repeat <= 0` raised TypeError on strings). from cron.jobs import normalize_repeat_value repeat_state = dict(job.get("repeat") or {}) repeat_state["times"] = normalize_repeat_value(a["repeat"]) @@ -928,16 +961,27 @@ def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: if job.get("state") != "paused": updates["state"] = "scheduled" updates["enabled"] = True + return None + + +# Validation order is behavior (first failing field wins): keep this sequence. +_UPDATE_STEPS = (_update_core_fields, _update_script_fields, _update_context_from, _update_run_fields) + + +def _action_update(job: Dict[str, Any], a: Dict[str, Any]) -> str: + updates: Dict[str, Any] = {} + for step in _UPDATE_STEPS: + error = step(job, a, updates) + if error: + return tool_error(error, success=False) if not updates: return tool_error("No updates provided.", success=False) - updated = update_job(job_id, updates) + updated = update_job(job["id"], updates) _notify_provider_jobs_changed_safe() - _upd_result: Dict[str, Any] = {"success": True, "job": _format_job(updated)} # An update can switch modes or delivery — echo the same guidance as create. - _upd_notes = _mode_guidance_notes(updated, _normalize_deliver_param(deliver)) - if _upd_notes: - _upd_result["guidance"] = _upd_notes - return _dumps(_upd_result) + return _dumps(_with_guidance( + {"success": True, "job": _format_job(updated)}, updated, _normalize_deliver_param(a["deliver"]) + )) # Actions that need no job_id, and job-bound actions (job resolved first). @@ -953,6 +997,31 @@ _JOB_ACTIONS = { } +def _resolve_job_or_error(job_id: str): + """``(job, None)`` or ``(None, json_error)`` for a job_id/name reference.""" + try: + job = resolve_job_ref(job_id) + except AmbiguousJobReference as exc: + return None, _dumps({ + "success": False, + "error": str(exc), + "matches": [ + { + "id": m["id"], + "name": m.get("name"), + "schedule": m.get("schedule_display"), + "next_run_at": m.get("next_run_at"), + } + for m in exc.matches + ], + }) + if not job: + return None, _dumps( + {"success": False, "error": f"Job with ID or name '{job_id}' not found. Use cronjob(action='list') to inspect jobs."}, + ) + return job, None + + def cronjob( action: str, job_id: Optional[str] = None, @@ -995,27 +1064,11 @@ def cronjob( if not job_id: return tool_error(f"job_id is required for action '{normalized}'", success=False) - - try: - job = resolve_job_ref(job_id) - except AmbiguousJobReference as exc: - return _dumps({ - "success": False, - "error": str(exc), - "matches": [ - { - "id": m["id"], - "name": m.get("name"), - "schedule": m.get("schedule_display"), - "next_run_at": m.get("next_run_at"), - } - for m in exc.matches - ], - }) - if not job: - return _dumps( - {"success": False, "error": f"Job with ID or name '{job_id}' not found. Use cronjob(action='list') to inspect jobs."}, - ) + # Job resolution precedes the action check (an unknown action on a + # missing job reports the missing job) — preserved ordering. + job, error = _resolve_job_or_error(job_id) + if error is not None: + return error handler = _JOB_ACTIONS.get(normalized) if handler is None: @@ -1130,6 +1183,18 @@ def check_cronjob_requirements() -> bool: from tools.registry import registry, tool_error +# Agent-facing arguments forwarded verbatim to cronjob(). model / provider / +# base_url are intentionally NOT in this list: per-job inference pins are +# user-owned (dashboard, `hermes cron create/edit --model`, or hand-edited +# jobs). The agent must not be able to point unattended spend at a different +# model. Programmatic callers of cronjob() itself retain the parameters. +_HANDLER_FORWARDED_ARGS = ( + "job_id", "prompt", "schedule", "name", "repeat", "deliver", "failure_deliver", "skill", "skills", + "reason", "script", "context_from", "continuity", "enabled_toolsets", "workdir", + "no_agent", "attach_to_session", +) + + def _cronjob_handler(args, **kw): """Model-tool dispatch for ``cronjob``: resolves the one model-facing ``monitor`` field into the stored ``monitor_script``/``monitor_url`` pair @@ -1139,33 +1204,12 @@ def _cronjob_handler(args, **kw): ) return cronjob( action=args.get("action", ""), - job_id=args.get("job_id"), - prompt=args.get("prompt"), - schedule=args.get("schedule"), - name=args.get("name"), - repeat=args.get("repeat"), - deliver=args.get("deliver"), - failure_deliver=args.get("failure_deliver"), include_disabled=args.get("include_disabled", True), - skill=args.get("skill"), - skills=args.get("skills"), - # model / provider / base_url are intentionally NOT read from the - # agent's arguments: per-job inference pins are user-owned (dashboard, - # `hermes cron create/edit --model`, or hand-edited jobs). The agent - # must not be able to point unattended spend at a different model. - # Programmatic callers of cronjob() itself retain the parameters. - reason=args.get("reason"), - script=args.get("script"), - context_from=args.get("context_from"), - continuity=args.get("continuity"), - enabled_toolsets=args.get("enabled_toolsets"), - workdir=args.get("workdir"), - no_agent=args.get("no_agent"), - attach_to_session=args.get("attach_to_session"), monitor_script=_mon_script, monitor_url=_mon_url, task_id=kw.get("task_id"), session_id=kw.get("session_id"), + **{key: args.get(key) for key in _HANDLER_FORWARDED_ARGS}, ) diff --git a/tools/desktop_ui.py b/tools/desktop_ui.py index 4ec636aa7d..46f5b2db71 100644 --- a/tools/desktop_ui.py +++ b/tools/desktop_ui.py @@ -33,7 +33,9 @@ def user_enabled(setting: str, default: bool) -> bool: The renderer mirrors these toggles onto the CONNECTED gateway's config, so this reads the user's real answer for local/SSH/URL/cloud gateways alike (an env var would only describe the process). ``check_fn``s use it to withdraw a tool from the - schema when the user switched the feature off. Unreadable config -> ``default``. + schema when the user switched the feature off — Hermes should not be told about + a surface it may not use. Unreadable config -> ``default`` so a shipped-on + feature does not vanish on a transient read error. """ try: from hermes_cli.config import load_config_readonly diff --git a/tools/lazy_deps.py b/tools/lazy_deps.py index 47fc4ead31..86b308d280 100644 --- a/tools/lazy_deps.py +++ b/tools/lazy_deps.py @@ -198,7 +198,7 @@ LAZY_DEPS: dict[str, tuple[str, ...]] = { # path for lean/partial installs. Call sites use prompt=False so read_file / # vision can never block on an input() prompt mid-session. "tool.vision": ("Pillow==12.3.0",), - "tool.doc_extract": ("firecrawl-anydoc==0.2.4",), # lockstep with pyproject + "tool.doc_extract": ("firecrawl-anydoc==0.2.4",), # imports as `anydoc`; lockstep with pyproject # MCP client SDK for the cua-driver; covers lean/broken-extra installs so # computer_use never dead-ends on `No module named 'mcp'`. "tool.computer_use": ( @@ -512,7 +512,10 @@ def _warm_installed_bytecode(specs: tuple[str, ...], target: Optional[Path]) -> on an installer. Best-effort; never invalidates a successful install.""" if sys.dont_write_bytecode: return - import compileall + try: + import compileall + except Exception: # pragma: no cover — stdlib, but never break an install + return for spec in specs: try: @@ -579,6 +582,8 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install try: # --compile-bytecode: uv writes no __pycache__ by default, so the # first import would recompile the backend AND its transitives. + # Covers the whole install; _warm_installed_bytecode is the + # belt-and-braces pass for the spec's own roots on any tier. r = _run([uv_bin, "pip", "install", "--compile-bytecode", *extra_args, *specs], timeout=timeout, env=uv_env) if r.returncode != 0: diff --git a/tools/registry.py b/tools/registry.py index 7c20959fb8..ae829360bc 100644 --- a/tools/registry.py +++ b/tools/registry.py @@ -22,6 +22,7 @@ import logging import sys import threading import time +from dataclasses import dataclass from pathlib import Path from typing import Callable, Dict, List, Optional, Set @@ -208,36 +209,26 @@ def _save_discovery_cache(cache: Dict[str, list]) -> None: logger.debug("Could not write tool discovery cache %s: %s", path, e) +@dataclass(eq=False, slots=True) class ToolEntry: - """Metadata for a single registered tool.""" + """Metadata for a single registered tool (identity semantics: registry + restore/CAS paths compare entries with ``is``).""" - __slots__ = ( - "name", "toolset", "schema", "handler", "check_fn", - "requires_env", "is_async", "description", "emoji", - "max_result_size_chars", "dynamic_schema_overrides", - ) - - def __init__(self, name, toolset, schema, handler, check_fn, - requires_env, is_async, description, emoji, - max_result_size_chars=None, dynamic_schema_overrides=None): - self.name = name - self.toolset = toolset - self.schema = schema - self.handler = handler - self.check_fn = check_fn - self.requires_env = requires_env - self.is_async = is_async - self.description = description - self.emoji = emoji - self.max_result_size_chars = max_result_size_chars - # Optional zero-arg callable returning a dict of schema overrides - # applied at get_definitions() time. Use for fields that depend on - # runtime config (e.g. delegate_task's description must reflect the - # user's current delegation.max_concurrent_children / max_spawn_depth - # so the model isn't told the wrong limits). The callable is invoked - # on every get_definitions() call; results are merged shallow on top - # of the base schema before the {"type": "function", ...} wrap. - self.dynamic_schema_overrides = dynamic_schema_overrides + name: str + toolset: str + schema: dict + handler: Callable + check_fn: Optional[Callable] + requires_env: list + is_async: bool + description: str + emoji: str + max_result_size_chars: int | float | None = None + # Zero-arg callable returning schema overrides merged (shallow) on top of + # the base schema at every get_definitions() call — for fields that track + # runtime config (e.g. delegate_task's description must reflect the current + # delegation.max_concurrent_children / max_spawn_depth). + dynamic_schema_overrides: Optional[Callable] = None class _PluginOverridePolicy: @@ -249,33 +240,32 @@ class _PluginOverridePolicy: self.allowed = bool(allowed) +_OVERRIDE_DENIED_MSG = ( + "Plugin module {owner!r} cannot override built-in tool {name!r} " + "without operator opt-in (allow_tool_override)." +) + + # --------------------------------------------------------------------------- # check_fn TTL cache # -# external state (Docker daemon, Modal SDK install, playwright binary -# availability). For a long-lived CLI or gateway process, calling them on -# every get_definitions() is pure waste — external state changes on human -# timescales. Cache results for ~30 s so env-var flips via ``hermes tools`` -# or live credential file changes propagate within a turn or two without -# requiring any explicit invalidation. +# check_fns probe external state (Docker daemon, Modal SDK, playwright binary) +# that changes on human timescales, so results are cached ~30 s: env-var flips +# via ``hermes tools`` still propagate within a turn or two with no explicit +# invalidation. # -# -# Transient-failure suppression (issue #21658 / #5304): these probes can flap. -# A single ``subprocess.run([docker, "version"], timeout=5)`` that times out -# under load returns False for one call, which would silently strip the entire -# terminal+file toolset from whatever agent is being built at that instant — -# most visibly a delegate_task subagent, which then reports "Tool read_file -# does not exist". To absorb such flakes WITHOUT pinning a permanently-stale -# "available" verdict, we remember the last time each check returned True and, -# when a fresh probe fails within a short grace window of that last success, -# we serve the last-good True instead of caching the failure. A failure that -# persists past the grace window is honored normally, so a backend that really -# went down stops advertising its tools. +# Transient-failure suppression: probes can flap (a ``docker version`` that +# times out under load), which would silently strip a whole toolset from the +# agent being built at that instant — most visibly a delegate_task subagent +# reporting "Tool read_file does not exist". So we remember each check's last +# success and, when a fresh probe fails within a short grace window of it, +# serve the last-good True WITHOUT caching the failure. A failure persisting +# past the window is honored, so a backend that really went down stops +# advertising its tools. # --------------------------------------------------------------------------- _CHECK_FN_TTL_SECONDS = 30.0 -# How long after a successful check a subsequent transient failure is treated -# as a flake (last-good True is served) rather than a real outage. Kept short +# Grace window after a success in which a failure counts as a flake; kept short # so a genuinely-down backend is reflected within a couple of turns. _CHECK_FN_FAILURE_GRACE_SECONDS = 60.0 _CHECK_FN_CACHE_MAX = 512 @@ -292,6 +282,10 @@ def no_cache_check_fn(fn: Callable) -> Callable: return fn +def _fn_label(fn: Callable) -> object: + return getattr(fn, "__qualname__", fn) + + def _prune_check_fn_caches(now: float) -> None: """Expire stale entries and cap profile-dimensional cache growth. @@ -313,15 +307,13 @@ def check_fn_cache_scope() -> Optional[str]: """Return the active profile key when availability is profile-scoped. Browser-controller availability is request-bound and can change on every - attach/detach. A fully bound browser-control request therefore bypasses both - this check cache and model_tools' outer definition cache; the same sentinel - is consumed by both layers. This prevents one Browser session's live tools - from leaking into any unrelated session. + attach/detach, so a fully bound browser-control request bypasses both this + cache and model_tools' outer definition cache (same sentinel for both + layers) — one Browser session's live tools must not leak into another. - Single-profile processes intentionally keep the historical process-wide - cache. A multiplex gateway installs a Hermes-home override for every - profile turn, so the canonical profile key is the stable isolation - boundary across repeated turns for that profile. + Single-profile processes keep the historical process-wide cache. A + multiplex gateway installs a Hermes-home override per profile turn, so the + canonical profile key is the stable isolation boundary. """ try: from gateway.session_context import get_session_env @@ -365,12 +357,12 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b # check_fns run before any profile secret scope exists, so # get_secret raises by design. The tool re-probes on the first # scoped turn — log without a traceback so this cannot be - # mistaken for a crashed check_fn (#100697). + # mistaken for a crashed check_fn. logger.debug( "check_fn %s hit the multiplex fail-closed path with no " "profile secret scope active; dependent tools re-probe on " "the first scoped turn", - getattr(fn, "__qualname__", fn), + _fn_label(fn), ) return False # The scope resolved but the read still failed closed: a genuinely @@ -378,7 +370,7 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b logger.warning( "check_fn %s raised UnscopedSecretError while the profile cache " "scope was resolved; dependent tools will be unavailable this turn", - getattr(fn, "__qualname__", fn), + _fn_label(fn), exc_info=True, ) return False @@ -386,7 +378,7 @@ def _run_check_fn_uncached(fn: Callable, *, unresolved_scope: bool = False) -> b detail = " while profile cache scope was unresolved" if unresolved_scope else "" logger.warning( "check_fn %s raised%s; dependent tools will be unavailable this turn", - getattr(fn, "__qualname__", fn), + _fn_label(fn), detail, exc_info=True, ) @@ -403,19 +395,17 @@ def _check_fn_cached(fn: Callable) -> bool: return _run_check_fn_uncached(fn, unresolved_scope=True) cache_key = (fn, scope) with _check_fn_cache_lock: - _prune_check_fn_caches(now) + _prune_check_fn_caches(now) # leaves only entries within TTL cached = _check_fn_cache.get(cache_key) if cached is not None: - ts, value = cached - if now - ts < _CHECK_FN_TTL_SECONDS: - return value + return cached[1] - raised = False try: value = bool(fn()) + outcome = "returned False" except Exception: value = False - raised = True + outcome = "raised" with _check_fn_cache_lock: _prune_check_fn_caches(now) @@ -432,9 +422,7 @@ def _check_fn_cached(fn: Callable) -> bool: logger.warning( "check_fn %s failed (%s) within %.0fs of last success; " "treating as transient and keeping tool(s) available", - getattr(fn, "__qualname__", fn), - "raised" if raised else "returned False", - _CHECK_FN_FAILURE_GRACE_SECONDS, + _fn_label(fn), outcome, _CHECK_FN_FAILURE_GRACE_SECONDS, ) return True @@ -442,8 +430,7 @@ def _check_fn_cached(fn: Callable) -> bool: # silent tool loss in quiet mode (subagents) is diagnosable. logger.warning( "check_fn %s %s; dependent tools will be unavailable this turn", - getattr(fn, "__qualname__", fn), - "raised" if raised else "returned False", + _fn_label(fn), outcome, ) _check_fn_cache[cache_key] = (now, False) return False @@ -458,27 +445,21 @@ def invalidate_check_fn_cache() -> None: def get_cached_check_fn_result(fn: Callable) -> Optional[bool]: - """Return the current cached verdict for *fn* if its TTL is still valid. + """Return the cached verdict for *fn* if its TTL is still valid, else None. - Unlike :func:`_check_fn_cached`, this NEVER executes the probe. It is for - read-only surfaces (e.g. dashboard status panels) that need the last-known - availability without triggering network / auth / SDK work inside a request - path. Returns ``None`` when there is no fresh cached verdict. + NEVER executes the probe: for read-only surfaces (dashboard status panels) + that must not trigger network / auth / SDK work inside a request path. """ now = time.monotonic() scope = check_fn_cache_scope() if scope == CHECK_FN_CACHE_BYPASS: - # Unresolved profile identity bypasses the cache entirely; there is no - # trustworthy cached verdict to report. + # Unresolved profile identity bypasses the cache; nothing trustworthy to report. return None with _check_fn_cache_lock: cached = _check_fn_cache.get((fn, scope)) - if cached is None: - return None - ts, value = cached - if now - ts < _CHECK_FN_TTL_SECONDS: - return value - return None + if cached is not None and now - cached[0] < _CHECK_FN_TTL_SECONDS: + return cached[1] + return None class ToolRegistry: @@ -517,6 +498,21 @@ class ToolRegistry: """Return the active profile's canonical registry scope.""" return hermes_home_key() + def _slot(self, scope: Optional[str], *, create: bool = False) -> Dict[str, ToolEntry]: + """The registration map for *scope*: global when None, else that profile's overlay.""" + if scope is None: + return self._tools + if create: + return self._scoped_tools.setdefault(scope, {}) + return self._scoped_tools.get(scope, {}) + + def _drop_toolset_aliases(self, toolset: str) -> None: + self._toolset_aliases = { + alias: target + for alias, target in self._toolset_aliases.items() + if target != toolset + } + def _merged_tools(self, scope: Optional[str] = None) -> Dict[str, ToolEntry]: """Return global tools overlaid with one profile's plugin tools.""" active_scope = scope or self.current_scope_key() @@ -583,8 +579,7 @@ class ToolRegistry: ) -> Optional[ToolEntry]: """Return the local slot state without following global fallback.""" with self._lock: - target = self._tools if scope is None else self._scoped_tools.get(scope, {}) - return target.get(name) + return self._slot(scope).get(name) def get_registered_toolset_names(self) -> List[str]: """Return sorted unique toolset names present in the registry.""" @@ -685,21 +680,14 @@ class ToolRegistry: return bool(policy and policy.allowed) def _plugin_owner_of(self, handler: Callable) -> Optional[str]: - """Return the plugin module namespace that defined *handler*, or None - if it was not defined in a loaded plugin module. + """Plugin namespace that DEFINED *handler* (None for built-in/MCP handlers). - Authorization is bound to where the handler was DEFINED - (``handler.__globals__["__name__"]``), which is fixed at definition - time and cannot drift with the call site, thread, or timing. Lambdas - and nested functions inherit the defining module's globals, so a - plugin cannot launder an override through a callback. Built-in/MCP - handlers live outside the plugin namespace and return None (unchanged - behavior). + Bound to ``handler.__globals__["__name__"]``, fixed at definition time so + it cannot drift with call site, thread, or timing; lambdas and nested + functions inherit it, so a plugin cannot launder an override via a callback. """ mod = self._callable_module(handler) - if not mod: - return None - return self._plugin_namespace_of_module(mod) + return self._plugin_namespace_of_module(mod) if mod else None @staticmethod def _callable_module(handler: Callable) -> str: @@ -778,13 +766,11 @@ class ToolRegistry: @staticmethod def _caller_module() -> str: - """Best-effort module name of whoever called the registry method that - invoked this helper (two frames up: this helper, then the registry - method itself, then the actual caller). + """Best-effort module name of the registry method's caller (two frames up). - ``deregister()`` takes only a tool name — unlike ``register()`` it has - no handler argument to bind authorization to via ``_plugin_owner_of``. - Frame inspection is the only way to know who is asking. + ``deregister()`` takes only a tool name — no handler to bind authorization + to via ``_plugin_owner_of`` — so frame inspection is the only way to know + who is asking. """ try: frame = sys._getframe(2) @@ -822,16 +808,15 @@ class ToolRegistry: if scope is None and owner is not None: scope = self._plugin_scope_of(owner) with self._lock: - target = ( - self._tools - if scope is None - else self._scoped_tools.setdefault(scope, {}) - ) + target = self._slot(scope, create=True) existing = ( self._tools.get(name) if scope is None else self._merged_tools(scope).get(name) ) + plugin_override_denied = ( + owner is not None and not self._plugin_override_allowed(scope, owner) + ) shadows_global = ( owner is not None and scope is not None @@ -847,17 +832,11 @@ class ToolRegistry: name, ) return - if not self._plugin_override_allowed(scope, owner): - raise PermissionError( - f"Plugin module {owner!r} cannot override built-in " - f"tool {name!r} without operator opt-in " - f"(allow_tool_override)." - ) + if plugin_override_denied: + raise PermissionError(_OVERRIDE_DENIED_MSG.format(owner=owner, name=name)) if existing and existing.toolset != toolset: if override: - if owner is not None and not self._plugin_override_allowed( - scope, owner - ): + if plugin_override_denied: logger.error( "Tool registration REJECTED: plugin %r attempted to " "override built-in tool %r (existing toolset %r) without " @@ -866,11 +845,7 @@ class ToolRegistry: "in config.yaml to allow it.", owner, name, existing.toolset, ) - raise PermissionError( - f"Plugin module {owner!r} cannot override built-in " - f"tool {name!r} without operator opt-in " - f"(allow_tool_override)." - ) + raise PermissionError(_OVERRIDE_DENIED_MSG.format(owner=owner, name=name)) # Explicit opt-in (or non-plugin caller): replace the tool. # Logged at INFO so the override is auditable in agent.log. logger.info( @@ -903,36 +878,27 @@ class ToolRegistry: max_result_size_chars=max_result_size_chars, dynamic_schema_overrides=dynamic_schema_overrides, ) - # Availability is now derived per-tool (_toolset_has_exposable_tools), - # so this map no longer gates a toolset. It is still consumed by + # Availability is derived per-tool (_toolset_has_exposable_tools), so + # this map no longer gates a toolset. It still feeds # get_toolset_requirements -> TOOLSET_REQUIREMENTS["check_fn"], which # banner.py reads (presence only, never called) to classify an - # already-unavailable toolset as lazy-init vs disabled. Keep the - # write path for that classification. + # already-unavailable toolset as lazy-init vs disabled. if scope is None and check_fn and toolset not in self._toolset_checks: self._toolset_checks[toolset] = check_fn self._generation += 1 def deregister(self, name: str, *, scope: Optional[str] = None) -> None: - """Remove a tool from the registry. + """Remove a tool; also drops the toolset check/aliases if it was the last + tool in its toolset (MCP nuke-and-repave on ``tools/list_changed``). - Also cleans up the toolset check if no other tools remain in the - same toolset. Used by MCP dynamic tool discovery to nuke-and-repave - when a server sends ``notifications/tools/list_changed``. + ``scope`` selects a profile overlay explicitly (multiplexed MCP tools live + in the owning profile's overlay). Plugin callers may not name another + scope; non-plugin callers without ``scope`` target the process-global map. - ``scope`` selects a profile overlay explicitly (multiplexed MCP tools - live in the owning profile's overlay). Plugin callers keep their own - scope and may not name another one; non-plugin callers without - ``scope`` keep the historical process-global target. - - Gated by the same operator opt-in policy ``register(override=True)`` - enforces. Without this, a plugin could bypass that gate entirely by - deregistering a tool it doesn't own and then calling plain - ``register()`` over the now-empty slot — ``register()`` only runs its - override check when an ``existing`` entry is present, so removing it - first skips the check altogether. MCP toolsets (``mcp-*``) are exempt: - dynamic tool discovery legitimately nukes-and-repaves its own tools on - every refresh and has no plugin-override concept. + Gated by the same opt-in as ``register(override=True)``: otherwise a plugin + could deregister a tool it doesn't own and re-register over the empty slot, + skipping the override check (which only runs when an entry exists). + ``mcp-*`` toolsets are exempt — discovery repaves its own tools per refresh. """ with self._lock: caller_mod = self._caller_module() @@ -949,31 +915,21 @@ class ToolRegistry: ) if scope is None: scope = caller_scope - target = ( - self._scoped_tools.get(scope, {}) - if scope is not None - else self._tools - ) + target = self._slot(scope) entry = target.get(name) - if entry is None and scope is not None: - if caller_owner is not None and name in self._tools: + if entry is None: + if scope is not None and caller_owner is not None and name in self._tools: raise PermissionError( f"Scoped plugin module {caller_mod!r} cannot deregister " f"process-global tool {name!r}; register a scoped " "override instead." ) return - if entry is None: - return if not entry.toolset.startswith("mcp-"): owner = self._plugin_owner_of(entry.handler) - # Ownership check: bind to the plugin package root - # (``hermes_plugins.{name}``), not the exact module string. - # A handler defined in ``hermes_plugins.pkg.handlers`` is - # still owned by the ``hermes_plugins.pkg`` package — exact - # string equality would wrongly block root-module cleanup code - # from removing tools registered by a submodule of the same - # plugin (egilewski review on #55840). + # Ownership binds to the plugin package root (``hermes_plugins.{name}``), + # not the exact module string: a handler defined in a submodule is + # still owned by the package, so root-module cleanup may remove it. same_plugin = bool(owner and caller_owner == owner) if ( caller_owner is not None @@ -1006,11 +962,7 @@ class ToolRegistry: ) if not toolset_still_exists: self._toolset_checks.pop(entry.toolset, None) - self._toolset_aliases = { - alias: target - for alias, target in self._toolset_aliases.items() - if target != entry.toolset - } + self._drop_toolset_aliases(entry.toolset) self._generation += 1 logger.debug("Deregistered tool: %s", name) @@ -1031,11 +983,7 @@ class ToolRegistry: must leave the newer entry untouched. """ with self._lock: - target = ( - self._tools - if scope is None - else self._scoped_tools.setdefault(scope, {}) - ) + target = self._slot(scope, create=True) if target.get(name) is not current: return False @@ -1072,11 +1020,7 @@ class ToolRegistry: for entries in self._scoped_tools.values() for entry in entries.values() ): - self._toolset_aliases = { - alias: target - for alias, target in self._toolset_aliases.items() - if target != toolset - } + self._drop_toolset_aliases(toolset) self._generation += 1 logger.debug("Restored tool registration: %s", name) return True @@ -1086,20 +1030,12 @@ class ToolRegistry: # ------------------------------------------------------------------ def get_definitions(self, tool_names: Set[str], quiet: bool = False) -> List[dict]: - """Return OpenAI-format tool schemas for the requested tool names. - - Only tools whose ``check_fn()`` returns True (or have no check_fn) - are included. ``check_fn()`` results are cached for ~30 s via - :func:`_check_fn_cached` to amortize repeat probes (check_terminal_ - requirements probes modal/docker, browser checks probe playwright, - etc.); TTL chosen so env-var changes (``hermes tools enable foo``) - still take effect in near-real-time without forcing a full cache - flush on every call. + """Return OpenAI-format schemas for the requested tools whose ``check_fn`` + passes (or is absent). Probes go through the ~30 s TTL cache + (:func:`_check_fn_cached`) so ``hermes tools enable`` still lands quickly. """ result = [] - # Per-call cache on top of the 30 s TTL — handles repeat probes of the - # same check_fn within one definitions pass without re-reading the - # TTL clock. + # Per-call memo on top of the TTL: one probe per distinct check_fn per pass. check_results: Dict[Callable, bool] = {} entries_by_name = {entry.name: entry for entry in self._snapshot_entries()} for name in sorted(tool_names): @@ -1113,13 +1049,10 @@ class ToolRegistry: if not quiet: logger.debug("Tool %s unavailable (check failed)", name) continue - # Ensure schema always has a "name" field — use entry.name as fallback schema_with_name = {**entry.schema, "name": entry.name} - # Apply runtime-dynamic overrides (e.g. delegate_task description - # depends on current delegation.max_concurrent_children / - # max_spawn_depth). Caller side (model_tools.get_tool_definitions) - # already keys its memo on config.yaml mtime + size, so changes - # to delegation.* in config invalidate the cache automatically. + # Runtime-dynamic overrides (e.g. delegate_task limits). The caller's + # memo (model_tools.get_tool_definitions) is keyed on config.yaml + # mtime+size, so config changes invalidate it automatically. if entry.dynamic_schema_overrides is not None: try: overrides = entry.dynamic_schema_overrides() @@ -1140,13 +1073,9 @@ class ToolRegistry: @staticmethod def _normalize_handler_result(name: str, result): - """Enforce the result shapes supported by the agent tool pipeline. - - Normal tool results are strings. The sole structured exception is the - multimodal envelope consumed by the agent executor. Returning every - other value as a string error keeps logging, hooks, budgeting, and - persistence from receiving values they cannot safely slice or size. - """ + """Results must be a string or the multimodal envelope; anything else + becomes a string error so logging/hooks/budgeting/persistence never + receive values they cannot slice or size.""" if isinstance(result, str): return _bound_json_error_result(result) if ( @@ -1177,14 +1106,9 @@ class ToolRegistry: scope: Optional[str] = None, **kwargs, ) -> str | dict: - """Execute a tool handler by name. - - * Async handlers are bridged automatically via ``_run_async()``. - * Handler results are normalized to a string or supported multimodal - envelope before leaving the registry. - * All exceptions are caught and returned as ``{"error": "..."}`` - for consistent error format. - """ + """Execute a tool handler by name: async handlers bridged via + ``_run_async()``, results normalized, every exception returned as + ``{"error": ...}``.""" entry = self.get_entry(name, scope=scope) if not entry: return tool_error(f"Unknown tool: {name}") @@ -1200,9 +1124,8 @@ class ToolRegistry: logger.exception( "Tool %s dispatch error: %s", name, _bound_error_text(str(e)) ) - # Route through the sanitizer so framing tokens / CDATA / fences - # in exception strings don't reach the model as structural noise. - # See model_tools._sanitize_tool_error for rationale. + # Sanitize so framing tokens / CDATA / fences in exception strings + # don't reach the model as structural noise. raw = f"Tool execution failed: {type(e).__name__}: {e}" try: from model_tools import _sanitize_tool_error @@ -1253,41 +1176,32 @@ class ToolRegistry: return {entry.name: entry.toolset for entry in self._snapshot_entries()} def is_toolset_available(self, toolset: str) -> bool: - """Check if a toolset has at least one exposable tool. - - Returns False (rather than crashing) when a per-tool check raises - an unexpected exception (e.g. network error, missing import, bad config). - """ - entries, _ = self._snapshot_state() - return self._toolset_has_exposable_tools(toolset, entries) + """True when a toolset has at least one exposable tool (never raises).""" + return self._toolset_has_exposable_tools(toolset, self._snapshot_entries()) def check_toolset_requirements(self) -> Dict[str, bool]: """Return ``{toolset: available_bool}`` for every toolset.""" - entries, _ = self._snapshot_state() - toolsets = sorted({entry.toolset for entry in entries}) + entries = self._snapshot_entries() return { toolset: self._toolset_has_exposable_tools(toolset, entries) - for toolset in toolsets + for toolset in sorted({entry.toolset for entry in entries}) } def get_available_toolsets(self) -> Dict[str, dict]: """Return toolset metadata for UI display.""" toolsets: Dict[str, dict] = {} - entries, _ = self._snapshot_state() + entries = self._snapshot_entries() for entry in entries: - ts = entry.toolset - if ts not in toolsets: - toolsets[ts] = { - "available": self._toolset_has_exposable_tools(ts, entries), + info = toolsets.get(entry.toolset) + if info is None: + info = toolsets[entry.toolset] = { + "available": self._toolset_has_exposable_tools(entry.toolset, entries), "tools": [], "description": "", "requirements": [], } - toolsets[ts]["tools"].append(entry.name) - if entry.requires_env: - for env in entry.requires_env: - if env not in toolsets[ts]["requirements"]: - toolsets[ts]["requirements"].append(env) + info["tools"].append(entry.name) + _extend_unique(info["requirements"], entry.requires_env or []) return toolsets def get_toolset_requirements(self) -> Dict[str, dict]: @@ -1295,27 +1209,22 @@ class ToolRegistry: result: Dict[str, dict] = {} entries, toolset_checks = self._snapshot_state() for entry in entries: - ts = entry.toolset - if ts not in result: - result[ts] = { - "name": ts, - "env_vars": [], - "check_fn": toolset_checks.get(ts), - "setup_url": None, - "tools": [], - } - if entry.name not in result[ts]["tools"]: - result[ts]["tools"].append(entry.name) - for env in entry.requires_env: - if env not in result[ts]["env_vars"]: - result[ts]["env_vars"].append(env) + info = result.setdefault(entry.toolset, { + "name": entry.toolset, + "env_vars": [], + "check_fn": toolset_checks.get(entry.toolset), + "setup_url": None, + "tools": [], + }) + _extend_unique(info["tools"], [entry.name]) + _extend_unique(info["env_vars"], entry.requires_env) return result def check_tool_availability(self, quiet: bool = False): """Return (available_toolsets, unavailable_info) like the old function.""" available = [] unavailable = [] - entries, _ = self._snapshot_state() + entries = self._snapshot_entries() for ts in sorted({entry.toolset for entry in entries}): ts_entries = [entry for entry in entries if entry.toolset == ts] if self._toolset_has_exposable_tools(ts, entries): @@ -1329,51 +1238,26 @@ class ToolRegistry: return available, unavailable +def _extend_unique(target: list, items) -> None: + for item in items: + if item not in target: + target.append(item) + + # Module-level singleton registry = ToolRegistry() -# --------------------------------------------------------------------------- -# Helpers for tool response serialization -# --------------------------------------------------------------------------- -# Every tool handler must return a JSON string. These helpers eliminate the -# boilerplate ``json.dumps({"error": msg}, ensure_ascii=False)`` that appears -# hundreds of times across tool files. -# -# Usage: -# from tools.registry import registry, tool_error, tool_result -# -# return tool_error("something went wrong") -# return tool_error("not found", code=404) -# return tool_result(success=True, data=payload) -# return tool_result(items) # pass a dict directly +# Tool handlers must return JSON strings; these replace the ubiquitous +# ``json.dumps({"error": msg}, ensure_ascii=False)`` boilerplate. def tool_error(message, **extra) -> str: - """Return a JSON error string for tool handlers. - - >>> tool_error("file not found") - '{"error": "file not found"}' - >>> tool_error("bad input", success=False) - '{"error": "bad input", "success": false}' - """ - # Bound the context-bound copy so a raw exception can't bloat history across retries. - result = {"error": _bound_error_text(str(message))} - if extra: - result.update(extra) - return json.dumps(result, ensure_ascii=False) + """``'{"error": "", **extra}'`` — the error body is bounded so a raw + exception can't bloat history across retries.""" + return json.dumps({"error": _bound_error_text(str(message)), **extra}, ensure_ascii=False) def tool_result(data=None, **kwargs) -> str: - """Return a JSON result string for tool handlers. - - Accepts a dict positional arg *or* keyword arguments (not both): - - >>> tool_result(success=True, count=42) - '{"success": true, "count": 42}' - >>> tool_result({"key": "value"}) - '{"key": "value"}' - """ - if data is not None: - return json.dumps(data, ensure_ascii=False) - return json.dumps(kwargs, ensure_ascii=False) + """JSON-encode a dict positional arg *or* keyword arguments (not both).""" + return json.dumps(data if data is not None else kwargs, ensure_ascii=False) diff --git a/tools/tool_backend_helpers.py b/tools/tool_backend_helpers.py index cbb1f67bd4..603a052a81 100644 --- a/tools/tool_backend_helpers.py +++ b/tools/tool_backend_helpers.py @@ -18,7 +18,8 @@ _VALID_MODAL_MODES = {"auto", "direct", "managed"} def managed_nous_tools_enabled(*, force_fresh: bool = False) -> bool: - """True when the user is entitled to the Nous Tool Gateway (coarse gate). + """True when the user is entitled to the Nous Tool Gateway (coarse gate: + paid Portal service access OR a live free tool pool — "is ANY managed tool usable"). Fails closed on unknown/error entitlement — never blocks startup. Per-category coverage is narrowed by callers via ``tool_gateway_entitled_for``. @@ -125,7 +126,9 @@ def resolve_modal_backend_state( def _scoped_credential(name: str) -> str: - """Read a credential env var under the active profile secret scope.""" + """Read a credential env var under the active profile secret scope. Raw env + fallback only if ``agent.secret_scope`` cannot import — a packaging edge must + never leave the caller without a key.""" try: from agent.secret_scope import get_secret @@ -280,6 +283,8 @@ def read_selection(section: str) -> str | None: text = str(value).strip().lower() if text: return text + # use_gateway: false with no name key is not a usable selection shape; + # per-capability web keys still count as configured via selection_exists(). return None @@ -295,8 +300,10 @@ def selection_exists(section: str) -> bool: return any(str(raw.get(key) or "").strip() for key in extra) -# Backends that once shipped in-tree but were removed; consulted by the startup -# config check and selection_error() so a stale selection gets a real message. +# Backends that once shipped in-tree but were removed. A config still pointing at +# one would otherwise fail silently at the FIRST tool call with a generic "no +# registered provider has that name" — no migration, no startup notice. Consulted +# by the startup config check and selection_error() so the user gets a real message. # Add removals here, never as one-off string checks at call sites, e.g. # "web": {"": "the backend was removed in vX.Y.Z (...)"}, REMOVED_BACKENDS: Dict[str, Dict[str, str]] = {} @@ -324,7 +331,8 @@ def selection_error(section: str, selection_name: str, failure: str) -> str: def fal_key_is_configured() -> bool: """True when FAL_KEY is set (scope/env, else ``.env`` for CLI paths that - run before dotenv loads) to a non-whitespace value.""" + run before dotenv loads) to a non-whitespace value — so tool-side and CLI + setup-time checks agree; whitespace-only counts as unset everywhere.""" value = _scoped_credential("FAL_KEY") or None if value is None: try: diff --git a/tools/tool_search.py b/tools/tool_search.py index 35f16f5735..64d1efaba4 100644 --- a/tools/tool_search.py +++ b/tools/tool_search.py @@ -38,11 +38,14 @@ from tools.tool_search_catalog import ( # noqa: F401 — re-exported public/tes _classify_source, _corpus_stats, _entry_search_text, + _fn, _listing_group_label, + _registry_entry, _short_desc, _stem, _tokenize, build_catalog, + build_catalog_listing, build_catalog_listing_with_form, search_catalog, ) @@ -77,7 +80,8 @@ class ToolSearchConfig: search_default_limit: int max_search_limit: int # Embedded name + short-description manifest so deferred tools stay - # discoverable. "auto"/"on" = include when it fits; "off" = bare bridge. + # DISCOVERABLE (like the skills listing) while full schemas stay deferred. + # "auto"/"on" = include when it fits (names-only, then bare); "off" = bare bridge. listing: str = "auto" # "auto" | "on" | "off" # Effective budget = min(listing_max_tokens, threshold_pct% of context). listing_max_tokens: int = 4000 @@ -98,32 +102,21 @@ class ToolSearchConfig: return cls(enabled="off" if raw is False else "auto", threshold_pct=5.0, search_default_limit=5, max_search_limit=25) - enabled = _tri_state(raw.get("enabled", "auto")) - threshold_pct = max(0.0, min(100.0, _safe_float(raw.get("threshold_pct"), 5.0))) - - max_search_limit = max(1, min(50, _safe_int(raw.get("max_search_limit"), 25))) - search_default_limit = max(1, min(max_search_limit, - _safe_int(raw.get("search_default_limit"), 5))) - - listing = _tri_state(raw.get("listing", "auto")) - listing_max_tokens = max(200, min(60000, _safe_int(raw.get("listing_max_tokens"), 4000))) - + max_search_limit = _clamped_int(raw.get("max_search_limit"), 25, 1, 50) defer_raw = raw.get("defer") - if isinstance(defer_raw, (list, tuple, set)): - defer_tools = frozenset( - str(n).strip() for n in defer_raw if str(n).strip() - ) - else: - defer_tools = None # curated default - return cls( - enabled=enabled, - threshold_pct=threshold_pct, - search_default_limit=search_default_limit, + enabled=_tri_state(raw.get("enabled", "auto")), + threshold_pct=max(0.0, min(100.0, _safe_float(raw.get("threshold_pct"), 5.0))), + search_default_limit=_clamped_int( + raw.get("search_default_limit"), 5, 1, max_search_limit), max_search_limit=max_search_limit, - listing=listing, - listing_max_tokens=listing_max_tokens, - defer_tools=defer_tools, + listing=_tri_state(raw.get("listing", "auto")), + listing_max_tokens=_clamped_int(raw.get("listing_max_tokens"), 4000, 200, 60000), + # A list replaces the curated default wholesale; anything else = curated. + defer_tools=( + frozenset(str(n).strip() for n in defer_raw if str(n).strip()) + if isinstance(defer_raw, (list, tuple, set)) else None + ), ) @@ -144,6 +137,11 @@ def _safe_int(value: Any, fallback: int) -> int: return fallback +def _clamped_int(value: Any, fallback: int, lo: int, hi: int) -> int: + """``_safe_int`` clamped to ``[lo, hi]`` (the fallback is clamped too).""" + return max(lo, min(hi, _safe_int(value, fallback))) + + def _safe_float(value: Any, fallback: float) -> float: try: return float(value) @@ -218,13 +216,12 @@ def is_deferrable_tool_name(name: str, defer_tools: Optional[frozenset] = None) return True if name in _core_tool_names(): return False + entry = _registry_entry(name) + if entry is None: + return False try: - from tools.registry import registry - entry = registry.get_entry(name) - if entry is None: - return False return entry.toolset.startswith("mcp-") or entry.toolset not in _DIRECT_SURFACE_TOOLSETS - except Exception: + except Exception: # malformed entry (no str toolset) is never deferrable return False @@ -233,11 +230,7 @@ def _describe_classification( defer_tools: Optional[frozenset] = None, ) -> Literal["available", "not_found", "not_deferrable"]: """Classify a describe name without treating unknown names as errors.""" - try: - from tools.registry import registry - entry = registry.get_entry(name) - except Exception: - return "not_found" + entry = _registry_entry(name) if entry is None: return "not_found" if defer_tools is not None and name in defer_tools: @@ -251,6 +244,11 @@ def _describe_classification( return "available" +def _tool_def_names(tool_defs: Iterable[Dict[str, Any]]) -> Iterable[str]: + """Function names of a tool-defs list (``""`` for a nameless def).""" + return (_fn(td).get("name", "") for td in tool_defs) + + def classify_tools( tool_defs: List[Dict[str, Any]], defer_tools: Optional[frozenset] = None, @@ -259,18 +257,19 @@ def classify_tools( dropped (they are re-added after classification).""" visible: List[Dict[str, Any]] = [] deferrable: List[Dict[str, Any]] = [] - for td in tool_defs: - fn = td.get("function") or {} - name = fn.get("name", "") + for td, name in zip(tool_defs, _tool_def_names(tool_defs)): if name in BRIDGE_TOOL_NAMES: continue - if is_deferrable_tool_name(name, defer_tools): - deferrable.append(td) - else: - visible.append(td) + (deferrable if is_deferrable_tool_name(name, defer_tools) else visible).append(td) return visible, deferrable +def _deferrable_in(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Deferrable subset of a pre-assembly ``tool_defs`` list under the current + (read-only) user config — the universe the bridge tools operate on.""" + return classify_tools(tool_defs, load_config_readonly().effective_defer_tools)[1] + + # --------------------------------------------------------------------------- # Token estimation and threshold gate # --------------------------------------------------------------------------- @@ -319,6 +318,57 @@ def listing_token_budget( # --------------------------------------------------------------------------- +def _bridge_schema(name: str, description: str, properties: Dict[str, Any], + required: List[str]) -> Dict[str, Any]: + """One OpenAI-style function schema (key order is part of the frozen bytes).""" + return { + "type": "function", + "function": { + "name": name, + "description": description, + "parameters": {"type": "object", "properties": properties, "required": required}, + }, + } + + +def _search_description(deferred_count: int, listing: Optional[str], listing_form: str) -> str: + """tool_search bridge description with the listing embedded; ``listing_form`` + picks the framing (see :func:`bridge_tool_schemas`).""" + desc = ( + f"Search {deferred_count} additional tools that are loaded on demand. " + "Takes a list of queries searched in parallel against the same " + "catalog; send one query per distinct capability you need. Returns " + "matching tool names grouped per query plus a shared map with each " + "tool's description. Follow with " + f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, " + f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this " + "system prompt are already available and do not need to be searched." + ) + if not listing: + return desc + if listing_form == "groups": + return desc + ( + "\n\nThe servers below are connected and their tools ARE available " + "through this bridge. For any request in these domains, search " + "here FIRST — do not claim the capability is unavailable and do " + "not substitute a generic tool (terminal/browser) without " + "searching.\n\n" + listing + ) + desc += ( + "\n\nEvery deferred capability is listed below. If a tool name " + "appears here, do NOT claim it is unavailable — load it with " + f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you " + "already see the exact name)." + ) + if listing_form == "mixed": + desc += ( + " For servers marked 'names not listed', the tools exist " + f"too — find them with `{TOOL_SEARCH_NAME}` before " + "concluding anything is missing." + ) + return desc + "\n\n" + listing + + def bridge_tool_schemas( deferred_count: int, listing: Optional[str] = None, @@ -329,111 +379,51 @@ def bridge_tool_schemas( tool_search description; ``listing_form`` picks the framing (per-tool forms say "skip search when you see the exact name", the "groups" summary says which domains exist and that search is mandatory).""" - desc_search = ( - f"Search {deferred_count} additional tools that are loaded on demand. " - "Takes a list of queries searched in parallel against the same " - "catalog; send one query per distinct capability you need. Returns " - "matching tool names grouped per query plus a shared map with each " - "tool's description. Follow with " - f"`{TOOL_DESCRIBE_NAME}` to load full parameter schemas, " - f"then `{TOOL_CALL_NAME}` to invoke. Tools listed at the top of this " - "system prompt are already available and do not need to be searched." - ) - if listing and listing_form == "groups": - desc_search += ( - "\n\nThe servers below are connected and their tools ARE available " - "through this bridge. For any request in these domains, search " - "here FIRST — do not claim the capability is unavailable and do " - "not substitute a generic tool (terminal/browser) without " - "searching.\n\n" + listing - ) - elif listing: - desc_search += ( - "\n\nEvery deferred capability is listed below. If a tool name " - "appears here, do NOT claim it is unavailable — load it with " - f"`{TOOL_DESCRIBE_NAME}` (skip `{TOOL_SEARCH_NAME}` when you " - "already see the exact name)." - ) - if listing_form == "mixed": - desc_search += ( - " For servers marked 'names not listed', the tools exist " - f"too — find them with `{TOOL_SEARCH_NAME}` before " - "concluding anything is missing." - ) - desc_search += "\n\n" + listing - desc_describe = ( - f"Load the full JSON schemas for tools returned by `{TOOL_SEARCH_NAME}`. " - f"Required before `{TOOL_CALL_NAME}` if a tool's parameters are unknown. " - "Batch every schema you need into one call." - ) - desc_call = ( - "Invoke a deferred tool by name with the given arguments. Argument shape " - f"matches the tool's schema (see `{TOOL_DESCRIBE_NAME}`). Policy, hooks, " - "and approvals run exactly as for any directly-listed tool." - ) - return [ - { - "type": "function", - "function": { - "name": TOOL_SEARCH_NAME, - "description": desc_search, - "parameters": { - "type": "object", - "properties": { - "queries": { - "type": "array", - "items": {"type": "string"}, - "description": "Search queries, each a few keywords describing one capability (e.g. ['create github issue', 'send slack message']). Searched in parallel; results come back grouped per query. A single string is accepted and treated as one query.", - }, - "limit": { - "type": "integer", - "description": "Maximum number of matches per query. Defaults to 5 and is clamped to the configured maximum (25 by default).", - }, - }, - "required": ["queries"], + _bridge_schema( + TOOL_SEARCH_NAME, + _search_description(deferred_count, listing, listing_form), + { + "queries": { + "type": "array", + "items": {"type": "string"}, + "description": "Search queries, each a few keywords describing one capability (e.g. ['create github issue', 'send slack message']). Searched in parallel; results come back grouped per query. A single string is accepted and treated as one query.", + }, + "limit": { + "type": "integer", + "description": "Maximum number of matches per query. Defaults to 5 and is clamped to the configured maximum (25 by default).", }, }, - }, - { - "type": "function", - "function": { - "name": TOOL_DESCRIBE_NAME, - "description": desc_describe, - "parameters": { - "type": "object", - "properties": { - "names": { - "type": "array", - "items": {"type": "string"}, - "description": "Exact tool names (as returned by tool_search). A single string is accepted and treated as one name.", - }, - }, - "required": ["names"], + ["queries"], + ), + _bridge_schema( + TOOL_DESCRIBE_NAME, + f"Load the full JSON schemas for tools returned by `{TOOL_SEARCH_NAME}`. " + f"Required before `{TOOL_CALL_NAME}` if a tool's parameters are unknown. " + "Batch every schema you need into one call.", + { + "names": { + "type": "array", + "items": {"type": "string"}, + "description": "Exact tool names (as returned by tool_search). A single string is accepted and treated as one name.", }, }, - }, - { - "type": "function", - "function": { - "name": TOOL_CALL_NAME, - "description": desc_call, - "parameters": { + ["names"], + ), + _bridge_schema( + TOOL_CALL_NAME, + "Invoke a deferred tool by name with the given arguments. Argument shape " + f"matches the tool's schema (see `{TOOL_DESCRIBE_NAME}`). Policy, hooks, " + "and approvals run exactly as for any directly-listed tool.", + { + "name": {"type": "string", "description": "Exact tool name to invoke."}, + "arguments": { "type": "object", - "properties": { - "name": { - "type": "string", - "description": "Exact tool name to invoke.", - }, - "arguments": { - "type": "object", - "description": "Arguments for the tool, matching its schema.", - }, - }, - "required": ["name", "arguments"], + "description": "Arguments for the tool, matching its schema.", }, }, - }, + ["name", "arguments"], + ), ] @@ -452,7 +442,8 @@ class AssemblyResult: deferred_tokens: int = 0 threshold_tokens: int = 0 # 0 = passthrough; 1 = bridge + per-tool listing (full/names/mixed); - # 2 = bare bridge / server-summary only. + # 2 = bare bridge / server-summary only — domains stay visible but + # individual tools are reachable only via tool_search. tier: int = 0 listing_form: str = "none" # "full" | "names" | "mixed" | "groups" | "none" @@ -469,8 +460,8 @@ def assemble_tool_defs( if config is None: config = load_config() - incoming = [td for td in tool_defs - if (td.get("function") or {}).get("name") not in BRIDGE_TOOL_NAMES] + incoming = [td for td, name in zip(tool_defs, _tool_def_names(tool_defs)) + if name not in BRIDGE_TOOL_NAMES] visible, deferrable = classify_tools(incoming, config.effective_defer_tools) if not deferrable: @@ -487,15 +478,13 @@ def assemble_tool_defs( tier=0, ) - listing = None - listing_form = "none" + listing, listing_form = None, "none" listing_budget = listing_token_budget(config, context_length) if config.listing != "off": listing, listing_form = build_catalog_listing_with_form( deferrable, max_tokens=listing_budget) bridge = bridge_tool_schemas(len(deferrable), listing=listing, listing_form=listing_form) - result = visible + bridge tier = 1 if listing_form in ("full", "names", "mixed") else 2 logger.info( @@ -506,7 +495,7 @@ def assemble_tool_defs( ) return AssemblyResult( - tool_defs=result, + tool_defs=visible + bridge, activated=True, deferred_count=len(deferrable), deferred_tokens=deferrable_tokens, @@ -531,12 +520,8 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]: trivial call without a ``tool_describe`` round-trip.""" schema = entry.schema if isinstance(entry.schema, dict) else {} fn = schema.get("function") - if not isinstance(fn, dict): - fn = {} - params = fn.get("parameters") - if not isinstance(params, dict): - params = {} - required = params.get("required") + params = fn.get("parameters") if isinstance(fn, dict) else None + required = params.get("required") if isinstance(params, dict) else None if not isinstance(required, list): required = [] return { @@ -549,24 +534,36 @@ def _shared_tool_record(entry: CatalogEntry) -> Dict[str, Any]: def _available_source_summary(catalog: List[CatalogEntry]) -> List[Dict[str, Any]]: """Deterministic ``[{name, tool_count}]`` of connected sources, attached to - empty query groups so a lexical miss is not read as a missing capability.""" + empty query groups so a lexical miss is not read as a missing capability + (adds nothing to the fixed per-turn prompt).""" counts = Counter(_listing_group_label(entry.source_name) for entry in catalog) return [{"name": name, "tool_count": counts[name]} for name in sorted(counts)] -def _string_list(raw: Any, *, dedupe: bool) -> Optional[List[str]]: - """Normalize a list-of-strings argument. A bare string (a common model slip) - counts as a one-item list; non-list input returns None.""" +def _string_list_arg( + args: Dict[str, Any], key: str, *, dedupe: bool, max_items: int, retry_hint: str, +) -> Tuple[Optional[List[str]], Optional[str]]: + """Read a list-of-strings bridge argument -> ``(items, error_json)``. A bare + string (a common model slip) counts as a one-item list. Rejects non-list + input, an empty list (after stripping blanks), and more than ``max_items`` + (bounds the work one bridge call can request).""" + raw = args.get(key) if isinstance(raw, str): raw = [raw] if not isinstance(raw, list): - return None + return None, tool_error(f"{key} is required and must be an array of strings") out: List[str] = [] for item in raw: text = str(item or "").strip() if text and (not dedupe or text not in out): out.append(text) - return out + if not out: + return None, tool_error( + f"{key} is required and must contain at least one non-empty string") + if len(out) > max_items: + return None, tool_error( + f"too many {key}: {len(out)} > max {max_items}. {retry_hint}") + return out, None def dispatch_tool_search(args: Dict[str, Any], @@ -585,27 +582,19 @@ def dispatch_tool_search(args: Dict[str, Any], if config is None: config = load_config() - queries = _string_list(args.get("queries"), dedupe=False) - if queries is None: - return tool_error("queries is required and must be an array of strings") - if not queries: - return tool_error("queries is required and must contain at least one non-empty string") - if len(queries) > _MAX_QUERIES_PER_CALL: - return tool_error( - f"too many queries: {len(queries)} > max {_MAX_QUERIES_PER_CALL}. " - "Retry with fewer, more targeted queries." - ) + queries, err = _string_list_arg( + args, "queries", dedupe=False, max_items=_MAX_QUERIES_PER_CALL, + retry_hint="Retry with fewer, more targeted queries.") + if err: + return err raw_limit = args.get("limit") if raw_limit is None: limit = config.search_default_limit else: - limit = max(1, min(config.max_search_limit, _safe_int(raw_limit, config.search_default_limit))) + limit = _clamped_int(raw_limit, config.search_default_limit, 1, config.max_search_limit) - _, deferrable = classify_tools( - current_tool_defs, load_config_readonly().effective_defer_tools - ) - catalog = build_catalog(deferrable) + catalog = build_catalog(_deferrable_in(current_tool_defs)) results: List[Dict[str, Any]] = [] tools_map: Dict[str, Dict[str, Any]] = {} @@ -614,8 +603,7 @@ def dispatch_tool_search(args: Dict[str, Any], for query in queries: hits = search_catalog(catalog, query, limit=limit, corpus_stats=corpus_stats) for h in hits: - if h.name not in tools_map: - tools_map[h.name] = _shared_tool_record(h) + tools_map.setdefault(h.name, _shared_tool_record(h)) group: Dict[str, Any] = {"query": query, "matches": [h.name for h in hits]} if not hits and catalog: group["available_sources"] = available_sources @@ -627,13 +615,12 @@ def dispatch_tool_search(args: Dict[str, Any], ) results.append(group) - result: Dict[str, Any] = { + return json.dumps({ "queries": queries, "total_available": len(catalog), "results": results, "tools": tools_map, - } - return json.dumps(result, ensure_ascii=False) + }, ensure_ascii=False) def dispatch_tool_describe(args: Dict[str, Any], @@ -651,25 +638,14 @@ def dispatch_tool_describe(args: Dict[str, Any], if config is None: config = load_config_readonly() - names = _string_list(args.get("names"), dedupe=True) - if names is None: - return tool_error("names is required and must be an array of strings") - if not names: - return tool_error("names is required and must contain at least one non-empty string") - if len(names) > _MAX_DESCRIBE_NAMES_PER_CALL: - return tool_error( - f"too many names: {len(names)} > max {_MAX_DESCRIBE_NAMES_PER_CALL}. " - "Retry with fewer names per call." - ) + names, err = _string_list_arg( + args, "names", dedupe=True, max_items=_MAX_DESCRIBE_NAMES_PER_CALL, + retry_hint="Retry with fewer names per call.") + if err: + return err - _, deferrable = classify_tools( - current_tool_defs, load_config_readonly().effective_defer_tools - ) - by_name: Dict[str, Dict[str, Any]] = {} - for td in deferrable: - fn = td.get("function") or {} - if fn.get("name"): - by_name[fn["name"]] = fn + deferrable = _deferrable_in(current_tool_defs) + by_name = {name: _fn(td) for td, name in zip(deferrable, _tool_def_names(deferrable)) if name} tools: Dict[str, Dict[str, Any]] = {} not_found: List[str] = [] @@ -707,7 +683,7 @@ def scoped_deferrable_names(tool_defs: List[Dict[str, Any]]) -> frozenset[str]: out-of-scope tool via the bridge.""" defer_tools = load_config_readonly().effective_defer_tools return frozenset( - name for name in ((td.get("function") or {}).get("name", "") for td in tool_defs) + name for name in _tool_def_names(tool_defs) if name and is_deferrable_tool_name(name, defer_tools) ) @@ -753,6 +729,7 @@ __all__ = [ "estimate_tokens_from_schemas", "should_activate", "build_catalog", + "build_catalog_listing", "build_catalog_listing_with_form", "listing_token_budget", "search_catalog", diff --git a/tools/tool_search_catalog.py b/tools/tool_search_catalog.py index 072607c97a..d745fd2bed 100644 --- a/tools/tool_search_catalog.py +++ b/tools/tool_search_catalog.py @@ -63,6 +63,22 @@ def _tokenize(text: str) -> List[str]: return [_stem(token.lower()) for token in _TOKEN_RE.findall(text)] +def _fn(td: Dict[str, Any]) -> Dict[str, Any]: + """The ``function`` block of a tool-def (``{}`` when absent/None).""" + return td.get("function") or {} + + +def _registry_entry(name: str) -> Any: + """Registry entry for ``name``; None when unregistered OR when the registry + is unavailable/raises (lookup failures must never fail a bridge call). + The import stays lazy: tests patch ``tools.registry.registry``.""" + try: + from tools.registry import registry + return registry.get_entry(name) + except Exception: + return None + + def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str: """Search-text blob: split name words + source label + description + top-level parameter names. Schema bodies are excluded (noise, no recall @@ -70,7 +86,7 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str: its IDF is ~0. The source label lets a service-name query ("linear") reach a tool whose own name omits the vendor. """ - fn = td.get("function") or {} + fn = _fn(td) name = fn.get("name", "") if name.startswith("mcp__"): name = name[len("mcp__"):] @@ -84,15 +100,12 @@ def _entry_search_text(td: Dict[str, Any], source_label: str = "") -> str: def _classify_source(name: str) -> Tuple[str, str]: """Return (source_kind, source_name) for a registered tool name.""" + entry = _registry_entry(name) + if entry is None: + return ("other", "") try: - from tools.registry import registry - entry = registry.get_entry(name) - if entry is None: - return ("other", "") - if entry.toolset.startswith("mcp-"): - return ("mcp", entry.toolset) - return ("plugin", entry.toolset) - except Exception: + return ("mcp" if entry.toolset.startswith("mcp-") else "plugin", entry.toolset) + except Exception: # malformed entry (no str toolset) return ("other", "") @@ -100,7 +113,7 @@ def build_catalog(tool_defs: List[Dict[str, Any]]) -> List[CatalogEntry]: """Build the deferred-tool catalog from the deferrable subset of tool-defs.""" catalog: List[CatalogEntry] = [] for td in tool_defs: - fn = td.get("function") or {} + fn = _fn(td) name = fn.get("name", "") if not name: continue @@ -123,7 +136,7 @@ def _bm25_score(query_tokens: List[str], doc_tokens: List[str], doc_freq: Dict[str, int], n_docs: int, k1: float = 1.5, b: float = 0.75) -> float: """Standard BM25 score for one query against one document (inlined; the - catalog is small enough that a dependency is not worth it).""" + catalog is bounded — typically < 500 tools — so a dependency is not worth it).""" if not doc_tokens: return 0.0 score = 0.0 @@ -204,7 +217,8 @@ _SENTENCE_END_RE = re.compile(r"(? str: """First sentence of a tool description, clipped to ``max_chars`` on a - word boundary. Linear-time on hostile input.""" + word boundary. ``e.g.``/``i.e.``/``etc.`` do not end a sentence; whitespace + normalization and the regex search stay linear-time on hostile input.""" text = " ".join((description or "").split()) if not text: return "" @@ -222,9 +236,17 @@ def _short_desc(description: str, max_chars: int = 60) -> str: def _listing_group_label(source_name: str) -> str: """Human-facing group heading for a toolset, e.g. ``mcp-github`` -> ``github``.""" label = source_name or "other" - if label.startswith("mcp-"): - label = label[4:] - return label + return label[4:] if label.startswith("mcp-") else label + + +def build_catalog_listing( + deferrable: List[Dict[str, Any]], + *, + max_tokens: int = 4000, +) -> Optional[str]: + """Render the deferred-catalog manifest; text only (see + :func:`build_catalog_listing_with_form` for the degradation ladder).""" + return build_catalog_listing_with_form(deferrable, max_tokens=max_tokens)[0] def build_catalog_listing_with_form( @@ -250,12 +272,13 @@ def build_catalog_listing_with_form( groups: Dict[str, List[Tuple[str, str]]] = {} for td in deferrable: - fn = td.get("function") or {} + fn = _fn(td) name = fn.get("name", "") if not name: continue - source, source_name = _classify_source(name) - label = _listing_group_label(source_name if source != "other" else "other") + # ``_classify_source`` returns ("other", "") for unregistered names and + # ``_listing_group_label("")`` is "other", so one call covers both. + label = _listing_group_label(_classify_source(name)[1]) groups.setdefault(label, []).append((name, _short_desc(fn.get("description", "")))) if not groups: @@ -269,8 +292,7 @@ def build_catalog_listing_with_form( f"discover via `{TOOL_SEARCH_NAME}`)") lines = [f"{label} tools ({len(tools)}):"] if mode == "full": - for name, desc in tools: - lines.append(f"- {name}: {desc}" if desc else f"- {name}") + lines.extend(f"- {name}: {desc}" if desc else f"- {name}" for name, desc in tools) else: lines.append(", ".join(name for name, _ in tools)) return "\n".join(lines) @@ -278,30 +300,26 @@ def build_catalog_listing_with_form( header = ("Deferred tool catalog (call schemas via " f"`{TOOL_DESCRIBE_NAME}`, invoke via `{TOOL_CALL_NAME}`):") - def assemble(modes: Dict[str, str]) -> str: - return "\n".join([header] + [render_group(lbl, modes[lbl]) - for lbl in sorted(groups)]) + def assemble_if_fits(modes: Dict[str, str]) -> Optional[str]: + text = "\n".join([header] + [render_group(lbl, modes[lbl]) for lbl in sorted(groups)]) + return text if math.ceil(len(text) / CHARS_PER_TOKEN) <= max_tokens else None - def fits(text: str) -> bool: - return math.ceil(len(text) / CHARS_PER_TOKEN) <= max_tokens - - # 1. Everything full. - modes = {lbl: "full" for lbl in groups} - if fits(assemble(modes)): - return assemble(modes), "full" - - # 2. Everything names-only. - modes = {lbl: "names" for lbl in groups} - if fits(assemble(modes)): - return assemble(modes), "names" + # 1. Everything full. 2. Everything names-only. + for mode in ("full", "names"): + modes = {lbl: mode for lbl in groups} + text = assemble_if_fits(modes) + if text is not None: + return text, mode # 3. Per-server degradation: collapse the LARGEST rendered groups first - # (deterministic: size then label). + # (deterministic: size then label) so one oversized server does not + # cost a small co-attached server its per-tool names. by_size = sorted(groups, key=lambda lbl: (-len(render_group(lbl, "names")), lbl)) for lbl in by_size: modes[lbl] = "summary" - if fits(assemble(modes)): + text = assemble_if_fits(modes) + if text is not None: form = "groups" if all(m == "summary" for m in modes.values()) else "mixed" - return assemble(modes), form + return text, form return None, "none" diff --git a/tools/tool_search_names.py b/tools/tool_search_names.py index f7350d85bb..bd775460da 100644 --- a/tools/tool_search_names.py +++ b/tools/tool_search_names.py @@ -1,5 +1,7 @@ """Reserved bridge tool names shared by tool_search and its catalog module.""" +# Reserved: a user/plugin/MCP tool may not take these names — the registry's +# override protection rejects such registrations. TOOL_SEARCH_NAME = "tool_search" TOOL_DESCRIBE_NAME = "tool_describe" TOOL_CALL_NAME = "tool_call" diff --git a/tools/tool_search_validation.py b/tools/tool_search_validation.py index 19cf84be16..c5ee9ba401 100644 --- a/tools/tool_search_validation.py +++ b/tools/tool_search_validation.py @@ -107,13 +107,11 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str if not isinstance(schema, dict): return None fn = schema.get("function") if schema.get("type") == "function" else schema - if not isinstance(fn, dict): - return None - params = fn.get("parameters") + params = fn.get("parameters") if isinstance(fn, dict) else None if not isinstance(params, dict): return None required = params.get("required") - if isinstance(required, list) and required: + if isinstance(required, list): missing = [r for r in required if isinstance(r, str) and r not in args] if missing: return _validation_error( @@ -131,7 +129,7 @@ def validate_deferred_call_args(name: str, args: Dict[str, Any]) -> Optional[str return None # Validate the repaired shape dispatch will see; copy because - # coerce_tool_args may normalize in place. + # coerce_tool_args may normalize in place (dispatch re-coerces canonically). candidate_args = dict(args) try: from model_tools import coerce_tool_args