Merge pull request #4519 from nesquena/release-v0.51.528-4454
Release v0.51.528 — isolated HERMES_HOME single-profile mode (#2698)
This commit is contained in:
commit
0dddca5b6b
12 changed files with 1016 additions and 38 deletions
|
|
@ -3,6 +3,12 @@
|
|||
|
||||
## [Unreleased]
|
||||
|
||||
## [v0.51.528] — 2026-06-20 — Release SM (isolated HERMES_HOME single-profile mode)
|
||||
|
||||
### Added
|
||||
|
||||
- **Pin the WebUI to a single profile by pointing `HERMES_HOME` at a profile directory (#2698).** When `HERMES_HOME` resolves to `~/.hermes/profiles/<name>` at startup, the WebUI now runs in isolated single-profile mode: it lists only that profile, rejects cross-profile switch/create/delete (403), scopes the WebUI state directory to `<HERMES_HOME>/webui` (so even a profile literally named `default` gets its own sessions/state rather than sharing the root profile's), forces `?all_profiles=1` aggregate reads off (no cross-profile session/project leak), and hides the multi-profile UI affordances (New-profile button, profile dropdown, Manage link). A normal install — `HERMES_HOME` unset or pointing at the base `~/.hermes` — is unaffected: isolation stays off and the state directory and multi-profile behavior are unchanged. Thanks @rodboev (state-dir contract folded in from @tomtong2015's #4449).
|
||||
|
||||
## [v0.51.527] — 2026-06-19 — Release SL (completion notifications survive a backgrounded tab)
|
||||
|
||||
### Fixed
|
||||
|
|
|
|||
|
|
@ -337,7 +337,7 @@ Full list of environment variables:
|
|||
| `HERMES_WEBUI_PYTHON` | auto-discovered | Python executable |
|
||||
| `HERMES_WEBUI_HOST` | `127.0.0.1` | Bind address (`0.0.0.0` for all IPv4, `::` for all IPv6, `::1` for IPv6 loopback) |
|
||||
| `HERMES_WEBUI_PORT` | `8787` | Port |
|
||||
| `HERMES_WEBUI_STATE_DIR` | `$HERMES_HOME/webui` (Windows default `%LOCALAPPDATA%\hermes\webui`, POSIX default `~/.hermes/webui`) | Where sessions and state are stored |
|
||||
| `HERMES_WEBUI_STATE_DIR` | `$HERMES_HOME/webui` (Windows default `%LOCALAPPDATA%\hermes\webui`, POSIX default `~/.hermes/webui`) | Where sessions and state are stored. **Note (upgrade):** the default now follows `HERMES_HOME` — if you previously relocated `HERMES_HOME` to a non-default base **without** setting `HERMES_WEBUI_STATE_DIR`, your WebUI state now resolves to `$HERMES_HOME/webui` instead of the old platform-default `~/.hermes/webui`. To keep using the old location, set `HERMES_WEBUI_STATE_DIR` to it (or move the directory). Installs with `HERMES_HOME` unset or at the default base are unaffected. |
|
||||
| `HERMES_WEBUI_DEFAULT_WORKSPACE` | `~/workspace` | Default workspace |
|
||||
| `HERMES_WEBUI_DEFAULT_MODEL` | *(provider default)* | Optional model override; leave unset to use the active Hermes provider default |
|
||||
| `HERMES_WEBUI_PASSWORD` | *(unset)* | Set to enable password authentication |
|
||||
|
|
@ -348,6 +348,7 @@ Full list of environment variables:
|
|||
| `HERMES_WEBUI_EXTENSION_STYLESHEET_URLS` | *(unset)* | Optional comma-separated same-origin stylesheet URLs to inject; see [WebUI Extensions](docs/EXTENSIONS.md) |
|
||||
| `HERMES_HOME` | Windows: `%LOCALAPPDATA%\hermes`; POSIX: `~/.hermes` | Base directory for Hermes state (affects all paths) |
|
||||
| `HERMES_CONFIG_PATH` | `$HERMES_HOME/config.yaml` | Path to Hermes config file |
|
||||
| `HERMES_WEBUI_SERVER_CWD` | *(unset)* | Working directory for the server process. Defaults to the agent dir; point it at a writable workspace when the agent dir is read-only so fallback relative writes land somewhere writable |
|
||||
| `HERMES_WEBUI_AGENT_CACHE_MAX` | `25` | Max live agent instances kept warm in the in-memory LRU. Each pins a full conversation transcript, so this is the dominant lever on resident memory — lower it on installs with many long sessions to cap RAM (at the cost of more cold reloads) |
|
||||
| `HERMES_WEBUI_SESSIONS_MAX` | `100` | Max compact `Session` objects held in the in-memory LRU. Lighter than the agent cache; lower it on installs with hundreds of sessions |
|
||||
|
||||
|
|
|
|||
|
|
@ -70,9 +70,10 @@ TLS_ENABLED = TLS_CERT is not None and TLS_KEY is not None
|
|||
|
||||
# ── State directory (env-overridable, never inside repo) ──────────────────────
|
||||
_DEFAULT_HERMES_HOME = _platform_default_hermes_home()
|
||||
_DEFAULT_STATE_HOME = Path(os.getenv("HERMES_HOME") or _DEFAULT_HERMES_HOME).expanduser()
|
||||
|
||||
STATE_DIR = (
|
||||
Path(os.getenv("HERMES_WEBUI_STATE_DIR", str(_DEFAULT_HERMES_HOME / "webui")))
|
||||
Path(os.getenv("HERMES_WEBUI_STATE_DIR", str(_DEFAULT_STATE_HOME / "webui")))
|
||||
.expanduser()
|
||||
.resolve()
|
||||
)
|
||||
|
|
|
|||
179
api/profiles.py
179
api/profiles.py
|
|
@ -33,6 +33,15 @@ _PROFILE_DIRS = [
|
|||
]
|
||||
_CLONE_CONFIG_FILES = ['config.yaml', '.env', 'SOUL.md']
|
||||
|
||||
# ── Snapshot initial HERMES_HOME before init_profile_state() mutates it ──────
|
||||
# _is_isolated_profile_mode() needs to detect whether HERMES_HOME *at startup*
|
||||
# points to a profile subdirectory (e.g., ~/.hermes/profiles/user1), not the
|
||||
# base home (~/.hermes). But init_profile_state() overwrites HERMES_HOME to the
|
||||
# base home, which would disable isolation detection. Snapshot it here at import
|
||||
# time (before init_profile_state runs) and use the snapshot in the detector.
|
||||
_INITIAL_HERMES_HOME = os.getenv('HERMES_HOME', '').strip()
|
||||
_ISOLATED_SYMLINK_WARNING_EMITTED = False
|
||||
|
||||
# ── Module state ────────────────────────────────────────────────────────────
|
||||
_active_profile = 'default'
|
||||
_profile_lock = threading.Lock()
|
||||
|
|
@ -116,6 +125,52 @@ def _unwrap_profile_home_to_base(home: Path) -> Path:
|
|||
return home
|
||||
|
||||
|
||||
def _is_isolated_profile_mode() -> bool:
|
||||
"""Detect isolated single-profile mode from HERMES_HOME env var.
|
||||
|
||||
Returns True when HERMES_HOME points at a concrete profile subdirectory
|
||||
(e.g., ~/.hermes/profiles/user1) rather than the base home (~/.hermes).
|
||||
This indicates a single-profile deployment where the WebUI should pin to
|
||||
that profile and reject cross-profile operations.
|
||||
|
||||
Isolation is derived only from the literal ``*/profiles/<name>`` shape of
|
||||
HERMES_HOME at startup. A normal multi-profile deployment points
|
||||
HERMES_HOME at the base home (``~/.hermes``), which does not match the
|
||||
shape, so isolation stays off. We deliberately do not key this off
|
||||
HERMES_BASE_HOME: that variable normally points at the base home already,
|
||||
so using it as an opt-out silently disables legitimate isolated-profile
|
||||
deployments.
|
||||
|
||||
Uses _INITIAL_HERMES_HOME (snapshotted at import time) to detect isolation,
|
||||
not the current os.environ value. init_profile_state() overwrites HERMES_HOME
|
||||
at startup, which would disable isolation detection if we read it here.
|
||||
"""
|
||||
hermes_home = _INITIAL_HERMES_HOME
|
||||
if not hermes_home:
|
||||
return False
|
||||
|
||||
p = Path(hermes_home).expanduser()
|
||||
# Check if this path looks like ~/.hermes/profiles/<name>
|
||||
# i.e., parent dir is named 'profiles' and grandparent exists
|
||||
if p.parent.name == 'profiles' and p.parent.parent.exists():
|
||||
return True
|
||||
if p.is_symlink():
|
||||
global _ISOLATED_SYMLINK_WARNING_EMITTED
|
||||
if not _ISOLATED_SYMLINK_WARNING_EMITTED:
|
||||
logger.warning(
|
||||
"HERMES_HOME symlink %s does not literally match */profiles/<name>; "
|
||||
"isolated profile mode is disabled unless the literal profile path is used.",
|
||||
p,
|
||||
)
|
||||
_ISOLATED_SYMLINK_WARNING_EMITTED = True
|
||||
return False
|
||||
|
||||
|
||||
def _isolated_profile_name() -> str:
|
||||
"""Return the profile directory name from _INITIAL_HERMES_HOME."""
|
||||
return Path(_INITIAL_HERMES_HOME).expanduser().name
|
||||
|
||||
|
||||
def _resolve_base_hermes_home() -> Path:
|
||||
"""Return the BASE ~/.hermes directory — the root that contains profiles/.
|
||||
|
||||
|
|
@ -281,9 +336,12 @@ def get_active_profile_name() -> str:
|
|||
"""Return the currently active profile name.
|
||||
|
||||
Priority:
|
||||
1. Thread-local (set per-request from hermes_profile cookie) — issue #798
|
||||
2. Process-level default (_active_profile)
|
||||
1. Isolated-profile deployment name from the configured HERMES_HOME path
|
||||
2. Thread-local (set per-request from hermes_profile cookie) — issue #798
|
||||
3. Process-level default (_active_profile)
|
||||
"""
|
||||
if _is_isolated_profile_mode():
|
||||
return _isolated_profile_name()
|
||||
tls_name = getattr(_tls, 'profile', None)
|
||||
if tls_name is not None:
|
||||
return tls_name
|
||||
|
|
@ -318,6 +376,17 @@ def _resolve_profile_home_for_name(name: str) -> Path:
|
|||
names fall back to the base home so traversal-shaped cookie values cannot
|
||||
influence filesystem paths.
|
||||
"""
|
||||
# In isolated mode, every logical profile lookup clamps to the configured
|
||||
# startup HERMES_HOME so callers cannot resolve a foreign profile path.
|
||||
if _is_isolated_profile_mode():
|
||||
isolated_name = _isolated_profile_name()
|
||||
isolated_home = Path(_INITIAL_HERMES_HOME).expanduser()
|
||||
if name and not _profiles_match(name, isolated_name):
|
||||
logger.warning(
|
||||
"Ignoring profile lookup %r in isolated profile mode; using pinned profile %r",
|
||||
name, isolated_name,
|
||||
)
|
||||
return isolated_home
|
||||
if not name or _is_root_profile(name):
|
||||
return _DEFAULT_HERMES_HOME
|
||||
if not _PROFILE_ID_RE.fullmatch(name):
|
||||
|
|
@ -331,6 +400,8 @@ def get_active_hermes_home() -> Path:
|
|||
Uses get_active_profile_name() so per-request TLS context (issue #798)
|
||||
is respected, not just the process-level global.
|
||||
"""
|
||||
if _is_isolated_profile_mode():
|
||||
return Path(_INITIAL_HERMES_HOME).expanduser()
|
||||
return _resolve_profile_home_for_name(get_active_profile_name())
|
||||
|
||||
|
||||
|
|
@ -378,6 +449,14 @@ def _home_for_scheduled_cron_job(job: dict) -> Path:
|
|||
fall back to the server default rather than crashing every scheduler tick.
|
||||
"""
|
||||
raw = str((job or {}).get('profile') or '').strip()
|
||||
if _is_isolated_profile_mode():
|
||||
active = _isolated_profile_name()
|
||||
if raw and not _profiles_match(raw, active):
|
||||
logger.warning(
|
||||
"Cron job %s references profile %r outside isolated profile %r; falling back to isolated home",
|
||||
(job or {}).get('id', '?'), raw, active,
|
||||
)
|
||||
return get_active_hermes_home()
|
||||
if not raw:
|
||||
return get_active_hermes_home()
|
||||
if _is_root_profile(raw):
|
||||
|
|
@ -428,6 +507,8 @@ def install_cron_scheduler_profile_isolation() -> None:
|
|||
return original(job, *args, **kwargs)
|
||||
finally:
|
||||
event_profile = str((job or {}).get("profile") or "").strip() or None
|
||||
if _is_isolated_profile_mode():
|
||||
event_profile = _isolated_profile_name()
|
||||
try:
|
||||
publish_session_list_changed("cron_complete", profile=event_profile)
|
||||
except TypeError:
|
||||
|
|
@ -965,8 +1046,12 @@ def init_profile_state() -> None:
|
|||
module-level cached paths. Called once from config.py after imports.
|
||||
"""
|
||||
global _active_profile
|
||||
_active_profile = _read_active_profile_file()
|
||||
home = get_active_hermes_home()
|
||||
if _is_isolated_profile_mode():
|
||||
_active_profile = _isolated_profile_name()
|
||||
home = Path(_INITIAL_HERMES_HOME).expanduser()
|
||||
else:
|
||||
_active_profile = _read_active_profile_file()
|
||||
home = get_active_hermes_home()
|
||||
_set_hermes_home(home)
|
||||
install_cron_scheduler_profile_isolation()
|
||||
_reload_dotenv(home)
|
||||
|
|
@ -978,6 +1063,9 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict:
|
|||
Validates the profile exists, updates process state, patches module caches,
|
||||
reloads .env, and reloads config.yaml.
|
||||
|
||||
In isolated profile mode, switching to a different profile is rejected (403).
|
||||
Switching to the isolated profile itself is allowed (idempotent).
|
||||
|
||||
Args:
|
||||
name: Profile name to switch to.
|
||||
process_wide: If True (default), updates the process-global
|
||||
|
|
@ -985,10 +1073,20 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict:
|
|||
WebUI where the profile is managed via cookie + thread-local (#798).
|
||||
|
||||
Returns: {'profiles': [...], 'active': name}
|
||||
Raises ValueError if profile doesn't exist or agent is busy.
|
||||
Raises ValueError when profile doesn't exist, RuntimeError when agent is running,
|
||||
PermissionError in isolated mode for cross-profile switches.
|
||||
"""
|
||||
global _active_profile
|
||||
|
||||
# In isolated profile mode, reject switching to other profiles
|
||||
if _is_isolated_profile_mode():
|
||||
active = _isolated_profile_name()
|
||||
if name != active:
|
||||
raise PermissionError(
|
||||
f"Profile switching is not allowed in isolated profile mode. "
|
||||
f"Currently pinned to profile '{active}'."
|
||||
)
|
||||
|
||||
# Import here to avoid circular import at module load
|
||||
from api.config import STREAMS, STREAMS_LOCK, reload_config
|
||||
|
||||
|
|
@ -1006,7 +1104,9 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict:
|
|||
)
|
||||
|
||||
# Resolve profile directory
|
||||
if _is_root_profile(name):
|
||||
if _is_isolated_profile_mode():
|
||||
home = Path(_INITIAL_HERMES_HOME).expanduser()
|
||||
elif _is_root_profile(name):
|
||||
home = _DEFAULT_HERMES_HOME
|
||||
else:
|
||||
home = _resolve_named_profile_home(name)
|
||||
|
|
@ -1281,6 +1381,9 @@ def _build_profile_rows_fast() -> list | None:
|
|||
def list_profiles_api() -> list:
|
||||
"""List all profiles with metadata, serialized for JSON response.
|
||||
|
||||
In isolated profile mode (HERMES_HOME points to ~/.hermes/profiles/<name>),
|
||||
returns only that single profile and skips other profiles entirely.
|
||||
|
||||
Fast path: build the rows from upstream's cheap per-profile helpers and skip
|
||||
``find_alias_for_profile`` (whose result the WebUI discards) — see
|
||||
``_build_profile_rows_fast``. Results are cached for a short TTL so rapid
|
||||
|
|
@ -1292,6 +1395,56 @@ def list_profiles_api() -> list:
|
|||
global _LIST_PROFILES_CACHE
|
||||
now = time.time()
|
||||
|
||||
# In isolated profile mode, return only the active (isolated) profile
|
||||
if _is_isolated_profile_mode():
|
||||
active = _isolated_profile_name()
|
||||
hermes_home = Path(_INITIAL_HERMES_HOME).expanduser()
|
||||
try:
|
||||
from hermes_cli.profiles import list_profiles
|
||||
infos = list_profiles()
|
||||
# When the isolated profile is literally named "default", upstream
|
||||
# can surface the base-home row first. Only trust a row whose path
|
||||
# resolves to the same directory as the isolated startup home.
|
||||
for p in infos:
|
||||
try:
|
||||
same_home = Path(p.path).expanduser().resolve() == hermes_home.resolve()
|
||||
except OSError:
|
||||
same_home = False
|
||||
if p.name == active and same_home:
|
||||
enabled_count, total_count = _get_profile_skills_stats(p.path)
|
||||
return [{
|
||||
'name': p.name,
|
||||
'path': str(p.path),
|
||||
'is_default': p.is_default,
|
||||
'is_active': True, # Always true in isolated mode
|
||||
'gateway_running': p.gateway_running,
|
||||
'model': p.model,
|
||||
'provider': p.provider,
|
||||
'has_env': p.has_env,
|
||||
'visible': _profile_visible_from_meta(p.path),
|
||||
'skill_count': enabled_count,
|
||||
'enabled_skills': enabled_count,
|
||||
'total_skills': total_count,
|
||||
}]
|
||||
except (ImportError, OSError, PermissionError):
|
||||
pass
|
||||
# Fallback: construct profile dict with actual active name and hermes_home path
|
||||
enabled_count, total_count = _get_profile_skills_stats(hermes_home)
|
||||
return [{
|
||||
'name': active,
|
||||
'path': str(hermes_home),
|
||||
'is_default': active == 'default',
|
||||
'is_active': True,
|
||||
'gateway_running': False,
|
||||
'model': None,
|
||||
'provider': None,
|
||||
'has_env': (hermes_home / '.env').exists(),
|
||||
'visible': _profile_visible_from_meta(hermes_home),
|
||||
'skill_count': enabled_count,
|
||||
'enabled_skills': enabled_count,
|
||||
'total_skills': total_count,
|
||||
}]
|
||||
|
||||
with _LIST_PROFILES_CACHE_LOCK:
|
||||
cached = _LIST_PROFILES_CACHE
|
||||
if cached is not None and now - cached[1] < _LIST_PROFILES_CACHE_TTL:
|
||||
|
|
@ -1709,7 +1862,12 @@ def create_profile_api(name: str, clone_from: str = None,
|
|||
api_key: str = None,
|
||||
default_model: str = None,
|
||||
model_provider: str = None) -> dict:
|
||||
"""Create a new profile. Returns the new profile info dict."""
|
||||
"""Create a new profile. Returns the new profile info dict.
|
||||
|
||||
In isolated profile mode, profile creation is rejected (403).
|
||||
"""
|
||||
if _is_isolated_profile_mode():
|
||||
raise PermissionError("Profile creation is not allowed in isolated profile mode.")
|
||||
_validate_profile_name(name)
|
||||
# Defense-in-depth: validate clone_from here too, even though routes.py
|
||||
# also validates it. Any caller that bypasses the HTTP layer gets protection.
|
||||
|
|
@ -1808,7 +1966,12 @@ def create_profile_api(name: str, clone_from: str = None,
|
|||
|
||||
|
||||
def delete_profile_api(name: str) -> dict:
|
||||
"""Delete a profile. Switches to default first if it's the active one."""
|
||||
"""Delete a profile. Switches to default first if it's the active one.
|
||||
|
||||
In isolated profile mode, profile deletion is rejected (403).
|
||||
"""
|
||||
if _is_isolated_profile_mode():
|
||||
raise PermissionError("Profile deletion is not allowed in isolated profile mode.")
|
||||
if _is_root_profile(name):
|
||||
raise ValueError("Cannot delete the default profile.")
|
||||
_validate_profile_name(name)
|
||||
|
|
|
|||
|
|
@ -429,6 +429,7 @@ def _visible_pinned_lineage_ids(session_rows) -> set[str]:
|
|||
# module keep resolving without per-call-site refactors.
|
||||
from api.profiles import ( # noqa: F401, E402 (re-export)
|
||||
_profiles_match,
|
||||
_is_isolated_profile_mode,
|
||||
get_active_profile_name,
|
||||
get_active_profile_name as _get_active_profile_name,
|
||||
)
|
||||
|
|
@ -445,6 +446,11 @@ def _all_profiles_query_flag(parsed_url) -> bool:
|
|||
return raw in ('1', 'true', 'yes', 'on')
|
||||
|
||||
|
||||
def _all_profiles_enabled(parsed_url) -> bool:
|
||||
"""Enable aggregate profile reads only when the request asks and mode allows it."""
|
||||
return _all_profiles_query_flag(parsed_url) and not _is_isolated_profile_mode()
|
||||
|
||||
|
||||
def _session_visible_to_active_profile(session_profile, handler=None) -> bool:
|
||||
"""Return whether a detail-load session belongs to the active profile.
|
||||
|
||||
|
|
@ -1927,7 +1933,7 @@ def _build_session_list_cache_payload(
|
|||
other_profile_count = 0
|
||||
else:
|
||||
scoped = [s for s in merged if _profiles_match(s.get("profile"), active_profile)]
|
||||
other_profile_count = len(merged) - len(scoped)
|
||||
other_profile_count = 0 if _is_isolated_profile_mode() else len(merged) - len(scoped)
|
||||
diag_stage("messaging_dedupe")
|
||||
scoped = _keep_latest_messaging_session_per_source(
|
||||
scoped,
|
||||
|
|
@ -8175,7 +8181,7 @@ def handle_get(handler, parsed) -> bool:
|
|||
show_cron_sessions = bool(settings.get("show_cron_sessions"))
|
||||
agent_session_source_filter = settings.get("agent_session_source_filter")
|
||||
active_profile = get_active_profile_name()
|
||||
all_profiles = _all_profiles_query_flag(parsed)
|
||||
all_profiles = _all_profiles_enabled(parsed)
|
||||
key = _session_list_cache_key(
|
||||
active_profile=active_profile,
|
||||
all_profiles=all_profiles,
|
||||
|
|
@ -8213,17 +8219,20 @@ def handle_get(handler, parsed) -> bool:
|
|||
from api.profiles import get_active_profile_name
|
||||
active_profile = get_active_profile_name()
|
||||
all_projects = load_projects()
|
||||
all_profiles = _all_profiles_query_flag(parsed)
|
||||
isolated_profile_mode = _is_isolated_profile_mode()
|
||||
all_profiles = _all_profiles_enabled(parsed)
|
||||
if all_profiles:
|
||||
scoped = all_projects
|
||||
other_profile_count = 0
|
||||
else:
|
||||
scoped = [p for p in all_projects
|
||||
if _profiles_match(p.get("profile"), active_profile)]
|
||||
other_profile_count = 0 if isolated_profile_mode else len(all_projects) - len(scoped)
|
||||
return j(handler, {
|
||||
"projects": scoped,
|
||||
"all_profiles": all_profiles,
|
||||
"active_profile": active_profile,
|
||||
"other_profile_count": len(all_projects) - len(scoped),
|
||||
"other_profile_count": other_profile_count,
|
||||
})
|
||||
|
||||
if parsed.path == "/api/prompts":
|
||||
|
|
@ -8586,11 +8595,15 @@ def handle_get(handler, parsed) -> bool:
|
|||
|
||||
# ── Profile API (GET) ──
|
||||
if parsed.path == "/api/profiles":
|
||||
from api.profiles import list_profiles_api, get_active_profile_name
|
||||
from api.profiles import list_profiles_api
|
||||
|
||||
return j(
|
||||
handler,
|
||||
{"profiles": list_profiles_api(), "active": get_active_profile_name()},
|
||||
{
|
||||
"profiles": list_profiles_api(),
|
||||
"active": get_active_profile_name(),
|
||||
"single_profile_mode": _is_isolated_profile_mode(),
|
||||
},
|
||||
)
|
||||
|
||||
if parsed.path == "/api/profile/active":
|
||||
|
|
@ -10064,6 +10077,8 @@ def handle_post(handler, parsed) -> bool:
|
|||
return j(handler, result, extra_headers={
|
||||
'Set-Cookie': build_profile_cookie(name, handler),
|
||||
})
|
||||
except PermissionError as e:
|
||||
return bad(handler, _sanitize_error(e), 403)
|
||||
except (ValueError, FileNotFoundError) as e:
|
||||
return bad(handler, _sanitize_error(e), 404)
|
||||
except RuntimeError as e:
|
||||
|
|
@ -10104,6 +10119,8 @@ def handle_post(handler, parsed) -> bool:
|
|||
model_provider=model_provider,
|
||||
)
|
||||
return j(handler, {"ok": True, "profile": result})
|
||||
except PermissionError as e:
|
||||
return bad(handler, _sanitize_error(e), 403)
|
||||
except (ValueError, FileExistsError, RuntimeError) as e:
|
||||
return bad(handler, str(e))
|
||||
|
||||
|
|
@ -10117,6 +10134,8 @@ def handle_post(handler, parsed) -> bool:
|
|||
_validate_profile_name(name)
|
||||
result = delete_profile_api(name)
|
||||
return j(handler, result)
|
||||
except PermissionError as e:
|
||||
return bad(handler, _sanitize_error(e), 403)
|
||||
except (ValueError, FileNotFoundError) as e:
|
||||
return bad(handler, _sanitize_error(e))
|
||||
except RuntimeError as e:
|
||||
|
|
@ -11177,7 +11196,7 @@ def _handle_sessions_search(handler, parsed):
|
|||
content_search = qs.get("content", ["1"])[0] == "1"
|
||||
from api.profiles import get_active_profile_name
|
||||
active_profile = get_active_profile_name()
|
||||
all_profiles = _all_profiles_query_flag(parsed)
|
||||
all_profiles = _all_profiles_enabled(parsed)
|
||||
sessions = all_sessions()
|
||||
if not all_profiles:
|
||||
sessions = [
|
||||
|
|
@ -18065,6 +18084,8 @@ def _handle_session_import_cli(handler, body):
|
|||
if requested_profile == "":
|
||||
return bad(handler, "invalid profile", 400)
|
||||
allow_all_profiles = _request_wants_all_profiles_import(body)
|
||||
if allow_all_profiles and _is_isolated_profile_mode():
|
||||
return bad(handler, "all_profiles import is not allowed in isolated profile mode", 403)
|
||||
if allow_all_profiles and not requested_profile:
|
||||
return bad(handler, "profile is required for all_profiles import", 400)
|
||||
|
||||
|
|
|
|||
|
|
@ -494,7 +494,8 @@ def main() -> int:
|
|||
|
||||
python_exe = ensure_python_has_webui_deps(discover_launcher_python(agent_dir), agent_dir)
|
||||
state_dir = Path(
|
||||
os.getenv("HERMES_WEBUI_STATE_DIR", str(Path.home() / ".hermes" / "webui"))
|
||||
os.getenv("HERMES_WEBUI_STATE_DIR")
|
||||
or Path(os.getenv("HERMES_HOME") or (Path.home() / ".hermes")) / "webui"
|
||||
).expanduser()
|
||||
state_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
|
|
@ -505,7 +506,8 @@ def main() -> int:
|
|||
if agent_dir:
|
||||
os.environ["HERMES_WEBUI_AGENT_DIR"] = str(agent_dir)
|
||||
|
||||
server_cwd = str(agent_dir or REPO_ROOT)
|
||||
# Let operators move fallback relative writes out of a read-only agent dir.
|
||||
server_cwd = os.environ.get("HERMES_WEBUI_SERVER_CWD", "").strip() or str(agent_dir or REPO_ROOT)
|
||||
server_path = str(REPO_ROOT / "server.py")
|
||||
# Scheme the server will advertise (HTTPS when TLS cert+key are configured).
|
||||
scheme = "https" if _tls_probe_enabled() else "http"
|
||||
|
|
@ -547,6 +549,8 @@ def main() -> int:
|
|||
# for SO_EXCLUSIVEADDRUSE.
|
||||
_CREATE_NEW_PROCESS_GROUP = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)
|
||||
subprocess.Popen([python_exe, server_path],
|
||||
cwd=server_cwd,
|
||||
env=os.environ.copy(),
|
||||
creationflags=_CREATE_NEW_PROCESS_GROUP)
|
||||
sys.exit(0)
|
||||
os.execv(python_exe, [python_exe, server_path])
|
||||
|
|
|
|||
|
|
@ -5566,17 +5566,28 @@ async function loadProfilesPanel() {
|
|||
const data = await api('/api/profiles');
|
||||
_profilesCache = data;
|
||||
panel.innerHTML = '';
|
||||
const explainer = document.createElement('div');
|
||||
explainer.className = 'profile-card profile-help-card';
|
||||
explainer.innerHTML = `
|
||||
<div class="profile-card-header">
|
||||
<div style="min-width:0;flex:1">
|
||||
<div class="profile-card-name">Profiles vs workspaces</div>
|
||||
<div class="profile-card-meta">Use profiles for how the agent works; use workspaces for what files it works on.</div>
|
||||
</div>
|
||||
</div>`;
|
||||
explainer.onclick = () => _renderProfileConceptHelp(data.active || 'default');
|
||||
panel.appendChild(explainer);
|
||||
|
||||
// Hide "New profile" button in single profile mode
|
||||
const newProfileBtn = document.querySelector('[onclick="openProfileCreate()"]');
|
||||
if (newProfileBtn) {
|
||||
newProfileBtn.style.display = data.single_profile_mode ? 'none' : '';
|
||||
}
|
||||
|
||||
// In single profile mode, don't show the explanatory card
|
||||
if (!data.single_profile_mode) {
|
||||
const explainer = document.createElement('div');
|
||||
explainer.className = 'profile-card profile-help-card';
|
||||
explainer.innerHTML = `
|
||||
<div class="profile-card-header">
|
||||
<div style="min-width:0;flex:1">
|
||||
<div class="profile-card-name">Profiles vs workspaces</div>
|
||||
<div class="profile-card-meta">Use profiles for how the agent works; use workspaces for what files it works on.</div>
|
||||
</div>
|
||||
</div>`;
|
||||
explainer.onclick = () => _renderProfileConceptHelp(data.active || 'default');
|
||||
panel.appendChild(explainer);
|
||||
}
|
||||
|
||||
if (!data.profiles || !data.profiles.length) {
|
||||
const emptyMsg = document.createElement('div');
|
||||
emptyMsg.style.cssText = 'padding:16px;color:var(--muted);font-size:12px';
|
||||
|
|
@ -5695,8 +5706,9 @@ function _setProfileHeaderButtons(mode, p, activeName){
|
|||
if (mode === 'read') {
|
||||
const isActive = p && p.name === activeName;
|
||||
const isDefault = !!(p && p.is_default);
|
||||
if (isActive) hide(actBtn); else show(actBtn);
|
||||
if (isDefault) hide(delBtn); else show(delBtn);
|
||||
const singleProfileMode = !!(_profilesCache && _profilesCache.single_profile_mode);
|
||||
if (isActive || singleProfileMode) hide(actBtn); else show(actBtn);
|
||||
if (isDefault || singleProfileMode) hide(delBtn); else show(delBtn);
|
||||
hide(cancelBtn); hide(saveBtn);
|
||||
} else if (mode === 'create') {
|
||||
hide(actBtn); hide(delBtn); show(cancelBtn); show(saveBtn);
|
||||
|
|
@ -5774,12 +5786,14 @@ function renderProfileDropdown(data) {
|
|||
};
|
||||
dd.appendChild(opt);
|
||||
}
|
||||
// Divider + Manage link
|
||||
const div = document.createElement('div'); div.className = 'ws-divider'; dd.appendChild(div);
|
||||
const mgmt = document.createElement('div'); mgmt.className = 'profile-opt ws-manage';
|
||||
mgmt.innerHTML = `${li('settings',12)} ${esc(t('manage_profiles'))}`;
|
||||
mgmt.onclick = () => { closeProfileDropdown(); mobileSwitchPanel('profiles'); };
|
||||
dd.appendChild(mgmt);
|
||||
// Divider + Manage link (hidden in single profile mode)
|
||||
if (!data.single_profile_mode) {
|
||||
const div = document.createElement('div'); div.className = 'ws-divider'; dd.appendChild(div);
|
||||
const mgmt = document.createElement('div'); mgmt.className = 'profile-opt ws-manage';
|
||||
mgmt.innerHTML = `${li('settings',12)} ${esc(t('manage_profiles'))}`;
|
||||
mgmt.onclick = () => { closeProfileDropdown(); mobileSwitchPanel('profiles'); };
|
||||
dd.appendChild(mgmt);
|
||||
}
|
||||
}
|
||||
|
||||
function toggleProfileDropdown() {
|
||||
|
|
@ -5789,6 +5803,11 @@ function toggleProfileDropdown() {
|
|||
closeWsDropdown(); // close workspace dropdown if open
|
||||
if(typeof closeModelDropdown==='function') closeModelDropdown();
|
||||
api('/api/profiles').then(data => {
|
||||
// In single profile mode, don't show profile dropdown at all
|
||||
if (data.single_profile_mode) {
|
||||
closeProfileDropdown();
|
||||
return;
|
||||
}
|
||||
renderProfileDropdown(data);
|
||||
dd.classList.add('open');
|
||||
_positionProfileDropdown();
|
||||
|
|
|
|||
|
|
@ -92,6 +92,9 @@ def clean_env(monkeypatch):
|
|||
"HERMES_WEBUI_HOST",
|
||||
"HERMES_WEBUI_PORT",
|
||||
"HERMES_WEBUI_AGENT_DIR",
|
||||
"HERMES_WEBUI_STATE_DIR",
|
||||
"HERMES_WEBUI_SERVER_CWD",
|
||||
"HERMES_HOME",
|
||||
):
|
||||
monkeypatch.delenv(name, raising=False)
|
||||
|
||||
|
|
@ -346,6 +349,27 @@ class TestMainForegroundRouting:
|
|||
bs.main()
|
||||
assert len(execv_calls) == 1
|
||||
|
||||
def test_foreground_defaults_state_dir_to_hermes_home_webui(self, stub_main_dependencies, clean_env, monkeypatch, tmp_path):
|
||||
bs = stub_main_dependencies
|
||||
hermes_home = tmp_path / ".hermes" / "profiles" / "webui"
|
||||
hermes_home.mkdir(parents=True, exist_ok=True)
|
||||
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
|
||||
monkeypatch.delenv("HERMES_WEBUI_STATE_DIR", raising=False)
|
||||
monkeypatch.setattr(sys, "argv", ["bootstrap.py", "--foreground"])
|
||||
monkeypatch.setattr(sys, "platform", "linux")
|
||||
monkeypatch.setattr(os, "chdir", lambda p: None)
|
||||
monkeypatch.setattr(os, "access", lambda path, mode: True)
|
||||
|
||||
def fake_execv(*_args):
|
||||
raise SystemExit(0)
|
||||
|
||||
monkeypatch.setattr(os, "execv", fake_execv)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
bs.main()
|
||||
|
||||
assert os.environ["HERMES_WEBUI_STATE_DIR"] == str(hermes_home / "webui")
|
||||
|
||||
|
||||
class TestForegroundEnvAndCwd:
|
||||
"""The post-execv server.py inherits os.environ and cwd from us."""
|
||||
|
|
@ -384,6 +408,49 @@ class TestForegroundEnvAndCwd:
|
|||
assert len(chdir_calls) == 1
|
||||
assert chdir_calls[0] == str(agent_dir)
|
||||
|
||||
def test_foreground_chdirs_to_server_cwd_override_before_launch(self, setup, monkeypatch, clean_env, tmp_path):
|
||||
bs, _agent_dir = setup
|
||||
workspace = tmp_path / "workspace"
|
||||
workspace.mkdir()
|
||||
monkeypatch.setenv("HERMES_WEBUI_SERVER_CWD", str(workspace))
|
||||
monkeypatch.setattr(sys, "argv", ["bootstrap.py", "--foreground"])
|
||||
monkeypatch.setattr(sys, "platform", "linux")
|
||||
|
||||
chdir_calls = []
|
||||
monkeypatch.setattr(os, "chdir", lambda p: chdir_calls.append(p))
|
||||
|
||||
def fake_execv(*_args):
|
||||
raise SystemExit(0)
|
||||
|
||||
monkeypatch.setattr(os, "execv", fake_execv)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
bs.main()
|
||||
|
||||
assert chdir_calls == [str(workspace)]
|
||||
|
||||
def test_windows_foreground_popen_uses_server_cwd_override(self, setup, monkeypatch, clean_env, tmp_path):
|
||||
bs, _agent_dir = setup
|
||||
workspace = tmp_path / "workspace-win"
|
||||
workspace.mkdir()
|
||||
monkeypatch.setenv("HERMES_WEBUI_SERVER_CWD", str(workspace))
|
||||
monkeypatch.setattr(sys, "argv", ["bootstrap.py", "--foreground"])
|
||||
monkeypatch.setattr(sys, "platform", "win32")
|
||||
monkeypatch.setattr(os, "chdir", lambda p: None)
|
||||
|
||||
popen_calls = []
|
||||
|
||||
def fake_popen(*args, **kwargs):
|
||||
popen_calls.append((args, kwargs))
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(subprocess, "Popen", fake_popen)
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
bs.main()
|
||||
|
||||
assert popen_calls[0][1]["cwd"] == str(workspace)
|
||||
|
||||
def test_foreground_exports_resolved_env_vars(self, setup, monkeypatch, clean_env):
|
||||
bs, agent_dir = setup
|
||||
monkeypatch.setattr(sys, "argv", [
|
||||
|
|
|
|||
|
|
@ -106,6 +106,22 @@ def test_all_profiles_query_flag_false_values():
|
|||
assert _all_profiles_query_flag(u) is False, f"path {path!r} should be false"
|
||||
|
||||
|
||||
def test_all_profiles_enabled_in_normal_mode(monkeypatch):
|
||||
"""The aggregate toggle still works outside isolated-profile mode."""
|
||||
import api.routes as routes
|
||||
|
||||
monkeypatch.setattr(routes, "_is_isolated_profile_mode", lambda: False)
|
||||
assert routes._all_profiles_enabled(urlparse('/api/sessions?all_profiles=1')) is True
|
||||
|
||||
|
||||
def test_all_profiles_disabled_in_isolated_mode(monkeypatch):
|
||||
"""An isolated deployment must ignore all_profiles=1 aggregate requests."""
|
||||
import api.routes as routes
|
||||
|
||||
monkeypatch.setattr(routes, "_is_isolated_profile_mode", lambda: True)
|
||||
assert routes._all_profiles_enabled(urlparse('/api/sessions?all_profiles=1')) is False
|
||||
|
||||
|
||||
# ── No client-side CLI bypass ──────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
575
tests/test_issue2698_isolated_hermes_home.py
Normal file
575
tests/test_issue2698_isolated_hermes_home.py
Normal file
|
|
@ -0,0 +1,575 @@
|
|||
"""
|
||||
Tests for issue #2698: HERMES_HOME isolated profile mode.
|
||||
|
||||
When HERMES_HOME points at a specific profile directory like ~/.hermes/profiles/user1,
|
||||
the WebUI should pin to that single profile: list only it, reject create/switch/delete
|
||||
of other profiles, and hide multi-profile UI affordances.
|
||||
"""
|
||||
|
||||
import os
|
||||
import io
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
|
||||
import api.profiles as _profiles_mod
|
||||
from api.profiles import (
|
||||
_is_isolated_profile_mode,
|
||||
clear_request_profile,
|
||||
list_profiles_api,
|
||||
create_profile_api,
|
||||
delete_profile_api,
|
||||
get_active_hermes_home,
|
||||
get_active_profile_name,
|
||||
init_profile_state,
|
||||
set_request_profile,
|
||||
switch_profile,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_profile_cache():
|
||||
"""Clear the profile list cache before and after each test to prevent cross-test leakage."""
|
||||
_profiles_mod._LIST_PROFILES_CACHE = None
|
||||
yield
|
||||
_profiles_mod._LIST_PROFILES_CACHE = None
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_hermes_home():
|
||||
"""Create a temporary .hermes directory structure for testing."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
home = Path(tmpdir) / ".hermes"
|
||||
home.mkdir()
|
||||
profiles_root = home / "profiles"
|
||||
profiles_root.mkdir()
|
||||
yield home
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def temp_single_profile():
|
||||
"""Create a temporary .hermes/profiles/user1 structure for isolated mode."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
home = Path(tmpdir) / ".hermes"
|
||||
home.mkdir()
|
||||
profiles_root = home / "profiles"
|
||||
profiles_root.mkdir()
|
||||
user1 = profiles_root / "user1"
|
||||
user1.mkdir()
|
||||
# Create required subdirs
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(user1 / subdir).mkdir(exist_ok=True)
|
||||
yield user1
|
||||
|
||||
|
||||
class TestIsolatedProfileModeDetection:
|
||||
"""Test _is_isolated_profile_mode() helper."""
|
||||
|
||||
def test_normal_mode_when_hermes_home_is_base(self, temp_hermes_home):
|
||||
"""Normal mode when HERMES_HOME points to base ~/.hermes."""
|
||||
with mock.patch.dict(os.environ, {"HERMES_HOME": str(temp_hermes_home)}):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", temp_hermes_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_hermes_home)):
|
||||
isolated = _is_isolated_profile_mode()
|
||||
assert isolated is False
|
||||
|
||||
def test_isolated_mode_when_hermes_home_is_profile_subdir(self, temp_single_profile):
|
||||
"""Isolated mode when HERMES_HOME points to ~/.hermes/profiles/user1."""
|
||||
# Ensure we're in the fixture context where temp_single_profile exists
|
||||
assert temp_single_profile.exists(), f"Test fixture path doesn't exist: {temp_single_profile}"
|
||||
assert temp_single_profile.parent.name == "profiles", f"Parent not named 'profiles': {temp_single_profile.parent}"
|
||||
|
||||
# _is_isolated_profile_mode() uses _INITIAL_HERMES_HOME (snapshotted at import time),
|
||||
# not the current os.environ value. Patch both for this test.
|
||||
with mock.patch.dict(os.environ, {"HERMES_HOME": str(temp_single_profile)}, clear=False):
|
||||
with mock.patch.dict(os.environ, {"HERMES_BASE_HOME": ""}, clear=False):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
isolated = _is_isolated_profile_mode()
|
||||
assert isolated is True, f"Expected isolated mode for {temp_single_profile}"
|
||||
|
||||
def test_hermes_base_home_does_not_disable_isolation(self, temp_single_profile):
|
||||
"""HERMES_BASE_HOME must not disable isolation for a profiles/<name> path."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": str(base_home),
|
||||
},
|
||||
):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
isolated = _is_isolated_profile_mode()
|
||||
assert isolated is True
|
||||
|
||||
|
||||
class TestListProfilesInIsolatedMode:
|
||||
"""Test list_profiles_api() returns only isolated profile when in isolated mode."""
|
||||
|
||||
def test_list_returns_all_profiles_in_normal_mode(self, temp_hermes_home):
|
||||
"""Normal mode lists all profiles."""
|
||||
# Create a few test profiles
|
||||
profiles_root = temp_hermes_home / "profiles"
|
||||
(profiles_root / "user1").mkdir()
|
||||
(profiles_root / "user2").mkdir()
|
||||
(profiles_root / "user3").mkdir()
|
||||
|
||||
# Create required subdirs for each
|
||||
for prof_dir in profiles_root.iterdir():
|
||||
if prof_dir.is_dir():
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(prof_dir / subdir).mkdir(exist_ok=True)
|
||||
|
||||
with mock.patch.dict(os.environ, {"HERMES_HOME": str(temp_hermes_home)}):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", temp_hermes_home):
|
||||
with mock.patch("api.profiles._is_isolated_profile_mode", return_value=False):
|
||||
# Mock _get_profile_skills_stats to avoid importing agent.skill_utils
|
||||
with mock.patch("api.profiles._get_profile_skills_stats", return_value=(0, 0)):
|
||||
# Mock _build_profile_rows_fast to return the expected profiles
|
||||
def mock_build_profiles():
|
||||
return [
|
||||
{'name': 'default', 'path': str(temp_hermes_home), 'is_default': True,
|
||||
'is_active': False, 'gateway_running': False, 'model': None, 'provider': None,
|
||||
'has_env': False, 'visible': True, 'skill_count': 0, 'enabled_skills': 0, 'total_skills': 0},
|
||||
{'name': 'user1', 'path': str(profiles_root / 'user1'), 'is_default': False,
|
||||
'is_active': False, 'gateway_running': False, 'model': None, 'provider': None,
|
||||
'has_env': False, 'visible': True, 'skill_count': 0, 'enabled_skills': 0, 'total_skills': 0},
|
||||
{'name': 'user2', 'path': str(profiles_root / 'user2'), 'is_default': False,
|
||||
'is_active': False, 'gateway_running': False, 'model': None, 'provider': None,
|
||||
'has_env': False, 'visible': True, 'skill_count': 0, 'enabled_skills': 0, 'total_skills': 0},
|
||||
{'name': 'user3', 'path': str(profiles_root / 'user3'), 'is_default': False,
|
||||
'is_active': False, 'gateway_running': False, 'model': None, 'provider': None,
|
||||
'has_env': False, 'visible': True, 'skill_count': 0, 'enabled_skills': 0, 'total_skills': 0},
|
||||
]
|
||||
with mock.patch("api.profiles._build_profile_rows_fast", side_effect=mock_build_profiles):
|
||||
profiles = list_profiles_api()
|
||||
# Should have at least 'default' plus the created profiles
|
||||
names = [p["name"] for p in profiles]
|
||||
assert "user1" in names
|
||||
assert "user2" in names
|
||||
assert "user3" in names
|
||||
|
||||
def test_list_returns_only_isolated_profile_in_isolated_mode(self, temp_single_profile):
|
||||
"""Isolated mode lists only the configured profile."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
# Create other profiles that should be hidden
|
||||
other_profiles = base_home / "profiles"
|
||||
(other_profiles / "user2").mkdir()
|
||||
(other_profiles / "user3").mkdir()
|
||||
for prof_dir in [other_profiles / "user2", other_profiles / "user3"]:
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(prof_dir / subdir).mkdir(exist_ok=True)
|
||||
|
||||
# Clear HERMES_BASE_HOME to allow isolated mode detection to fire
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
# Mock _get_profile_skills_stats to avoid importing agent.skill_utils
|
||||
with mock.patch("api.profiles._get_profile_skills_stats", return_value=(0, 0)):
|
||||
profiles = list_profiles_api()
|
||||
# Should only have user1
|
||||
assert len(profiles) == 1
|
||||
assert profiles[0]["name"] == "user1"
|
||||
|
||||
def test_list_includes_single_profile_mode_flag(self, temp_single_profile):
|
||||
"""Response includes single_profile_mode: true in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
with mock.patch.dict(os.environ, {"HERMES_HOME": str(temp_single_profile)}):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
with mock.patch("api.profiles._is_isolated_profile_mode", return_value=True):
|
||||
with mock.patch("api.profiles._resolve_base_hermes_home", return_value=base_home):
|
||||
# Mock _get_profile_skills_stats to avoid importing agent.skill_utils
|
||||
with mock.patch("api.profiles._get_profile_skills_stats", return_value=(0, 0)):
|
||||
profiles = list_profiles_api()
|
||||
# Check for single_profile_mode flag in response structure
|
||||
# For now, profiles should be a list; the flag will be in routes.py response
|
||||
assert len(profiles) == 1
|
||||
|
||||
|
||||
class TestIsolatedRuntimePinning:
|
||||
"""Test that isolated mode pins both detection and active runtime home."""
|
||||
|
||||
def test_get_active_profile_name_ignores_tls_and_global_in_isolated_mode(self, temp_single_profile):
|
||||
"""Isolated mode must ignore request TLS and process-global active profile."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
with mock.patch("api.profiles._active_profile", "other_profile"):
|
||||
set_request_profile("tls_profile")
|
||||
try:
|
||||
assert get_active_profile_name() == "user1"
|
||||
finally:
|
||||
clear_request_profile()
|
||||
|
||||
@pytest.mark.parametrize("active_profile_contents", [None, "user2"])
|
||||
def test_init_profile_state_pins_runtime_home_to_isolated_profile(
|
||||
self,
|
||||
temp_single_profile,
|
||||
monkeypatch,
|
||||
active_profile_contents,
|
||||
):
|
||||
"""init_profile_state must ignore the base active_profile file in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
other_profile = base_home / "profiles" / "user2"
|
||||
other_profile.mkdir()
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(other_profile / subdir).mkdir(exist_ok=True)
|
||||
|
||||
active_profile_file = base_home / "active_profile"
|
||||
if active_profile_contents is None:
|
||||
active_profile_file.unlink(missing_ok=True)
|
||||
else:
|
||||
active_profile_file.write_text(active_profile_contents, encoding="utf-8")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", base_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setattr(_profiles_mod, "_active_profile", "default")
|
||||
monkeypatch.setattr(_profiles_mod, "install_cron_scheduler_profile_isolation", lambda: None)
|
||||
|
||||
reloaded = []
|
||||
real_reload_dotenv = _profiles_mod._reload_dotenv
|
||||
|
||||
def _record_reload(home):
|
||||
reloaded.append(home)
|
||||
return real_reload_dotenv(home)
|
||||
|
||||
monkeypatch.setattr(_profiles_mod, "_reload_dotenv", _record_reload)
|
||||
|
||||
init_profile_state()
|
||||
|
||||
assert _profiles_mod._active_profile == "user1"
|
||||
assert get_active_profile_name() == "user1"
|
||||
assert get_active_hermes_home() == temp_single_profile
|
||||
assert Path(os.environ["HERMES_HOME"]) == temp_single_profile
|
||||
assert reloaded == [temp_single_profile]
|
||||
|
||||
def test_get_active_hermes_home_keeps_profiles_default_pinned(self, temp_hermes_home, monkeypatch):
|
||||
"""An isolated profiles/default path must not collapse back to the base home."""
|
||||
isolated_default = temp_hermes_home / "profiles" / "default"
|
||||
isolated_default.mkdir()
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(isolated_default / subdir).mkdir(exist_ok=True)
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", temp_hermes_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setattr(_profiles_mod, "_active_profile", "other_profile")
|
||||
|
||||
assert get_active_profile_name() == "default"
|
||||
assert get_active_hermes_home() == isolated_default
|
||||
|
||||
def test_explicit_profile_resolution_for_isolated_default_uses_pinned_home(self, temp_hermes_home, monkeypatch):
|
||||
"""Explicit default-profile resolution must stay on the isolated home."""
|
||||
from api.profiles import get_hermes_home_for_profile, _resolve_profile_home_for_name
|
||||
|
||||
isolated_default = temp_hermes_home / "profiles" / "default"
|
||||
isolated_default.mkdir()
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(isolated_default / subdir).mkdir(exist_ok=True)
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", temp_hermes_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(isolated_default))
|
||||
|
||||
assert _resolve_profile_home_for_name("default") == isolated_default
|
||||
assert get_hermes_home_for_profile("default") == isolated_default
|
||||
assert get_hermes_home_for_profile("default") != temp_hermes_home
|
||||
|
||||
def test_explicit_foreign_profile_resolution_stays_on_isolated_home(self, temp_single_profile, monkeypatch):
|
||||
"""Explicit profile lookups must not escape the isolated startup home."""
|
||||
from api.profiles import get_hermes_home_for_profile, _resolve_profile_home_for_name
|
||||
|
||||
base_home = temp_single_profile.parent.parent
|
||||
monkeypatch.setenv("HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", base_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(temp_single_profile))
|
||||
|
||||
assert _resolve_profile_home_for_name("other_profile") == temp_single_profile
|
||||
assert get_hermes_home_for_profile("other_profile") == temp_single_profile
|
||||
|
||||
def test_list_profiles_prefers_matching_isolated_default_home(self, temp_hermes_home, monkeypatch):
|
||||
"""The isolated profiles/default row must not collapse to the base-home row."""
|
||||
isolated_default = temp_hermes_home / "profiles" / "default"
|
||||
isolated_default.mkdir(parents=True)
|
||||
for subdir in ["memories", "sessions", "skills", "skins", "logs", "plans", "workspace", "cron"]:
|
||||
(isolated_default / subdir).mkdir(exist_ok=True)
|
||||
|
||||
class _Info:
|
||||
def __init__(self, name, path, is_default):
|
||||
self.name = name
|
||||
self.path = path
|
||||
self.is_default = is_default
|
||||
self.gateway_running = False
|
||||
self.model = None
|
||||
self.provider = None
|
||||
self.has_env = False
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", temp_hermes_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setattr(_profiles_mod, "_get_profile_skills_stats", lambda _path: (0, 0))
|
||||
|
||||
# Inject a STUB hermes_cli.profiles into sys.modules rather than
|
||||
# mock.patch("hermes_cli.profiles.list_profiles", ...): that string
|
||||
# target forces a real import of hermes_cli, which is NOT installed in
|
||||
# CI (the bundled agent isn't on the WebUI test path) -> ModuleNotFoundError.
|
||||
# list_profiles_api() does `from hermes_cli.profiles import list_profiles`
|
||||
# internally, so a stub module satisfies it at the WebUI boundary. (#4454 CI)
|
||||
stub_profiles = types.ModuleType("hermes_cli.profiles")
|
||||
stub_profiles.list_profiles = lambda: [
|
||||
_Info("default", temp_hermes_home, True),
|
||||
_Info("default", isolated_default, False),
|
||||
]
|
||||
stub_pkg = sys.modules.get("hermes_cli") or types.ModuleType("hermes_cli")
|
||||
monkeypatch.setitem(sys.modules, "hermes_cli", stub_pkg)
|
||||
monkeypatch.setitem(sys.modules, "hermes_cli.profiles", stub_profiles)
|
||||
|
||||
rows = list_profiles_api()
|
||||
|
||||
assert len(rows) == 1
|
||||
assert Path(rows[0]["path"]) == isolated_default
|
||||
|
||||
def test_list_profiles_isolated_fallback_does_not_reenter_root_profile_lookup(self, temp_single_profile, monkeypatch):
|
||||
"""The isolated fallback must not recurse back through _is_root_profile."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
monkeypatch.setenv("HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", base_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setattr(_profiles_mod, "_get_profile_skills_stats", lambda _path: (0, 0))
|
||||
monkeypatch.setattr(
|
||||
_profiles_mod,
|
||||
"_is_root_profile",
|
||||
lambda _name: (_ for _ in ()).throw(AssertionError("_is_root_profile should not run in isolated fallback")),
|
||||
)
|
||||
|
||||
with mock.patch.dict(sys.modules, {"hermes_cli": None, "hermes_cli.profiles": None}):
|
||||
rows = list_profiles_api()
|
||||
|
||||
assert len(rows) == 1
|
||||
assert rows[0]["name"] == "user1"
|
||||
assert rows[0]["is_default"] is False
|
||||
|
||||
|
||||
class TestProfileMutationsInIsolatedMode:
|
||||
"""Test that create/delete/switch are rejected (403) in isolated mode."""
|
||||
|
||||
def test_create_profile_rejected_in_isolated_mode(self, temp_single_profile):
|
||||
"""create_profile_api should reject creation in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
# Clear HERMES_BASE_HOME to allow isolated mode detection to fire
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
with pytest.raises(PermissionError, match=".*isolated.*|.*single.*"):
|
||||
create_profile_api("newprofile")
|
||||
|
||||
def test_delete_profile_rejected_in_isolated_mode(self, temp_single_profile):
|
||||
"""delete_profile_api should reject deletion in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
# Clear HERMES_BASE_HOME to allow isolated mode detection to fire
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
with pytest.raises(PermissionError, match=".*isolated.*|.*single.*"):
|
||||
delete_profile_api("user1")
|
||||
|
||||
def test_switch_to_different_profile_rejected(self, temp_single_profile):
|
||||
"""switch_profile should reject switching to another profile in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
with pytest.raises(PermissionError, match=".*isolated.*|.*pinned.*"):
|
||||
switch_profile("other_user")
|
||||
|
||||
def test_switch_to_same_profile_idempotent(self, temp_single_profile):
|
||||
"""switch_profile to the isolated profile itself should pass through."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
env_dict = {
|
||||
"HERMES_HOME": str(temp_single_profile),
|
||||
"HERMES_BASE_HOME": "",
|
||||
}
|
||||
with mock.patch.dict(os.environ, env_dict, clear=False):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", base_home):
|
||||
with mock.patch("api.profiles._INITIAL_HERMES_HOME", str(temp_single_profile)):
|
||||
# Should not raise PermissionError; may fail downstream
|
||||
# for other reasons (missing hermes_cli), but the isolation
|
||||
# guard must pass for the same-name case.
|
||||
try:
|
||||
switch_profile("user1")
|
||||
except PermissionError:
|
||||
pytest.fail("switch_profile should allow switching to the isolated profile itself")
|
||||
except (ImportError, ValueError, RuntimeError):
|
||||
pass # expected in test env without hermes_cli
|
||||
|
||||
def test_switch_to_same_default_profile_keeps_pinned_home(self, temp_hermes_home, monkeypatch, tmp_path):
|
||||
"""A same-name switch for isolated profiles/default must keep using the pinned home."""
|
||||
isolated_default = temp_hermes_home / "profiles" / "default"
|
||||
isolated_default.mkdir(parents=True)
|
||||
(isolated_default / "workspace").mkdir()
|
||||
base_workspace = tmp_path / "base-workspace"
|
||||
isolated_workspace = tmp_path / "isolated-workspace"
|
||||
base_workspace.mkdir()
|
||||
isolated_workspace.mkdir()
|
||||
(temp_hermes_home / "config.yaml").write_text(f"workspace: {base_workspace}\n", encoding="utf-8")
|
||||
(isolated_default / "config.yaml").write_text(f"workspace: {isolated_workspace}\n", encoding="utf-8")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", temp_hermes_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(isolated_default))
|
||||
monkeypatch.setattr(_profiles_mod, "list_profiles_api", lambda: [])
|
||||
|
||||
result = switch_profile("default", process_wide=False)
|
||||
|
||||
assert result["default_workspace"] == str(isolated_workspace.resolve())
|
||||
|
||||
def test_scheduled_cron_jobs_stay_pinned_to_isolated_home(self, temp_single_profile, monkeypatch):
|
||||
"""Scheduler jobs must not resolve foreign profile homes in isolated mode."""
|
||||
base_home = temp_single_profile.parent.parent
|
||||
monkeypatch.setenv("HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", base_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(temp_single_profile))
|
||||
|
||||
assert _profiles_mod._home_for_scheduled_cron_job({"id": "job2698", "profile": "other"}) == temp_single_profile
|
||||
|
||||
def test_cli_import_all_profiles_is_rejected_in_isolated_mode(self, monkeypatch):
|
||||
"""Direct all_profiles CLI imports must not bypass isolated-profile boundaries."""
|
||||
import api.routes as routes
|
||||
|
||||
captured = {}
|
||||
|
||||
class _Handler:
|
||||
def __init__(self):
|
||||
self.wfile = io.BytesIO()
|
||||
|
||||
def send_response(self, status):
|
||||
self.status = status
|
||||
|
||||
def send_header(self, key, value):
|
||||
pass
|
||||
|
||||
def end_headers(self):
|
||||
pass
|
||||
|
||||
monkeypatch.setattr(routes, "_is_isolated_profile_mode", lambda: True)
|
||||
monkeypatch.setattr(
|
||||
routes,
|
||||
"bad",
|
||||
lambda h, m, c=400: (captured.__setitem__("bad", (m, c)), True)[1],
|
||||
)
|
||||
|
||||
routes._handle_session_import_cli(
|
||||
_Handler(),
|
||||
{"session_id": "foreign-cli-2698", "all_profiles": 1, "profile": "other"},
|
||||
)
|
||||
|
||||
assert captured["bad"] == (
|
||||
"all_profiles import is not allowed in isolated profile mode",
|
||||
403,
|
||||
)
|
||||
|
||||
def test_scheduler_publishes_isolated_profile_after_foreign_job_profile(self, temp_single_profile, monkeypatch):
|
||||
"""Scheduled cron completion must publish the isolated profile identity."""
|
||||
events = []
|
||||
base_home = temp_single_profile.parent.parent
|
||||
|
||||
cron_pkg = types.ModuleType("cron")
|
||||
cron_pkg.__path__ = []
|
||||
cron_scheduler = types.ModuleType("cron.scheduler")
|
||||
cron_scheduler.run_job = lambda job: events.append(("run", job["id"])) or "ok"
|
||||
|
||||
class _Ctx:
|
||||
def __init__(self, home):
|
||||
self.home = str(home)
|
||||
|
||||
def __enter__(self):
|
||||
events.append(("enter", self.home))
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
events.append(("exit", self.home))
|
||||
return False
|
||||
|
||||
monkeypatch.setitem(sys.modules, "cron", cron_pkg)
|
||||
monkeypatch.setitem(sys.modules, "cron.scheduler", cron_scheduler)
|
||||
monkeypatch.setenv("HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setenv("HERMES_BASE_HOME", "")
|
||||
monkeypatch.setattr(_profiles_mod, "_DEFAULT_HERMES_HOME", base_home)
|
||||
monkeypatch.setattr(_profiles_mod, "_INITIAL_HERMES_HOME", str(temp_single_profile))
|
||||
monkeypatch.setattr(_profiles_mod, "cron_profile_context_for_home", _Ctx)
|
||||
monkeypatch.setattr(
|
||||
_profiles_mod,
|
||||
"publish_session_list_changed",
|
||||
lambda reason, profile=None: events.append(("publish", reason, profile)),
|
||||
)
|
||||
|
||||
_profiles_mod.install_cron_scheduler_profile_isolation()
|
||||
|
||||
assert cron_scheduler.run_job({"id": "job2698", "profile": "other"}) == "ok"
|
||||
assert events[-1] == ("publish", "cron_complete", "user1")
|
||||
|
||||
|
||||
class TestNormalModePreservation:
|
||||
"""Test that normal mode behavior is completely unchanged."""
|
||||
|
||||
def test_normal_mode_profile_operations_work(self, temp_hermes_home):
|
||||
"""Normal mode allows profile creation and deletion."""
|
||||
with mock.patch.dict(os.environ, {"HERMES_HOME": str(temp_hermes_home)}):
|
||||
with mock.patch("api.profiles._DEFAULT_HERMES_HOME", temp_hermes_home):
|
||||
with mock.patch("api.profiles._is_isolated_profile_mode", return_value=False):
|
||||
# Normal mode should not raise errors for create/delete operations
|
||||
# (though they may fail for other reasons in this test environment)
|
||||
try:
|
||||
# Just verify the isolation guard doesn't trigger
|
||||
from api.profiles import create_profile_api
|
||||
# The actual call might fail due to missing hermes_cli,
|
||||
# but should NOT fail with an "isolated mode" error
|
||||
try:
|
||||
create_profile_api("testprof1")
|
||||
except ValueError as e:
|
||||
# Should be a different error, not about isolation
|
||||
assert "isolated" not in str(e).lower()
|
||||
assert "single" not in str(e).lower()
|
||||
except ImportError:
|
||||
# hermes_cli not available, skip
|
||||
pass
|
||||
|
||||
|
||||
def test_profiles_panel_hides_delete_controls_in_single_profile_mode():
|
||||
panels_js = (Path(__file__).resolve().parents[1] / "static" / "panels.js").read_text(encoding="utf-8")
|
||||
|
||||
assert "const singleProfileMode = !!(_profilesCache && _profilesCache.single_profile_mode);" in panels_js
|
||||
assert "if (isDefault || singleProfileMode) hide(delBtn); else show(delBtn);" in panels_js
|
||||
|
|
@ -5,6 +5,8 @@ import types
|
|||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
|
@ -21,6 +23,15 @@ def _profile_row(name: str, path: Path, *, is_default: bool = False):
|
|||
)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_profile_rows_cache():
|
||||
import api.profiles as profiles
|
||||
|
||||
profiles._LIST_PROFILES_CACHE = None
|
||||
yield
|
||||
profiles._LIST_PROFILES_CACHE = None
|
||||
|
||||
|
||||
def _install_fake_hermes_profiles(monkeypatch, rows):
|
||||
hermes_cli = types.ModuleType("hermes_cli")
|
||||
profiles_mod = types.ModuleType("hermes_cli.profiles")
|
||||
|
|
|
|||
94
tests/test_issue4449_state_dir_contract.py
Normal file
94
tests/test_issue4449_state_dir_contract.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""Regression tests for the #4449 state-dir contract folded into #4454.
|
||||
|
||||
`api.config.STATE_DIR` is a module-level global computed from the environment at
|
||||
import time. Exercising how it's derived requires importing config under a
|
||||
specific HERMES_HOME / HERMES_WEBUI_STATE_DIR. We do this in a SUBPROCESS rather
|
||||
than `importlib.reload(config)` in-process: reloading config inside the shared
|
||||
pytest process leaks recomputed globals (STATE_DIR/SESSION_DIR) and stale
|
||||
references into later tests (the cancel/stream/health/state-isolation suites),
|
||||
which is fragile no matter how carefully env is restored. A subprocess gives a
|
||||
truly isolated import and can't pollute the test process.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
_PROBE = (
|
||||
"import api.config as c; "
|
||||
"print(c.STATE_DIR)"
|
||||
)
|
||||
|
||||
|
||||
def _state_dir_for_env(**env_overrides) -> Path:
|
||||
"""Import api.config in a fresh subprocess under the given env and return
|
||||
the resolved STATE_DIR it computes. None-valued overrides unset the var."""
|
||||
env = dict(os.environ)
|
||||
# Start from a clean slate for the two vars under test so the parent
|
||||
# process's pytest values don't bleed in.
|
||||
env.pop("HERMES_HOME", None)
|
||||
env.pop("HERMES_WEBUI_STATE_DIR", None)
|
||||
for key, value in env_overrides.items():
|
||||
if value is None:
|
||||
env.pop(key, None)
|
||||
else:
|
||||
env[key] = str(value)
|
||||
out = subprocess.run(
|
||||
[sys.executable, "-c", _PROBE],
|
||||
cwd=str(REPO_ROOT),
|
||||
env=env,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=60,
|
||||
)
|
||||
assert out.returncode == 0, f"probe failed: {out.stderr}"
|
||||
return Path(out.stdout.strip()).resolve()
|
||||
|
||||
|
||||
def _platform_default_state_dir() -> Path:
|
||||
"""What STATE_DIR resolves to with HERMES_HOME + STATE_DIR both unset."""
|
||||
return _state_dir_for_env()
|
||||
|
||||
|
||||
def test_config_state_dir_defaults_to_hermes_home_webui(tmp_path):
|
||||
hermes_home = tmp_path / ".hermes" / "profiles" / "isolated"
|
||||
hermes_home.mkdir(parents=True)
|
||||
|
||||
state_dir = _state_dir_for_env(HERMES_HOME=hermes_home)
|
||||
assert state_dir == (hermes_home / "webui").resolve()
|
||||
|
||||
|
||||
def test_config_state_dir_unchanged_for_normal_install_hermes_home_unset():
|
||||
"""Backward-compat: with HERMES_HOME unset, STATE_DIR stays at the platform
|
||||
default `<~/.hermes>/webui` — a normal install's state must NOT relocate
|
||||
(the #4449/#4454 state-dir move only affects an explicitly-set HERMES_HOME).
|
||||
|
||||
Cross-check: the unset-HERMES_HOME result must NOT equal the result of
|
||||
pointing HERMES_HOME at an arbitrary other base — i.e. the default is
|
||||
genuinely the platform home, not whatever the test environment injected."""
|
||||
default = _platform_default_state_dir()
|
||||
assert default.name == "webui"
|
||||
# Pointing HERMES_HOME elsewhere produces a DIFFERENT dir, proving the unset
|
||||
# case resolves to the platform default rather than echoing an injected base.
|
||||
elsewhere = _state_dir_for_env(HERMES_HOME="/tmp/hermes-4449-elsewhere-base")
|
||||
assert elsewhere == Path("/tmp/hermes-4449-elsewhere-base/webui").resolve()
|
||||
assert default != elsewhere
|
||||
|
||||
|
||||
def test_config_state_dir_explicit_override_takes_precedence(tmp_path):
|
||||
"""HERMES_WEBUI_STATE_DIR always wins over the HERMES_HOME-derived default,
|
||||
so an operator who pinned a state dir keeps it even in isolated mode."""
|
||||
hermes_home = tmp_path / ".hermes" / "profiles" / "isolated"
|
||||
hermes_home.mkdir(parents=True)
|
||||
explicit = tmp_path / "custom-state"
|
||||
|
||||
state_dir = _state_dir_for_env(
|
||||
HERMES_HOME=hermes_home,
|
||||
HERMES_WEBUI_STATE_DIR=explicit,
|
||||
)
|
||||
assert state_dir == explicit.resolve()
|
||||
Loading…
Reference in a new issue