fix(registry): derive toolset availability from per-tool checks

Doctor and banner used the first check_fn registered for a toolset, so
desktop-only read_terminal gated the whole terminal toolset even though
terminal and process still expose at runtime.

Fixes #54820
This commit is contained in:
xxxigm 2026-06-29 17:32:28 +07:00 committed by kshitij
parent c969090878
commit 6e84257717

View file

@ -251,6 +251,33 @@ class ToolRegistry:
logger.debug("Toolset %s check raised; marking unavailable", toolset)
return False
def _toolset_has_exposable_tools(
self,
toolset: str,
entries: List[ToolEntry],
*,
quiet: bool = False,
) -> bool:
"""Return True when at least one tool in *toolset* would be exposed.
Mirrors :meth:`get_tool_definitions` per-tool filtering so doctor,
banners, and other toolset-level surfaces agree with runtime exposure.
Mixed toolsets (e.g. ``terminal`` plus desktop-only ``read_terminal``)
must not be gated solely by the first registered ``check_fn``.
"""
del quiet # reserved for parity with check_tool_availability signature
check_results: Dict[Callable, bool] = {}
for entry in entries:
if entry.toolset != toolset:
continue
if not entry.check_fn:
return True
if entry.check_fn not in check_results:
check_results[entry.check_fn] = _check_fn_cached(entry.check_fn)
if check_results[entry.check_fn]:
return True
return False
def get_entry(self, name: str) -> Optional[ToolEntry]:
"""Return a registered tool entry by name, or None."""
with self._lock:
@ -567,35 +594,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's requirements are met.
"""Check if a toolset has at least one exposable tool.
Returns False (rather than crashing) when the check function raises
Returns False (rather than crashing) when a per-tool check raises
an unexpected exception (e.g. network error, missing import, bad config).
"""
with self._lock:
check = self._toolset_checks.get(toolset)
return self._evaluate_toolset_check(toolset, check)
entries, _ = self._snapshot_state()
return self._toolset_has_exposable_tools(toolset, entries)
def check_toolset_requirements(self) -> Dict[str, bool]:
"""Return ``{toolset: available_bool}`` for every toolset."""
entries, toolset_checks = self._snapshot_state()
entries, _ = self._snapshot_state()
toolsets = sorted({entry.toolset for entry in entries})
return {
toolset: self._evaluate_toolset_check(toolset, toolset_checks.get(toolset))
toolset: self._toolset_has_exposable_tools(toolset, entries)
for toolset in toolsets
}
def get_available_toolsets(self) -> Dict[str, dict]:
"""Return toolset metadata for UI display."""
toolsets: Dict[str, dict] = {}
entries, toolset_checks = self._snapshot_state()
entries, _ = self._snapshot_state()
for entry in entries:
ts = entry.toolset
if ts not in toolsets:
toolsets[ts] = {
"available": self._evaluate_toolset_check(
ts, toolset_checks.get(ts)
),
"available": self._toolset_has_exposable_tools(ts, entries),
"tools": [],
"description": "",
"requirements": [],
@ -632,20 +656,16 @@ class ToolRegistry:
"""Return (available_toolsets, unavailable_info) like the old function."""
available = []
unavailable = []
seen = set()
entries, toolset_checks = self._snapshot_state()
for entry in entries:
ts = entry.toolset
if ts in seen:
continue
seen.add(ts)
if self._evaluate_toolset_check(ts, toolset_checks.get(ts)):
entries, _ = self._snapshot_state()
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, quiet=quiet):
available.append(ts)
else:
unavailable.append({
"name": ts,
"env_vars": entry.requires_env,
"tools": [e.name for e in entries if e.toolset == ts],
"env_vars": ts_entries[0].requires_env if ts_entries else [],
"tools": [entry.name for entry in ts_entries],
})
return available, unavailable