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

Merge branch 'simp/r3-16-E' into simp/r3-16

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:34:36 -07:00
parent 90e9e43f13 27287e6c88
commit f8e4064654
4 changed files with 784 additions and 1659 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/active_sessions.py
+142 -261
View File
@@ -1,6 +1,6 @@
"""Cross-process active chat session leases.
The session database records persisted conversations. This module records
The session database records persisted conversations; this module records
currently open chat surfaces, including idle CLI/TUI sessions that have not
written a transcript row yet.
"""
@@ -9,11 +9,12 @@ from __future__ import annotations
import json
import logging
import collections
import math
import os
import time
import uuid
from contextlib import contextmanager
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator, Optional
@@ -31,32 +32,16 @@ def coerce_max_concurrent_sessions(value: Any, key: str = "max_concurrent_sessio
"""Return a positive integer cap, or None when disabled/invalid."""
if value is None:
return None
if isinstance(value, bool):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
)
return None
try:
if isinstance(value, float):
if not value.is_integer():
raise ValueError(value)
parsed = int(value)
elif isinstance(value, str):
parsed = int(value.strip(), 10)
else:
parsed = int(value)
if isinstance(value, bool) or (isinstance(value, float) and not value.is_integer()):
raise ValueError(value)
parsed = int(value.strip(), 10) if isinstance(value, str) else int(value)
except (TypeError, ValueError):
logger.warning(
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)",
key,
value,
"Ignoring invalid %s=%r (expected a positive integer; 0/null disables)", key, value
)
return None
if parsed <= 0:
return None
return parsed
return parsed if parsed > 0 else None
def resolve_max_concurrent_sessions(config: Any) -> Optional[int]:
@@ -88,10 +73,7 @@ def summarize_holders(entries: list[dict[str, Any]]) -> str:
"""Compact "who is holding the slots" phrase, e.g. ``desktop x4, cli``."""
if not entries:
return ""
counts: dict[str, int] = {}
for entry in entries:
surface = str(entry.get("surface") or "unknown")
counts[surface] = counts.get(surface, 0) + 1
counts = collections.Counter(str(e.get("surface") or "unknown") for e in entries)
held = ", ".join(
f"{surface} x{n}" if n > 1 else surface
for surface, n in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))
@@ -103,9 +85,7 @@ def summarize_holders(entries: list[dict[str, Any]]) -> str:
def active_session_limit_message(
active_count: int,
max_sessions: int,
entries: Optional[list[dict[str, Any]]] = None,
active_count: int, max_sessions: int, entries: Optional[list[dict[str, Any]]] = None
) -> str:
# Name the holders: slots are shared across CLI, desktop/TUI and gateway,
# so the rejected surface is usually NOT the one squatting on them.
@@ -117,29 +97,24 @@ def active_session_limit_message(
)
def _registry_home(registry_home: str | Path | None = None) -> Path:
return Path(registry_home) if registry_home is not None else Path(get_hermes_home())
# Machine-readable refusal reasons: the reason is the contract, the message is
# for people. Capacity = "busy, come back later"; ownership = "this session
# has a live owner and writing would interleave with theirs".
# Machine-readable refusal reasons (the reason is the contract, the message is for
# people). Capacity = "busy, come back later"; ownership = "a live owner exists and
# writing would interleave with theirs".
SESSION_NOT_OWNED = "SESSION_NOT_OWNED"
MAX_CONCURRENT_SESSIONS = "MAX_CONCURRENT_SESSIONS"
# Ownership could not be PROVEN (registry unreadable/corrupt). Deliberately
# distinct from SESSION_NOT_OWNED: collapsing "can't tell" into a silent
# go-ahead is the fail-open hole that let two writers share one session.
# Ownership could not be PROVEN (registry unreadable/corrupt). Deliberately distinct
# from SESSION_NOT_OWNED: treating "can't tell" as a go-ahead is the fail-open hole
# that let two writers share one session.
SESSION_COORDINATION_UNAVAILABLE = "SESSION_COORDINATION_UNAVAILABLE"
# Advertised through the gateway. A module constant, not a config flag: it is
# true because try_acquire_active_session checks atomically, so it cannot drift
# out of step with the enforcement without this file changing.
# Advertised through the gateway. A module constant, not a config flag: it holds
# because try_acquire_active_session checks atomically, so it cannot drift from the
# enforcement without this file changing.
PER_SESSION_EXCLUSIVE_SUBMIT = True
class ActiveSessionRefusal(str):
"""A refusal message (``str`` subclass, so existing callers are untouched)
that also carries a machine-readable ``reason``."""
"""Refusal message (a ``str``, so callers are untouched) with a machine-readable ``reason``."""
reason: str
@@ -151,9 +126,8 @@ class ActiveSessionRefusal(str):
def _is_same_writer(entry: dict[str, Any], metadata: Optional[dict[str, Any]]) -> bool:
"""True when an existing lease belongs to the very caller re-acquiring it.
Identity is (pid, live_session_id): pid alone would let two live sessions in
one process steal each other's lease; the live id alone would let another
process with an equal id do the same."""
Identity is (pid, live_session_id): pid alone lets two live sessions in one process
steal each other's lease; the live id alone lets another process with an equal id."""
try:
if int(entry.get("pid") or -1) != os.getpid():
return False
@@ -161,9 +135,7 @@ def _is_same_writer(entry: dict[str, Any], metadata: Optional[dict[str, Any]]) -
return False
existing_live = str((entry.get("metadata") or {}).get("live_session_id") or "")
incoming_live = str((metadata or {}).get("live_session_id") or "")
if not existing_live or not incoming_live:
return False
return existing_live == incoming_live
return bool(existing_live and incoming_live) and existing_live == incoming_live
def session_already_owned_message(session_id: str, entry: dict[str, Any]) -> str:
@@ -178,38 +150,35 @@ def session_already_owned_message(session_id: str, entry: dict[str, Any]) -> str
)
def _state_dir(registry_home: str | Path | None = None) -> Path:
return _registry_home(registry_home) / "runtime"
def _registry_home(registry_home: str | Path | None = None) -> Path:
return Path(registry_home) if registry_home is not None else Path(get_hermes_home())
def _state_path(registry_home: str | Path | None = None) -> Path:
return _state_dir(registry_home) / "active_sessions.json"
return _registry_home(registry_home) / "runtime" / "active_sessions.json"
def _lock_path(registry_home: str | Path | None = None) -> Path:
return _state_dir(registry_home) / "active_sessions.lock"
return _registry_home(registry_home) / "runtime" / "active_sessions.lock"
def _lease_paths(
lease: Optional["ActiveSessionLease"] = None,
registry_home: str | Path | None = None,
lease: Optional["ActiveSessionLease"] = None, registry_home: str | Path | None = None
) -> tuple[Path, Path]:
if lease is not None and lease.state_path is not None and lease.lock_path is not None:
return lease.state_path, lease.lock_path
home = _registry_home(registry_home)
return _state_path(home), _lock_path(home)
return home / "runtime" / "active_sessions.json", home / "runtime" / "active_sessions.lock"
def _flock(fh, *, lock: bool) -> None:
"""Exclusive whole-file lock/unlock on ``fh`` (fcntl on POSIX, msvcrt on Windows)."""
if os.name == "nt":
import msvcrt
fh.seek(0)
msvcrt.locking(fh.fileno(), msvcrt.LK_LOCK if lock else msvcrt.LK_UNLCK, 1)
else:
import fcntl
fcntl.flock(fh.fileno(), fcntl.LOCK_EX if lock else fcntl.LOCK_UN)
@@ -230,19 +199,17 @@ class _FileLock:
return self
def __exit__(self, exc_type, exc, tb):
if self._fh is None:
return
try:
_flock(self._fh, lock=False)
except Exception:
pass
try:
self._fh.close()
finally:
self._fh = None
fh, self._fh = self._fh, None
if fh is not None:
with suppress(Exception):
_flock(fh, lock=False)
fh.close()
def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
def invalid(what: str) -> ActiveSessionRegistryError:
return ActiveSessionRegistryError(f"active session registry {what}: {path}")
try:
with open(path, "r", encoding="utf-8") as fh:
data = json.load(fh)
@@ -250,25 +217,19 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
return []
except Exception as exc:
if strict:
raise ActiveSessionRegistryError(
f"active session registry unreadable: {path}"
) from exc
raise invalid("unreadable") from exc
logger.warning("Ignoring corrupt active session registry at %s", path)
return []
entries = data.get("entries") if isinstance(data, dict) else data
if not isinstance(entries, list):
if strict:
raise ActiveSessionRegistryError(
f"active session registry has invalid shape: {path}"
)
raise invalid("has invalid shape")
return []
valid = [entry for entry in entries if isinstance(entry, dict)]
if not strict:
return valid
if len(valid) != len(entries):
raise ActiveSessionRegistryError(
f"active session registry contains invalid entries: {path}"
)
raise invalid("contains invalid entries")
seen_leases: set[str] = set()
for entry in valid:
lease_id = entry.get("lease_id")
@@ -285,9 +246,7 @@ def _read_entries(path: Path, *, strict: bool = False) -> list[dict[str, Any]]:
(lambda: not _valid_process_start(entry.get("process_start_time")), "an invalid process start time"),
):
if bad():
raise ActiveSessionRegistryError(
f"active session registry contains {what}: {path}"
)
raise invalid(f"contains {what}")
seen_leases.add(lease_id)
return valid
@@ -332,11 +291,10 @@ def _write_entries(path: Path, entries: list[dict[str, Any]]) -> None:
def _process_start_time(pid: int) -> Optional[float]:
# Pair pid with process create_time when psutil can read it, so a recycled
# pid does not keep a stale lease alive indefinitely.
# Pair pid with create_time when psutil can read it, so a recycled pid does not
# keep a stale lease alive indefinitely.
try:
import psutil # type: ignore
return float(psutil.Process(pid).create_time())
except Exception:
return None
@@ -352,23 +310,20 @@ def _optional_float(value: Any) -> Optional[float]:
def _pid_liveness(pid: Any, process_start_time: Any = None, *, lenient: bool = False) -> Optional[bool]:
"""Return True/False for live/dead, or None when liveness is unknowable.
``lenient`` never returns None: an unparseable pid or a failed existence
probe counts as dead, an unreadable current start time as alive.
"""
"""True/False for live/dead, or None when unknowable. ``lenient`` never returns None:
an unparseable pid or failed existence probe counts as dead, an unreadable start as alive."""
unknown_dead = False if lenient else None
try:
pid_int = int(pid)
except (TypeError, ValueError):
return False if lenient else None
pid_int = 0
if pid_int <= 0:
return False if lenient else None
return unknown_dead
try:
from gateway.status import _pid_exists
exists = bool(_pid_exists(pid_int))
except Exception:
return False if lenient else None
return unknown_dead
if not exists:
return False
expected_start = _optional_float(process_start_time)
@@ -380,26 +335,17 @@ def _pid_liveness(pid: Any, process_start_time: Any = None, *, lenient: bool = F
return abs(current_start - expected_start) < 0.001
def _pid_alive(pid: Any, process_start_time: Any = None) -> bool:
return bool(_pid_liveness(pid, process_start_time, lenient=True))
def _prune_dead(
entries: list[dict[str, Any]], *, strict: bool = False
) -> list[dict[str, Any]]:
def _prune_dead(entries: list[dict[str, Any]], *, strict: bool = False) -> list[dict[str, Any]]:
"""Keep entries whose owner is alive; tracked/strict entries must be provably so."""
live: list[dict[str, Any]] = []
for entry in entries:
tracked = bool(entry.get("track_liveness"))
if strict or tracked:
state = _pid_liveness(entry.get("pid"), entry.get("process_start_time"))
if state is None:
raise ActiveSessionRegistryError(
"active session owner liveness is unknown"
)
if state:
live.append(entry)
continue
if _pid_alive(entry.get("pid"), entry.get("process_start_time")):
tracked = strict or bool(entry.get("track_liveness"))
state = _pid_liveness(
entry.get("pid"), entry.get("process_start_time"), lenient=not tracked
)
if state is None:
raise ActiveSessionRegistryError("active session owner liveness is unknown")
if state:
live.append(entry)
return live
@@ -411,9 +357,9 @@ class ActiveSessionLease:
surface: str
enabled: bool = True
released: bool = False
# Pinned at acquisition: a lease acquired under the root HERMES_HOME must
# release against the same registry even when release() runs inside a
# profile-home override, or phantom leases fill the session cap.
# Pinned at acquisition: a lease taken under the root HERMES_HOME must release
# against the same registry even inside a profile-home override, or phantom
# leases fill the session cap.
state_path: Optional[Path] = None
lock_path: Optional[Path] = None
track_liveness: bool = False
@@ -428,8 +374,19 @@ def _clean_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
return {str(k): v for k, v in metadata.items() if isinstance(k, str)}
def _without_lease(entries: list[dict[str, Any]], lease_id: str) -> list[dict[str, Any]]:
return [e for e in entries if str(e.get("lease_id") or "") != lease_id]
def _drop_lease(
state_path: Path, entries: list[dict[str, Any]], lease_id: str
) -> list[dict[str, Any]]:
"""Remove ``lease_id`` from ``entries``, writing the registry only if it was present."""
kept = [e for e in entries if str(e.get("lease_id") or "") != lease_id]
if len(kept) != len(entries):
_write_entries(state_path, kept)
return kept
def _holds_session(entries: list[dict[str, Any]], session_id: str) -> bool:
target = str(session_id or "")
return bool(target) and any(str(e.get("session_id") or "") == target for e in entries)
def _read_live_entries(
@@ -437,9 +394,8 @@ def _read_live_entries(
) -> Optional[tuple[list[dict[str, Any]], list[dict[str, Any]]]]:
"""``(raw, pruned)`` from the registry, or None when it is unreadable.
Liveness-tracked callers re-raise instead: they must not proceed on an
unprovable registry. Untracked callers get ``warn`` logged and decide
how to degrade themselves.
Liveness-tracked callers re-raise instead (they must not proceed on an unprovable
registry); untracked callers get ``warn`` logged and decide how to degrade.
"""
try:
raw_entries = _read_entries(state_path, strict=True)
@@ -452,12 +408,8 @@ def _read_live_entries(
def _lease_entry(
*,
lease_id: str,
session_id: str,
surface: str,
metadata: Optional[dict[str, Any]] = None,
track_liveness: bool = False,
*, lease_id: str, session_id: str, surface: str,
metadata: Optional[dict[str, Any]] = None, track_liveness: bool = False,
) -> dict[str, Any]:
now = time.time()
entry: dict[str, Any] = {
@@ -477,23 +429,15 @@ def _lease_entry(
def try_acquire_active_session(
*,
session_id: str,
surface: str,
config: Any,
metadata: Optional[dict[str, Any]] = None,
registry_home: str | Path | None = None,
track_liveness: bool = False,
*, session_id: str, surface: str, config: Any, metadata: Optional[dict[str, Any]] = None,
registry_home: str | Path | None = None, track_liveness: bool = False,
) -> tuple[Optional[ActiveSessionLease], Optional[str]]:
"""Acquire an active-session slot.
"""Acquire an active-session slot: ``(lease, None)`` or ``(None, ActiveSessionRefusal)``.
Per-session exclusivity is CORRECTNESS, enforced unconditionally: at most
one live owner per stored session. ``max_concurrent_sessions`` is resource
POLICY and applies only when configured. ``registry_home`` lets
profile-scoped backends share the owning profile's registry.
Returns ``(lease, None)`` or ``(None, ActiveSessionRefusal)``. Ownership
uncertainty fails CLOSED with ``SESSION_COORDINATION_UNAVAILABLE``.
Per-session exclusivity is CORRECTNESS, enforced unconditionally (at most one live
owner per stored session); ``max_concurrent_sessions`` is resource POLICY, applied
only when configured. ``registry_home`` lets profile-scoped backends share the owning
profile's registry. Ownership uncertainty fails CLOSED (SESSION_COORDINATION_UNAVAILABLE).
"""
max_sessions = resolve_max_concurrent_sessions(config)
lease_id = uuid.uuid4().hex
@@ -503,28 +447,17 @@ def try_acquire_active_session(
# refuses empty session ids): hand back a no-op lease.
if not key and not track_liveness:
return ActiveSessionLease(
lease_id=lease_id,
session_id=key,
surface=str(surface),
enabled=False,
lease_id=lease_id, session_id=key, surface=str(surface), enabled=False
), None
entry = _lease_entry(
lease_id=lease_id,
session_id=key,
surface=str(surface),
metadata=metadata,
lease_id=lease_id, session_id=key, surface=str(surface), metadata=metadata,
track_liveness=track_liveness,
)
state_path, lock_path = _lease_paths(registry_home=registry_home)
lease = ActiveSessionLease(
lease_id=lease_id,
session_id=key,
surface=str(surface),
state_path=state_path,
lock_path=lock_path,
track_liveness=track_liveness,
lease_id=lease_id, session_id=key, surface=str(surface), state_path=state_path,
lock_path=lock_path, track_liveness=track_liveness,
)
with _FileLock(lock_path):
# A capacity cap could degrade open; exclusivity cannot: "could not
@@ -536,11 +469,9 @@ def try_acquire_active_session(
)
if loaded is None:
return None, ActiveSessionRefusal(
(
"Hermes could not read the active-session registry at "
f"{state_path}, so it cannot prove this session has no other "
"live owner. Fix or remove that file and try again."
),
"Hermes could not read the active-session registry at "
f"{state_path}, so it cannot prove this session has no other "
"live owner. Fix or remove that file and try again.",
SESSION_COORDINATION_UNAVAILABLE,
)
raw_entries, entries = loaded
@@ -548,6 +479,11 @@ def try_acquire_active_session(
if pruned:
logger.info("Pruned %d stale active session lease(s)", pruned)
def refuse(message: str, reason: str, log: str, *args) -> tuple[None, ActiveSessionRefusal]:
_write_entries(state_path, entries) # persist the prune even when refusing
logger.info(log, *args)
return None, ActiveSessionRefusal(message, reason)
# Correctness first, under the same lock that just pruned dead owners.
# An empty key is exempt: treating "" as an identity would make every
# unsaved draft exclude every other one.
@@ -555,7 +491,6 @@ def try_acquire_active_session(
for index, existing in enumerate(entries):
if str(existing.get("session_id") or "") != key:
continue
# The same writer is not a second writer: a live session that
# leaked its lease reference would otherwise be fenced out of
# its own session permanently (pruning only removes entries
@@ -564,34 +499,20 @@ def try_acquire_active_session(
entries[index] = entry
_write_entries(state_path, entries)
return lease, None
_write_entries(state_path, entries)
logger.info(
return refuse(
session_already_owned_message(key, existing), SESSION_NOT_OWNED,
"Refused active session %s: already held by pid=%s surface=%s",
key,
existing.get("pid"),
existing.get("surface"),
)
return None, ActiveSessionRefusal(
session_already_owned_message(key, existing),
SESSION_NOT_OWNED,
key, existing.get("pid"), existing.get("surface"),
)
# Capacity second, and only when an operator asked for one.
if max_sessions is not None:
active_count = len(entries)
if active_count >= max_sessions:
_write_entries(state_path, entries)
logger.info(
"Active session limit reached: active=%d max=%d surface=%s",
active_count,
max_sessions,
surface,
)
return None, ActiveSessionRefusal(
active_session_limit_message(active_count, max_sessions, entries),
MAX_CONCURRENT_SESSIONS,
)
if max_sessions is not None and len(entries) >= max_sessions:
return refuse(
active_session_limit_message(len(entries), max_sessions, entries),
MAX_CONCURRENT_SESSIONS,
"Active session limit reached: active=%d max=%d surface=%s",
len(entries), max_sessions, surface,
)
entries.append(entry)
_write_entries(state_path, entries)
@@ -600,7 +521,7 @@ def try_acquire_active_session(
def release_active_session(lease: ActiveSessionLease) -> None:
# Prefer the registry the lease was acquired against: the caller may be
# running under a profile HERMES_HOME override (#85431).
# running under a profile HERMES_HOME override.
state_path, lock_path = _lease_paths(lease)
with _FileLock(lock_path):
if lease.released:
@@ -611,24 +532,16 @@ def release_active_session(lease: ActiveSessionLease) -> None:
"releasing an untracked lease",
)
if loaded is not None:
entries = loaded[1]
kept = _without_lease(entries, lease.lease_id)
if len(kept) != len(entries):
_write_entries(state_path, kept)
_drop_lease(state_path, loaded[1], lease.lease_id)
lease.released = True
def transfer_active_session(
lease: ActiveSessionLease,
*,
session_id: str,
metadata: Optional[dict[str, Any]] = None,
lease: ActiveSessionLease, *, session_id: str, metadata: Optional[dict[str, Any]] = None
) -> bool:
"""Move an existing lease to a new session id without dropping the slot."""
new_session_id = str(session_id or "")
if not new_session_id:
return False
if lease.released:
if not new_session_id or lease.released:
return False
if not lease.enabled:
lease.session_id = new_session_id
@@ -648,40 +561,30 @@ def transfer_active_session(
if loaded is None:
return False
entries = loaded[1]
updated = False
for entry in entries:
if str(entry.get("lease_id") or "") != lease.lease_id:
continue
entry["session_id"] = new_session_id
entry["updated_at"] = time.time()
own = next((e for e in entries if str(e.get("lease_id") or "") == lease.lease_id), None)
if own is not None:
own["session_id"] = new_session_id
own["updated_at"] = time.time()
if metadata:
entry["metadata"] = _clean_metadata(metadata)
updated = True
break
if not updated and lease.track_liveness:
entries.append(
_lease_entry(
lease_id=lease.lease_id,
session_id=new_session_id,
surface=lease.surface,
metadata=metadata,
track_liveness=True,
)
)
updated = True
if updated:
_write_entries(state_path, entries)
lease.session_id = new_session_id
return updated
own["metadata"] = _clean_metadata(metadata)
elif lease.track_liveness:
entries.append(_lease_entry(
lease_id=lease.lease_id, session_id=new_session_id, surface=lease.surface,
metadata=metadata, track_liveness=True,
))
else:
return False
_write_entries(state_path, entries)
lease.session_id = new_session_id
return True
def release_orphaned_leases(live_lease_ids: set[str]) -> int:
"""Drop this process's registry entries that no live session owns.
``_prune_dead`` only reclaims leases of dead processes; a days-long server
never trips it, so a lease whose session skipped teardown is held until
restart. The owning process is the only authority on its own leases —
exact, no heartbeat on the turn path, no staleness threshold.
``_prune_dead`` only reclaims leases of dead processes, so on a days-long server a
lease whose session skipped teardown is held until restart. The owning process is the
only authority on its own leases — exact, no heartbeat on the turn path, no threshold.
"""
pid = os.getpid()
state_path = _state_path()
@@ -698,10 +601,8 @@ def release_orphaned_leases(live_lease_ids: set[str]) -> int:
return 0
entries = loaded[1]
kept = [
entry
for entry in entries
if entry.get("pid") != pid
or str(entry.get("lease_id") or "") in live_lease_ids
entry for entry in entries
if entry.get("pid") != pid or str(entry.get("lease_id") or "") in live_lease_ids
]
dropped = len(entries) - len(kept)
if dropped:
@@ -724,52 +625,32 @@ def active_session_registry_snapshot(
@contextmanager
def active_session_liveness_guard(
session_id: str,
*,
registry_home: str | Path | None = None,
session_id: str, *, registry_home: str | Path | None = None
) -> Iterator[bool]:
"""Hold the registry lock while reporting whether ``session_id`` is leased,
so no new backend can acquire a lease between the check and the caller's
``end_session`` write."""
target = str(session_id or "")
"""Hold the registry lock while reporting whether ``session_id`` is leased, so no
new backend can acquire a lease between the check and the caller's ``end_session``."""
state_path, lock_path = _lease_paths(registry_home=registry_home)
with _FileLock(lock_path):
entries = _prune_dead(_read_entries(state_path, strict=True), strict=True)
_write_entries(state_path, entries)
yield bool(target) and any(
str(entry.get("session_id") or "") == target for entry in entries
)
yield _holds_session(entries, session_id)
@contextmanager
def release_active_session_liveness_guard(
lease: ActiveSessionLease,
session_id: str,
lease: ActiveSessionLease, session_id: str
) -> Iterator[bool]:
"""Remove ``lease`` and hold its registry lock through a lifecycle write,
making cleanup one atomic ownership decision (release, check siblings,
end the durable row) before any new backend can reopen it."""
"""Remove ``lease`` and hold its registry lock through a lifecycle write, making
cleanup one atomic decision (release, check siblings, end the durable row)."""
if not lease.enabled or lease.released:
with active_session_liveness_guard(
session_id, registry_home=_registry_home_for_lease(lease)
) as active:
home = lease.state_path.parent.parent if lease.state_path is not None else None
with active_session_liveness_guard(session_id, registry_home=home) as active:
yield active
return
target = str(session_id or "")
state_path, lock_path = _lease_paths(lease)
with _FileLock(lock_path):
entries = _prune_dead(_read_entries(state_path, strict=True), strict=True)
kept = _without_lease(entries, lease.lease_id)
if len(kept) != len(entries):
_write_entries(state_path, kept)
kept = _drop_lease(state_path, entries, lease.lease_id)
lease.released = True
yield bool(target) and any(
str(entry.get("session_id") or "") == target for entry in kept
)
def _registry_home_for_lease(lease: ActiveSessionLease) -> Path | None:
if lease.state_path is None:
return None
return lease.state_path.parent.parent
yield _holds_session(kept, session_id)
hermes_cli/agent_import.py
+182 -432
View File
@@ -1,8 +1,8 @@
"""hermes import-agent — import Claude Code / Codex CLI setups into Hermes.
Secrets are NEVER imported: credential files (.credentials.json, auth.json) are ignored, and MCP
server env vars with secret-looking names (KEY, TOKEN, SECRET, PASSWORD, ...) are stripped and
reported so the user can re-add them deliberately via ``hermes setup`` or config.yaml.
Secrets are NEVER imported: credential files are never read, and MCP env vars with secret-looking
names (KEY, TOKEN, SECRET, PASSWORD, ...) are stripped and reported so the user re-adds them via
``hermes setup`` or config.yaml.
"""
from __future__ import annotations
@@ -13,42 +13,28 @@ import re
import shutil
import sys
import time
import tomllib
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple
import yaml
from utils import atomic_write_text, atomic_yaml_write
logger = logging.getLogger(__name__)
# Same entry delimiter as the Hermes memory store and the openclaw migration
# script — memories/MEMORY.md entries are separated by bare "§" lines.
# Entry delimiter of the Hermes memory store (memories/MEMORY.md) and the openclaw script.
ENTRY_DELIMITER = "\n§\n"
# Character budget for merged memory files (matches the openclaw script's
# default memory limit).
# Character budget for merged memory files (openclaw script default).
MEMORY_CHAR_LIMIT = 20_000
SUPPORTED_AGENTS = ("claude-code", "codex")
_AGENT_DEFAULT_DIRS = {
"claude-code": ".claude",
"codex": ".codex",
}
_SKILL_CATEGORY = {
"claude-code": "claude-code-imports",
"codex": "codex-imports",
}
_AGENT_DEFAULT_DIRS = {"claude-code": ".claude", "codex": ".codex"}
_SKILL_CATEGORY = {"claude-code": "claude-code-imports", "codex": "codex-imports"}
# Env var names that look like credentials — never copied into config.yaml.
_SECRET_KEY_RE = re.compile(
r"(?:^|_)(?:API[_-]?KEY|APIKEY|TOKEN|SECRET|PASSWORD|PASSWD|CREDENTIALS?|"
r"AUTH|PRIVATE[_-]?KEY|ACCESS[_-]?KEY)(?:_|$)|KEY$",
re.IGNORECASE,
)
# Files inside the source tree that hold credentials — never read.
_CREDENTIAL_FILENAMES = (".credentials.json", "auth.json", "credentials.json")
r"AUTH|PRIVATE[_-]?KEY|ACCESS[_-]?KEY)(?:_|$)|KEY$", re.IGNORECASE)
def is_secret_key(key: str) -> bool:
@@ -65,92 +51,59 @@ def read_text(path: Path) -> str:
class ConfigReadError(RuntimeError):
"""An existing config file is present but cannot be read or parsed.
Signals that a read-modify-write round trip must be abandoned: the caller has no idea what the
file holds, so writing a merged result back would replace real settings with only the keys it
merged.
"""
"""Existing config file present but unreadable/unparseable: the read-modify-write round trip
must be abandoned, else the merged result would replace real settings with only merged keys."""
def load_yaml_file(path: Path) -> Dict[str, Any]:
"""Load a YAML mapping, distinguishing "absent" from "unreadable".
- Absent, or present but empty -> ``{}``; first-time creation still works. - Present but
unreadable, unparseable, or not a mapping -> raise :class:`ConfigReadError` so the caller
refuses and leaves the file byte-identical.
"""
import yaml
"""Load a YAML mapping: absent/empty -> ``{}``; unreadable/unparseable/non-mapping ->
:class:`ConfigReadError` so the caller refuses and leaves the file byte-identical."""
if not path.exists():
return {}
fix_hint = "Fix it with `hermes config edit` (or move it aside), then re-run the import."
def refusal(detail: str) -> ConfigReadError:
return ConfigReadError(f"Refusing to overwrite {path}: {detail}")
try:
raw = read_text(path)
except OSError as exc:
raise ConfigReadError(
f"Refusing to overwrite {path}: the existing file cannot be read "
f"({exc}). Fix the file permissions or move it aside first."
) from exc
raise refusal(f"the existing file cannot be read ({exc}). "
"Fix the file permissions or move it aside first.") from exc
try:
data = yaml.safe_load(raw)
except yaml.YAMLError as exc:
raise ConfigReadError(
f"Refusing to overwrite {path}: the existing file is not valid YAML ({exc}). {fix_hint}"
) from exc
# An empty file parses to None — a legitimate state with nothing to lose.
if data is None:
raise refusal(f"the existing file is not valid YAML ({exc}). {fix_hint}") from exc
if data is None: # empty file — a legitimate state with nothing to lose
return {}
if not isinstance(data, dict):
raise ConfigReadError(
f"Refusing to overwrite {path}: expected the existing file to hold a "
f"YAML mapping but found {type(data).__name__}. {fix_hint}"
)
raise refusal(f"expected the existing file to hold a YAML mapping but found "
f"{type(data).__name__}. {fix_hint}")
return data
def dump_yaml_file(path: Path, data: Dict[str, Any]) -> None:
"""Write ``data`` as YAML atomically (temp file + fsync + rename).
Only ever reached after :func:`load_yaml_file` has successfully read the same path, so the
mapping being written is the real file's content plus the merged section — never a silently-
empty stand-in.
"""
"""Atomic YAML write; only reached after :func:`load_yaml_file` succeeded on the same path."""
atomic_yaml_write(path, data)
# ---------------------------------------------------------------------------
# Memory-entry primitives (ported from openclaw_to_hermes.py)
# ---------------------------------------------------------------------------
def extract_markdown_entries(text: str) -> List[str]:
"""Split a markdown document into individual memory entries.
Headings become context prefixes; bullets and paragraphs become entries; code blocks and
tables are skipped.
"""
"""Split markdown into memory entries: headings become context prefixes; bullets and
paragraphs become entries; code blocks and tables are skipped."""
entries: List[str] = []
headings: List[str] = []
paragraph_lines: List[str] = []
def context_prefix() -> str:
filtered = [
h for h in headings
if h and not re.search(
r"\b(MEMORY|USER|SOUL|AGENTS|TOOLS|IDENTITY|CLAUDE)\.md\b",
h, re.I,
)
]
return " > ".join(filtered)
def add_entry(content: str) -> None:
prefix = context_prefix()
prefix = " > ".join(
h for h in headings
if h and not re.search(r"\b(MEMORY|USER|SOUL|AGENTS|TOOLS|IDENTITY|CLAUDE)\.md\b",
h, re.I))
entries.append(f"{prefix}: {content}" if prefix else content)
def flush_paragraph() -> None:
nonlocal paragraph_lines
block = " ".join(line.strip() for line in paragraph_lines).strip()
paragraph_lines = []
paragraph_lines.clear()
if block:
add_entry(block)
@@ -158,98 +111,54 @@ def extract_markdown_entries(text: str) -> List[str]:
for raw_line in (text or "").splitlines():
line = raw_line.rstrip()
stripped = line.strip()
if stripped.startswith("```"):
in_code_block = not in_code_block
flush_paragraph()
continue
if in_code_block:
continue
heading_match = re.match(r"^(#{1,6})\s+(.*\S)\s*$", stripped)
if heading_match:
flush_paragraph()
level = len(heading_match.group(1))
value = heading_match.group(2).strip()
while len(headings) >= level:
headings.pop()
headings.append(value)
# Drop deeper (and same-level) headings, then push this one.
headings[len(heading_match.group(1)) - 1:] = [heading_match.group(2).strip()]
continue
bullet_match = re.match(r"^\s*(?:[-*]|\d+\.)\s+(.*\S)\s*$", line)
if bullet_match:
flush_paragraph()
add_entry(bullet_match.group(1).strip())
continue
if not stripped:
flush_paragraph()
if not stripped or (stripped.startswith("|") and stripped.endswith("|")):
flush_paragraph() # blank line or table row ends the paragraph
continue
if stripped.startswith("|") and stripped.endswith("|"):
flush_paragraph()
continue
paragraph_lines.append(stripped)
flush_paragraph()
deduped: List[str] = []
seen = set()
for entry in entries:
normalized = normalize_text(entry)
if not normalized or normalized in seen:
continue
seen.add(normalized)
deduped.append(entry.strip())
if normalized and normalized not in seen:
seen.add(normalized)
deduped.append(entry.strip())
return deduped
def parse_existing_memory_entries(path: Path) -> List[str]:
"""Parse the DESTINATION memory store into entries.
``memories/MEMORY.md`` is the entry-delimited store written by ``MemoryStore._write_file``
(tools/memory_tool.py), not a markdown document, so this splits on ``ENTRY_DELIMITER`` only —
exactly what ``MemoryStore._parse_entries`` does.
Do NOT fall back to :func:`extract_markdown_entries` here. That extractor is correct for
CLAUDE.md / AGENTS.md *sources*, but it drops fenced code blocks and table rows and splits a
block into one entry per bullet — and the merged result is written straight back over the user's
store, so the loss is permanent.
"""
"""Parse the DESTINATION memory store (``ENTRY_DELIMITER``-split, as ``MemoryStore`` does).
Do NOT fall back to :func:`extract_markdown_entries`: it drops code blocks/table rows and
splits bullets, and the merged result overwrites the user's store — the loss is permanent."""
if not path.exists():
return []
raw = read_text(path)
if not raw.strip():
return []
return [e.strip() for e in raw.split(ENTRY_DELIMITER) if e.strip()]
return [e.strip() for e in read_text(path).split(ENTRY_DELIMITER) if e.strip()]
def backup_memory_file(path: Path) -> Optional[Path]:
"""Snapshot ``path`` before a destructive rewrite; return the backup path.
Restores parity with the openclaw migration script this module was ported from, which calls
``maybe_backup(destination)`` before rewriting a memory store. Uses the same
``<name>.bak.<unix_ts>`` naming as ``MemoryStore._backup_drifted_file``. Returns None when there
is nothing to back up.
"""
if not path.exists():
return None
backup = path.with_suffix(path.suffix + f".bak.{int(time.time())}")
shutil.copy2(path, backup)
return backup
def merge_entries(
existing: Sequence[str],
incoming: Sequence[str],
limit: int,
) -> Tuple[List[str], Dict[str, int]]:
def merge_entries(existing: Sequence[str], incoming: Sequence[str],
limit: int) -> Tuple[List[str], Dict[str, int]]:
merged = list(existing)
seen = {normalize_text(e) for e in existing if e.strip()}
stats = {"existing": len(existing), "added": 0, "duplicates": 0, "overflowed": 0}
current_len = len(ENTRY_DELIMITER.join(merged)) if merged else 0
current_len = len(ENTRY_DELIMITER.join(merged))
for entry in incoming:
normalized = normalize_text(entry)
if not normalized:
@@ -257,10 +166,8 @@ def merge_entries(
if normalized in seen:
stats["duplicates"] += 1
continue
candidate_len = (
len(entry) if not merged
else current_len + len(ENTRY_DELIMITER) + len(entry)
)
candidate_len = (len(entry) if not merged
else current_len + len(ENTRY_DELIMITER) + len(entry))
if candidate_len > limit:
stats["overflowed"] += 1
continue
@@ -271,43 +178,22 @@ def merge_entries(
return merged, stats
# ---------------------------------------------------------------------------
# Claude Code permission rules → Hermes command patterns
# ---------------------------------------------------------------------------
_BASH_RULE_RE = re.compile(r"^Bash\((?P<inner>.*)\)$")
def claude_rule_to_command_pattern(rule: str) -> Optional[str]:
"""Convert a Claude Code ``Bash(...)`` permission rule into a Hermes glob.
``Bash(npm run test:*)`` -> ``npm run test*`` (Claude ':*' is a prefix match). Bare ``Bash``
and non-Bash rules (``Read(...)``, ``WebFetch(...)``) return None: the former is too broad to
import, the latter gate Claude-specific tools with no command-allowlist equivalent.
"""
rule = (rule or "").strip()
m = _BASH_RULE_RE.match(rule)
if not m:
return None
inner = m.group("inner").strip()
"""``Bash(npm run test:*)`` -> ``npm run test*`` (Claude ':*' is a prefix match). Bare ``Bash``
(too broad) and non-Bash rules (Claude-only tools, no allowlist equivalent) return None."""
m = _BASH_RULE_RE.match((rule or "").strip())
inner = m.group("inner").strip() if m else ""
if not inner:
return None
if inner.endswith(":*"):
inner = inner[:-2] + "*"
return inner
# ---------------------------------------------------------------------------
# Detection
# ---------------------------------------------------------------------------
def default_source_dir(agent: str) -> Path:
return Path.home() / _AGENT_DEFAULT_DIRS[agent]
return inner[:-2] + "*" if inner.endswith(":*") else inner
def detect_agents() -> List[str]:
"""Return the list of supported agents whose default dirs exist."""
return [a for a in SUPPORTED_AGENTS if default_source_dir(a).is_dir()]
return [a for a in SUPPORTED_AGENTS if (Path.home() / _AGENT_DEFAULT_DIRS[a]).is_dir()]
def sanitize_mcp_env(env: Any) -> Tuple[Dict[str, str], List[str]]:
@@ -318,13 +204,6 @@ def sanitize_mcp_env(env: Any) -> Tuple[Dict[str, str], List[str]]:
return kept, [str(k) for k in env if str(k) not in kept]
def _copy_skill_dir(skill_dir: Path, destination: Path) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
if destination.exists():
shutil.rmtree(destination)
shutil.copytree(skill_dir, destination)
def _translate_mcp_server(name: str, srv: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
"""Map one Claude/Codex MCP server entry to Hermes shape; returns (server, stripped secret paths)."""
hermes_srv: Dict[str, Any] = {}
@@ -343,35 +222,20 @@ def _translate_mcp_server(name: str, srv: Dict[str, Any]) -> Tuple[Dict[str, Any
hermes_srv["url"] = srv["url"]
headers = srv.get("headers")
if isinstance(headers, dict):
kept_headers = {
k: v for k, v in headers.items()
if not is_secret_key(str(k)) and "authorization" not in str(k).lower()
}
kept_headers = {k: v for k, v in headers.items()
if not is_secret_key(str(k)) and "authorization" not in str(k).lower()}
if kept_headers:
hermes_srv["headers"] = kept_headers
stripped.extend(f"mcp_servers.{name}.headers.{k}" for k in headers if k not in kept_headers)
return hermes_srv, stripped
# ---------------------------------------------------------------------------
# Importer
# ---------------------------------------------------------------------------
class AgentImporter:
"""Detect/parse/map/apply importer for a single agent source tree.
"""Detect/parse/map/apply importer for one agent source tree. ``execute=False`` runs the full
plan without touching disk; every item is recorded as imported/skipped/conflict/error."""
``execute=False`` runs the full plan without touching disk (dry run). Every item is recorded
as imported/skipped/conflict/error with a reason so the CLI can print a per-item report.
"""
def __init__(
self,
agent: str,
source_root: Path,
target_root: Path,
execute: bool = False,
overwrite: bool = False,
) -> None:
def __init__(self, agent: str, source_root: Path, target_root: Path,
execute: bool = False, overwrite: bool = False) -> None:
if agent not in SUPPORTED_AGENTS:
raise ValueError(f"Unsupported agent: {agent!r}")
self.agent = agent
@@ -382,27 +246,16 @@ class AgentImporter:
self.items: List[Dict[str, Any]] = []
self.stripped_secrets: List[str] = []
# -- reporting ---------------------------------------------------------
def record(self, kind: str, source, destination, status: str,
reason: str = "", **details) -> None:
item: Dict[str, Any] = {
"kind": kind,
"source": str(source) if source else None,
"destination": str(destination) if destination else None,
"status": status,
"reason": reason,
}
item.update(details)
self.items.append(item)
self.items.append({"kind": kind, "source": str(source) if source else None,
"destination": str(destination) if destination else None,
"status": status, "reason": reason, **details})
def load_target_config(self, kind: str, source, destination: Path
) -> Optional[Dict[str, Any]]:
"""Read the destination config.yaml, or record a refusal and return None.
Deliberately runs in dry-run too: ``--dry-run`` must report the refusal, not preview an
``imported`` that would destroy the config.
"""
def load_target_config(self, kind: str, source,
destination: Path) -> Optional[Dict[str, Any]]:
"""Read the destination config.yaml, or record a refusal and return None. Runs in dry-run
too: ``--dry-run`` must report the refusal, not preview an ``imported`` that destroys it."""
try:
return load_yaml_file(destination)
except ConfigReadError as exc:
@@ -412,88 +265,74 @@ class AgentImporter:
def apply(self, kind: str, source, destination, would: str, action,
details: Optional[Dict[str, Any]] = None) -> None:
"""Record ``imported`` (reason ``would`` in dry-run); in execute mode run ``action`` first.
``action`` may add keys to ``details`` (shared by reference) and returns an error string
to record ``error`` instead, or None on success.
"""
(recorded as ``error``) or None."""
details = details or {}
if not self.execute:
self.record(kind, source, destination, "imported", would, **details)
return
error = action()
self.record(kind, source, destination, "error" if error else "imported", error or "", **details)
error = action() if self.execute else None
self.record(kind, source, destination, "error" if error else "imported",
error or ("" if self.execute else would), **details)
def build_report(self) -> Dict[str, Any]:
summary = {"imported": 0, "skipped": 0, "conflict": 0, "error": 0}
for item in self.items:
summary[item["status"]] = summary.get(item["status"], 0) + 1
report: Dict[str, Any] = {
"agent": self.agent,
"source": str(self.source_root),
"target": str(self.target_root),
"dry_run": not self.execute,
"items": self.items,
"summary": summary,
}
report: Dict[str, Any] = {"agent": self.agent, "source": str(self.source_root),
"target": str(self.target_root), "dry_run": not self.execute,
"items": self.items, "summary": summary}
if self.stripped_secrets:
report["stripped_secrets"] = sorted(set(self.stripped_secrets))
return report
# -- orchestration -----------------------------------------------------
def run(self) -> Dict[str, Any]:
if not self.source_root.is_dir():
self.record("source", self.source_root, None, "error",
"Source directory does not exist")
return self.build_report()
{"claude-code": self._run_claude_code, "codex": self._run_codex}[self.agent]()
else:
{"claude-code": self._run_claude_code, "codex": self._run_codex}[self.agent]()
return self.build_report()
def _run_claude_code(self) -> None:
settings = self._load_claude_settings()
self.import_context_file(self.source_root / "CLAUDE.md", kind="claude-md")
self.import_permission_allowlist(settings)
self.import_permission_denylist(settings)
self.import_mcp_servers(self._claude_mcp_servers(settings), kind="mcp-servers")
settings = self._load_source_mapping(
"settings", self.source_root / "settings.json", json.loads, self._JSON_ERRORS,
record_missing=True, non_mapping_error="settings.json is not a JSON object")
self.import_context_file(self.source_root / "CLAUDE.md", "claude-md")
self._import_permission_rules(settings, "allow")
self._import_permission_rules(settings, "deny")
# mcpServers: ~/.claude.json (preferred; lives NEXT TO ~/.claude/) then settings.json
claude_json = self._load_source_mapping(
"mcp-servers", self.source_root.parent / ".claude.json", json.loads, self._JSON_ERRORS)
servers: Dict[str, Any] = {}
for source in (claude_json.get("mcpServers"), settings.get("mcpServers")):
if isinstance(source, dict):
for name, srv in source.items():
servers.setdefault(name, srv)
self.import_mcp_servers(servers, kind="mcp-servers")
self.import_skills(self.source_root / "skills")
commands_dir = self.source_root / "commands"
if commands_dir.is_dir() and any(commands_dir.glob("*.md")):
self.record(
"slash-commands", commands_dir, None, "skipped",
"Claude slash commands have no direct Hermes equivalent — "
"consider converting them into skills",
)
self.record("slash-commands", commands_dir, None, "skipped",
"Claude slash commands have no direct Hermes equivalent — "
"consider converting them into skills")
def _run_codex(self) -> None:
config = self._load_codex_config()
self.import_context_file(self.source_root / "AGENTS.md", kind="agents-md")
config = self._load_source_mapping("config", self.source_root / "config.toml",
tomllib.loads, Exception, record_missing=True)
self.import_context_file(self.source_root / "AGENTS.md", "agents-md")
mcp = config.get("mcp_servers")
self.import_mcp_servers(mcp if isinstance(mcp, dict) else {}, kind="mcp-servers")
self.import_memories_dir(self.source_root / "memories")
memories_dir = self.source_root / "memories"
self._import_markdown_files(
"memories", memories_dir,
sorted(memories_dir.glob("*.md")) if memories_dir.is_dir() else None,
"No memories directory found")
self.import_skills(self.source_root / "skills")
# -- parsers (fail soft: bad files become per-item error records) -------
_JSON_ERRORS = (json.JSONDecodeError, OSError)
def _load_claude_settings(self) -> Dict[str, Any]:
return self._load_source_mapping(
"settings", self.source_root / "settings.json", json.loads, self._JSON_ERRORS,
record_missing=True, non_mapping_error="settings.json is not a JSON object",
)
def _load_codex_config(self) -> Dict[str, Any]:
import tomllib
return self._load_source_mapping(
"config", self.source_root / "config.toml", tomllib.loads, Exception,
record_missing=True,
)
def _load_source_mapping(self, kind: str, path: Path, parse, errors, *,
record_missing: bool = False,
non_mapping_error: str = "") -> Dict[str, Any]:
"""Parse ``path`` into a mapping; problems become per-item error records, result ``{}``."""
"""Parse ``path`` into a mapping; problems become per-item error records and ``{}``."""
if not path.exists():
if record_missing:
self.record(kind, None, None, "skipped", f"No {path.name} found")
@@ -509,46 +348,20 @@ class AgentImporter:
self.record(kind, path, None, "error", non_mapping_error)
return {}
def _claude_mcp_servers(self, settings: Dict[str, Any]) -> Dict[str, Any]:
"""Collect mcpServers from ~/.claude.json (preferred) and settings.json."""
# ~/.claude.json lives NEXT TO ~/.claude/, not inside it
data = self._load_source_mapping(
"mcp-servers", self.source_root.parent / ".claude.json", json.loads, self._JSON_ERRORS)
servers: Dict[str, Any] = {}
if isinstance(data.get("mcpServers"), dict):
servers.update(data["mcpServers"])
from_settings = settings.get("mcpServers")
if isinstance(from_settings, dict):
for name, srv in from_settings.items():
servers.setdefault(name, srv)
return servers
# -- mappers -------------------------------------------------------------
def import_context_file(self, source: Path, kind: str) -> None:
"""CLAUDE.md / AGENTS.md → memory entries in memories/MEMORY.md."""
if not source.exists():
self.record(kind, None, self.target_root / "memories" / "MEMORY.md",
"skipped", f"No {source.name} found")
return
self._import_markdown_files(kind, source, [source], single_file=True)
def import_memories_dir(self, memories_dir: Path) -> None:
"""codex memories/*.md → memory entries in memories/MEMORY.md."""
if not memories_dir.is_dir():
self.record("memories", None, self.target_root / "memories" / "MEMORY.md",
"skipped", "No memories directory found")
return
self._import_markdown_files("memories", memories_dir, sorted(memories_dir.glob("*.md")))
def _import_markdown_files(self, kind: str, source: Path, files: List[Path],
single_file: bool = False) -> None:
"""Extract entries from ``files`` and merge them into memories/MEMORY.md.
self._import_markdown_files(kind, source, [source] if source.exists() else None,
f"No {source.name} found", single_file=True)
def _import_markdown_files(self, kind: str, source: Path, files: Optional[List[Path]],
missing_reason: str, single_file: bool = False) -> None:
"""Extract entries from ``files`` (None = source missing) and merge into memories/MEMORY.md.
An unreadable file records an error; a directory import then still reports "no entries"
when nothing was extracted, while a single-file import stops at the error.
"""
when nothing was extracted, while a single-file import stops at the error."""
destination = self.target_root / "memories" / "MEMORY.md"
if files is None:
self.record(kind, None, destination, "skipped", missing_reason)
return
incoming: List[str] = []
failed = False
for md_file in files:
@@ -561,35 +374,29 @@ class AgentImporter:
if not (failed and single_file):
self.record(kind, source, destination, "skipped", "No importable entries found")
return
self._merge_memory_entries(kind, source, destination, incoming)
def _merge_memory_entries(self, kind: str, source: Path,
destination: Path, incoming: List[str]) -> None:
existing = parse_existing_memory_entries(destination)
merged, stats = merge_entries(existing, incoming, MEMORY_CHAR_LIMIT)
details = {
"existing_entries": stats["existing"],
"added_entries": stats["added"],
"duplicate_entries": stats["duplicates"],
"overflowed_entries": stats["overflowed"],
}
details = {"existing_entries": stats["existing"], "added_entries": stats["added"],
"duplicate_entries": stats["duplicates"],
"overflowed_entries": stats["overflowed"]}
if stats["added"] == 0:
self.record(kind, source, destination, "skipped", "No new entries to import", **details)
return
def write() -> Optional[str]:
destination.parent.mkdir(parents=True, exist_ok=True)
# Snapshot first (``<name>.bak.<unix_ts>``, as MemoryStore does); never rewrite the
# store when the safety net failed.
step = "back up existing"
try:
backup = backup_memory_file(destination)
except OSError as exc:
# Never rewrite the store when the safety net failed.
return f"Could not back up existing memory file: {exc}"
if backup is not None:
details["backup"] = str(backup)
try:
if destination.exists():
backup = destination.with_suffix(f"{destination.suffix}.bak.{int(time.time())}")
shutil.copy2(destination, backup)
details["backup"] = str(backup)
step = "write merged"
atomic_write_text(destination, ENTRY_DELIMITER.join(merged) + ("\n" if merged else ""))
except OSError as exc:
return f"Could not write merged memory file: {exc}"
return f"Could not {step} memory file: {exc}"
return None
self.apply(kind, source, destination, "Would merge entries", write, details)
@@ -597,18 +404,10 @@ class AgentImporter:
# (settings key, item kind, config path, dry-run tracks unmapped rules)
_PERMISSION_RULES = {
"allow": ("command-allowlist", ("command_allowlist",), True),
"deny": ("command-denylist", ("approvals", "deny"), False),
}
def import_permission_allowlist(self, settings: Dict[str, Any]) -> None:
"""settings.json permissions.allow → config.yaml command_allowlist."""
self._import_permission_rules(settings, "allow")
def import_permission_denylist(self, settings: Dict[str, Any]) -> None:
"""settings.json permissions.deny → config.yaml approvals.deny."""
self._import_permission_rules(settings, "deny")
"deny": ("command-denylist", ("approvals", "deny"), False)}
def _import_permission_rules(self, settings: Dict[str, Any], key: str) -> None:
"""settings.json permissions.allow/deny → config.yaml command_allowlist / approvals.deny."""
kind, config_path, track_unmapped = self._PERMISSION_RULES[key]
label = f"settings.json permissions.{key}"
destination = self.target_root / "config.yaml"
@@ -617,18 +416,9 @@ class AgentImporter:
if not isinstance(rules, list) or not rules:
self.record(kind, None, destination, "skipped", f"No permissions.{key} rules found")
return
patterns: List[str] = []
skipped_rules: List[str] = []
for rule in rules:
if not isinstance(rule, str):
continue
pattern = claude_rule_to_command_pattern(rule)
if pattern:
patterns.append(pattern)
else:
skipped_rules.append(rule)
patterns = sorted(dict.fromkeys(patterns))
mapped = [(r, claude_rule_to_command_pattern(r)) for r in rules if isinstance(r, str)]
patterns = sorted(dict.fromkeys(p for _, p in mapped if p))
skipped_rules = [r for r, p in mapped if not p]
unmapped: Dict[str, Any] = {"unmapped_rules": skipped_rules} if track_unmapped else {}
if not patterns:
self.record(kind, None, destination, "skipped",
@@ -636,7 +426,6 @@ class AgentImporter:
return
if not skipped_rules:
unmapped = {}
config = self.load_target_config(kind, label, destination)
if config is None:
return
@@ -644,18 +433,15 @@ class AgentImporter:
parent: Dict[str, Any] = config
for part in config_path[:-1]:
child = parent.get(part)
if not isinstance(child, dict):
child = {}
parent[part] = child
parent = child
parent[part] = parent = child if isinstance(child, dict) else {}
current = parent.get(config_path[-1], [])
if not isinstance(current, list):
current = []
current = current if isinstance(current, list) else []
merged = sorted(dict.fromkeys(list(current) + patterns))
added = [p for p in merged if p not in current]
if not added:
self.record(kind, label, destination, "skipped", "All patterns already present")
return
def write() -> None:
parent[config_path[-1]] = merged
dump_yaml_file(destination, config)
@@ -669,13 +455,11 @@ class AgentImporter:
if not servers:
self.record(kind, None, destination, "skipped", "No MCP servers found")
return
config = self.load_target_config(kind, None, destination)
if config is None:
return
existing = config.get("mcp_servers")
if not isinstance(existing, dict):
existing = {}
existing = existing if isinstance(existing, dict) else {}
added = 0
for name, srv in servers.items():
if not isinstance(srv, dict):
@@ -685,32 +469,26 @@ class AgentImporter:
self.record(kind, name, f"mcp_servers.{name}", "conflict",
"MCP server already exists in Hermes config")
continue
hermes_srv, stripped = _translate_mcp_server(name, srv)
self.stripped_secrets.extend(stripped)
if not hermes_srv:
self.record(kind, name, None, "skipped", "Server has neither a command nor a url")
continue
existing[name] = hermes_srv
added += 1
self.record(kind, name, f"config.yaml mcp_servers.{name}", "imported")
if added > 0 and self.execute:
config["mcp_servers"] = existing
dump_yaml_file(destination, config)
def import_skills(self, source_root: Path) -> None:
"""skills/<name>/SKILL.md dirs → HERMES_HOME/skills/<category>/<name>."""
category = _SKILL_CATEGORY[self.agent]
destination_root = self.target_root / "skills" / category
destination_root = self.target_root / "skills" / _SKILL_CATEGORY[self.agent]
if not source_root.is_dir():
self.record("skills", None, destination_root, "skipped", "No skills directory found")
return
skill_dirs = [
p for p in sorted(source_root.iterdir())
if p.is_dir() and (p / "SKILL.md").exists()
]
skill_dirs = [p for p in sorted(source_root.iterdir())
if p.is_dir() and (p / "SKILL.md").exists()]
if not skill_dirs:
self.record("skills", source_root, destination_root, "skipped",
"No skills with SKILL.md found")
@@ -721,26 +499,25 @@ class AgentImporter:
self.record("skill", skill_dir, destination, "conflict",
"Destination skill already exists")
continue
self.apply("skill", skill_dir, destination, "Would copy skill directory",
lambda: _copy_skill_dir(skill_dir, destination))
def copy(skill_dir=skill_dir, destination=destination) -> None:
destination.parent.mkdir(parents=True, exist_ok=True)
if destination.exists():
shutil.rmtree(destination)
shutil.copytree(skill_dir, destination)
self.apply("skill", skill_dir, destination, "Would copy skill directory", copy)
# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------
def import_agent_command(args) -> None:
"""Handle ``hermes import-agent`` (invoked from hermes_cli.main)."""
from hermes_cli.config import get_config_path, load_config, save_config
from hermes_constants import get_hermes_home
from hermes_cli.setup import (
Colors, color, print_header, print_info, print_success, print_error, prompt_yes_no,
)
from hermes_cli.setup import (Colors, color, print_header, print_info, print_success,
print_error, prompt_yes_no)
agent, explicit_source = args.agent, args.source
dry_run, overwrite, auto_yes = args.dry_run, args.overwrite, args.yes
agent, explicit_source, overwrite = args.agent, args.source, args.overwrite
# -- detect --------------------------------------------------------------
if agent is None:
detected = detect_agents()
if not detected:
@@ -754,20 +531,17 @@ def import_agent_command(args) -> None:
print_info("Pick one: hermes import-agent claude-code or hermes import-agent codex")
return
agent = detected[0]
source_dir = Path(explicit_source) if explicit_source else default_source_dir(agent)
source_dir = Path(explicit_source or Path.home() / _AGENT_DEFAULT_DIRS[agent])
print()
print(color("┌─────────────────────────────────────────────────────────┐", Colors.MAGENTA))
print(color("│ ⚕ Hermes — Import From Another Agent │", Colors.MAGENTA))
print(color("└─────────────────────────────────────────────────────────┘", Colors.MAGENTA))
if not source_dir.is_dir():
print()
print_error(f"Agent directory not found: {source_dir}")
print_info(f"Specify a custom path: hermes import-agent {agent} --source /path/to/{_AGENT_DEFAULT_DIRS[agent]}")
return
hermes_home = get_hermes_home()
print()
print_header("Import Settings")
@@ -776,48 +550,41 @@ def import_agent_command(args) -> None:
print_info(f"Target: {hermes_home}")
print_info(f"Overwrite: {'yes' if overwrite else 'no (skip conflicts)'}")
print_info("Secrets: never imported — run 'hermes setup' for credentials")
# Ensure config.yaml exists before the import tries to merge into it
config_path = get_config_path()
if not config_path.exists():
if not get_config_path().exists():
save_config(load_config())
# -- Phase 1: preview (always) --------------------------------------------
def run_import(execute: bool) -> Dict[str, Any]:
return AgentImporter(
agent=agent,
source_root=source_dir.resolve(),
target_root=hermes_home.resolve(),
execute=execute,
overwrite=overwrite,
).run()
def run_import(execute: bool, phase: str) -> Optional[Dict[str, Any]]:
"""Run the importer; on failure print the error and return None."""
try:
return AgentImporter(agent, source_dir.resolve(), hermes_home.resolve(),
execute=execute, overwrite=overwrite).run()
except Exception as e:
print()
print_error(f"Import{phase} failed: {e}")
logger.debug(f"import-agent{phase} error", exc_info=True)
return None
try:
preview = run_import(execute=False)
except Exception as e:
print()
print_error(f"Import preview failed: {e}")
logger.debug("import-agent preview error", exc_info=True)
# Phase 1: preview (always)
preview = run_import(False, " preview")
if preview is None:
return
summary = preview.get("summary", {})
if summary.get("imported", 0) == 0 and summary.get("conflict", 0) == 0:
print()
print_info(f"Nothing to import from {agent}.")
print_import_report(preview, dry_run=True)
return
print()
print_header(f"Import Preview — {summary.get('imported', 0)} item(s) would be imported")
print_info("No changes have been made yet. Review the list below:")
print_import_report(preview, dry_run=True)
if dry_run:
if args.dry_run:
return
# -- Phase 2: confirm and execute -----------------------------------------
# Phase 2: confirm and execute
print()
if not auto_yes:
if not args.yes:
if not sys.stdin.isatty():
print_info("Non-interactive session — preview only.")
print_info(f"To execute, re-run with: hermes import-agent {agent} --yes")
@@ -825,15 +592,9 @@ def import_agent_command(args) -> None:
if not prompt_yes_no("Proceed with import?", default=True):
print_info("Import cancelled.")
return
try:
report = run_import(execute=True)
except Exception as e:
print()
print_error(f"Import failed: {e}")
logger.debug("import-agent error", exc_info=True)
report = run_import(True, "")
if report is None:
return
print_import_report(report, dry_run=False)
print()
print_success("Import complete.")
@@ -845,47 +606,36 @@ def print_import_report(report: Dict[str, Any], dry_run: bool) -> None:
"""Print a formatted per-item import report (claw-migrate style)."""
from hermes_cli.setup import Colors, color, print_header, print_info
summary = report.get("summary", {})
print()
print_header("Dry Run Results" if dry_run else "Import Results")
if dry_run:
print_header("Dry Run Results")
print_info("No files were modified. This is a preview of what would happen.")
else:
print_header("Import Results")
print()
items = report.get("items", [])
# (status, colour, group heading, summary label)
groups = (
("imported", Colors.GREEN, "✓ Would import" if dry_run else "✓ Imported"),
("conflict", Colors.YELLOW, "⚠ Conflicts (skipped — use --overwrite to force)"),
("skipped", Colors.DIM, "─ Skipped"),
("error", Colors.RED, "✗ Errors"),
)
for status, col, label in groups:
group_items = [i for i in items if i.get("status") == status]
("imported", Colors.GREEN, "✓ Would import" if dry_run else "✓ Imported",
"would import" if dry_run else "imported"),
("conflict", Colors.YELLOW, "⚠ Conflicts (skipped — use --overwrite to force)",
"conflict(s)"),
("skipped", Colors.DIM, "─ Skipped", "skipped"),
("error", Colors.RED, "✗ Errors", "error(s)"))
for status, col, label, _ in groups:
group_items = [i for i in report.get("items", []) if i.get("status") == status]
if not group_items:
continue
print(color(f" {label}:", col))
for item in group_items:
tail = (
"→ " + str(item.get("destination") or "").replace(str(Path.home()), "~")
if status == "imported" else f" {item.get('reason', '')}"
)
tail = ("→ " + str(item.get("destination") or "").replace(str(Path.home()), "~")
if status == "imported" else f" {item.get('reason', '')}")
print(f" {item.get('kind', 'unknown'):<22s} {tail}")
print()
stripped = report.get("stripped_secrets") or []
if stripped:
if stripped := report.get("stripped_secrets"):
print(color(" ⚷ Secrets stripped (never imported):", Colors.YELLOW))
for name in stripped:
print(f" {name}")
print_info("Re-add credentials deliberately via 'hermes setup' or ~/.hermes/.env.")
print()
labels = (
("imported", "would import" if dry_run else "imported"),
("conflict", "conflict(s)"), ("skipped", "skipped"), ("error", "error(s)"),
)
parts = [f"{summary[k]} {label}" for k, label in labels if summary.get(k)]
summary = report.get("summary", {})
parts = [f"{summary[k]} {label}" for k, _, _, label in groups if summary.get(k)]
if parts:
print_info(f"Summary: {', '.join(parts)}")
hermes_cli/agent_plugins.py
+81 -180
View File
@@ -2,23 +2,22 @@
from __future__ import annotations
import ipaddress
import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Mapping, Tuple
from urllib.parse import urlsplit
from agent.skill_utils import yaml_load
PLUGIN_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json"
MCP_SCHEMA_V1 = "https://agent-plugins.org/schemas/1.0.0/mcp.schema.json"
_PLUGIN_FIELDS = {
"$schema", "name", "version", "description", "author", "homepage", "repository", "license",
"keywords", "extensions",
}
_PLUGIN_FIELDS = {"$schema", "name", "version", "description", "author", "homepage", "repository",
"license", "keywords", "extensions"}
_AUTHOR_FIELDS = {"name", "email", "url"}
_STDIO_FIELDS = {"type", "command", "args", "env", "cwd"}
_REMOTE_FIELDS = {"type", "url", "headers"}
@@ -62,8 +61,7 @@ class AgentPluginPackage:
def _inside(path: Path, root: Path) -> bool:
try:
path.resolve(strict=False).relative_to(root.resolve(strict=True))
return True
return path.resolve(strict=False).is_relative_to(root.resolve(strict=True))
except (OSError, RuntimeError, ValueError):
return False
@@ -92,52 +90,38 @@ def _validate_manifest(root: Path) -> tuple[dict, list[AgentPluginDiagnostic]]:
raise AgentPluginError("plugin.json must be a regular file within the plugin root")
manifest = _read_json_object(manifest_path, label="plugin.json")
diagnostics: list[AgentPluginDiagnostic] = []
for field in sorted(set(manifest) - _PLUGIN_FIELDS):
diagnostics.append(AgentPluginDiagnostic("manifest", f"ignored unknown top-level field: {field}"))
manifest.pop(field)
if manifest.get("$schema") != PLUGIN_SCHEMA_V1:
raise AgentPluginError("plugin.json declares an unsupported or missing Agent Plugins schema")
name = manifest.get("name")
if not isinstance(name, str) or not 1 <= len(name) <= 64 or _PLUGIN_NAME_RE.fullmatch(name) is None:
raise AgentPluginError("plugin.json name does not satisfy v1 constraints")
for field in ("version", "description", "homepage", "repository", "license"):
if field in manifest and not isinstance(manifest[field], str):
raise AgentPluginError(f"plugin.json {field} must be a string")
if "keywords" in manifest:
keywords = manifest["keywords"]
if not isinstance(keywords, list) or not _all_str(keywords):
raise AgentPluginError("plugin.json keywords must be an array of strings")
if "author" in manifest:
author = manifest["author"]
if not isinstance(author, dict):
raise AgentPluginError("plugin.json author must be an object")
if set(author) - _AUTHOR_FIELDS or not _all_str(author.values()):
raise AgentPluginError("plugin.json author may contain only string name, email, and url fields")
if "extensions" in manifest:
extensions = manifest["extensions"]
if not isinstance(extensions, dict):
diagnostics.append(AgentPluginDiagnostic("manifest", "ignored non-object extensions field"))
manifest.pop("extensions")
elif any(not isinstance(value, dict) for value in extensions.values()):
raise AgentPluginError("plugin.json extension namespace values must be objects")
keywords = manifest.get("keywords", [])
if not isinstance(keywords, list) or not _all_str(keywords):
raise AgentPluginError("plugin.json keywords must be an array of strings")
author = manifest.get("author", {})
if not isinstance(author, dict):
raise AgentPluginError("plugin.json author must be an object")
if set(author) - _AUTHOR_FIELDS or not _all_str(author.values()):
raise AgentPluginError("plugin.json author may contain only string name, email, and url fields")
extensions = manifest.get("extensions", {})
if not isinstance(extensions, dict):
diagnostics.append(AgentPluginDiagnostic("manifest", "ignored non-object extensions field"))
manifest.pop("extensions")
elif any(not isinstance(value, dict) for value in extensions.values()):
raise AgentPluginError("plugin.json extension namespace values must be objects")
return manifest, diagnostics
def _valid_skill_frontmatter(frontmatter: Mapping[str, Any], directory_name: str) -> str | None:
name = frontmatter.get("name")
if (
not isinstance(name, str)
or name != directory_name
or not 1 <= len(name) <= 64
or _SKILL_NAME_RE.fullmatch(name) is None
):
if (not isinstance(name, str) or name != directory_name or not 1 <= len(name) <= 64
or _SKILL_NAME_RE.fullmatch(name) is None):
return "name must match the directory and satisfy Agent Skills constraints"
description = frontmatter.get("description")
if not isinstance(description, str) or not 1 <= len(description) <= 1024:
@@ -184,15 +168,11 @@ def _discover_skills(root: Path, diagnostics: list[AgentPluginDiagnostic]) -> tu
except OSError as exc:
diagnostics.append(AgentPluginDiagnostic("skills", f"cannot list skills: {exc}"))
return ()
skills: list[AgentPluginSkill] = []
for child in children:
skill_md = child / "SKILL.md"
if not child.is_dir() or not skill_md.exists():
continue
scope = f"skill:{child.name}"
error: str | None = None
frontmatter: dict = {}
if not _inside(skill_md, root) or not skill_md.is_file():
error = "SKILL.md must be a regular in-root file"
else:
@@ -203,17 +183,12 @@ def _discover_skills(root: Path, diagnostics: list[AgentPluginDiagnostic]) -> tu
else:
error = _valid_skill_frontmatter(frontmatter, child.name)
if error:
diagnostics.append(AgentPluginDiagnostic(scope, error))
diagnostics.append(AgentPluginDiagnostic(f"skill:{child.name}", error))
continue
skills.append(
AgentPluginSkill(
name=child.name,
description=frontmatter["description"],
root=child.resolve(strict=True),
skill_md=skill_md.resolve(strict=True),
frontmatter=dict(frontmatter),
)
)
skills.append(AgentPluginSkill(
name=child.name, description=frontmatter["description"],
root=child.resolve(strict=True), skill_md=skill_md.resolve(strict=True),
frontmatter=dict(frontmatter)))
return tuple(skills)
@@ -222,9 +197,8 @@ def _expand(value: str, plugin_root: Path, data_root: Path) -> str:
return _PLACEHOLDER_RE.sub(lambda match: replacements[match.group(1)], value)
def _resolve_scoped_path(
value: str, plugin_root: Path, data_root: Path, *, expand_placeholders: bool = True
) -> Path:
def _resolve_scoped_path(value: str, plugin_root: Path, data_root: Path, *,
expand_placeholders: bool = True) -> Path:
expanded = _expand(value, plugin_root, data_root) if expand_placeholders else value
if value.startswith("./"):
base, candidate = plugin_root, plugin_root / expanded[2:]
@@ -243,28 +217,17 @@ def _resolve_scoped_path(
def _validate_headers(headers: object) -> bool:
if headers is None:
return True
if not isinstance(headers, dict):
return False
return headers is None
names = [name for name in headers if isinstance(name, str) and _HEADER_NAME_RE.fullmatch(name)]
return (
len(names) == len(headers)
and len({name.lower() for name in names}) == len(names)
and all(isinstance(v, str) and "\r" not in v and "\n" not in v for v in headers.values())
)
values = headers.values()
return (len(names) == len(headers) and len({name.lower() for name in names}) == len(names)
and all(isinstance(v, str) and "\r" not in v and "\n" not in v for v in values))
def _validate_remote_url(url: object) -> str:
"""Validate a portable remote MCP URL per the v1 spec and return it.
Rules (Agent Plugins v1 §7.2.1): absolute http(s) URL, no user information, no fragment; non-
loopback endpoints must use HTTPS. HTTP is allowed only when the host is exactly ``localhost``
or an IP literal in a loopback range. No placeholder or environment expansion is performed.
"""
from urllib.parse import urlsplit
"""Validate a portable remote MCP URL per Agent Plugins v1 §7.2.1 and return it: absolute
http(s), no user info, no fragment; HTTP only for ``localhost``/loopback IP. No expansion."""
if not isinstance(url, str) or not url:
raise ValueError("url must be a non-empty string")
try:
@@ -282,8 +245,6 @@ def _validate_remote_url(url: object) -> str:
if not host:
raise ValueError("url must have a host")
if scheme == "http" and host != "localhost":
import ipaddress
try:
loopback = ipaddress.ip_address(host).is_loopback
except ValueError:
@@ -294,13 +255,8 @@ def _validate_remote_url(url: object) -> str:
def _translate_remote(config: Mapping[str, Any]) -> Dict[str, Any]:
"""Translate a portable ``streamable-http`` entry into native MCP config.
``strict_redirect_headers`` tells the runtime to drop configured headers on any cross-origin
redirect, as the v1 spec requires for portable packages (no forwarding to another origin
without explicit user authorization).
"""
"""Translate a portable ``streamable-http`` entry into native MCP config. The v1 spec requires
``strict_redirect_headers``: drop configured headers on any cross-origin redirect."""
if set(config) - _REMOTE_FIELDS:
raise ValueError("unknown remote field")
url = _validate_remote_url(config.get("url"))
@@ -312,122 +268,83 @@ def _translate_remote(config: Mapping[str, Any]) -> Dict[str, Any]:
return translated
def _translate_stdio(config: Mapping[str, Any], plugin_root: Path, data_root: Path) -> Dict[str, Any]:
def _translate_stdio(config: Mapping[str, Any], plugin_root: Path, data_root: Path,
create_data: bool = False) -> Dict[str, Any]:
if set(config) - _STDIO_FIELDS:
raise ValueError("unknown stdio field")
command = config.get("command")
if not isinstance(command, str) or not command or "\x00" in command:
raise ValueError("command must be a non-empty executable token")
if command.startswith("./"):
command_value = str(_resolve_scoped_path(command, plugin_root, data_root, expand_placeholders=False))
command = str(_resolve_scoped_path(command, plugin_root, data_root,
expand_placeholders=False))
elif any(character.isspace() for character in command):
raise ValueError("command must contain one executable token")
elif "/" in command or "\\" in command or command in {".", ".."}:
raise ValueError("command must be a bare executable or begin with ./")
else:
command_value = command
args = config.get("args", [])
if not isinstance(args, list) or not _all_str(args):
raise ValueError("args must be an array of strings")
env = config.get("env", {})
if not _str_map(env):
raise ValueError("env must map string keys to string values")
env_keys = {key.upper() if os.name == "nt" else key for key in env}
if "PLUGIN_ROOT" in env_keys or "PLUGIN_DATA" in env_keys:
if {"PLUGIN_ROOT", "PLUGIN_DATA"} & {key.upper() if os.name == "nt" else key for key in env}:
raise ValueError("PLUGIN_ROOT and PLUGIN_DATA are reserved")
cwd = config.get("cwd")
if cwd is None:
cwd_value = plugin_root
elif not isinstance(cwd, str):
if cwd is not None and not isinstance(cwd, str):
raise ValueError("cwd must be a string")
else:
cwd_value = _resolve_scoped_path(cwd, plugin_root, data_root)
translated_env = {key: _expand(value, plugin_root, data_root) for key, value in env.items()}
translated_env["PLUGIN_ROOT"] = str(plugin_root)
translated_env["PLUGIN_DATA"] = str(data_root)
return {
"command": command_value,
"args": [_expand(value, plugin_root, data_root) for value in args],
"env": translated_env,
"cwd": str(cwd_value),
}
def _translate_stdio_server(server: Mapping[str, Any], root: Path, data_root: Path, create_data: bool) -> Dict[str, Any]:
translated_server = _translate_stdio(server, root, data_root)
cwd_path = plugin_root if cwd is None else _resolve_scoped_path(cwd, plugin_root, data_root)
if create_data:
data_root.mkdir(parents=True, exist_ok=True)
cwd_path = Path(translated_server["cwd"])
# The MCP client starts stdio servers with this cwd. Create only data-root
# descendants; plugin-root paths remain package-owned and are never made
# writable as a side effect of discovery.
# The MCP client starts stdio servers with this cwd. Create only data-root descendants;
# plugin-root paths remain package-owned and are never made writable by discovery.
if cwd_path.is_relative_to(data_root):
cwd_path.mkdir(parents=True, exist_ok=True)
return translated_server
translated_env = {**{key: _expand(value, plugin_root, data_root) for key, value in env.items()},
"PLUGIN_ROOT": str(plugin_root), "PLUGIN_DATA": str(data_root)}
return {"command": command, "args": [_expand(value, plugin_root, data_root) for value in args],
"env": translated_env, "cwd": str(cwd_path)}
def _reject_sse(server: Mapping[str, Any]) -> None:
if (
set(server) - _REMOTE_FIELDS
or not isinstance(server.get("url"), str)
or not server.get("url")
or not _validate_headers(server.get("headers"))
):
if (set(server) - _REMOTE_FIELDS or not isinstance(server.get("url"), str)
or not server.get("url") or not _validate_headers(server.get("headers"))):
raise ValueError("invalid remote entry")
raise ValueError("portable sse transport is not supported")
def _discover_mcp(
root: Path,
data_root: Path,
diagnostics: list[AgentPluginDiagnostic],
*,
create_data: bool = True,
) -> Dict[str, Dict[str, Any]]:
def _discover_mcp(root: Path, data_root: Path, diagnostics: list[AgentPluginDiagnostic], *,
create_data: bool = True) -> Dict[str, Dict[str, Any]]:
mcp_path = root / "mcp.json"
if not mcp_path.exists() and not mcp_path.is_symlink():
return {}
def _fail(message: str) -> Dict[str, Dict[str, Any]]:
diagnostics.append(AgentPluginDiagnostic("mcp", message))
return {}
if not _inside(mcp_path, root) or not mcp_path.is_file():
return _fail("mcp.json must be a regular in-root file")
try:
try: # file-level problems (incl. AgentPluginError from the JSON read) -> one "mcp" diagnostic
if not _inside(mcp_path, root) or not mcp_path.is_file():
raise ValueError("mcp.json must be a regular in-root file")
config = _read_json_object(mcp_path, label="mcp.json")
except AgentPluginError as exc:
return _fail(str(exc))
if set(config) != {"$schema", "mcpServers"}:
return _fail("mcp.json has an invalid top-level shape")
if config.get("$schema") != MCP_SCHEMA_V1:
return _fail("mcp.json declares an unsupported schema")
servers = config.get("mcpServers")
if not isinstance(servers, dict):
return _fail("mcpServers must be an object")
translators = {
"stdio": lambda server: _translate_stdio_server(server, root, data_root, create_data),
"streamable-http": _translate_remote,
"sse": _reject_sse,
}
if set(config) != {"$schema", "mcpServers"}:
raise ValueError("mcp.json has an invalid top-level shape")
if config.get("$schema") != MCP_SCHEMA_V1:
raise ValueError("mcp.json declares an unsupported schema")
servers = config.get("mcpServers")
if not isinstance(servers, dict):
raise ValueError("mcpServers must be an object")
except ValueError as exc:
diagnostics.append(AgentPluginDiagnostic("mcp", str(exc)))
return {}
translators = {"stdio": lambda server: _translate_stdio(server, root, data_root, create_data),
"streamable-http": _translate_remote, "sse": _reject_sse}
translated: Dict[str, Dict[str, Any]] = {}
for name, server in servers.items():
scope = f"mcp:{name}"
if not isinstance(name, str) or not name or not isinstance(server, dict):
diagnostics.append(AgentPluginDiagnostic(scope, "invalid server entry"))
continue
translate = translators.get(server.get("type"))
if translate is None:
diagnostics.append(AgentPluginDiagnostic(scope, "unknown MCP server type"))
continue
try:
if not isinstance(name, str) or not name or not isinstance(server, dict):
raise ValueError("invalid server entry")
translate = translators.get(server.get("type"))
if translate is None:
raise ValueError("unknown MCP server type")
translated[name] = translate(server)
except (OSError, ValueError) as exc:
diagnostics.append(AgentPluginDiagnostic(scope, str(exc)))
diagnostics.append(AgentPluginDiagnostic(f"mcp:{name}", str(exc)))
return translated
@@ -440,39 +357,23 @@ def _validate_root(plugin_root: Path) -> tuple[Path, dict, list[AgentPluginDiagn
def load_agent_plugin(plugin_root: Path, data_root: Path) -> AgentPluginPackage:
"""Validate and translate one installed Agent Plugins v1 package."""
root, manifest, diagnostics = _validate_root(plugin_root)
resolved_data = Path(data_root).resolve(strict=False)
skills = _discover_skills(root, diagnostics)
mcp_servers = _discover_mcp(root, resolved_data, diagnostics)
return AgentPluginPackage(
name=manifest["name"],
version=manifest.get("version", ""),
description=manifest.get("description", ""),
root=root,
data_root=resolved_data,
manifest=dict(manifest),
skills=skills,
mcp_servers=mcp_servers,
diagnostics=tuple(diagnostics),
)
return AgentPluginPackage( # skills are discovered before MCP: diagnostics keep that order
name=manifest["name"], version=manifest.get("version", ""),
description=manifest.get("description", ""), root=root, data_root=resolved_data,
manifest=dict(manifest), skills=_discover_skills(root, diagnostics),
mcp_servers=_discover_mcp(root, resolved_data, diagnostics), diagnostics=tuple(diagnostics))
def read_agent_plugin_manifest(plugin_root: Path) -> tuple[dict, tuple[AgentPluginDiagnostic, ...]]:
"""Validate only root ``plugin.json`` without discovering components."""
_root, manifest, diagnostics = _validate_root(plugin_root)
return manifest, tuple(diagnostics)
def has_enabled_agent_plugin_mcp(raw_config: Mapping[str, Any]) -> bool:
"""Compatibility wrapper for the shared PluginManager MCP probe.
Directory scanning belongs to :mod:`hermes_cli.plugins` so startup gating and full plugin
discovery cannot drift apart. Keep this import-compatible entry point for callers that used the
original helper.
"""
"""Import-compatible wrapper for the shared PluginManager MCP probe; directory scanning lives
in :mod:`hermes_cli.plugins` so startup gating and full plugin discovery cannot drift apart."""
from hermes_cli.plugins import has_enabled_agent_plugin_mcp as _probe
return _probe(raw_config)
hermes_cli/browser_connect.py
+379 -786
View File
File diff suppressed because it is too large Load Diff