fix(gateway): make stale scoped-lock removal atomic via tombstone rename

Replace the unlink()+O_EXCL sequence in acquire_scoped_lock with an
atomic os.replace() of the stale lock to a <lock>.stale tombstone
followed by the existing O_EXCL create. With plain unlink(), two racing
starters could both judge the lock stale and the second unlink() would
silently delete the first racer's freshly-created lock — both would then
'win'. os.replace() guarantees exactly one racer claims the stale file;
the loser gets FileNotFoundError and falls through to O_EXCL, which
admits at most one winner. Tombstones are cleaned up immediately;
behavior is otherwise identical.

gateway/status.py

@@ -1184,10 +1184,27 @@ def acquire_scoped_lock(scope: str, identity: str, metadata: Optional[dict[str,
                     except (OSError, PermissionError):
                         pass
         if stale:
+            # Remove the stale lock ATOMICALLY by renaming it to a tombstone
+            # instead of unlinking. With unlink()+O_EXCL, two racing starters
+            # could both observe "removed" (the second unlink() silently
+            # deleting the first racer's freshly-created lock) and both win.
+            # os.replace() is atomic: exactly one racer claims the stale
+            # file; the loser gets FileNotFoundError and falls through to
+            # the O_EXCL create below, where at most one process succeeds.
+            tombstone = lock_path.with_name(lock_path.name + ".stale")
             try:
-                lock_path.unlink(missing_ok=True)
+                os.replace(lock_path, tombstone)
+            except FileNotFoundError:
+                # Another racer already claimed the stale lock (and may have
+                # created a fresh one) — let O_EXCL below decide the winner.
+                pass
             except OSError:
                 pass
+            else:
+                try:
+                    tombstone.unlink(missing_ok=True)
+                except OSError:
+                    pass
         else:
             return False, existing
 

tests/gateway/test_status.py

@@ -876,6 +876,100 @@ class TestScopedLocks:
         assert payload["pid"] == os.getpid()
         assert payload["metadata"]["platform"] == "telegram"
 
+    def test_acquire_scoped_lock_atomic_removal_leaves_no_tombstone(self, tmp_path, monkeypatch):
+        """Stale lock removal renames to a tombstone then cleans it up — the
+        happy path must behave exactly like the old unlink()-based removal."""
+        monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))
+        lock_path = tmp_path / "locks" / "telegram-bot-token-2bb80d537b1da3e3.lock"
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_path.write_text(json.dumps({
+            "pid": 99999,
+            "start_time": 123,
+            "kind": "hermes-gateway",
+        }))
+        monkeypatch.setattr(status, "_pid_exists", lambda pid: False)
+
+        acquired, existing = status.acquire_scoped_lock("telegram-bot-token", "secret", metadata={"platform": "telegram"})
+
+        assert acquired is True
+        payload = json.loads(lock_path.read_text())
+        assert payload["pid"] == os.getpid()
+        assert not (lock_path.parent / (lock_path.name + ".stale")).exists()
+
+    def test_acquire_scoped_lock_race_second_acquirer_loses(self, tmp_path, monkeypatch):
+        """Two racing starters both observe the same stale lock. The loser's
+        os.replace() hits FileNotFoundError (winner already claimed the stale
+        file) and the winner's FRESH lock must survive: the loser must fall
+        through to O_EXCL and lose, not clobber it like the old unlink() did."""
+        monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))
+        lock_path = tmp_path / "locks" / "telegram-bot-token-2bb80d537b1da3e3.lock"
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        stale_record = {
+            "pid": 99999,
+            "start_time": 123,
+            "kind": "hermes-gateway",
+        }
+        lock_path.write_text(json.dumps(stale_record))
+        monkeypatch.setattr(status, "_pid_exists", lambda pid: False)
+
+        winner_record = {
+            "pid": 424242,
+            "start_time": 456,
+            "kind": "hermes-gateway",
+            "scope": "telegram-bot-token",
+        }
+        real_replace = os.replace
+
+        def racing_replace(src, dst, *args, **kwargs):
+            if str(src) == str(lock_path):
+                # Simulate the winner completing removal + O_EXCL create
+                # between our staleness check and our removal attempt.
+                lock_path.write_text(json.dumps(winner_record))
+                raise FileNotFoundError(2, "No such file or directory", str(src))
+            return real_replace(src, dst, *args, **kwargs)
+
+        monkeypatch.setattr(status.os, "replace", racing_replace)
+
+        acquired, existing = status.acquire_scoped_lock("telegram-bot-token", "secret", metadata={"platform": "telegram"})
+
+        assert acquired is False
+        assert existing is not None
+        assert existing["pid"] == 424242
+        # The winner's fresh lock must be untouched on disk.
+        assert json.loads(lock_path.read_text())["pid"] == 424242
+
+    def test_acquire_scoped_lock_two_sequential_acquirers_one_winner(self, tmp_path, monkeypatch):
+        """Sequential end-to-end: first acquirer replaces a stale lock and
+        wins; a second acquirer (different identity, same lock file) sees the
+        first's LIVE lock and must lose without touching it."""
+        monkeypatch.setenv("HERMES_GATEWAY_LOCK_DIR", str(tmp_path / "locks"))
+        lock_path = tmp_path / "locks" / "telegram-bot-token-2bb80d537b1da3e3.lock"
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_path.write_text(json.dumps({
+            "pid": 99999,
+            "start_time": 123,
+            "kind": "hermes-gateway",
+        }))
+        monkeypatch.setattr(status, "_pid_exists", lambda pid: pid != 99999)
+
+        acquired1, _ = status.acquire_scoped_lock("telegram-bot-token", "secret", metadata={"platform": "telegram"})
+        assert acquired1 is True
+        first_payload = json.loads(lock_path.read_text())
+        assert first_payload["pid"] == os.getpid()
+
+        # Second acquirer: pretend the current record belongs to a different
+        # live process so the self-ownership fast path doesn't kick in.
+        rewritten = dict(first_payload)
+        rewritten["pid"] = 424242
+        lock_path.write_text(json.dumps(rewritten))
+        monkeypatch.setattr(status, "_get_process_start_time", lambda pid: rewritten.get("start_time"))
+        monkeypatch.setattr(status, "_looks_like_gateway_process", lambda pid: True)
+
+        acquired2, existing2 = status.acquire_scoped_lock("telegram-bot-token", "secret", metadata={"platform": "telegram"})
+        assert acquired2 is False
+        assert existing2 is not None and existing2["pid"] == 424242
+        assert json.loads(lock_path.read_text())["pid"] == 424242
+
     def test_acquire_scoped_lock_keeps_lock_when_cmdline_unreadable_but_record_is_gateway(self, tmp_path, monkeypatch):
         """Windows regression: ps unavailable so cmdline cannot be read.