DasPoschi/hermes-webui
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Harden worktree removal safeguards

Browse Source
This commit is contained in:
Frank Song
2026-05-13 09:49:15 +08:00
parent 93b7d35bfa
commit 46c62851ad
6 changed files with 209 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
api/routes.py
View File

@@ -4273,36 +4273,6 @@ def handle_post(handler, parsed) -> bool:
logger.debug("Failed to close workspace terminal after workspace update")
set_last_workspace(new_ws)
return j(handler, {"session": s.compact() | {"messages": s.messages}})
if parsed.path == "/api/session/worktree/status":
query = parse_qs(parsed.query)
sid = query.get("session_id", [""])[0]
if not sid:
return bad(handler, "session_id is required", status=400)
try:
s = get_session(sid, metadata_only=True)
except KeyError:
return bad(handler, "Session not found", status=404)
try:
from api.worktrees import worktree_status_for_session
return j(handler, {"status": worktree_status_for_session(s)})
except ValueError as exc:
return bad(handler, str(exc), status=400)
except Exception as exc:
logger.exception("failed to read worktree status for session %s", sid)
return bad(handler, _sanitize_error(exc), status=500)
if parsed.path == "/api/session/compress/status":
query = parse_qs(parsed.query)
return _handle_session_compress_status(handler, query.get("session_id", [""])[0])
if parsed.path == "/api/session":
import time as _time
_t0 = _time.monotonic()
_debug_slow = os.environ.get("HERMES_DEBUG_SLOW", "")
if parsed.path == "/api/session/worktree/remove":
sid = body.get("session_id", "")
if not sid or not isinstance(sid, str) or not sid.strip():

21
api/worktrees.py
View File

@@ -236,7 +236,7 @@ def remove_worktree_for_session(session, *, force: bool = False) -> dict:
if status["locked_by_terminal"]:
raise ValueError("Worktree is locked by an active terminal session")
# Guard: dirty / untracked files without force
# Guard: local changes and unpushed commits without explicit force.
if status["dirty"] and not force:
raise ValueError(
"Worktree has uncommitted changes. Use force=true to override."
@@ -251,17 +251,26 @@ def remove_worktree_for_session(session, *, force: bool = False) -> dict:
f"Worktree has {status['untracked_count']} untracked file(s). "
"Use force=true to override."
)
ahead = int((status.get("ahead_behind") or {}).get("ahead") or 0)
if ahead > 0:
if force:
warnings.append(f"{ahead} unpushed commit(s) will be removed.")
else:
raise ValueError(
f"Worktree has {ahead} unpushed commit(s). "
"Use force=true to override."
)
# Remove the worktree — must run from the repo root, not the worktree dir
repo_root = getattr(session, "worktree_repo_root", None)
if not repo_root:
raise ValueError("Session missing worktree_repo_root")
try:
result = _run_git(
["worktree", "remove", "--force", str(worktree_path)],
str(repo_root),
timeout=10,
)
remove_args = ["worktree", "remove"]
if force:
remove_args.append("--force")
remove_args.append(str(worktree_path))
result = _run_git(remove_args, str(repo_root), timeout=10)
except (OSError, subprocess.TimeoutExpired) as exc:
raise ValueError(f"Failed to remove worktree: {exc}") from exc

1
static/i18n.js
View File

@@ -440,6 +440,7 @@ const LOCALES = {
session_worktree_remove_status_failed: 'Failed to read worktree status: ',
session_worktree_remove_locked_by_stream: 'Cannot remove — an active streaming session is using this worktree.',
session_worktree_remove_locked_by_terminal: 'Cannot remove — an active terminal session is using this worktree.',
session_worktree_remove_unsafe_blocked: 'Resolve local changes or unpushed commits before removing this worktree.',
session_worktree_remove_dirty_warning: 'WARNING: This worktree has uncommitted changes which will be lost.',
session_worktree_remove_untracked_warning: (count) => `${count} untracked file(s) will be permanently deleted.`,
session_worktree_remove_ahead_warning: (ahead) => `${ahead} unpushed commit(s) will be lost.`,

13
static/sessions.js
View File

@@ -3201,6 +3201,16 @@ async function removeWorktree(session){
if(status.ahead_behind&&status.ahead_behind.ahead>0){
details+='\n'+t('session_worktree_remove_ahead_warning',status.ahead_behind.ahead);
}
if(status.dirty||status.untracked_count>0||(status.ahead_behind&&status.ahead_behind.ahead>0)){
showToast(t('session_worktree_remove_failed')+t('session_worktree_remove_unsafe_blocked'),0,'error');
await showConfirmDialog({
message:details,
confirmLabel:t('dialog_confirm_btn'),
danger:true,
focusCancel:true
});
return;
}
}
const ok=await showConfirmDialog({
message:details,
@@ -3208,11 +3218,10 @@ async function removeWorktree(session){
danger:true
});
if(!ok)return;
const force=(status.dirty||status.untracked_count>0);
try{
const result=await api('/api/session/worktree/remove',{
method:'POST',
body:JSON.stringify({session_id:session.session_id, force:force})
body:JSON.stringify({session_id:session.session_id, force:false})
});
const warn=result.warnings&&result.warnings.length?(' '+result.warnings.join(' ')):'';
showToast(t('session_worktree_removed')+warn);

12
tests/test_issue2057_worktree_ui_static.py
View File

@@ -68,3 +68,15 @@ def test_worktree_archive_delete_api_responses_are_explicit():
assert '"worktree_retained": True' in src
assert '{"ok": True, **worktree_retained}' in src
assert '{"ok": True, "session": s.compact(), **_worktree_retained_payload(s)}' in src
def test_remove_worktree_ui_does_not_force_unsafe_status_by_default():
src = read("static/sessions.js")
i18n = read("static/i18n.js")
assert "async function removeWorktree(session)" in src
assert "status.dirty||status.untracked_count>0||(status.ahead_behind&&status.ahead_behind.ahead>0)" in src
assert "session_worktree_remove_unsafe_blocked" in src
assert "session_worktree_remove_unsafe_blocked" in i18n
assert "Resolve local changes or unpushed commits before removing this worktree." in i18n
assert "JSON.stringify({session_id:session.session_id, force:false})" in src
assert "const force=(status.dirty||status.untracked_count>0)" not in src

170
tests/test_worktree_remove.py
View File

@@ -3,6 +3,8 @@
from types import SimpleNamespace
from pathlib import Path
import pytest
import api.models as models
import api.routes as routes
import api.worktrees as worktrees
@@ -76,6 +78,167 @@ def test_remove_clean_worktree_succeeds(tmp_path):
assert not wt_path.exists()
def test_remove_clean_worktree_does_not_force(tmp_path, monkeypatch):
from api.models import Session
worktree_path = tmp_path / "wt_clean"
worktree_path.mkdir()
repo_root = tmp_path / "repo"
repo_root.mkdir()
s = Session(
session_id="testcleanforce",
title="Clean",
workspace=str(worktree_path),
worktree_path=str(worktree_path),
worktree_branch="hermes/testcleanforce",
worktree_repo_root=str(repo_root),
)
monkeypatch.setattr(worktrees, "worktree_status_for_session", lambda session: {
"exists": True,
"dirty": False,
"untracked_count": 0,
"ahead_behind": {"ahead": 0, "behind": 0, "available": False, "upstream": None},
"locked_by_stream": False,
"locked_by_terminal": False,
})
calls = []
def fake_run_git(args, cwd, timeout=2):
calls.append(args)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(worktrees, "_run_git", fake_run_git)
result = worktrees.remove_worktree_for_session(s, force=False)
assert result["ok"] is True
assert calls[0] == ["worktree", "remove", str(worktree_path.resolve())]
def test_remove_dirty_worktree_without_force_is_rejected(tmp_path, monkeypatch):
from api.models import Session
worktree_path = tmp_path / "wt_dirty"
worktree_path.mkdir()
repo_root = tmp_path / "repo"
repo_root.mkdir()
s = Session(
session_id="testdirty",
title="Dirty",
workspace=str(worktree_path),
worktree_path=str(worktree_path),
worktree_branch="hermes/testdirty",
worktree_repo_root=str(repo_root),
)
monkeypatch.setattr(worktrees, "worktree_status_for_session", lambda session: {
"exists": True,
"dirty": True,
"untracked_count": 0,
"ahead_behind": {"ahead": 0, "behind": 0, "available": False, "upstream": None},
"locked_by_stream": False,
"locked_by_terminal": False,
})
monkeypatch.setattr(worktrees, "_run_git", lambda *args, **kwargs: pytest.fail("git remove should not run"))
with pytest.raises(ValueError, match="uncommitted changes"):
worktrees.remove_worktree_for_session(s, force=False)
def test_remove_untracked_worktree_without_force_is_rejected(tmp_path, monkeypatch):
from api.models import Session
worktree_path = tmp_path / "wt_untracked"
worktree_path.mkdir()
repo_root = tmp_path / "repo"
repo_root.mkdir()
s = Session(
session_id="testuntracked",
title="Untracked",
workspace=str(worktree_path),
worktree_path=str(worktree_path),
worktree_branch="hermes/testuntracked",
worktree_repo_root=str(repo_root),
)
monkeypatch.setattr(worktrees, "worktree_status_for_session", lambda session: {
"exists": True,
"dirty": False,
"untracked_count": 2,
"ahead_behind": {"ahead": 0, "behind": 0, "available": False, "upstream": None},
"locked_by_stream": False,
"locked_by_terminal": False,
})
monkeypatch.setattr(worktrees, "_run_git", lambda *args, **kwargs: pytest.fail("git remove should not run"))
with pytest.raises(ValueError, match="untracked"):
worktrees.remove_worktree_for_session(s, force=False)
def test_remove_ahead_worktree_without_force_is_rejected(tmp_path, monkeypatch):
from api.models import Session
worktree_path = tmp_path / "wt_ahead"
worktree_path.mkdir()
repo_root = tmp_path / "repo"
repo_root.mkdir()
s = Session(
session_id="testahead",
title="Ahead",
workspace=str(worktree_path),
worktree_path=str(worktree_path),
worktree_branch="hermes/testahead",
worktree_repo_root=str(repo_root),
)
monkeypatch.setattr(worktrees, "worktree_status_for_session", lambda session: {
"exists": True,
"dirty": False,
"untracked_count": 0,
"ahead_behind": {"ahead": 1, "behind": 0, "available": True, "upstream": "origin/main"},
"locked_by_stream": False,
"locked_by_terminal": False,
})
monkeypatch.setattr(worktrees, "_run_git", lambda *args, **kwargs: pytest.fail("git remove should not run"))
with pytest.raises(ValueError, match="unpushed"):
worktrees.remove_worktree_for_session(s, force=False)
def test_remove_force_warns_and_uses_git_force(tmp_path, monkeypatch):
from api.models import Session
worktree_path = tmp_path / "wt_force"
worktree_path.mkdir()
repo_root = tmp_path / "repo"
repo_root.mkdir()
s = Session(
session_id="testforce",
title="Force",
workspace=str(worktree_path),
worktree_path=str(worktree_path),
worktree_branch="hermes/testforce",
worktree_repo_root=str(repo_root),
)
monkeypatch.setattr(worktrees, "worktree_status_for_session", lambda session: {
"exists": True,
"dirty": True,
"untracked_count": 3,
"ahead_behind": {"ahead": 2, "behind": 0, "available": True, "upstream": "origin/main"},
"locked_by_stream": False,
"locked_by_terminal": False,
})
calls = []
def fake_run_git(args, cwd, timeout=2):
calls.append(args)
return SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(worktrees, "_run_git", fake_run_git)
result = worktrees.remove_worktree_for_session(s, force=True)
assert result["ok"] is True
assert calls[0] == ["worktree", "remove", "--force", str(worktree_path.resolve())]
assert "untracked file" in " ".join(result["warnings"])
assert "unpushed commit" in " ".join(result["warnings"])
def test_remove_worktree_not_exists(tmp_path):
from api.models import Session
@@ -163,3 +326,10 @@ def test_remove_missing_session_returns_404(tmp_path, monkeypatch):
routes.handle_post(object(), SimpleNamespace(path="/api/session/worktree/remove"))
assert captured["status"] == 404
assert "not found" in captured["payload"].get("error", "").lower()
def test_post_router_does_not_expose_read_only_worktree_or_compress_status():
src = Path("api/routes.py").read_text(encoding="utf-8")
post_body = src[src.index("def handle_post"):src.index('if parsed.path == "/api/session/worktree/remove"')]
assert '"/api/session/worktree/status"' not in post_body
assert '"/api/session/compress/status"' not in post_body