feat(extensions): project native capabilities into catalog

This commit is contained in:
Xubin Ren 2026-07-26 16:27:20 +08:00
parent 863b02d215
commit 1e573c75ae
13 changed files with 742 additions and 10 deletions

View File

@ -89,6 +89,11 @@ and are exposed only through declared host interfaces. Existing workspace,
network, SSRF, and shell restrictions continue to apply to host-provided
operations.
Untrusted packages remain visible in the catalog with an inactive state. They
do not own active contributions and their runtime is not imported. Built-in
capabilities are trusted by construction; installed and workspace packages
need an explicit trusted entry or an allowed workspace trust policy.
The root `extensions` config controls explicit search paths, allow/deny policy,
per-extension enablement, package-owned config, and workspace trust. Discovery
does not import extension code. Installation does not imply workspace trust,

View File

@ -86,9 +86,21 @@ class ToolLoader:
def load(self, ctx: Any, registry: ToolRegistry, *, scope: str = "core") -> list[str]:
registered: list[str] = []
builtin_names: set[str] = set()
sources = [(self.discover(), False), (self._discover_plugins().values(), True)]
sources = [
(
(("nanobot.core", tool_cls) for tool_cls in self.discover()),
False,
),
(
(
(f"legacy.tool.{entry_point_name}", tool_cls)
for entry_point_name, tool_cls in self._discover_plugins().items()
),
True,
),
]
for source, is_plugin_source in sources:
for tool_cls in source:
for owner, tool_cls in source:
cls_label = tool_cls.__name__
try:
if scope not in getattr(tool_cls, "_scopes", {"core"}):
@ -109,7 +121,7 @@ class ToolLoader:
"Tool name collision: %s from %s overwrites existing",
tool.name, cls_label,
)
registry.register(tool)
registry.register(tool, owner=owner)
registered.append(tool.name)
if not is_plugin_source:
builtin_names.add(tool.name)

View File

@ -963,7 +963,7 @@ async def connect_mcp_servers(
)
continue
wrapper = MCPToolWrapper(session, name, tool_def, tool_timeout=cfg.tool_timeout)
registry.register(wrapper)
registry.register(wrapper, owner=f"nanobot.mcp.{name}")
logger.debug("MCP: registered tool '{}' from server '{}'", wrapper.name, name)
registered_count += 1
if enabled_tools:
@ -1000,7 +1000,7 @@ async def connect_mcp_servers(
wrapper = MCPResourceWrapper(
session, name, resource, resource_timeout=cfg.tool_timeout
)
registry.register(wrapper)
registry.register(wrapper, owner=f"nanobot.mcp.{name}")
registered_count += 1
logger.debug(
"MCP: registered resource '{}' from server '{}'",
@ -1018,7 +1018,7 @@ async def connect_mcp_servers(
wrapper = MCPPromptWrapper(
session, name, prompt, prompt_timeout=cfg.tool_timeout
)
registry.register(wrapper)
registry.register(wrapper, owner=f"nanobot.mcp.{name}")
registered_count += 1
logger.debug(
"MCP: registered prompt '{}' from server '{}'",

View File

@ -25,22 +25,29 @@ class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
self._owners: dict[str, str] = {}
self._cached_definitions: list[dict[str, Any]] | None = None
def register(self, tool: Tool) -> None:
def register(self, tool: Tool, *, owner: str = "nanobot.core") -> None:
"""Register a tool."""
self._tools[tool.name] = tool
self._owners[tool.name] = owner
self._cached_definitions = None
def unregister(self, name: str) -> None:
"""Unregister a tool by name."""
self._tools.pop(name, None)
self._owners.pop(name, None)
self._cached_definitions = None
def get(self, name: str) -> Tool | None:
"""Get a tool by name."""
return self._tools.get(name)
def owner(self, name: str) -> str | None:
"""Return the extension ID that registered a tool."""
return self._owners.get(name)
def get_runtime_context_providers(self) -> list[RuntimeContextProvider]:
"""Return tool-owned providers in stable tool-name order."""
providers: list[RuntimeContextProvider] = []

View File

@ -64,16 +64,45 @@ class CommandRouter:
self._priority: dict[str, Handler] = {}
self._exact: dict[str, Handler] = {}
self._prefix: list[tuple[str, Handler]] = []
self._owners: dict[tuple[str, str], str] = {}
def priority(self, cmd: str, handler: Handler) -> None:
def priority(
self,
cmd: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._priority[cmd] = handler
self._owners[("priority", cmd)] = owner
def exact(self, cmd: str, handler: Handler) -> None:
def exact(
self,
cmd: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._exact[cmd] = handler
self._owners[("exact", cmd)] = owner
def prefix(self, pfx: str, handler: Handler) -> None:
def prefix(
self,
pfx: str,
handler: Handler,
*,
owner: str = "nanobot.core",
) -> None:
self._prefix.append((pfx, handler))
self._prefix.sort(key=lambda p: len(p[0]), reverse=True)
self._owners[("prefix", pfx)] = owner
def registrations(self) -> tuple[tuple[str, str, str], ...]:
"""Return ``(tier, command, owner)`` rows for extension inspection."""
return tuple(
(tier, command, owner)
for (tier, command), owner in sorted(self._owners.items())
)
def is_priority(self, text: str) -> bool:
return normalize_command_text(text).lower() in self._priority

View File

@ -1,5 +1,6 @@
"""First-class extension metadata and discovery primitives."""
from nanobot.extensions.catalog import ExtensionCatalog, build_extension_catalog
from nanobot.extensions.codec import (
MANIFEST_FILENAME,
ManifestFormatError,
@ -18,6 +19,7 @@ from nanobot.extensions.manifest import (
ExtensionPermission,
ExtensionRuntime,
)
from nanobot.extensions.native import discover_native_extensions
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
@ -33,6 +35,7 @@ __all__ = [
"ContributionKind",
"DependencyKind",
"ExtensionCandidate",
"ExtensionCatalog",
"ExtensionContribution",
"ExtensionDependency",
"ExtensionDiagnostic",
@ -46,7 +49,9 @@ __all__ = [
"MANIFEST_FILENAME",
"ManifestFormatError",
"ResolvedContribution",
"build_extension_catalog",
"dump_manifest",
"discover_native_extensions",
"load_manifest",
"manifest_from_mapping",
"manifest_to_mapping",

View File

@ -0,0 +1,136 @@
"""Assemble native and installed extensions into one inspectable catalog."""
from __future__ import annotations
from dataclasses import dataclass, replace
from pathlib import Path
from typing import TYPE_CHECKING, Iterable
from nanobot.extensions.discovery import (
ExtensionDiscoveryResult,
discover_manifest_root,
)
from nanobot.extensions.native import discover_native_extensions
from nanobot.extensions.registry import (
ExtensionCandidate,
ExtensionDiagnostic,
ExtensionPolicy,
ExtensionRegistry,
ExtensionScope,
ExtensionSnapshot,
)
if TYPE_CHECKING:
from nanobot.agent.skills import SkillsLoader
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.command.router import CommandRouter
from nanobot.config.schema import Config
@dataclass(frozen=True, slots=True)
class ExtensionCatalog:
"""Discovered candidates plus the active, policy-resolved snapshot."""
candidates: tuple[ExtensionCandidate, ...]
snapshot: ExtensionSnapshot
diagnostics: tuple[ExtensionDiagnostic, ...]
def build_extension_catalog(
config: Config,
*,
skills: SkillsLoader | None = None,
tools: ToolRegistry | None = None,
commands: CommandRouter | None = None,
user_root: Path | None = None,
) -> ExtensionCatalog:
"""Build the authoritative extension view without executing plugin code."""
native = discover_native_extensions(
config,
skills=skills,
tools=tools,
commands=commands,
)
discoveries = [native]
if config.extensions.enabled:
discoveries.extend(
_external_discoveries(
config,
user_root=user_root or Path.home() / ".nanobot" / "extensions",
)
)
candidates = tuple(
_apply_entry_config(config, candidate)
for result in discoveries
for candidate in result.candidates
)
discovery_diagnostics = tuple(
diagnostic
for result in discoveries
for diagnostic in result.diagnostics
)
registry = ExtensionRegistry(
ExtensionPolicy(
allow=frozenset(config.extensions.allow),
deny=frozenset(config.extensions.deny),
)
)
registry_diagnostics: list[ExtensionDiagnostic] = []
for candidate in candidates:
try:
registry.register(candidate)
except ValueError as exc:
registry_diagnostics.append(
ExtensionDiagnostic(
code="duplicate_installation",
extension_id=candidate.manifest.id,
message=str(exc),
)
)
snapshot = registry.snapshot()
diagnostics = (
discovery_diagnostics
+ tuple(registry_diagnostics)
+ snapshot.diagnostics
)
return ExtensionCatalog(candidates, snapshot, diagnostics)
def _external_discoveries(
config: Config,
*,
user_root: Path,
) -> Iterable[ExtensionDiscoveryResult]:
yield discover_manifest_root(
user_root,
scope=ExtensionScope.USER,
)
for raw_path in config.extensions.paths:
yield discover_manifest_root(
Path(raw_path).expanduser(),
scope=ExtensionScope.USER,
)
workspace_root = config.workspace_path / ".nanobot" / "extensions"
workspace_trust = config.extensions.workspace_trust
if workspace_trust != "deny":
yield discover_manifest_root(
workspace_root,
scope=ExtensionScope.WORKSPACE,
trusted=workspace_trust == "allow",
)
def _apply_entry_config(
config: Config,
candidate: ExtensionCandidate,
) -> ExtensionCandidate:
entry = config.extensions.entries.get(candidate.manifest.id)
if entry is None or candidate.scope is ExtensionScope.BUILTIN:
return candidate
return replace(
candidate,
enabled=entry.enabled,
trusted=candidate.trusted or entry.trusted,
)

View File

@ -0,0 +1,339 @@
"""Project existing nanobot registries into the extension control plane."""
from __future__ import annotations
import re
from collections import defaultdict
from pathlib import Path
from typing import TYPE_CHECKING, Iterable
from packaging.requirements import Requirement
from nanobot import __version__
from nanobot.extensions.discovery import ExtensionDiscoveryResult
from nanobot.extensions.manifest import (
ContributionKind,
DependencyKind,
ExtensionContribution,
ExtensionDependency,
ExtensionManifest,
ExtensionRuntime,
)
from nanobot.extensions.registry import ExtensionCandidate, ExtensionScope
if TYPE_CHECKING:
from nanobot.agent.skills import SkillsLoader
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.command.router import CommandRouter
from nanobot.config.schema import Config
_NON_ID_CHARACTER = re.compile(r"[^a-z0-9._-]+")
def discover_native_extensions(
config: Config,
*,
skills: SkillsLoader | None = None,
tools: ToolRegistry | None = None,
commands: CommandRouter | None = None,
) -> ExtensionDiscoveryResult:
"""Return built-in, workspace, and configured capabilities as extensions."""
candidates = [
*_channel_candidates(),
*_provider_candidates(),
*_transcription_candidates(),
*_image_generation_candidates(),
*_mcp_candidates(config),
]
if skills is not None:
candidates.extend(_skill_candidates(skills))
if tools is not None:
candidates.extend(_tool_candidates(tools))
if commands is not None:
candidates.extend(_command_candidates(commands))
return ExtensionDiscoveryResult(candidates=_merge_candidates(candidates))
def _channel_candidates() -> list[ExtensionCandidate]:
from nanobot.channels.registry import discover_plugins
candidates = []
for name, plugin in sorted(discover_plugins().items()):
contributions = [
ExtensionContribution(
kind=ContributionKind.CHANNEL,
name=name,
target=plugin.runtime,
description=plugin.display_name,
)
]
if plugin.webui:
contributions.append(
ExtensionContribution(
kind=ContributionKind.WEBUI,
name=f"channel-{name}",
target=plugin.webui,
)
)
candidates.append(
_candidate(
f"nanobot.channel.{name}",
plugin.display_name,
contributions,
dependencies=_python_dependencies(plugin.dependencies),
)
)
return candidates
def _provider_candidates() -> list[ExtensionCandidate]:
from nanobot.providers.registry import PROVIDERS
candidates = []
for spec in PROVIDERS:
if spec.settings_alias_for or spec.is_transcription_only:
continue
candidates.append(
_candidate(
f"nanobot.provider.{spec.name}",
spec.label,
(
ExtensionContribution(
kind=ContributionKind.LLM_PROVIDER,
name=spec.name,
target=spec.backend,
),
),
)
)
return candidates
def _transcription_candidates() -> list[ExtensionCandidate]:
from nanobot.audio.transcription_registry import TRANSCRIPTION_PROVIDERS
return [
_candidate(
f"nanobot.transcription.{spec.name}",
f"{spec.name} transcription",
(
ExtensionContribution(
kind=ContributionKind.TRANSCRIPTION_PROVIDER,
name=spec.name,
target=spec.adapter,
),
),
)
for spec in TRANSCRIPTION_PROVIDERS
]
def _image_generation_candidates() -> list[ExtensionCandidate]:
from nanobot.providers.image_generation import image_gen_provider_names
return [
_candidate(
f"nanobot.image-generation.{name}",
f"{name} image generation",
(
ExtensionContribution(
kind=ContributionKind.IMAGE_GENERATION_PROVIDER,
name=name,
),
),
)
for name in image_gen_provider_names()
]
def _mcp_candidates(config: Config) -> list[ExtensionCandidate]:
return [
_candidate(
f"nanobot.mcp.{_identifier(name)}",
name,
(
ExtensionContribution(
kind=ContributionKind.MCP_SERVER,
name=_identifier(name),
target=name,
),
),
scope=ExtensionScope.USER,
)
for name in sorted(config.tools.mcp_servers)
]
def _skill_candidates(skills: SkillsLoader) -> list[ExtensionCandidate]:
candidates = []
for entry in skills.list_skills(filter_unavailable=False):
name = entry["name"]
metadata = skills.get_skill_metadata(name) or {}
requirements = skills.get_skill_requirements(name)
dependencies = tuple(
ExtensionDependency(DependencyKind.EXECUTABLE, value)
for value in requirements["bins"]
) + tuple(
ExtensionDependency(DependencyKind.ENVIRONMENT, value)
for value in requirements["env"]
)
scope = (
ExtensionScope.WORKSPACE
if entry["source"] == "workspace"
else ExtensionScope.BUILTIN
)
candidates.append(
_candidate(
f"nanobot.skill.{_identifier(name)}",
name,
(
ExtensionContribution(
kind=ContributionKind.SKILL,
name=_identifier(name),
target=entry["path"],
description=str(metadata.get("description") or name),
),
),
dependencies=dependencies,
scope=scope,
location=Path(entry["path"]).parent,
)
)
return candidates
def _tool_candidates(tools: ToolRegistry) -> list[ExtensionCandidate]:
grouped: dict[str, list[ExtensionContribution]] = defaultdict(list)
for name in tools.tool_names:
owner = tools.owner(name) or "nanobot.core"
tool = tools.get(name)
grouped[owner].append(
ExtensionContribution(
kind=ContributionKind.TOOL,
name=_identifier(name),
description=tool.description if tool is not None else "",
)
)
return [
_candidate(
owner,
owner,
contributions,
scope=(
ExtensionScope.USER
if owner.startswith(("legacy.", "nanobot.mcp."))
else ExtensionScope.BUILTIN
),
)
for owner, contributions in sorted(grouped.items())
]
def _command_candidates(commands: CommandRouter) -> list[ExtensionCandidate]:
grouped: dict[str, list[ExtensionContribution]] = defaultdict(list)
for _tier, command, owner in commands.registrations():
grouped[owner].append(
ExtensionContribution(
kind=ContributionKind.COMMAND,
name=_identifier(command.lstrip("/").rstrip()),
target=command,
)
)
return [
_candidate(owner, owner, contributions)
for owner, contributions in sorted(grouped.items())
]
def _candidate(
extension_id: str,
name: str,
contributions: Iterable[ExtensionContribution],
*,
dependencies: tuple[ExtensionDependency, ...] = (),
scope: ExtensionScope = ExtensionScope.BUILTIN,
location: Path | None = None,
) -> ExtensionCandidate:
return ExtensionCandidate(
manifest=ExtensionManifest(
id=_identifier(extension_id),
name=name,
version=__version__,
runtime=ExtensionRuntime.PYTHON,
contributions=tuple(contributions),
dependencies=dependencies,
),
scope=scope,
location=location,
enabled=True,
trusted=True,
)
def _python_dependencies(
requirements: tuple[str, ...],
) -> tuple[ExtensionDependency, ...]:
dependencies = []
for raw in requirements:
requirement = Requirement(raw)
name = requirement.name
if requirement.extras:
name += f"[{','.join(sorted(requirement.extras))}]"
specifier = str(requirement.specifier)
if requirement.marker:
specifier += f"; {requirement.marker}"
dependencies.append(
ExtensionDependency(
kind=DependencyKind.PYTHON,
name=name,
specifier=specifier,
)
)
return tuple(dependencies)
def _merge_candidates(
candidates: Iterable[ExtensionCandidate],
) -> tuple[ExtensionCandidate, ...]:
merged: dict[tuple[str, ExtensionScope], ExtensionCandidate] = {}
for candidate in candidates:
key = (candidate.manifest.id, candidate.scope)
existing = merged.get(key)
if existing is None:
merged[key] = candidate
continue
manifest = existing.manifest
incoming = candidate.manifest
merged[key] = ExtensionCandidate(
manifest=ExtensionManifest(
id=manifest.id,
name=manifest.name,
version=manifest.version,
runtime=manifest.runtime,
contributions=manifest.contributions + incoming.contributions,
description=manifest.description or incoming.description,
dependencies=tuple(
dict.fromkeys(manifest.dependencies + incoming.dependencies)
),
permissions=tuple(
dict.fromkeys(manifest.permissions + incoming.permissions)
),
homepage=manifest.homepage or incoming.homepage,
license=manifest.license or incoming.license,
),
scope=existing.scope,
location=existing.location or candidate.location,
enabled=existing.enabled and candidate.enabled,
trusted=existing.trusted and candidate.trusted,
)
return tuple(
sorted(
merged.values(),
key=lambda item: (item.scope, item.manifest.id),
)
)
def _identifier(value: str) -> str:
normalized = _NON_ID_CHARACTER.sub("-", value.strip().lower()).strip("._-")
return normalized or "unnamed"

View File

@ -62,6 +62,7 @@ class ExtensionPolicy:
extension_id = candidate.manifest.id
return (
candidate.enabled
and (candidate.scope is ExtensionScope.BUILTIN or candidate.trusted)
and extension_id not in self.deny
and (not self.allow or extension_id in self.allow)
)

View File

@ -0,0 +1,73 @@
import json
from unittest.mock import patch
from nanobot.config.schema import Config
from nanobot.extensions import build_extension_catalog
def _write_extension(root, extension_id: str) -> None:
package = root / extension_id
package.mkdir(parents=True)
(package / "nanobot.extension.json").write_text(
json.dumps(
{
"id": extension_id,
"name": extension_id,
"version": "1.0.0",
"runtime": "python",
"contributions": [{"kind": "tool", "name": f"{extension_id}_tool"}],
}
)
)
def _catalog(config: Config, user_root):
with (
patch("nanobot.channels.registry.discover_plugins", return_value={}),
patch("nanobot.providers.registry.PROVIDERS", ()),
patch("nanobot.audio.transcription_registry.TRANSCRIPTION_PROVIDERS", ()),
patch(
"nanobot.providers.image_generation.image_gen_provider_names",
return_value=(),
),
):
return build_extension_catalog(config, user_root=user_root)
def test_installed_extension_requires_explicit_trust(tmp_path) -> None:
_write_extension(tmp_path, "acme")
catalog = _catalog(Config(), tmp_path)
assert [item.manifest.id for item in catalog.candidates] == ["acme"]
assert catalog.snapshot.extensions == ()
def test_entry_config_trust_activates_installed_extension(tmp_path) -> None:
_write_extension(tmp_path, "acme")
config = Config.model_validate(
{
"extensions": {
"entries": {
"acme": {
"enabled": True,
"trusted": True,
}
}
}
}
)
catalog = _catalog(config, tmp_path)
assert [item.manifest.id for item in catalog.snapshot.extensions] == ["acme"]
assert catalog.snapshot.contributions[0].contribution.name == "acme_tool"
def test_extensions_disabled_keeps_native_catalog_only(tmp_path) -> None:
_write_extension(tmp_path, "acme")
config = Config.model_validate({"extensions": {"enabled": False}})
catalog = _catalog(config, tmp_path)
assert all(item.manifest.id != "acme" for item in catalog.candidates)

View File

@ -0,0 +1,96 @@
from unittest.mock import patch
from nanobot.agent.skills import SkillsLoader
from nanobot.agent.tools.base import Tool
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.command.router import CommandRouter
from nanobot.config.schema import Config, MCPServerConfig
from nanobot.extensions import ContributionKind, discover_native_extensions
class _Tool(Tool):
@property
def name(self) -> str:
return "acme_tool"
@property
def description(self) -> str:
return "Acme tool"
@property
def parameters(self) -> dict:
return {"type": "object"}
async def execute(self, **kwargs):
return kwargs
def test_native_inventory_preserves_runtime_ownership(tmp_path) -> None:
tools = ToolRegistry()
tools.register(_Tool(), owner="acme.extension")
commands = CommandRouter()
async def _handler(_ctx):
return None
commands.exact("/acme", _handler, owner="acme.extension")
config = Config()
config.tools.mcp_servers["docs"] = MCPServerConfig(url="https://example.com")
with (
patch("nanobot.channels.registry.discover_plugins", return_value={}),
patch("nanobot.providers.registry.PROVIDERS", ()),
patch("nanobot.audio.transcription_registry.TRANSCRIPTION_PROVIDERS", ()),
patch(
"nanobot.providers.image_generation.image_gen_provider_names",
return_value=(),
),
):
result = discover_native_extensions(
config,
skills=SkillsLoader(tmp_path, builtin_skills_dir=tmp_path / "missing"),
tools=tools,
commands=commands,
)
extensions = {item.manifest.id: item for item in result.candidates}
assert {
contribution.kind
for contribution in extensions["acme.extension"].manifest.contributions
} == {ContributionKind.TOOL, ContributionKind.COMMAND}
assert "nanobot.mcp.docs" in extensions
def test_native_inventory_projects_workspace_skill(tmp_path) -> None:
skill_dir = tmp_path / "skills" / "release"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
"---\n"
"name: release\n"
"description: Prepare a release.\n"
"metadata:\n"
" nanobot:\n"
" requires:\n"
" bins: [gh]\n"
"---\n"
)
with (
patch("nanobot.channels.registry.discover_plugins", return_value={}),
patch("nanobot.providers.registry.PROVIDERS", ()),
patch("nanobot.audio.transcription_registry.TRANSCRIPTION_PROVIDERS", ()),
patch(
"nanobot.providers.image_generation.image_gen_provider_names",
return_value=(),
),
):
result = discover_native_extensions(
Config(),
skills=SkillsLoader(tmp_path, builtin_skills_dir=tmp_path / "missing"),
)
skill = next(
item for item in result.candidates if item.manifest.id == "nanobot.skill.release"
)
assert skill.scope.name == "WORKSPACE"
assert skill.manifest.contributions[0].description == "Prepare a release."

View File

@ -16,6 +16,7 @@ def _candidate(
scope: ExtensionScope,
contribution_name: str = "",
replaces: tuple[str, ...] = (),
trusted: bool = True,
) -> ExtensionCandidate:
contributions = (
ExtensionContribution(
@ -33,6 +34,7 @@ def _candidate(
contributions=contributions,
),
scope=scope,
trusted=trusted,
)
@ -75,6 +77,23 @@ def test_policy_filters_extensions_before_contribution_resolution() -> None:
] == ["allowed_tool"]
def test_untrusted_external_extension_is_visible_to_discovery_but_not_active() -> None:
registry = ExtensionRegistry()
registry.register(
_candidate(
"untrusted",
scope=ExtensionScope.USER,
contribution_name="unsafe_tool",
trusted=False,
)
)
snapshot = registry.snapshot()
assert snapshot.extensions == ()
assert snapshot.contributions == ()
def test_conflicting_contribution_does_not_silently_replace_owner() -> None:
registry = ExtensionRegistry()
registry.register(

View File

@ -108,6 +108,16 @@ def test_suggest_name_updates_after_register_and_unregister() -> None:
assert registry._suggest_name("read-file") == "readFile"
def test_registry_tracks_and_removes_tool_owner() -> None:
registry = ToolRegistry()
registry.register(_FakeTool("custom"), owner="acme.tools")
assert registry.owner("custom") == "acme.tools"
registry.unregister("custom")
assert registry.owner("custom") is None
def test_prepare_call_read_file_rejects_non_object_params_with_actionable_hint() -> None:
registry = ToolRegistry()
registry.register(_FakeTool("read_file"))