fix: recover from stale deleted workspaces

This commit is contained in:
dobby-d-elf
2026-05-12 05:52:16 -06:00
parent 15d620392f
commit e03c197cdf
3 changed files with 136 additions and 11 deletions
+40 -1
View File
@@ -6967,7 +6967,7 @@ def _handle_chat_start(handler, body, diag=None):
attachments = _normalize_chat_attachments(body.get("attachments") or [])[:20]
diag.stage("resolve_workspace") if diag else None
try:
workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace))
workspace = _resolve_chat_workspace_with_recovery(s, body.get("workspace"))
except ValueError as e:
return bad(handler, str(e))
requested_model = body.get("model") or s.model
@@ -7000,6 +7000,45 @@ def _handle_chat_start(handler, body, diag=None):
def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str:
"""Resolve a chat workspace, recovering stale implicit session paths.
If the browser explicitly sent a workspace, preserve the existing strict
validation behaviour and surface any error to the user.
If the browser omitted ``workspace`` and the session's stored workspace now
points at a deleted directory (common after old test workspaces are cleaned
up), fall back to the current last/default workspace and persist the repair
so the chat becomes usable again.
"""
explicit = requested_workspace not in (None, "")
candidate = requested_workspace if explicit else getattr(s, "workspace", None)
try:
return str(resolve_trusted_workspace(candidate))
except ValueError:
if explicit:
raise
fallback = str(resolve_trusted_workspace(get_last_workspace()))
stale = str(candidate or "").strip()
if stale and fallback != stale:
logger.warning(
"Recovered stale session workspace for %s: %s -> %s",
getattr(s, "session_id", "unknown"),
stale,
fallback,
)
s.workspace = fallback
try:
s.save()
except Exception:
logger.debug(
"Failed to persist recovered workspace for session %s",
getattr(s, "session_id", "unknown"),
)
return fallback
raise
def _normalize_chat_attachments(raw_attachments):
"""Normalize attachment payloads from the browser.
+24 -10
View File
@@ -20,11 +20,26 @@ logger = logging.getLogger(__name__)
from api.config import (
WORKSPACES_FILE as _GLOBAL_WS_FILE,
LAST_WORKSPACE_FILE as _GLOBAL_LW_FILE,
DEFAULT_WORKSPACE as _BOOT_DEFAULT_WORKSPACE,
MAX_FILE_BYTES, IMAGE_EXTS, MD_EXTS
)
def _current_default_workspace() -> Path:
"""Return the live default workspace from api.config.
``api.config.DEFAULT_WORKSPACE`` is mutable at runtime (for example after
``save_settings()``). Importing it once into this module bakes in a stale
snapshot that can diverge from the actual current default and leak deleted
test workspaces back into live sessions.
"""
try:
from api import config as _config
return Path(_config.DEFAULT_WORKSPACE).expanduser().resolve()
except Exception:
return Path.home().expanduser().resolve()
# ── Profile-aware path resolution ───────────────────────────────────────────
def _profile_state_dir() -> Path:
@@ -64,7 +79,7 @@ def _profile_default_workspace() -> str:
2. 'default_workspace' — alternate explicit key
3. 'terminal.cwd' — hermes-agent terminal working dir (most common)
Falls back to the boot-time DEFAULT_WORKSPACE constant.
Falls back to the live DEFAULT_WORKSPACE from api.config.
"""
try:
from api.config import get_config
@@ -86,7 +101,7 @@ def _profile_default_workspace() -> str:
return str(p)
except (ImportError, Exception):
logger.debug("Failed to load profile default workspace config")
return str(_BOOT_DEFAULT_WORKSPACE)
return str(_current_default_workspace())
# ── Public API ──────────────────────────────────────────────────────────────
@@ -427,7 +442,7 @@ def _trusted_workspace_roots() -> list[Path]:
roots.append(p)
add(Path.home())
add(_BOOT_DEFAULT_WORKSPACE)
add(_current_default_workspace())
for w in load_workspaces():
add(w.get("path"))
roots.sort(key=lambda p: len(str(p)))
@@ -536,11 +551,10 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
/boot, /proc, /sys, /dev, /root on Linux/macOS; Windows system dirs).
This prevents even admin-saved workspaces from pointing at OS internals.
None/empty path falls back to the boot-time DEFAULT_WORKSPACE, which is always
trusted (it was validated at server startup).
None/empty path falls back to the current DEFAULT_WORKSPACE.
"""
if path in (None, ""):
return Path(_BOOT_DEFAULT_WORKSPACE).expanduser().resolve()
return _current_default_workspace()
candidate = Path(path).expanduser().resolve()
@@ -571,14 +585,14 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
except Exception:
pass
# (C) Trusted if it is equal to or under the boot-time DEFAULT_WORKSPACE.
# (C) Trusted if it is equal to or under the current DEFAULT_WORKSPACE.
# In Docker deployments HERMES_WEBUI_DEFAULT_WORKSPACE is often set to a
# volume mount outside the user's home (e.g. /data/workspace). That path
# was already validated at server startup, so any sub-path of it is safe
# without requiring the user to add it to the workspace list manually.
try:
boot_default = Path(_BOOT_DEFAULT_WORKSPACE).expanduser().resolve()
candidate.relative_to(boot_default)
current_default = _current_default_workspace()
candidate.relative_to(current_default)
return candidate
except ValueError:
pass
+72
View File
@@ -0,0 +1,72 @@
from pathlib import Path
from types import SimpleNamespace
import pytest
from api import config as api_config
from api import routes, workspace
def test_profile_default_workspace_uses_live_config_default(monkeypatch, tmp_path):
live_default = tmp_path / "live-default"
live_default.mkdir()
monkeypatch.setattr(api_config, "DEFAULT_WORKSPACE", live_default)
monkeypatch.setattr(api_config, "get_config", lambda: {})
assert workspace._profile_default_workspace() == str(live_default.resolve())
assert workspace.resolve_trusted_workspace(None) == live_default.resolve()
def test_resolve_chat_workspace_with_recovery_repairs_missing_implicit_workspace(monkeypatch, tmp_path):
fallback = tmp_path / "fallback"
fallback.mkdir()
stale = tmp_path / "deleted-workspace"
def fake_resolve(value):
if value == str(stale):
raise ValueError(f"Path does not exist: {stale}")
return Path(value).resolve()
saved = {"count": 0}
def fake_save():
saved["count"] += 1
session = SimpleNamespace(session_id="sess-1", workspace=str(stale), save=fake_save)
monkeypatch.setattr(routes, "resolve_trusted_workspace", fake_resolve)
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
resolved = routes._resolve_chat_workspace_with_recovery(session, None)
assert resolved == str(fallback.resolve())
assert session.workspace == str(fallback.resolve())
assert saved["count"] == 1
def test_resolve_chat_workspace_with_recovery_preserves_explicit_errors(monkeypatch, tmp_path):
fallback = tmp_path / "fallback"
fallback.mkdir()
stale = tmp_path / "deleted-workspace"
def fake_resolve(value):
if value == str(stale):
raise ValueError(f"Path does not exist: {stale}")
return Path(value).resolve()
saved = {"count": 0}
def fake_save():
saved["count"] += 1
session = SimpleNamespace(session_id="sess-2", workspace=str(fallback), save=fake_save)
monkeypatch.setattr(routes, "resolve_trusted_workspace", fake_resolve)
monkeypatch.setattr(routes, "get_last_workspace", lambda: str(fallback))
with pytest.raises(ValueError, match="Path does not exist"):
routes._resolve_chat_workspace_with_recovery(session, str(stale))
assert session.workspace == str(fallback)
assert saved["count"] == 0