464 lines
18 KiB
Python
464 lines
18 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,
|
|
},
|
|
)
|
|
|
|
|
|
@method("profiles.describe")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Full configuration snapshot of one profile, for an editor UI.
|
|
|
|
Params: ``name`` (required). Result:
|
|
``{name, description, soul, model: {provider, default}, skills:
|
|
[{name, enabled}], toolsets: [{name, description, tool_count, enabled}]}``
|
|
|
|
Skill enablement mirrors the disabled-list model (installed = enabled
|
|
unless in ``skills.disabled``). Toolset enablement reports the profile's
|
|
``tools.enabled_toolsets`` pin, or every toolset enabled when unpinned.
|
|
All reads are scoped to the profile via the HERMES_HOME override.
|
|
"""
|
|
name = str(params.get("name") or "").strip()
|
|
if not name:
|
|
return _err(rid, 4063, "name required")
|
|
try:
|
|
from pathlib import Path
|
|
|
|
from hermes_cli.profiles import get_profile_dir
|
|
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
|
|
|
profile_dir = Path(get_profile_dir(name))
|
|
if not profile_dir.is_dir():
|
|
return _err(rid, 4064, f"profile '{name}' not found")
|
|
|
|
token = set_hermes_home_override(str(profile_dir))
|
|
try:
|
|
from hermes_cli.config import load_config
|
|
from hermes_cli.skills_config import get_disabled_skills
|
|
|
|
cfg = load_config() or {}
|
|
disabled = {s.lower() for s in get_disabled_skills(cfg)}
|
|
|
|
installed = []
|
|
skills_root = profile_dir / "skills"
|
|
if skills_root.is_dir():
|
|
for md in sorted(skills_root.rglob("SKILL.md")):
|
|
skill_name = md.parent.name
|
|
installed.append(
|
|
{"name": skill_name, "enabled": skill_name.lower() not in disabled}
|
|
)
|
|
|
|
from toolsets import get_all_toolsets, get_toolset_info
|
|
|
|
tools_cfg = cfg.get("tools") if isinstance(cfg.get("tools"), dict) else {}
|
|
pinned = tools_cfg.get("enabled_toolsets")
|
|
pinned_set = (
|
|
{str(t).strip() for t in pinned if str(t).strip()}
|
|
if isinstance(pinned, list)
|
|
else None
|
|
)
|
|
toolsets_out = []
|
|
for ts_name in sorted(get_all_toolsets().keys()):
|
|
info = get_toolset_info(ts_name) or {}
|
|
toolsets_out.append(
|
|
{
|
|
"name": ts_name,
|
|
"description": info.get("description") or "",
|
|
"tool_count": len(info.get("tools") or []),
|
|
"enabled": True if pinned_set is None else ts_name in pinned_set,
|
|
}
|
|
)
|
|
|
|
soul_path = profile_dir / "SOUL.md"
|
|
soul = ""
|
|
try:
|
|
if soul_path.is_file():
|
|
soul = soul_path.read_text(encoding="utf-8", errors="replace")
|
|
except Exception:
|
|
pass
|
|
|
|
model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
|
|
|
|
description = ""
|
|
try:
|
|
from hermes_cli.profiles import read_profile_meta
|
|
|
|
description = str(read_profile_meta(profile_dir).get("description") or "")
|
|
except Exception:
|
|
pass
|
|
|
|
return _ok(
|
|
rid,
|
|
{
|
|
"name": name,
|
|
"description": description,
|
|
"soul": soul,
|
|
"model": {
|
|
"provider": str(model_cfg.get("provider") or ""),
|
|
"default": str(model_cfg.get("default") or ""),
|
|
},
|
|
"skills": installed,
|
|
"toolsets": toolsets_out,
|
|
"toolsets_pinned": pinned_set is not None,
|
|
},
|
|
)
|
|
finally:
|
|
reset_hermes_home_override(token)
|
|
except Exception as e:
|
|
return _err(rid, 5063, str(e))
|
|
|
|
|
|
@method("profiles.configure")
|
|
def _(rid, params: dict) -> dict:
|
|
"""Apply configuration changes to a profile (editor Save).
|
|
|
|
Params: ``name`` (required) plus any of:
|
|
``description`` (str), ``soul`` (str, full SOUL.md replacement),
|
|
``model`` + ``provider`` (both required together),
|
|
``disabled_skills`` (list[str], replace semantics),
|
|
``enabled_toolsets`` (list[str], replace semantics; empty list clears
|
|
the pin so every toolset is enabled again).
|
|
|
|
Each section is applied independently and best-effort; the result
|
|
reports per-section success so a UI can surface partial failures.
|
|
"""
|
|
name = str(params.get("name") or "").strip()
|
|
if not name:
|
|
return _err(rid, 4063, "name required")
|
|
try:
|
|
from pathlib import Path
|
|
|
|
from hermes_cli.profiles import get_profile_dir
|
|
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
|
|
|
profile_dir = Path(get_profile_dir(name))
|
|
if not profile_dir.is_dir():
|
|
return _err(rid, 4064, f"profile '{name}' not found")
|
|
|
|
applied = {}
|
|
|
|
if isinstance(params.get("soul"), str):
|
|
try:
|
|
(profile_dir / "SOUL.md").write_text(params["soul"], encoding="utf-8")
|
|
applied["soul"] = True
|
|
except Exception:
|
|
applied["soul"] = False
|
|
|
|
if isinstance(params.get("description"), str):
|
|
try:
|
|
from hermes_cli.profiles import write_profile_meta
|
|
|
|
write_profile_meta(
|
|
profile_dir,
|
|
description=params["description"].strip(),
|
|
description_auto=False,
|
|
)
|
|
applied["description"] = True
|
|
except Exception:
|
|
applied["description"] = False
|
|
|
|
model = str(params.get("model") or "").strip()
|
|
provider = str(params.get("provider") or "").strip()
|
|
if model and provider:
|
|
try:
|
|
from hermes_cli.web_routers.profiles import _write_profile_model
|
|
|
|
_write_profile_model(profile_dir, provider, model)
|
|
applied["model"] = True
|
|
except Exception:
|
|
applied["model"] = False
|
|
|
|
needs_cfg = isinstance(params.get("disabled_skills"), list) or isinstance(
|
|
params.get("enabled_toolsets"), list
|
|
)
|
|
if needs_cfg:
|
|
token = set_hermes_home_override(str(profile_dir))
|
|
try:
|
|
from hermes_cli.config import load_config, save_config
|
|
|
|
cfg = load_config() or {}
|
|
|
|
if isinstance(params.get("disabled_skills"), list):
|
|
try:
|
|
from hermes_cli.skills_config import save_disabled_skills
|
|
|
|
wanted = {
|
|
str(s).strip()
|
|
for s in params["disabled_skills"]
|
|
if str(s).strip()
|
|
}
|
|
save_disabled_skills(cfg, wanted)
|
|
applied["skills"] = True
|
|
cfg = load_config() or {}
|
|
except Exception:
|
|
applied["skills"] = False
|
|
|
|
if isinstance(params.get("enabled_toolsets"), list):
|
|
try:
|
|
wanted = [str(t).strip() for t in params["enabled_toolsets"] if str(t).strip()]
|
|
tools_cfg = cfg.get("tools") if isinstance(cfg.get("tools"), dict) else {}
|
|
if wanted:
|
|
tools_cfg["enabled_toolsets"] = sorted(set(wanted))
|
|
else:
|
|
tools_cfg.pop("enabled_toolsets", None)
|
|
cfg["tools"] = tools_cfg
|
|
save_config(cfg)
|
|
applied["toolsets"] = True
|
|
except Exception:
|
|
applied["toolsets"] = False
|
|
finally:
|
|
reset_hermes_home_override(token)
|
|
|
|
return _ok(rid, {"ok": all(applied.values()) if applied else True, "applied": applied})
|
|
except Exception as e:
|
|
return _err(rid, 5064, str(e))
|
|
|
|
|
|
def register(server) -> None:
|
|
_registry.install(server)
|