mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-08-06 01:18:45 +00:00
139 lines
4.2 KiB
Python
139 lines
4.2 KiB
Python
"""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.preflight import evaluate_dependencies
|
|
from nanobot.extensions.registry import (
|
|
ExtensionCandidate,
|
|
ExtensionDiagnostic,
|
|
ExtensionPolicy,
|
|
ExtensionRegistry,
|
|
ExtensionScope,
|
|
ExtensionSnapshot,
|
|
)
|
|
from nanobot.extensions.store import ExtensionStore
|
|
|
|
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:
|
|
external_root = user_root or Path.home() / ".nanobot" / "extensions"
|
|
discoveries.append(ExtensionStore(external_root).discover())
|
|
discoveries.extend(
|
|
_external_discoveries(config, user_root=external_root)
|
|
)
|
|
|
|
candidates = tuple(
|
|
_apply_entry_config(config, candidate)
|
|
for result in discoveries
|
|
for candidate in result.candidates
|
|
)
|
|
candidates, dependency_diagnostics = evaluate_dependencies(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
|
|
+ dependency_diagnostics
|
|
+ tuple(registry_diagnostics)
|
|
+ snapshot.diagnostics
|
|
)
|
|
return ExtensionCatalog(candidates, snapshot, diagnostics)
|
|
|
|
|
|
def _external_discoveries(
|
|
config: Config,
|
|
*,
|
|
user_root: Path,
|
|
) -> Iterable[ExtensionDiscoveryResult]:
|
|
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,
|
|
granted_permissions=(
|
|
candidate.granted_permissions | frozenset(entry.permissions)
|
|
),
|
|
)
|