Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208"""Session-bound Python callable inventory for optional native plugins.
The kernel owner preimports ``PluginNamespace(session)`` as ``plugin``, thenawaits ``plugin.refresh()`` when the native inventory changes. No discovery orI/O runs at import time. Commands and hooks are not Python exports.
Wire contract (over the existing ``rlm.host_request`` transport):* ``plugin.catalog`` receives ``{session}`` and returns ``{session, revision, exports: [{plugin, generation, name, description, mode}]}``.* ``plugin.invoke`` receives ``{session, generation, plugin, export, input}``. Only keyword arguments are accepted; ``input`` is their JSON object, not an args/kwargs envelope. The unwrapped host result is returned unchanged.* Modes are ``unary`` and ``streaming``. Streaming handles explicitly raise ``NotImplementedError``; this bridge never discards updates to fake unary I/O.
Refresh validates a whole catalog before one atomic publication. Older repliescannot overwrite newer published refreshes or revisions. Saved callables retaintheir session and generation; the host remains authoritative for retirement."""
from __future__ import annotations
import jsonimport keywordfrom collections.abc import Awaitable, Callablefrom dataclasses import dataclassfrom types import MappingProxyTypefrom typing import Any, Mapping
__all__ = ["PluginNamespace", "PluginExport"]
Transport = Callable[[str, dict[str, Any]], Awaitable[Any]]
async def _native_request(kind: str, payload: dict[str, Any]) -> Any: from . import host_request
return await host_request(kind, payload)
def _name(value: Any) -> bool: return ( isinstance(value, str) and value.isidentifier() and not value.startswith("_") and not keyword.iskeyword(value) )
def _json_keys(value: Any) -> None: # json.dumps rejects non-JSON values and cycles, but silently coerces some # object keys. Reject those rather than changing a caller's input keys. if isinstance(value, dict): for key, child in value.items(): if not isinstance(key, str): raise TypeError("plugin input object keys must be strings") _json_keys(child) elif isinstance(value, (list, tuple)): for child in value: _json_keys(child)
@dataclass(frozen=True)class PluginExport: """Immutable declared target. Its identity never follows a reload."""
session: str plugin: str generation: str name: str description: str mode: str
@dataclass(frozen=True)class _CallableExport: descriptor: PluginExport _transport: Transport
@property def __doc__(self) -> str: return self.descriptor.description
async def __call__(self, **kwargs: Any) -> Any: if self.descriptor.mode == "streaming": raise NotImplementedError("Python plugin exports do not support streaming yet") # Snapshot input before yielding; the transport only sees structured JSON. encoded = json.dumps(kwargs, ensure_ascii=False, allow_nan=False) _json_keys(kwargs) target = self.descriptor return await self._transport("plugin.invoke", { "session": target.session, "generation": target.generation, "plugin": target.plugin, "export": target.name, "input": json.loads(encoded), })
class _PluginView: def __init__(self, owner: PluginNamespace, name: str) -> None: self._owner = owner self._name = name
def __getattr__(self, name: str) -> _CallableExport: try: return self._owner._exports[self._name][name] except KeyError: raise AttributeError(name) from None
def __dir__(self) -> list[str]: return sorted(self._owner._exports.get(self._name, {}))
class PluginNamespace: """Current exports for one explicitly supplied session, with an injectable transport.
``inventory`` returns immutable descriptors. An absent/disabled plugin has no attribute or inventory entry. A previously saved plugin view follows current membership, but a previously saved callable always invokes its old target. Host/transport exceptions propagate unchanged. No local files are written. """
def __init__(self, session: str, *, transport: Transport = _native_request) -> None: if not isinstance(session, str) or not session: raise ValueError("plugin namespace requires a non-empty session identity") self._session = session self._transport = transport self._exports: Mapping[str, Mapping[str, _CallableExport]] = MappingProxyType({}) self._revision = -1 self._serial = 0 self._published = 0
@property def session(self) -> str: return self._session
@property def revision(self) -> int | None: return self._revision if self._revision >= 0 else None
@property def inventory(self) -> tuple[PluginExport, ...]: return tuple( exports[name].descriptor for plugin, exports in sorted(self._exports.items()) for name in sorted(exports) )
def __getattr__(self, name: str) -> _PluginView: if name not in self._exports: raise AttributeError(name) return _PluginView(self, name)
def __dir__(self) -> list[str]: return sorted({"refresh", "inventory", "session", "revision", *self._exports})
async def refresh(self) -> bool: """Fetch and validate a catalog; return whether this reply was published.
A malformed or wrong-session reply raises ValueError without changing the inventory. An out-of-order reply returns False. Revisions must be nonnegative integers and equal revisions must describe equal inventories. """ self._serial += 1 serial = self._serial catalog = await self._transport("plugin.catalog", {"session": self._session}) if not isinstance(catalog, dict) or catalog.get("session") != self._session: raise ValueError("plugin catalog session mismatch or malformed catalog") revision = catalog.get("revision") entries = catalog.get("exports") if type(revision) is not int or revision < 0 or not isinstance(entries, list): raise ValueError("plugin catalog requires a revision and exports list") replacement: dict[str, dict[str, _CallableExport]] = {} generations: dict[str, str] = {} for entry in entries: if not isinstance(entry, dict): raise ValueError("malformed plugin export descriptor") plugin, name = entry.get("plugin"), entry.get("name") generation = entry.get("generation") description, mode = entry.get("description"), entry.get("mode") if ( not _name(plugin) or not _name(name) or plugin in {"refresh", "inventory", "session", "revision"} or not isinstance(generation, str) or not generation or not isinstance(description, str) or mode not in ("unary", "streaming") or entry.get("session", self._session) != self._session or entry.get("kind", "export") != "export" ): raise ValueError("malformed or wrong-session plugin export descriptor") exports = replacement.setdefault(plugin, {}) if name in exports or generations.get(plugin, generation) != generation: raise ValueError("duplicate export or mixed plugin generations") generations[plugin] = generation exports[name] = _CallableExport(PluginExport( self._session, plugin, generation, name, description, mode, ), self._transport) # There are no await points between validation and publication. if serial < self._published or revision < self._revision: return False current = {p: dict(e) for p, e in self._exports.items()} if revision == self._revision and replacement != current: raise ValueError("plugin catalog changed without a new revision") self._exports = MappingProxyType({ plugin: MappingProxyType(exports) for plugin, exports in replacement.items() }) self._revision = revision self._published = serial return True