4094ab610d
* feat(platform): resolver core with locate/inspect/probe tiers and ordered candidates Every resource lookup needs one result shape and one cost contract. `locate` reads metadata only, `inspect` may open files and call OS APIs in-process, `probe` is fresh and the only tier that may spawn or connect. `Resolution.candidates` keeps probe order so fan-out consumers can try every present binary. Linear NS-921. * feat(platform): AppResolver over AppDef with plist, PE, registry, and server.json sources Desktop apps need presence, version, and liveness as separate observations. The runtime file's bearer token is parsed, used for one request, and discarded inside the probe; no public type carries it. Endpoints are accepted only when loopback with a numeric port. * refactor(copilot): gh candidates through locate_command and the Homebrew table First consumer of the resolver. The gh token probe still tries every present binary in order; the allowlist loses its two copilot_auth rows. * feat(platform): availability() over an application declaration locate() + inspect() only, never probes; the fail-closed _version in app.py treats a vendor's plist/PE/registry entry as untrusted input. Salvaged from PR #118122; reads any object with requires_app, min_version, app_for(os) — nothing here imports the MCP catalog. * feat(platform): application declarations parsed into AppDef per OS The parser slice of PR #118122's catalog manifest, re-homed as a catalog-free module: whoever owns an MCP server declares the app it fronts per OS and what it needs, and registers it here. Stdlib + hermes_platform.resolver only. register/lookup/clear are the one seam the MCP check_fn and the skill gate both read. * feat(mcp): check_fn honours a registered application declaration _make_check_fn ANDs the declared app's availability into the connection-alive check; with nothing registered for the server the behaviour is the pre-PR3 connection check. Provenance is explicit registration, not endpoint matching. Returns a plain bool: the registry caches bool(fn()). * feat(skills): requires_apps gate through registered declarations Offer-time filter beside environments:; names resolve through hermes_platform.declaration, an unknown name hides the skill (fail closed). The disk snapshot carries requires_apps and the fast path re-evaluates it (snapshot version bumped to 3): app presence is a host fact that changes without SKILL.md changing. * docs: application declarations page The plugin-facing schema reference: app: and requires: blocks, availability() states, and the two gates that read the registry. Registered under Extending > Plugins in the docs sidebar. * test(platform): declaration parser, availability, gates The PR3 app-block tests re-homed off the catalog: fixtures are dicts passed to parse_declaration, the check_fn gate keys on explicit registration (not endpoint matching), and the import-hygiene probe now covers hermes_platform.declaration and resolver.availability.
95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
"""Whether a catalog entry can be offered on this host, from its `app` and `requires` blocks.
|
|
|
|
`availability` runs `locate` and `inspect` only. It never probes, never spawns, never connects, and holds
|
|
no cache: the callers that need a TTL (the tool registry) already have one.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import re
|
|
from typing import Literal, Protocol
|
|
|
|
from hermes_platform.host import facts
|
|
from hermes_platform.resolver.app import AppDef, AppResolver
|
|
from hermes_platform.resolver.core import CheckState
|
|
|
|
AvailabilityState = Literal[
|
|
"available",
|
|
"installed_not_running",
|
|
"missing_app",
|
|
"version_too_old",
|
|
"unsupported_os",
|
|
"no_requirements",
|
|
]
|
|
|
|
OFFERABLE: frozenset[str] = frozenset({"available", "no_requirements"})
|
|
|
|
|
|
class _HasRequirements(Protocol):
|
|
"""The slice of a catalog entry `availability` reads; keeps this module free of `hermes_cli` imports."""
|
|
|
|
@property
|
|
def requires_app(self) -> bool: ...
|
|
|
|
@property
|
|
def min_version(self) -> str | None: ...
|
|
|
|
def app_for(self, os_family: str) -> AppDef | None: ...
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Availability:
|
|
state: AvailabilityState
|
|
version: str | None = None
|
|
path: str | None = None
|
|
min_version: str | None = None
|
|
|
|
@property
|
|
def offerable(self) -> bool:
|
|
return self.state in OFFERABLE
|
|
|
|
def __bool__(self) -> bool:
|
|
raise TypeError("Availability is not a boolean; read .offerable or .state")
|
|
|
|
def as_dict(self) -> dict:
|
|
return {"state": self.state, "version": self.version, "path": self.path, "min_version": self.min_version}
|
|
|
|
|
|
def version_at_least(found: str, minimum: str) -> bool:
|
|
"""Compare dot-separated decimal components, failing closed on invalid input."""
|
|
|
|
def parts(version: str) -> list[int] | None:
|
|
if not re.fullmatch(r"[0-9]{1,9}(?:\.[0-9]{1,9})*", version):
|
|
return None
|
|
return [int(component) for component in version.split(".")]
|
|
|
|
a, b = parts(found), parts(minimum)
|
|
if a is None or b is None:
|
|
return False
|
|
width = max(len(a), len(b))
|
|
a += [0] * (width - len(a))
|
|
b += [0] * (width - len(b))
|
|
return a >= b
|
|
|
|
|
|
def availability(entry: _HasRequirements, *, os_family: str | None = None) -> Availability:
|
|
if not entry.requires_app:
|
|
return Availability("no_requirements")
|
|
osf = os_family or facts.os_family()
|
|
definition = entry.app_for(osf)
|
|
if definition is None:
|
|
return Availability("unsupported_os", min_version=entry.min_version)
|
|
resolver = AppResolver(definition)
|
|
res = resolver.locate()
|
|
looked_at = res.command[0] if res.command else (res.candidates[0].value if res.candidates else None)
|
|
if not res.found:
|
|
return Availability("missing_app", path=looked_at, min_version=entry.min_version)
|
|
version = None
|
|
if entry.min_version or definition.version_kind != "none":
|
|
obs = resolver.inspect(res).version
|
|
version = obs.value if obs.state is CheckState.PRESENT else None
|
|
if entry.min_version and (version is None or not version_at_least(version, entry.min_version)):
|
|
return Availability("version_too_old", version=version, path=looked_at, min_version=entry.min_version)
|
|
return Availability("available", version=version, path=looked_at, min_version=entry.min_version)
|