ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

feat(plugins): add namespaced config and durable state bridge

Browse Source
This commit is contained in:
Teknium
2026-08-08 12:43:03 -07:00
parent 223e74345e
commit 6bf93c0e38
3 changed files with 707 additions and 259 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/rfcs/plugin-config-state-bridge.md
+95 -259
View File
@@ -1,308 +1,144 @@
# Plugin Config & State Bridge — Design Proposal # Plugin Config & State Bridge
**Branch:** `feat/plugin-config-state-bridge` **Status:** config + state slice implemented by #64227
**Target:** `hermes_cli/plugins.py` (PluginContext), `hermes_cli/config.py`, `cron/scheduler.py`
**Concrete consumer:** kanban-advanced plugin (config overlay, cron provisioning, dashboard)
--- **Original design:** Topher Ross (@thebizfixer), RFC PR #58542
## Summary **Concrete consumer:** kanban-advanced
Today plugins that need to read/write Hermes configuration or manage cron jobs ## Scope
must shell out to CLI commands (`hermes config set`, `hermes cron create`) or
manipulate YAML/config files directly. This is fragile across platform
differences (Windows path separators, MSYS vs native Python), Hermes version
bumps, and concurrent access.
This proposal adds four capabilities to the `PluginContext` API so plugins can This slice adds two native `PluginContext` capabilities:
integrate with Hermes' config and cron systems through stable, typed interfaces
without reaching into core internals.
--- - typed, namespace-jailed settings via `ctx.get_config()` and `ctx.set_config()`;
- atomic, profile-scoped runtime data via `ctx.state`.
## Proposal 1: `ctx.get_config()` / `ctx.set_config()` Config schema registration, config defaults, and the cron facade from the
original RFC remain separate follow-up work. No core model tool is added.
### Current state ## Config API
Plugins read `config.yaml` via `hermes_cli.config.load_config()`, parse it
manually, and write back via direct file manipulation. This bypasses Hermes'
own config manager — no schema validation, no atomic writes, no migration
compatibility.
### Proposed API
```python ```python
class PluginContext: def register(ctx):
def get_config(self, key: str, default: Any = None) -> Any: endpoint = ctx.get_config("api_url", default="https://example.invalid")
"""Read a config value by dotted key (e.g. 'kanban.dispatch_stale_timeout_seconds'). retries = ctx.get_config("retry.attempts", default=3)
Returns the default if the key is unset or the config file is missing. ctx.set_config("api_url", "https://api.example.com")
Reads through Hermes' own config loader so layered configs (env overrides, ctx.set_config("retry.attempts", 5)
profile-specific merges) are respected.
"""
def set_config(self, key: str, value: Any) -> None:
"""Write a config value by dotted key.
Writes through Hermes' config manager — atomic file write, schema
validation for known keys, migration compatibility. Raises ValueError
if the key path is invalid or the value fails schema validation.
"""
``` ```
### Implementation sketch Keys are **relative to the calling plugin**. The example above reads and writes:
```python ```yaml
# hermes_cli/plugins.py — PluginContext additions plugins:
def get_config(self, key: str, default: Any = None) -> Any: entries:
from hermes_cli.config import load_config, cfg_get <effective-plugin-id>:
try: settings:
config = load_config() api_url: https://api.example.com
return cfg_get(config, key, default=default) retry:
except Exception: attempts: 5
return default
def set_config(self, key: str, value: Any) -> None:
# Deferred to #64227 implementation.
# When delivered, writes through the config manager with namespace jail
# enforcement (see § Namespace jail below).
raise NotImplementedError(
"ctx.set_config is pending implementation per #64227"
)
``` ```
> **Note:** `cfg_set` does not exist in the current codebase. The read path `<effective-plugin-id>` is `manifest.key` when present, otherwise
> (`cfg_get` at `hermes_cli/config.py:6627`) is available. `set_config` will be `manifest.name`. `settings` is the canonical namespace chosen after the issue
> delivered by #64227's implementation with the namespace jail described below. discussion in #64227/#67531. For migration safety, reads fall back to the former
`plugins.entries.<id>.config.*` subtree only when the canonical value is absent.
### Concrete use case (kanban-advanced) Writes always target `settings`; they do not rewrite or delete legacy values.
Our `config_overlay.py` currently does:
```python
# Current: fragile YAML manipulation
config = yaml.safe_load(Path(config_path).read_text())
config["kanban"]["dispatch_stale_timeout_seconds"] = 14400
Path(config_path).write_text(yaml.dump(config))
```
With this API:
```python
ctx.set_config("kanban.dispatch_stale_timeout_seconds", 14400)
ctx.set_config("kanban.auto_decompose", False)
```
### Platform safety
On Windows, our config writes hit `re.sub` backslash escape bugs because
paths like `C:\Users\Owner` contain `\U` which Python interprets as a
Unicode escape in replacement strings. Going through Hermes' own config
manager eliminates this class of bug.
### Namespace jail ### Namespace jail
`set_config` keys are strictly namespaced to `plugins.entries.<id>.config.*`. The API does not accept full config paths. A plugin can never use it to inspect
A plugin may not read or write config outside its own entry. Enforcement rules: or change arbitrary Hermes configuration.
- **Key prefix:** all writes must start with `plugins.entries.<plugin_id>.config.` Accepted:
- **Cross-plugin rejection:** writing another plugin's config subtree is rejected
with a logged error
- **Path-traversal rejection:** `../../security.approval_mode` and similar
escape attempts are rejected with a logged error
- **Read allow-list:** `ctx.get_config` can only read from
`plugins.entries.<id>.config.*` plus a small read-only allow-list
(e.g. active profile name, already available via `ctx.profile_name`)
Acceptance criteria (from #64227):
- Fixture plugin round-trips config + state through the bridge on a temp
`HERMES_HOME` (real file I/O, no mocks)
- Namespace-escape attempt (`../../security.*`, cross-plugin key, path
traversal) rejected with a logged error
- `hermes doctor` reports a plugin config value violating its registered schema
---
## Proposal 2: `ctx.register_config_schema()`
### Current state
Plugins with configuration (like kanban-advanced's `kanban-config.yaml`) have
JSON Schemas that are invisible to `hermes doctor`, `hermes config check`, and
the setup wizard. Users only discover configuration errors at runtime.
### Proposed API
```python ```python
class PluginContext: ctx.get_config("endpoint")
def register_config_schema(self, schema: dict) -> None: ctx.set_config("retry.policy", {"attempts": 3})
"""Register a JSON Schema for this plugin's config namespace.
The schema validates keys under plugins.entries.<plugin_id>.config.
After registration, `hermes config check` validates the plugin's
config against this schema, and `hermes setup` can walk plugin
config interactively.
"""
``` ```
### Manifest support (plugin.yaml) Rejected with `ValueError` and a warning log:
```yaml
# plugin.yaml — optional schema declaration
provides_config_schema: schema/kanban-config.schema.json
```
When present, the schema is loaded and registered automatically during plugin
init — no explicit `register_config_schema()` call needed in `__init__.py`.
### Concrete use case (kanban-advanced)
Our kanban-config has 30+ keys with validation rules. Currently validation is
done in our own scripts only. With this, `hermes doctor` would catch
misconfigurations before they cause runtime failures.
---
## Proposal 3: `ctx.cron` — Cron API access
### Current state
Plugins that manage cron jobs (kanban-advanced provisions 5+ crons for
auto_unblock, board_keeper, lifecycle, dashboard keepalive) must shell out to
`hermes cron create/list/remove` via `subprocess.run()`. This is fragile:
- Platform-dependent (bash vs cmd, path resolution)
- No structured error handling
- No idempotency guarantees
- Cannot inspect job state programmatically
### Proposed API
```python ```python
class PluginContext: ctx.get_config("security.approval_mode")
@property ctx.set_config("model.provider", "attacker-proxy")
def cron(self) -> "PluginCronFacade": ctx.set_config("plugins.entries.other.settings.token", "...")
"""Return a facade for managing cron jobs owned by this plugin. ctx.set_config("../../security.approval_mode", "always_allow")
ctx.set_config(r"..\..\model.provider", "attacker-proxy")
All jobs created through this facade are tagged with the plugin's
name, enabling bulk operations (list plugin jobs, remove all on
uninstall).
"""
if self._cron is None:
from hermes_cli.plugins import PluginCronFacade
self._cron = PluginCronFacade(
plugin_id=self.manifest.key or self.manifest.name
)
return self._cron
class PluginCronFacade:
def create(self, name: str, schedule: str, *,
prompt: str | None = None,
script: str | None = None,
deliver: str = "local", skills: list[str] | None = None,
idempotency_key: str | None = None) -> str:
"""Create a cron job. Returns the job ID.
Uses the live ``cron/jobs.py:create_job()`` API (L1039-1234).
Pass ``prompt`` for LLM-driven jobs, ``script`` for shell-script
watchdog jobs (``no_agent=True``). When idempotency_key is set,
returns existing job ID if a match exists.
"""
def list(self, mine: bool = True) -> list[dict]:
"""List cron jobs. When ``mine=True``, returns only jobs tagged
with this plugin's id."""
def remove(self, job_id: str) -> bool:
"""Remove a cron job by ID."""
def get(self, job_id: str) -> dict | None:
"""Get a single job's details."""
def pause(self, job_id: str) -> bool: ...
def resume(self, job_id: str) -> bool: ...
``` ```
### Implementation sketch There is no global read allowlist: `ctx.profile_name` already exposes the only
small host fact requested by the RFC. Settings writes use Hermes'
profile-aware config loader/saver and atomic YAML replacement. The bridge
validates the existing YAML before writing so malformed config is never
silently replaced. Every operation resolves the active context-local
`HERMES_HOME`, so one globally loaded plugin context follows multiplexed
profile turns without crossing profile data.
The `PluginCronFacade` delegates to `cron/jobs.py:create_job()` — the same ## Durable state API
function that powers `hermes cron create`. It adds plugin-namespacing via the
`name` prefix convention (`{plugin_id}:job-name`) and bulk cleanup on uninstall.
### Concrete use case (kanban-advanced) Use state for plugin-owned runtime data such as cursors, dedupe sets, and
Our `kanban_handoff.py` currently does: caches. Do not put those values in user-owned config.
```bash
hermes cron create "30s" --name "auto_unblock" \
--script scripts/auto_unblock.sh --deliver local
```
With this API:
```python ```python
ctx.cron.create( def register(ctx):
name="kanban-advanced:auto_unblock", cursor = ctx.state.get("cursor", default={"page": 0})
schedule="30s", ctx.state.set("cursor", {"page": cursor["page"] + 1})
script="scripts/auto_unblock.sh",
deliver="local",
idempotency_key="kanban-advanced-auto_unblock",
)
``` ```
--- The facade stores one JSON object at:
## Proposal 4: Config defaults in `plugin.yaml` ```text
<HERMES_HOME>/plugin-data/<plugin-data-namespace>/state.json
### Current state
Plugins that need non-default Hermes config values (kanban-advanced needs
`dispatch_stale_timeout_seconds: 14400`, `auto_decompose: false`,
`BLOCK_RECURRENCE_LIMIT: 5`) must apply them via bootstrap scripts. If the
user forgets to run bootstrap, the system runs with unsafe defaults.
### Proposed manifest field
```yaml
# plugin.yaml
provides_config_defaults:
kanban.dispatch_stale_timeout_seconds: 14400
kanban.auto_decompose: false
``` ```
On `hermes plugins install`, Hermes prompts the user with a diff of proposed Portable Agent Plugins use their existing `PLUGIN_DATA` namespace exactly.
config changes. On `hermes plugins update`, new defaults merge in (existing Native and nested plugin ids use the same collision-resistant, Windows-safe
user overrides preserved). The prompt shows: namespace algorithm. `ctx.state.data_dir` exposes the directory and
``` `ctx.state.path` exposes the JSON file when a plugin needs to inspect its own
Plugin 'kanban-advanced' recommends these config changes: location.
kanban.dispatch_stale_timeout_seconds: 900 → 14400 ### State guarantees
kanban.auto_decompose: true → false
Apply? [Y/n] - **Profile isolation:** the data root resolves from the active context-local
``` Hermes home on every operation.
- **Atomic replacement:** state writes use temp-file + `fsync` + `os.replace`.
- **Concurrent updates:** a sibling lock file serializes read-modify-write across
threads and processes (`fcntl` on POSIX, `msvcrt` on Windows).
- **Quota:** the complete serialized state is limited to 10 MiB per plugin. A
rejected update leaves the previous file untouched.
- **Fail closed:** malformed/non-object JSON is reported and never overwritten.
- **Typed values:** values must be JSON-serializable.
### Design constraints State keys are 1–128 characters and may contain letters, numbers, `_`, `-`,
- **Never silently override user config** — always prompt `.`, or `:`. Path separators and `..` are rejected.
- **First-install vs update** — first install applies all defaults; update
only applies NEW keys that don't exist in user config
- **Opt-out per key** — users can add keys to a `plugins.entries.<id>.config_defaults_skip`
list to reject specific defaults permanently
--- ## State vs. config
## Cross-cutting concerns | Data | API | Ownership | Example |
|---|---|---|---|
| User-visible behavior | `ctx.get_config` / `ctx.set_config` | User/plugin settings in `config.yaml` | endpoint, timeout, feature mode |
| Runtime bookkeeping | `ctx.state.get` / `ctx.state.set` | Plugin data under `plugin-data/` | cursor, cache, dedupe ids |
### Why these belong in PluginContext and not as separate CLI commands Both APIs are additive. Existing plugins that perform their own file I/O keep
working, but new plugins should use this bridge for stable profile and Windows
semantics.
1. **Atomicity** — config writes go through the same config manager that ## Verification contract
handles migrations, schema validation, and concurrent access
2. **Platform safety** — bypasses shell-level path issues on Windows
3. **Plugin lifecycle** — cron jobs tagged by plugin can be bulk-removed on
`hermes plugins remove`
4. **No new env vars** — follows the AGENTS.md rule: behavioral settings in
config.yaml, not env vars
### Backward compatibility The implementation is covered with real temporary-Hermes-home tests for:
All four additions are purely additive: - fixture-plugin discovery and config/state round trips;
- Existing plugins that shell out to CLI continue to work - canonical `settings` writes and legacy `config` read fallback;
- New methods are opt-in - direct global, cross-plugin, POSIX traversal, and Windows traversal rejection;
- Config schema registration doesn't affect existing config validation - concurrent settings writes without lost siblings;
- cross-thread and cross-process state updates;
--- - atomic quota rejection and malformed-state/config preservation;
- two-profile isolation after the ambient profile changes;
- Unicode and Windows-style path values.
## Related ## Related
- [#64227](https://github.com/NousResearch/hermes-agent/issues/64227) — upstream implementation issue adopting this RFC as its design basis - [Issue #64227](https://github.com/NousResearch/hermes-agent/issues/64227)
- kanban-advanced planned features: `plugin/data/references/planned-features.md` - [RFC PR #58542](https://github.com/NousResearch/hermes-agent/pull/58542) by Topher Ross
- Hermes AGENTS.md: "The core is a narrow waist; capability lives at the edges" - #67531 — standalone plugin settings namespace discussion
- Discord plugin interface expansion discussion
hermes_cli/plugins.py
+284
View File
@@ -38,15 +38,19 @@ import hashlib
import importlib.metadata import importlib.metadata
import importlib.util import importlib.util
import inspect import inspect
import json
import logging import logging
import os import os
import re
import sys import sys
import threading import threading
import types import types
from contextlib import contextmanager
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Union from typing import Any, Callable, Dict, List, Mapping, Optional, Set, Union
from hermes_constants import get_hermes_home from hermes_constants import get_hermes_home
from utils import env_var_enabled, fast_safe_load from utils import env_var_enabled, fast_safe_load
from hermes_cli.config import cfg_get from hermes_cli.config import cfg_get
@@ -365,6 +369,187 @@ class LoadedPlugin:
# PluginContext – handed to each plugin's ``register()`` function # PluginContext – handed to each plugin's ``register()`` function
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
_PLUGIN_SETTING_SEGMENT_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$")
_PLUGIN_SETTING_RESERVED_ROOTS = frozenset({"model", "plugins", "security", "settings"})
_PLUGIN_STATE_KEY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
_PLUGIN_STATE_QUOTA_BYTES = 10 * 1024 * 1024
_PLUGIN_STATE_LOCKS: Dict[str, threading.RLock] = {}
_PLUGIN_STATE_LOCKS_GUARD = threading.Lock()
def _plugin_relative_segments(key: str) -> tuple[str, ...]:
"""Validate and split a plugin-relative settings key.
The public API accepts only relative keys (``endpoint`` or
``retry.policy``). Full Hermes paths, traversal syntax, and the security-
sensitive core roots called out in #64227 are rejected before any config
read occurs.
"""
if not isinstance(key, str):
raise ValueError("Expected a plugin-relative config key string")
segments = tuple(key.split("."))
if (
not key
or "/" in key
or "\\" in key
or any(
not _PLUGIN_SETTING_SEGMENT_RE.fullmatch(segment) for segment in segments
)
or segments[0].lower() in _PLUGIN_SETTING_RESERVED_ROOTS
):
raise ValueError(
"Expected a plugin-relative config key such as 'endpoint' or "
"'retry.policy'; global, cross-plugin, and traversal paths are forbidden"
)
return segments
def _nested_plugin_value(root: object, segments: tuple[str, ...], default: Any) -> Any:
current = root
for segment in segments:
if not isinstance(current, Mapping) or segment not in current:
return default
current = current[segment]
return current
def _nested_plugin_mapping(segments: tuple[str, ...], value: Any) -> dict[str, Any]:
nested: Any = value
for segment in reversed(segments):
nested = {segment: nested}
return nested
def _plugin_data_namespace(plugin_id: str, skill_namespace: str) -> str:
"""Return one Windows-safe directory component for plugin-owned data."""
candidate = skill_namespace or plugin_id
if (
skill_namespace
and candidate.startswith("agent-plugin-")
and re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_-]{0,191}", candidate)
):
# Portable Agent Plugins already receive this exact PLUGIN_DATA path.
return candidate
# Reuse the portable namespace algorithm for native/nested ids too. Its
# fixed prefix avoids Windows reserved device names (CON, NUL, COM1...),
# while the digest prevents collisions after unsafe characters are folded.
return _portable_skill_namespace(candidate)
def _state_thread_lock(path: Path) -> threading.RLock:
key = str(path.resolve(strict=False))
with _PLUGIN_STATE_LOCKS_GUARD:
return _PLUGIN_STATE_LOCKS.setdefault(key, threading.RLock())
@contextmanager
def _locked_plugin_state(path: Path):
"""Serialize state read-modify-write across threads and processes.
``fcntl`` is used on POSIX and ``msvcrt`` on native Windows. The lock is
kept in a sibling file because atomic replacement changes the inode/file
handle of the target itself.
"""
lock_path = path.with_name(f".{path.name}.lock")
thread_lock = _state_thread_lock(lock_path)
with thread_lock:
lock_path.parent.mkdir(parents=True, exist_ok=True)
with open(lock_path, "a+b") as handle:
if os.name == "nt": # pragma: no cover - exercised on Windows CI
import msvcrt
if handle.seek(0, os.SEEK_END) == 0:
handle.write(b"\0")
handle.flush()
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_LOCK, 1)
else:
import fcntl
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
try:
yield
finally:
if os.name == "nt": # pragma: no cover - exercised on Windows CI
handle.seek(0)
msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
class PluginState:
"""Atomic, quota-bounded JSON key/value state owned by one plugin."""
def __init__(self, plugin_id: str, skill_namespace: str = "") -> None:
self._data_namespace = _plugin_data_namespace(plugin_id, skill_namespace)
@property
def data_dir(self) -> Path:
"""Profile-scoped directory matching portable plugins' PLUGIN_DATA."""
return get_hermes_home() / "plugin-data" / self._data_namespace
@property
def path(self) -> Path:
return self.data_dir / "state.json"
@property
def quota_bytes(self) -> int:
return _PLUGIN_STATE_QUOTA_BYTES
@staticmethod
def _validate_key(key: str) -> None:
if (
not isinstance(key, str)
or not _PLUGIN_STATE_KEY_RE.fullmatch(key)
or ".." in key
):
raise ValueError(
"Plugin state keys must be 1-128 characters using letters, "
"numbers, '_', '-', '.', or ':' (without '..')"
)
def _read_unlocked(self) -> dict[str, Any]:
try:
with open(self.path, encoding="utf-8") as handle:
data = json.load(handle)
except FileNotFoundError:
return {}
except (OSError, ValueError) as exc:
raise RuntimeError(f"Cannot parse plugin state {self.path}: {exc}") from exc
if not isinstance(data, dict):
raise RuntimeError(
f"Cannot parse plugin state {self.path}: root must be an object"
)
return data
def get(self, key: str, default: Any = None) -> Any:
"""Read a JSON value, returning *default* when the key is absent."""
self._validate_key(key)
with _locked_plugin_state(self.path):
return self._read_unlocked().get(key, default)
def set(self, key: str, value: Any) -> None:
"""Atomically set one JSON value without dropping concurrent updates."""
self._validate_key(key)
with _locked_plugin_state(self.path):
data = self._read_unlocked()
data[key] = value
try:
encoded = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
except (TypeError, ValueError) as exc:
raise ValueError(
f"Plugin state value for {key!r} is not JSON-serializable"
) from exc
if len(encoded) > self.quota_bytes:
raise ValueError(
f"Plugin state quota exceeded: {len(encoded)} bytes is greater "
f"than the {self.quota_bytes}-byte per-plugin quota"
)
from utils import atomic_json_write
atomic_json_write(self.path, data, mode=0o600)
class PluginContext: class PluginContext:
"""Facade given to plugins so they can register tools and hooks.""" """Facade given to plugins so they can register tools and hooks."""
@@ -374,6 +559,105 @@ class PluginContext:
# Lazy-built host-owned LLM facade — see ctx.llm property below. # Lazy-built host-owned LLM facade — see ctx.llm property below.
self._llm: Any = None self._llm: Any = None
self._subagent_lifecycle: Any = None self._subagent_lifecycle: Any = None
self._state: PluginState | None = None
@property
def plugin_id(self) -> str:
"""Return the effective registry id used for this plugin's namespaces."""
return self.manifest.key or self.manifest.name
# -- namespaced config and durable state --------------------------------
def get_config(self, key: str, default: Any = None) -> Any:
"""Read ``plugins.entries.<plugin_id>.settings.<key>``.
``key`` is always plugin-relative. For migration compatibility, a
missing canonical value falls back to the former ``config`` subtree;
no global config paths are exposed.
"""
try:
segments = _plugin_relative_segments(key)
except ValueError:
logger.warning(
"Rejected config path %r from plugin %s", key, self.plugin_id
)
raise
from hermes_cli.config import load_config_readonly
config = load_config_readonly() or {}
plugins = config.get("plugins") if isinstance(config, Mapping) else None
entries = plugins.get("entries") if isinstance(plugins, Mapping) else None
entry = entries.get(self.plugin_id) if isinstance(entries, Mapping) else None
if not isinstance(entry, Mapping):
return default
missing = object()
value = _nested_plugin_value(entry.get("settings"), segments, missing)
if value is not missing:
return value
return _nested_plugin_value(entry.get("config"), segments, default)
def set_config(self, key: str, value: Any) -> None:
"""Atomically write one value in this plugin's ``settings`` subtree."""
try:
segments = _plugin_relative_segments(key)
except ValueError:
logger.warning(
"Rejected config path %r from plugin %s", key, self.plugin_id
)
raise
from hermes_cli import config as config_mod
if config_mod.is_managed():
raise PermissionError(
"Plugin settings cannot be changed in a managed install"
)
from hermes_cli import managed_scope
dotted_path = ".".join((
"plugins",
"entries",
self.plugin_id,
"settings",
*segments,
))
if managed_scope.is_key_managed(dotted_path):
raise PermissionError(
f"Plugin setting {dotted_path!r} is administrator-managed"
)
partial = {
"plugins": {
"entries": {
self.plugin_id: {
"settings": _nested_plugin_mapping(segments, value),
}
}
}
}
full_path = ("plugins", "entries", self.plugin_id, "settings", *segments)
# The lock covers the merge read plus atomic save, preventing sibling
# plugin writes from racing between those two steps.
# Serialize bridge-to-bridge writes across processes as well as
# threads. Other Hermes config writers still retain their existing
# atomic-replace semantics; this lock specifically prevents two
# plugin read/merge/write transactions from dropping siblings.
with _locked_plugin_state(config_mod.get_config_path()):
with config_mod._CONFIG_LOCK:
# Fail closed on malformed YAML. save_config's raw-cache reader
# intentionally degrades parse failures to {}, which is safe for
# reads but destructive for read-modify-write.
config_mod.read_user_config_raw()
config_mod.save_config(
partial,
preserve_keys={full_path},
merge_existing=True,
)
@property
def state(self) -> PluginState:
"""Return this plugin's profile-scoped durable JSON state facade."""
if self._state is None:
self._state = PluginState(self.plugin_id, self.manifest.skill_namespace)
return self._state
# -- host-owned LLM access ---------------------------------------------- # -- host-owned LLM access ----------------------------------------------
tests/hermes_cli/test_plugin_config_state_bridge.py
+328
View File
@@ -0,0 +1,328 @@
"""End-to-end coverage for the profile-scoped plugin config/state bridge (#64227)."""
from __future__ import annotations
import json
import os
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import pytest
import yaml
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
def _context(
*, name: str = "fixture-plugin", key: str = "", namespace: str = ""
) -> PluginContext:
return PluginContext(
PluginManifest(name=name, key=key, skill_namespace=namespace),
PluginManager(),
)
def _in_home(home: Path, fn, *args):
token = set_hermes_home_override(home)
try:
return fn(*args)
finally:
reset_hermes_home_override(token)
@pytest.fixture
def isolated_home(tmp_path: Path):
home = tmp_path / "profile"
token = set_hermes_home_override(home)
try:
yield home
finally:
reset_hermes_home_override(token)
def test_config_round_trip_uses_canonical_settings_namespace(
isolated_home: Path,
) -> None:
ctx = _context(key="category/fixture-plugin")
assert ctx.get_config("api_url", default="unset") == "unset"
ctx.set_config("api_url", r"C:\Users\Owner\Hermes 🚀")
ctx.set_config("retry.policy", {"attempts": 3, "enabled": True})
raw = yaml.safe_load((isolated_home / "config.yaml").read_text(encoding="utf-8"))
settings = raw["plugins"]["entries"]["category/fixture-plugin"]["settings"]
assert settings == {
"api_url": r"C:\Users\Owner\Hermes 🚀",
"retry": {"policy": {"attempts": 3, "enabled": True}},
}
assert ctx.get_config("api_url") == r"C:\Users\Owner\Hermes 🚀"
assert ctx.get_config("retry.policy") == {"attempts": 3, "enabled": True}
def test_config_reads_legacy_config_namespace_until_canonical_value_is_set(
isolated_home: Path,
) -> None:
path = isolated_home / "config.yaml"
path.parent.mkdir(parents=True)
path.write_text(
yaml.safe_dump({
"plugins": {
"entries": {
"fixture-plugin": {
"config": {"endpoint": "legacy", "legacy_only": 7},
"settings": {"endpoint": "canonical"},
}
}
}
}),
encoding="utf-8",
)
ctx = _context()
assert ctx.get_config("endpoint") == "canonical"
assert ctx.get_config("legacy_only") == 7
ctx.set_config("legacy_only", 8)
assert ctx.get_config("legacy_only") == 8
raw = yaml.safe_load(path.read_text(encoding="utf-8"))
entry = raw["plugins"]["entries"]["fixture-plugin"]
assert entry["config"]["legacy_only"] == 7
assert entry["settings"]["legacy_only"] == 8
@pytest.mark.parametrize(
"key",
[
"../../security.approval_mode",
r"..\..\model.provider",
"security.approval_mode",
"model.provider",
"plugins.entries.other.settings.token",
"settings.endpoint",
"",
],
)
def test_config_rejects_global_cross_plugin_and_traversal_paths(
isolated_home: Path, key: str, caplog: pytest.LogCaptureFixture
) -> None:
ctx = _context()
with pytest.raises(ValueError, match="plugin-relative config key"):
ctx.set_config(key, "blocked")
assert not (isolated_home / "config.yaml").exists()
assert "Rejected config path" in caplog.text
def test_config_read_rejects_escape_instead_of_exposing_global_config(
isolated_home: Path,
) -> None:
path = isolated_home / "config.yaml"
path.parent.mkdir(parents=True)
path.write_text("security:\n approval_mode: always_allow\n", encoding="utf-8")
ctx = _context()
with pytest.raises(ValueError, match="plugin-relative config key"):
ctx.get_config("security.approval_mode")
def test_config_parse_failure_is_not_silently_overwritten(isolated_home: Path) -> None:
path = isolated_home / "config.yaml"
path.parent.mkdir(parents=True)
broken = "plugins:\n entries: [unterminated\n"
path.write_text(broken, encoding="utf-8")
with pytest.raises(Exception, match="while parsing|expected"):
_context().set_config("endpoint", "safe")
assert path.read_text(encoding="utf-8") == broken
def test_concurrent_config_writes_do_not_drop_sibling_settings(
isolated_home: Path,
) -> None:
ctx = _context()
with ThreadPoolExecutor(max_workers=8) as pool:
list(
pool.map(
lambda i: _in_home(isolated_home, ctx.set_config, f"worker_{i}", i),
range(24),
)
)
assert {f"worker_{i}": ctx.get_config(f"worker_{i}") for i in range(24)} == {
f"worker_{i}": i for i in range(24)
}
def test_config_cross_process_lock_preserves_every_setting(isolated_home: Path) -> None:
script = """
import sys
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
ctx = PluginContext(PluginManifest(name='fixture-plugin'), PluginManager())
for i in range(int(sys.argv[1]), int(sys.argv[2])):
ctx.set_config(f'process_{i}', i)
"""
env = dict(os.environ, HERMES_HOME=str(isolated_home))
processes = [
subprocess.Popen(
[sys.executable, "-c", script, str(start), str(start + 20)],
cwd=Path(__file__).resolve().parents[2],
env=env,
)
for start in (0, 20)
]
assert [process.wait(timeout=30) for process in processes] == [0, 0]
ctx = _context()
assert {f"process_{i}": ctx.get_config(f"process_{i}") for i in range(40)} == {
f"process_{i}": i for i in range(40)
}
def test_state_round_trip_is_atomic_and_aligned_with_plugin_data(
isolated_home: Path,
) -> None:
ctx = _context(namespace="agent-plugin-fixture-a1b2c3d4")
assert ctx.state.get("cursor", default="start") == "start"
ctx.state.set("cursor", {"page": 2, "path": r"C:\Users\Owner"})
state_path = (
isolated_home / "plugin-data" / "agent-plugin-fixture-a1b2c3d4" / "state.json"
)
assert ctx.state.data_dir == state_path.parent
assert json.loads(state_path.read_text(encoding="utf-8")) == {
"cursor": {"page": 2, "path": r"C:\Users\Owner"}
}
assert ctx.state.get("cursor") == {"page": 2, "path": r"C:\Users\Owner"}
assert not list(state_path.parent.glob("*.tmp"))
def test_native_state_namespace_is_windows_safe_and_cannot_traverse(
isolated_home: Path,
) -> None:
contexts = [_context(name="CON"), _context(key="../../other-plugin")]
for index, ctx in enumerate(contexts):
ctx.state.set("value", index)
assert ctx.state.data_dir.parent == isolated_home / "plugin-data"
assert ctx.state.data_dir.name.startswith("agent-plugin-")
assert ctx.state.data_dir.name.upper() not in {"CON", "NUL", "COM1", "LPT1"}
def test_concurrent_state_updates_do_not_drop_keys(isolated_home: Path) -> None:
ctx = _context()
with ThreadPoolExecutor(max_workers=8) as pool:
list(
pool.map(
lambda i: _in_home(isolated_home, ctx.state.set, f"cursor_{i}", i),
range(40),
)
)
state = json.loads(ctx.state.path.read_text(encoding="utf-8"))
assert state == {f"cursor_{i}": i for i in range(40)}
def test_state_cross_process_lock_preserves_every_update(isolated_home: Path) -> None:
script = """
import sys
from hermes_cli.plugins import PluginContext, PluginManager, PluginManifest
ctx = PluginContext(PluginManifest(name='fixture-plugin'), PluginManager())
for i in range(int(sys.argv[1]), int(sys.argv[2])):
ctx.state.set(f'process_{i}', i)
"""
env = dict(os.environ, HERMES_HOME=str(isolated_home))
processes = [
subprocess.Popen(
[sys.executable, "-c", script, str(start), str(start + 20)],
cwd=Path(__file__).resolve().parents[2],
env=env,
)
for start in (0, 20)
]
assert [process.wait(timeout=30) for process in processes] == [0, 0]
state = json.loads(_context().state.path.read_text(encoding="utf-8"))
assert state == {f"process_{i}": i for i in range(40)}
def test_state_quota_failure_preserves_previous_file(isolated_home: Path) -> None:
ctx = _context()
ctx.state.set("cursor", "safe")
before = ctx.state.path.read_bytes()
with pytest.raises(ValueError, match="quota"):
ctx.state.set("oversized", "x" * (ctx.state.quota_bytes + 1))
assert ctx.state.path.read_bytes() == before
assert ctx.state.get("cursor") == "safe"
def test_corrupt_state_is_not_silently_overwritten(isolated_home: Path) -> None:
ctx = _context()
ctx.state.data_dir.mkdir(parents=True)
ctx.state.path.write_text('{"cursor":', encoding="utf-8")
with pytest.raises(RuntimeError, match="Cannot parse plugin state"):
ctx.state.set("cursor", "replacement")
assert ctx.state.path.read_text(encoding="utf-8") == '{"cursor":'
def test_config_and_state_follow_context_local_profile_scope(tmp_path: Path) -> None:
homes = [tmp_path / "profiles" / "alpha", tmp_path / "profiles" / "beta"]
ctx = _context(namespace="fixture-plugin-data")
for index, home in enumerate(homes):
home.mkdir(parents=True)
token = set_hermes_home_override(home)
try:
ctx.set_config("profile_value", index)
ctx.state.set("profile_value", index)
finally:
reset_hermes_home_override(token)
# One globally-loaded plugin context follows each multiplexed turn's
# context-local profile instead of pinning the startup profile.
for index, home in enumerate(homes):
token = set_hermes_home_override(home)
try:
assert ctx.get_config("profile_value") == index
assert ctx.state.get("profile_value") == index
finally:
reset_hermes_home_override(token)
def test_fixture_plugin_round_trips_bridge_during_real_discovery(
isolated_home: Path,
) -> None:
plugin_dir = isolated_home / "plugins" / "bridge-fixture"
plugin_dir.mkdir(parents=True)
(plugin_dir / "plugin.yaml").write_text(
"name: bridge-fixture\nversion: 1.0.0\n", encoding="utf-8"
)
(plugin_dir / "__init__.py").write_text(
"def register(ctx):\n"
" old = ctx.get_config('loads', default=0)\n"
" ctx.set_config('loads', old + 1)\n"
" ctx.state.set('registered', {'profile': ctx.profile_name})\n",
encoding="utf-8",
)
(isolated_home / "config.yaml").write_text(
"plugins:\n enabled:\n - bridge-fixture\n", encoding="utf-8"
)
manager = PluginManager()
manager.discover_and_load()
assert manager._plugins["bridge-fixture"].enabled is True
ctx = _context(name="bridge-fixture")
assert ctx.get_config("loads") == 1
assert ctx.state.get("registered")["profile"] in {"custom", "default"}