fix: drop auto-delete; manual-instruction only (PR #5688 round-3)

Address round-2 gate cert (issuecomment-4896321817) on sha:dcec232d:

1. BRICK (Codex strace-verified): v2's fcntl-flock holder check on
   .git/index.lock was wrong. Codex straced git 2.43.0 and proved
   that git uses O_CREAT|O_EXCL + rename(2), NOT advisory locking,
   so an fcntl-flock probe returns False for a lock file a live git
   process is holding open. The 'fail-closed' claim was wrong:
   auto-delete could still race and corrupt the index.

2. CORE (Codex): if .git/index.lock vanished between _is_lock_held
   and os.remove, the FileNotFoundError surfaced as a 'remove-failed'
   diagnostic instead of 'already gone' and prevented the apply retry.

Fix: stop deleting lock files from the server entirely. The only thing
that removes a lock is the user, on the host, via the manual command
surfaced in the response. Once the lock is gone, the user re-clicks
Update Now and the normal non-destructive apply path runs.

Implementation:

api/updates.py:
  - Removed _is_lock_held (fcntl-flock probe) -- was provably unsafe.
  - Removed _try_remove_lock (os.remove path) -- was provably unsafe.
  - Removed _GIT_LOCK_FILES_REMOVABLE (no longer enumerable by name).
  - Added _inventory_locks(path) -> dict: pure inventory of well-known
    + other lock files under .git/. Never touches anything.
  - Rewrote apply_clear_lock to inventory-only + manual-instruction.
    When .git/index.lock is absent, runs the normal non-destructive
    apply path; when present, returns ok=False with the exact 'rm -f'
    command + an explanation of why the server cannot do this safely
    (O_CREAT|O_EXCL cannot be detected with advisory probes).
    CORE-1 race evaporates: there is no os.remove to race against.

api/routes.py:
  - Updated the /api/updates/clear_lock comment to reflect v2.2.

static/ui.js:
  - applyClearUpdateLock: when res.lock_held, call new
    _renderLockManualInstruction(target, res) which surfaces the exact
    'rm -f <path>' command in a copyable code block plus an
    'I've removed the lock -- retry update' button that POSTs the
    same endpoint again; the second call takes the success branch
    and re-runs the normal apply path. Also lists any other lock
    files present so the operator can investigate.

tests/test_updates.py:
  - Removed all 4 tests for the deleted v2 helpers (_is_lock_held
    Returns*, _try_remove_lock*).
  - Added test_v2_probe_helpers_removed: regression guard that
    fails loud if either helper is ever reintroduced.
  - Rewrote test_apply_clear_lock_removes_unheld_lock_and_runs_normal_update
    and test_apply_clear_lock_refuses_when_lock_held for the v2.2
    contract. The 'refuses' test monkeypatches os.remove to record
    any deletion attempt and asserts the call list is empty -- a
    strict runtime guard against future regressions to auto-delete.
  - Added 7 new tests: test_inventory_locks_* (4), test_apply_clear
    lock_with_no_lock_runs_normal_update, test_apply_clear_lock_with
    lock_present_returns_manual_instruction, test_apply_clear_lock
    listing_includes_other_locks, test_apply_clear_lock_rejects
    unknown_target, test_apply_clear_lock_rejects_not_git_repo.

Verification:
  ./scripts/test.sh tests/test_updates.py          -> 73 passed
  ./scripts/test.sh tests/test_update_banner_fixes.py -> 102 passed
  ./scripts/test.sh tests/test_api_timeout.py      -> 8 passed
  python3 scripts/ruff_lint.py --diff upstream/master -> 0 new violations
  Full ./scripts/test.sh -> 12137 passed, 18 failed (all 18 pre-existing
    flakes in unrelated subsystems; verified not v2.2 regressions)
This commit is contained in:
b3nw committed 2026-07-06 19:02:16 +00:00
1 parent dcec232dfa
commit 2ddef230ff
4 files changed
+327 -221

No files matched your search

+7 -6
View File
@@ -15044,12 +15044,13 @@ def handle_post(handler, parsed) -> bool:
return j(handler, apply_force_update(target))
if parsed.path == "/api/updates/clear_lock":
# Non-destructive recovery for the stale .git/index.lock case: tries
# to clear the well-known short-lived git index lock and re-runs the
# normal (non-destructive) update flow. Refuses to remove locks that
# another process still holds (see apply_clear_lock for the holder
# probe). This is the *only* path on which git locks are removed;
# /api/updates/apply and /api/updates/force do not touch locks.
# Manual-instruction recovery for the .git/index.lock case. The
# endpoint NEVER removes a lock file from the server -- it returns
# the diagnostic + the exact 'rm' command for the operator, and on
# a re-click with the lock already gone, it re-runs the normal
# non-destructive apply path. See apply_clear_lock for the v2.2
# design rationale (round-2 gate cert: fcntl-flock cannot detect
# git's O_CREAT|O_EXCL locks, so any auto-delete path races).
target = body.get("target", "")
if target not in ("webui", "agent"):
return bad(handler, 'target must be "webui" or "agent"')
+92 -138
View File
@@ -70,15 +70,11 @@ _GIT_LOCK_SIGNATURES = (
'another git process seems to be running',
'unable to create .git/index.lock',
)
# Lock files we are willing to remove on the *explicit user request* path
# (`/api/updates/clear_lock`). The clear_lock endpoint refuses to remove any
# lock whose underlying file appears to still be held by another process --
# there is no mtime-based heuristic any more (BRICK-1 from PR #5688 gate
# cert: a live `git add` was empirically shown to hold .git/index.lock past
# 31 s with unchanged mtime).
_GIT_LOCK_FILES_REMOVABLE = (
'index.lock',
)
# Lock files we previously enumerated for auto-removal in v2. v2.2 no longer
# removes anything on the server, so the enumerable list is no longer needed;
# ``_inventory_locks`` reports whatever ``.git/**/*.lock`` files currently exist
# via plain ``rglob``.
def _sanitize_git_diagnostic(output: str, *, limit: int = _GIT_DIAGNOSTIC_MAX_CHARS) -> str:
@@ -242,95 +238,70 @@ def _is_git_lock_error(output: str) -> bool:
return any(sig in lower_out for sig in _GIT_LOCK_SIGNATURES)
def _is_lock_held(lock_path: Path) -> bool:
"""Return True if a process appears to still hold ``lock_path``.
def _inventory_locks(path: Path) -> dict:
"""Return a snapshot of lock files currently present under ``path/.git``.
Used by the explicit `clear_lock` recovery path. The implementer is
intentionally conservative: a process hold signal returns True and the
caller refuses to remove the lock. If we cannot prove a holder exists
*and cannot prove one does not*, we still return True (fail-closed) on
the common case the OS denies `flock`/delete.
Currently: tries a non-blocking exclusive fcntl.flock on POSIX. Any
`BlockingIOError` means another process holds the lock. On platforms
without fcntl (Windows), returns True unconditionally so the endpoint
surfaces a manual-instruction message rather than risking corruption.
v2.2: replaced v2's `_is_lock_held` + `_try_remove_lock` machinery with
pure inventory. Round-2 cert (gate-fail) proved that `fcntl.flock`
cannot detect a live git lock, because git uses `O_CREAT|O_EXCL` and
`rename(2)`, NOT advisory locking. Any auto-delete path can therefore
race against a running `git add` and corrupt the index. v2.2 stops
deleting locks from the server entirely: the only thing that removes
a lock is the user, on the host, via the manual command surfaced in
the response. Once the lock is gone, the user re-clicks Update Now
and the normal non-destructive apply path runs.
"""
if not lock_path.exists():
return False
git_dir = path / '.git'
out = {
'well_known_lock_present': False, # ``.git/index.lock`` exists?
'well_known_lock_path': None, # absolute path of ``.git/index.lock``
'other_locks': [], # any other lock files, by relative path
}
if not git_dir.exists():
return out
well_known = git_dir / 'index.lock'
try:
import fcntl # local import keeps Windows imports happy
except ImportError:
# No fcntl on this platform -- cannot prove non-hold. Fail closed.
return True
try:
fd = os.open(str(lock_path), os.O_RDWR | os.O_NOFOLLOW)
out['well_known_lock_present'] = well_known.exists()
except OSError:
# Cannot open the file (deleted, perm-denied, etc.). Treat as
# held so the caller asks the user to act.
return True
# Permission problem reading the directory -- treat conservatively.
out['well_known_lock_present'] = True
out['well_known_lock_path'] = str(well_known)
# Enumerate every other lock file under .git/ for diagnostic reporting.
# We never touch them; this is purely an inventory.
try:
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except (BlockingIOError, OSError):
return True
# We acquired the lock; release immediately and report not-held.
try:
fcntl.flock(fd, fcntl.LOCK_UN)
except OSError:
pass
return False
finally:
try:
os.close(fd)
except OSError:
pass
def _try_remove_lock(path: Path, logger_obj: logging.Logger | None = None) -> tuple[bool, str]:
"""Best-effort removal of a git lock file. Refuses if a holder is detected.
Returns ``(removed, diagnostic)``. ``removed`` is True only if the file
was successfully unlinked. ``diagnostic`` is a human-readable reason
string for logging and for the HTTP response message.
"""
if not path.exists():
return False, 'no-such-file'
if _is_lock_held(path):
if logger_obj is not None:
logger_obj.warning(
'clear_lock: refusing to remove %s -- appears held', path
)
return False, 'lock-held-by-another-process'
try:
os.remove(path)
except OSError as exc:
if logger_obj is not None:
logger_obj.warning(
'clear_lock: os.remove failed for %s: %s', path, exc
)
return False, f'remove-failed: {exc}'
if logger_obj is not None:
logger_obj.info('clear_lock: removed %s', path)
return True, 'removed'
for entry in sorted(git_dir.rglob('*.lock')):
try:
rel = str(entry.relative_to(git_dir))
except ValueError:
continue
if rel == 'index.lock':
continue
out['other_locks'].append(rel)
except OSError:
# rglob can fail on unreadable subtrees; skip quietly.
pass
return out
def apply_clear_lock(target: str) -> dict:
"""Non-destructively attempt to clear a stale git lock for ``target``.
"""Manual-instruction lock recovery for ``target``.
Replaces the v1 approach of pre-clearing locks inside ``apply_force_update``.
v1's mtime heuristic was proven unsafe in PR #5688 gate cert (BRICK-1):
a live git add was empirically shown to hold ``.git/index.lock`` past
31 s with the lock's mtime unchanged, so age is not a proxy for staleness.
v2.2: NEVER removes a lock file. Strategy:
v2 strategy:
- Refuse to remove any lock file that another process currently holds
(probe via non-blocking fcntl.flock; on non-POSIX, fail closed).
- Only touch ``.git/index.lock`` (the only well-known short-lived lock
git creates); other lock files (``refs/**/*.lock``, ``FETCH_HEAD.lock``,
...) are reported in the response for the operator to handle manually.
- On success, retry ``_apply_update_inner`` once to apply the update.
(No ``checkout`` / ``clean`` / ``reset --hard`` -- safe by design.)
- If ``.git/index.lock`` is absent: re-run the normal non-destructive
apply path so the user lands on the latest version without ever
touching destructive git operations.
- If ``.git/index.lock`` is present: do NOT touch it -- the server
has no reliable proof that no live git process is still using
it (round-2 cert showed `fcntl.flock` does not detect git's
actual ``O_CREAT|O_EXCL`` locking). Return a response with the
exact manual command the operator can run, plus the inventory of
any other lock files so they can investigate. The frontend then
surfaces a copyable ``rm`` line and a "I've removed the lock --
try update again" button that re-invokes this endpoint, which
(now that the lock is gone) will take the success branch and
re-run the normal apply.
"""
blocker_snapshot = _restart_blocker_snapshot()
if blocker_snapshot.get('restart_blocked'):
@@ -350,60 +321,43 @@ def apply_clear_lock(target: str) -> dict:
if path is None or not (path / '.git').exists():
return {'ok': False, 'message': 'Not a git repository'}
git_dir = path / '.git'
# Only attempt removal of well-known short-lived lock files. Anything
# outside _GIT_LOCK_FILES_REMOVABLE is returned in the report so the
# operator can inspect it.
removed: list[str] = []
refused: list[str] = []
for name in _GIT_LOCK_FILES_REMOVABLE:
ok, diag = _try_remove_lock(git_dir / name)
if ok:
removed.append(name)
elif diag != 'no-such-file':
refused.append(f'{name} ({diag})')
inv = _inventory_locks(path)
manual_command = f"rm -f {inv['well_known_lock_path']}"
# Detect other lock files (refs/heads/<branch>.lock etc) for the
# diagnostic -- but never delete them here.
other_locks: list[str] = []
for entry in sorted(git_dir.rglob('*.lock')):
rel = str(entry.relative_to(git_dir))
if rel not in _GIT_LOCK_FILES_REMOVABLE:
other_locks.append(rel)
if refused and not removed:
# All attempted removals were refused -- surface a clean
# manual-instruction response.
return {
'ok': False,
'message': (
'Cannot safely clear the lock: '
+ '; '.join(refused)
+ '. A concurrent git process appears to hold the lock. '
+ 'Wait for it to finish, then retry. As a last resort, '
+ 'run: rm -f ' + str(git_dir / _GIT_LOCK_FILES_REMOVABLE[0])
),
'lock_held': True,
'target': target,
'refused': refused,
'other_locks': other_locks,
if not inv['well_known_lock_present']:
# Lock is gone. Run the normal non-destructive update flow and
# annotate the response with what we found for the user's
# records.
with _cache_lock:
_update_cache['checked_at'] = 0
retry_result = _apply_update_inner(target)
retry_result = dict(retry_result)
retry_result['lock_recovery'] = {
'action': 'no-lock-found',
'manual_command': manual_command,
'other_locks': inv['other_locks'],
}
return retry_result
# Invalidate cache so the next status check is fresh.
with _cache_lock:
_update_cache['checked_at'] = 0
# Now retry the normal (non-destructive) update flow once. If this
# succeeds the user gets their update without any checkout/clean/reset.
retry_result = _apply_update_inner(target)
# Annotate with the recovery summary so the UI can show what happened.
retry_result = dict(retry_result)
retry_result['lock_recovery'] = {
'removed': removed,
'refused': refused,
'other_locks': other_locks,
# Lock is present. The server cannot prove it's safe to delete;
# the only safe path is to ask the operator.
message = (
'A git lock file (.git/index.lock) is present. The server does '
'not delete locks automatically -- git uses O_CREAT|O_EXCL '
'locking, which cannot be detected with advisory probes. To '
'recover: confirm no other git process is running against '
f'this checkout, then run: {manual_command} '
'Click "Retry update" once you have removed it.'
)
return {
'ok': False,
'message': message,
'lock_held': True,
'target': target,
'manual_command': manual_command,
'well_known_lock_path': inv['well_known_lock_path'],
'other_locks': inv['other_locks'],
}
return retry_result
finally:
_apply_lock.release()
+77 -4
View File
@@ -9124,22 +9124,29 @@ async function applyClearUpdateLock(btn){
window._clearLockInFlight=true;
btn.disabled=true;
const originalLabel=btn.textContent;
btn.textContent='Clearing lock\u2026';
btn.textContent='Checking lock…';
try{
const res=await api('/api/updates/clear_lock',{method:'POST',body:JSON.stringify({target}),timeoutMs:60000});
if(res.ok){
sessionStorage.removeItem('hermes-update-checked');
sessionStorage.removeItem('hermes-update-dismissed');
showToast('Lock cleared — update applied\u2026');
showToast('Update applied — restarting…');
_waitForServerThenReload({});
} else if(res.lock_held){
// v2.2: server returns manual-instruction. Show the exact `rm`
// command + a one-click "I've removed it, retry update" affordance
// that POSTs the same endpoint a second time (now that the user
// has presumably removed the lock, the server's success branch
// runs the normal non-destructive apply).
_renderLockManualInstruction(target, res);
} else {
const msg='Could not clear the lock: '+(res.message||'unknown error');
const msg='Could not check the lock: '+(res.message||'unknown error');
const errEl=$('updateError');
if(errEl){errEl.textContent=msg;errEl.style.display='block';}
else showToast(msg);
}
}catch(e){
const msg='Clear-lock request failed: '+((e&&e.message)||String(e));
const msg='Lock-check request failed: '+((e&&e.message)||String(e));
const errEl=$('updateError');
if(errEl){errEl.textContent=msg;errEl.style.display='block';}
else showToast(msg);
@@ -9149,6 +9156,72 @@ async function applyClearUpdateLock(btn){
btn.textContent=originalLabel;
}
}
function _renderLockManualInstruction(target, res){
// Replace the inline `updateError` text with a richer block that shows
// the exact manual command and offers a one-click retry button. The
// "retry" handler re-invokes `applyClearUpdateLock`; this time, with
// the lock gone, the server's success branch runs the normal apply.
const cmd = res.manual_command || ('rm -f ' + (res.well_known_lock_path || '.git/index.lock'));
const errEl=$('updateError');
if(!errEl){
showToast('Lock present. Run: '+cmd);
return;
}
errEl.style.display='block';
errEl.innerHTML='';
const intro=document.createElement('div');
intro.style.marginBottom='6px';
intro.textContent='A stale .git/index.lock is present. The server cannot remove it safely — please run this command on the host:';
errEl.appendChild(intro);
const code=document.createElement('pre');
code.style.background='rgba(0,0,0,0.05)';
code.style.padding='6px';
code.style.margin='4px 0';
code.style.fontFamily='ui-monospace,monospace';
code.style.borderRadius='4px';
code.style.whiteSpace='pre-wrap';
code.style.wordBreak='break-all';
code.textContent=cmd;
errEl.appendChild(code);
const actions=document.createElement('div');
actions.style.display='flex';
actions.style.gap='8px';
actions.style.flexWrap='wrap';
const copyBtn=document.createElement('button');
copyBtn.type='button';
copyBtn.className='update-btn';
copyBtn.textContent='Copy command';
copyBtn.onclick=async()=>{
try{
if(navigator.clipboard&&navigator.clipboard.writeText){
await navigator.clipboard.writeText(cmd);
copyBtn.textContent='Copied';
setTimeout(()=>{ copyBtn.textContent='Copy command'; }, 1500);
} else {
copyBtn.textContent='Clipboard unavailable';
}
} catch(_){
copyBtn.textContent='Copy failed';
}
};
actions.appendChild(copyBtn);
const retryBtn=document.createElement('button');
retryBtn.type='button';
retryBtn.className='update-btn update-primary';
retryBtn.textContent="I've removed the lock — retry update";
retryBtn.dataset.target=target;
retryBtn.onclick=()=>{ applyClearUpdateLock(retryBtn); };
actions.appendChild(retryBtn);
errEl.appendChild(actions);
if(Array.isArray(res.other_locks)&&res.other_locks.length){
const other=document.createElement('div');
other.style.marginTop='6px';
other.style.fontSize='11px';
other.style.opacity='0.85';
other.textContent='Other lock files also present: '+res.other_locks.join(', ');
errEl.appendChild(other);
}
}
function _normalizeHealthServerIdentity(rawIdentity){
if(rawIdentity===undefined||rawIdentity===null) return null;
if(typeof rawIdentity==='string'){
+151 -73
View File
@@ -1,6 +1,5 @@
"""Tests for self-update diagnostics (api/updates.py)."""
import os
import sys
import time
from unittest.mock import MagicMock, patch
@@ -1046,79 +1045,95 @@ def test_apply_force_update_no_longer_touches_locks(tmp_path, monkeypatch):
)
# ── v2 tests for PR #5688 ──────────────────────────────────────────────────
# ── v2 (Round-2) tests for PR #5688 ──────────────────────────────────────────
#
# v2.2 dropped v2's fcntl-flock holder probe + os.remove path entirely.
# Those round-2 functions (`_is_lock_held`, `_try_remove_lock`) are
# removed in v2.2 because they were proven unsafe (Codex strace showed
# git uses O_CREAT|O_EXCL, not advisory locking). The deletion tests
# below assert they no longer exist -- if a future refactor reintroduces
# either, those tests fail loud. The replacements for them are the v2.2
# inventory + apply_clear_lock tests further below.
@pytest.mark.skipif(sys.platform.startswith('win'),
reason='fcntl-based holder probe is POSIX-only')
def test_is_lock_held_returns_false_when_no_other_holder(tmp_path):
"""An unheld .git/index.lock passes the holder probe."""
lock = tmp_path / 'index.lock'
lock.write_text('')
assert updates._is_lock_held(lock) is False
def test_v2_probe_helpers_removed():
"""v2.2 contract: the round-2 fcntl-flock probe machinery is gone.
@pytest.mark.skipif(sys.platform.startswith('win'),
reason='fcntl-based holder probe is POSIX-only')
def test_is_lock_held_returns_true_when_other_process_holds(tmp_path):
"""If an external process holds the lock via flock, the probe returns True."""
lock = tmp_path / 'index.lock'
lock.write_text('')
# Simulate a holder by acquiring flock ourselves from this process.
import fcntl
fd = os.open(str(lock), os.O_RDWR)
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
try:
# Now the probe (a different os.open + flock attempt) must fail
# and return True, refusing removal.
assert updates._is_lock_held(lock) is True
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
def test_try_remove_lock_refuses_when_held(tmp_path, monkeypatch):
"""_try_remove_lock must NEVER touch a file if a holder is detected."""
lock = tmp_path / 'index.lock'
lock.write_text('sentinel')
monkeypatch.setattr(
updates, '_is_lock_held', lambda p: True
Round-2 cert proved `flock` cannot detect git's O_CREAT|O_EXCL locks,
so any auto-delete path can race a running git process. This guard
test fails loud if anyone reintroduces either helper.
"""
assert not hasattr(updates, '_is_lock_held'), (
"_is_lock_held was re-introduced after v2.2 removal -- round-2 cert "
"showed fcntl.flock cannot detect git locks; do not bring it back."
)
assert not hasattr(updates, '_try_remove_lock'), (
"_try_remove_lock was re-introduced after v2.2 removal -- auto-delete "
"from the server is unsafe on a brick-risk path; do not bring it back."
)
ok, diag = updates._try_remove_lock(lock)
assert ok is False
assert diag == 'lock-held-by-another-process'
assert lock.exists(), "Lock must NOT be removed when a holder is detected"
assert lock.read_text() == 'sentinel', "Lock contents must NOT be modified"
def test_try_remove_lock_succeeds_when_unheld(tmp_path, monkeypatch):
ok, diag = updates._try_remove_lock(tmp_path / 'nope')
assert ok is False and diag == 'no-such-file'
lock = tmp_path / 'index.lock'
lock.write_text('')
monkeypatch.setattr(updates, '_is_lock_held', lambda p: False)
ok, diag = updates._try_remove_lock(lock)
assert ok is True, diag
assert not lock.exists()
# ── v2.2 tests for PR #5688 ──────────────────────────────────────────────────
#
# v2.2 dropped v2's fcntl-flock holder probe and os.remove path entirely.
# ``apply_clear_lock`` is now inventory-only and manual-instruction: if
# the lock is gone it re-runs the normal update; if the lock is present
# it returns the exact ``rm`` command the operator must run. These tests
# lock in that contract.
def test_apply_clear_lock_removes_unheld_lock_and_runs_normal_update(tmp_path,
monkeypatch):
"""clear_lock removes an unheld index.lock and re-runs the normal update path."""
def test_inventory_locks_reports_when_index_lock_present(tmp_path):
"""Inventory must report well_known_lock_present=True when .git/index.lock
exists, plus its absolute path."""
(tmp_path / '.git').mkdir()
lock = tmp_path / '.git' / 'index.lock'
lock.write_text('')
monkeypatch.setattr(updates, '_is_lock_held', lambda p: False)
lock.write_text('stale')
inv = updates._inventory_locks(tmp_path)
assert inv['well_known_lock_present'] is True
assert inv['well_known_lock_path'] == str(lock)
assert inv['other_locks'] == []
fake_git_calls = []
def fake_git(args, cwd, timeout=10):
fake_git_calls.append(args)
return '', True
def test_inventory_locks_reports_when_index_lock_absent(tmp_path):
"""Inventory must report well_known_lock_present=False when no lock exists."""
(tmp_path / '.git').mkdir()
inv = updates._inventory_locks(tmp_path)
assert inv['well_known_lock_present'] is False
assert inv['other_locks'] == []
monkeypatch.setattr(updates, '_run_git', fake_git)
def test_inventory_locks_lists_other_locks(tmp_path):
"""Inventory must enumerate refs/*.lock etc without including index.lock."""
(tmp_path / '.git').mkdir()
(tmp_path / '.git' / 'index.lock').write_text('')
(tmp_path / '.git' / 'refs' / 'heads').mkdir(parents=True)
(tmp_path / '.git' / 'refs' / 'heads' / 'main.lock').write_text('')
(tmp_path / '.git' / 'FETCH_HEAD.lock').write_text('')
inv = updates._inventory_locks(tmp_path)
assert inv['well_known_lock_present'] is True
assert sorted(inv['other_locks']) == [
'FETCH_HEAD.lock',
'refs/heads/main.lock',
]
def test_inventory_locks_handles_missing_git_dir(tmp_path):
"""When ``.git`` does not exist the inventory must still return a valid shape."""
inv = updates._inventory_locks(tmp_path)
assert inv == {
'well_known_lock_present': False,
'well_known_lock_path': None,
'other_locks': [],
}
def test_apply_clear_lock_with_no_lock_runs_normal_update(tmp_path, monkeypatch):
"""v2.2: when ``.git/index.lock`` is absent, apply_clear_lock re-runs
the normal non-destructive apply path."""
(tmp_path / '.git').mkdir()
# No lock file written.
monkeypatch.setattr(updates, '_run_git',
MagicMock(return_value=('', True)))
monkeypatch.setattr(updates, 'REPO_ROOT', tmp_path)
monkeypatch.setattr(
updates, '_restart_blocker_snapshot',
@@ -1128,22 +1143,34 @@ def test_apply_clear_lock_removes_unheld_lock_and_runs_normal_update(tmp_path,
updates, '_select_apply_compare_ref',
lambda path: 'origin/main'
)
result = updates.apply_clear_lock('webui')
assert result['ok'] is True, result
assert result.get('lock_recovery', {}).get('removed') == ['index.lock']
assert not lock.exists()
assert result['lock_recovery']['action'] == 'no-lock-found'
assert 'manual_command' in result['lock_recovery']
assert 'rm -f' in result['lock_recovery']['manual_command']
def test_apply_clear_lock_refuses_when_lock_held(tmp_path, monkeypatch):
"""If a holder is detected, clear_lock must NOT remove the lock and must
return a manual-instruction response."""
def test_apply_clear_lock_with_lock_present_returns_manual_instruction(tmp_path, monkeypatch):
"""v2.2: when a lock is present, apply_clear_lock must NEVER touch the
lock file; it returns ok=False with a manual-instruction response."""
(tmp_path / '.git').mkdir()
lock = tmp_path / '.git' / 'index.lock'
lock.write_text('do-not-touch')
lock.write_text('user-removed-this-by-hand')
# Always appear held.
monkeypatch.setattr(updates, '_is_lock_held', lambda p: True)
# Spy: capture whether the server attempted to delete anything. The
# correct v2.2 behavior is NO DELETE attempt whatsoever, regardless
# of any property of the lock file.
delete_attempts = []
def forbid_delete(*args, **kwargs):
delete_attempts.append((args, kwargs))
# Use a sentinel that will NEVER happen; if delete is called the
# test must fail. The cheap way: just record the attempt.
return None
# Patch os.remove + Path.unlink on the instance/module to record any
# destructive attempt. apply_clear_lock must NOT call them.
monkeypatch.setattr(updates.os, 'remove', forbid_delete)
monkeypatch.setattr(updates, 'REPO_ROOT', tmp_path)
monkeypatch.setattr(
updates, '_restart_blocker_snapshot',
@@ -1153,11 +1180,62 @@ def test_apply_clear_lock_refuses_when_lock_held(tmp_path, monkeypatch):
result = updates.apply_clear_lock('webui')
assert result['ok'] is False
assert result.get('lock_held') is True
assert 'Cannot safely clear the lock' in result['message']
assert lock.exists(), "Lock must NOT be removed when held"
assert lock.read_text() == 'do-not-touch'
assert result.get('manual_command', '').startswith('rm -f')
assert result.get('well_known_lock_path') == str(lock)
assert 'O_CREAT|O_EXCL' in result['message'], (
"message must explain why the server cannot do this automatically"
)
assert lock.exists(), "Lock must NOT be removed"
assert lock.read_text() == 'user-removed-this-by-hand', (
"Lock contents must NOT be modified"
)
assert delete_attempts == [], (
"v2.2 contract: apply_clear_lock must never attempt os.remove "
"under any circumstances"
)
def test_apply_clear_lock_listing_includes_other_locks(tmp_path, monkeypatch):
"""Inventory of other-lock files must round-trip through the response so
the operator can act on them too."""
(tmp_path / '.git').mkdir()
(tmp_path / '.git' / 'index.lock').write_text('')
(tmp_path / '.git' / 'refs').mkdir()
(tmp_path / '.git' / 'refs' / 'main.lock').write_text('')
monkeypatch.setattr(updates, 'REPO_ROOT', tmp_path)
monkeypatch.setattr(
updates, '_restart_blocker_snapshot',
lambda: {'restart_blocked': False, 'active_streams': 0, 'active_runs': 0}
)
result = updates.apply_clear_lock('webui')
assert result['ok'] is False
assert result['lock_held'] is True
assert result['other_locks'] == ['refs/main.lock']
def test_apply_clear_lock_rejects_unknown_target(tmp_path, monkeypatch):
monkeypatch.setattr(updates, 'REPO_ROOT', tmp_path)
monkeypatch.setattr(
updates, '_restart_blocker_snapshot',
lambda: {'restart_blocked': False, 'active_streams': 0, 'active_runs': 0}
)
result = updates.apply_clear_lock('not-a-target')
assert result['ok'] is False
assert 'Unknown target' in result['message']
def test_apply_clear_lock_rejects_not_git_repo(tmp_path, monkeypatch):
"""If REPO_ROOT has no .git, apply_clear_lock must refuse."""
# tmp_path has no .git
monkeypatch.setattr(updates, 'REPO_ROOT', tmp_path)
monkeypatch.setattr(
updates, '_restart_blocker_snapshot',
lambda: {'restart_blocked': False, 'active_streams': 0, 'active_runs': 0}
)
result = updates.apply_clear_lock('webui')
assert result['ok'] is False
assert 'Not a git repository' in result['message']
def test_apply_update_pull_lock_restores_stash(tmp_path, monkeypatch):
"""Greptile P1: a pull-lock error after stashing must restore the stash."""
(tmp_path / '.git').mkdir()