255 lines
9.7 KiB
Python
255 lines
9.7 KiB
Python
"""Profile JSON-RPC handlers — the ws twin of the dashboard's /api/profiles.
|
|
|
|
Motivation: desktop plugins reach the backend exclusively through the
|
|
generic ws JSON-RPC door (`host.request`). Profile enumeration/creation
|
|
previously lived only on the dashboard REST router, which plugins cannot
|
|
reach, so anything "one chat per agent profile"-shaped (bot rosters,
|
|
profile pickers, team panes) was impossible to build as a plugin. These
|
|
handlers delegate to the same `hermes_cli.profiles` primitives the REST
|
|
endpoints use.
|
|
|
|
Handlers are rebound onto server.py's globals at install time — see
|
|
method_ctx.py. They may reference server.py module globals (`_ok`,
|
|
`_err`, `is_truthy_value`, ...) that are not imported here.
|
|
"""
|
|
|
|
from .method_ctx import HandlerRegistry
|
|
|
|
_registry = HandlerRegistry()
|
|
method = _registry.method
|
|
|
|
|
|
@method("profiles.list")
|
|
def _(rid, params: dict) -> dict:
|
|
"""List Hermes profiles (name, path, model, description, skill count).
|
|
|
|
``include_sessions`` (default true) additionally reports each profile's
|
|
most recent conversation as ``last_session`` so a roster UI can paint
|
|
per-agent previews without N follow-up calls.
|
|
|
|
NOTE: helpers must be nested — install() rebinds this handler's
|
|
__globals__ onto server.py, so module-level names here are invisible.
|
|
"""
|
|
|
|
def _latest_profile_session_row(profile_path):
|
|
"""Most recent human-facing session in a profile's state.db, or None.
|
|
|
|
Mirrors session.list's deny-list (drops ``tool`` sub-agent rows and
|
|
``kanban`` dispatcher workers). Best-effort: any failure (missing
|
|
state.db, locked db, older schema) degrades to None rather than
|
|
failing the whole profiles.list call.
|
|
"""
|
|
try:
|
|
from pathlib import Path
|
|
|
|
db_path = Path(profile_path) / "state.db"
|
|
if not db_path.exists():
|
|
return None
|
|
from hermes_state import SessionDB
|
|
|
|
deny = frozenset({"kanban", "tool"})
|
|
db = SessionDB(db_path=db_path)
|
|
try:
|
|
for s in db.list_sessions_rich(
|
|
source=None, limit=20, order_by_last_active=True, compact_rows=True
|
|
):
|
|
if (s.get("source") or "").strip().lower() in deny:
|
|
continue
|
|
return {
|
|
"id": s["id"],
|
|
"title": s.get("title") or "",
|
|
"preview": s.get("preview") or "",
|
|
"started_at": s.get("started_at") or 0,
|
|
"last_active": s.get("last_active") or s.get("started_at") or 0,
|
|
"message_count": s.get("message_count") or 0,
|
|
}
|
|
finally:
|
|
try:
|
|
db.close()
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
try:
|
|
from hermes_cli.profiles import list_profiles
|
|
|
|
include_sessions = is_truthy_value(params.get("include_sessions", True))
|
|
out = []
|
|
for p in list_profiles():
|
|
row = {
|
|
"name": p.name,
|
|
"path": str(p.path),
|
|
"is_default": bool(p.is_default),
|
|
"model": p.model,
|
|
"provider": p.provider,
|
|
"description": getattr(p, "description", "") or "",
|
|
"skill_count": getattr(p, "skill_count", 0) or 0,
|
|
}
|
|
if include_sessions:
|
|
row["last_session"] = _latest_profile_session_row(p.path)
|
|
out.append(row)
|
|
return _ok(rid, {"profiles": out})
|
|
except Exception as e:
|
|
return _err(rid, 5061, str(e))
|
|
|
|
|
|
@method("profiles.create")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Create a profile — the ws twin of POST /api/profiles.
|
|
|
|
Params: ``name`` (required, lowercase slug), ``description``,
|
|
``clone_from`` (source profile; omitted = fresh profile with bundled
|
|
skills), ``clone_all``, ``no_skills``, ``soul`` (SOUL.md content),
|
|
``model`` + ``provider`` (optional model pin, best-effort), and
|
|
``mirror_credentials`` (default true) — copy the launch profile's
|
|
``.env`` and ``auth.json`` into the new profile, and inherit its
|
|
model.provider/model.default when no explicit pin is given.
|
|
|
|
Credential mirroring exists because ``create_profile()`` deliberately
|
|
seeds a comment-only ``.env`` and never copies ``auth.json`` (OAuth
|
|
tokens / credential pools), so a profile created headlessly from a
|
|
plugin was born with NO inference provider — the first message failed
|
|
with "No inference provider configured" and there is no interactive
|
|
``hermes setup`` in that flow to recover. A profile spawned as an
|
|
always-available teammate must be able to think out of the box; callers
|
|
that want an isolated/credential-free profile pass
|
|
``mirror_credentials: false``.
|
|
"""
|
|
|
|
def _has_real_env_content(env_path) -> bool:
|
|
"""True when .env has any non-comment, non-blank line."""
|
|
try:
|
|
for line in env_path.read_text(encoding="utf-8", errors="replace").splitlines():
|
|
stripped = line.strip()
|
|
if stripped and not stripped.startswith("#"):
|
|
return True
|
|
except Exception:
|
|
pass
|
|
return False
|
|
|
|
name = str(params.get("name") or "").strip()
|
|
if not name:
|
|
return _err(rid, 4061, "name required")
|
|
try:
|
|
from hermes_cli import profiles as profiles_mod
|
|
|
|
clone_from = str(params.get("clone_from") or "").strip() or None
|
|
clone_all = is_truthy_value(params.get("clone_all", False))
|
|
path = profiles_mod.create_profile(
|
|
name=name,
|
|
clone_from=clone_from,
|
|
clone_all=clone_all,
|
|
clone_config=bool(clone_from) and not clone_all,
|
|
no_skills=is_truthy_value(params.get("no_skills", False)),
|
|
description=str(params.get("description") or "").strip() or None,
|
|
)
|
|
except (ValueError, FileExistsError, FileNotFoundError) as e:
|
|
return _err(rid, 4062, str(e))
|
|
except Exception as e:
|
|
return _err(rid, 5062, str(e))
|
|
|
|
# Mirror the CLI/REST create flow: fresh profiles get the bundled
|
|
# skills; safe alias wrapper. Both best-effort.
|
|
try:
|
|
if not clone_from:
|
|
profiles_mod.seed_profile_skills(path, quiet=True)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
if not profiles_mod.check_alias_collision(name):
|
|
profiles_mod.create_wrapper_script(name)
|
|
except Exception:
|
|
pass
|
|
|
|
soul = params.get("soul")
|
|
soul_written = False
|
|
if isinstance(soul, str) and soul.strip():
|
|
try:
|
|
(path / "SOUL.md").write_text(soul, encoding="utf-8")
|
|
soul_written = True
|
|
except Exception:
|
|
pass
|
|
|
|
# Credential + provider mirroring (default ON): a headless-created
|
|
# profile must be able to run a first turn. Copy the launch profile's
|
|
# .env (only over the seeded comment-only stub — never clobber real
|
|
# secrets a clone brought along) and auth.json (only when absent), then
|
|
# inherit model.provider/model.default unless the caller pinned a model.
|
|
mirrored = {"env": False, "auth": False, "model_inherited": False}
|
|
if is_truthy_value(params.get("mirror_credentials", True)):
|
|
import shutil
|
|
|
|
from hermes_constants import get_hermes_home
|
|
|
|
launch_home = get_hermes_home()
|
|
try:
|
|
src_env = launch_home / ".env"
|
|
dst_env = path / ".env"
|
|
if src_env.is_file() and _has_real_env_content(src_env) and not _has_real_env_content(dst_env):
|
|
shutil.copy2(src_env, dst_env)
|
|
try:
|
|
os.chmod(str(dst_env), 0o600)
|
|
except OSError:
|
|
pass
|
|
mirrored["env"] = True
|
|
except Exception:
|
|
pass
|
|
try:
|
|
src_auth = launch_home / "auth.json"
|
|
dst_auth = path / "auth.json"
|
|
if src_auth.is_file() and not dst_auth.exists():
|
|
shutil.copy2(src_auth, dst_auth)
|
|
try:
|
|
os.chmod(str(dst_auth), 0o600)
|
|
except OSError:
|
|
pass
|
|
mirrored["auth"] = True
|
|
except Exception:
|
|
pass
|
|
|
|
model = str(params.get("model") or "").strip()
|
|
provider = str(params.get("provider") or "").strip()
|
|
model_set = False
|
|
if model and provider:
|
|
try:
|
|
from hermes_cli.web_routers.profiles import _write_profile_model
|
|
|
|
_write_profile_model(path, provider, model)
|
|
model_set = True
|
|
except Exception:
|
|
pass
|
|
elif is_truthy_value(params.get("mirror_credentials", True)) and not (path / "config.yaml").exists():
|
|
# No explicit pin and no cloned config: inherit the launch profile's
|
|
# provider+model so the first turn resolves. Same writer as the pin.
|
|
try:
|
|
from hermes_cli.config import load_config_readonly
|
|
from hermes_cli.web_routers.profiles import _write_profile_model
|
|
|
|
cfg = load_config_readonly() or {}
|
|
model_cfg = cfg.get("model") or {}
|
|
inherited_provider = str(model_cfg.get("provider") or "")
|
|
inherited_model = str(model_cfg.get("default") or "")
|
|
if inherited_provider and inherited_model:
|
|
_write_profile_model(path, inherited_provider, inherited_model)
|
|
mirrored["model_inherited"] = True
|
|
except Exception:
|
|
pass
|
|
|
|
return _ok(
|
|
rid,
|
|
{
|
|
"ok": True,
|
|
"name": name,
|
|
"path": str(path),
|
|
"soul_written": soul_written,
|
|
"model_set": model_set,
|
|
"mirrored": mirrored,
|
|
},
|
|
)
|
|
|
|
|
|
def register(server) -> None:
|
|
_registry.install(server)
|