fix(groups): order an approval's live check and dispatch with Stop

Greptile 'Approval can pass after Stop': the live-attempt read ran before
the policy lock and the approval was dispatched outside it, so a Stop from
another window could commit between the check and the dispatch and the
member's prompt still got answered. The check and dispatch now hold the
lock Stop takes, and Stop retires the stopped attempts' cached approvals
under it, so a concurrent Allow is either delivered while the attempt
still runs or refused.

tests/tui_gateway/test_hosted_room_approval_lifetime.py

@@ -256,3 +256,63 @@ def test_stale_local_approval_cannot_resolve_replacement_request(tmp_path: Path,
     assert service.status("room-1")["pending_actions"][0]["request_id"] == (
         "approval-B"
     )
+
+
+def test_allow_paused_after_its_live_read_cannot_answer_after_a_concurrent_stop(pending_approval, monkeypatch):
+    """An Allow whose live-attempt read raced another window's Stop is ordered with that Stop."""
+    import threading
+
+    service, task, _lease = pending_approval
+    statuses_at_dispatch = []
+
+    class _RecordingRPC(_FakeRPC):
+        def approve(self, **kwargs):
+            statuses_at_dispatch.append(driver.get_task(service.db_path, task["identity"])["status"])
+            return super().approve(**kwargs)
+
+    service.rpc = service.runtime.rpc = _RecordingRPC()
+    for hook in ("_interrupt_stopping_task", "_settle_stopping_completion", "_peer_stop_acknowledged"):
+        monkeypatch.setattr(service.runtime, hook, lambda *_args: False)
+    read, stop_progress = threading.Event(), threading.Event()
+    approver = threading.current_thread()
+    real_list_tasks = driver.list_tasks
+
+    def list_tasks(*args, **kwargs):
+        rows = real_list_tasks(*args, **kwargs)
+        if threading.current_thread() is approver and not read.is_set():
+            read.set()
+            assert stop_progress.wait(10), "Stop neither finished nor waited for this Allow"
+        return rows
+
+    class _SignallingLock:
+        """Stop waiting on this lock is as far as it can get while the Allow holds it."""
+        def __init__(self, lock):
+            self._lock = lock
+
+        def __enter__(self):
+            if not self._lock.acquire(blocking=False):
+                if threading.current_thread() is not approver:
+                    stop_progress.set()
+                self._lock.acquire()
+            return self
+
+        def __exit__(self, *exc):
+            self._lock.release()
+
+    monkeypatch.setattr(driver, "list_tasks", list_tasks)
+    service._approval_stop_lock = _SignallingLock(threading.RLock())
+    stopper = threading.Thread(target=lambda: (read.wait(10), service.stop_room("room", cancel_id="stop"),
+                                               stop_progress.set()))
+    stopper.start()
+    try:
+        result = service.approve_room_task(
+            "room", member_id="ops", task_id=task["identity"].task_id,
+            execution_generation=task["execution_generation"], request_id="approval", choice="once")
+    except RuntimeError as exc:
+        result = str(exc)
+    stopper.join(10)
+    assert not stopper.is_alive()
+    # Either the member got its answer while the attempt still ran, or the Allow was refused.
+    assert statuses_at_dispatch in ([], ["running"]), (statuses_at_dispatch, result)
+    assert driver.get_task(service.db_path, task["identity"])["status"] == "stopping"
+    assert ("room", "ops") not in service._pending_actions

tui_gateway/hosted_room_service.py

@@ -134,6 +134,10 @@ class HostedRoomService:
         self.discussion_limits = _discussion_limits(self.root)
         hosted_rooms.prune_disbanded_rooms(self.db_path)
         self._policy_lock = threading.RLock()
+        # Orders Stop with an approval's live check and dispatch: an Allow either reaches the member
+        # before Stop marks its attempt stopping, or sees the stop and is refused. Taken before
+        # ``_policy_lock``, never inside it.
+        self._approval_stop_lock = threading.RLock()
         self._pending_actions: dict[tuple[str, str], dict[str, Any]] = {}
         self.policy_checkpoint = HostedRoomPolicyCheckpoint(
             self.db_path,
@@ -698,6 +702,10 @@ class HostedRoomService:
 
     def stop_room(
         self, room_id: str, *, cancel_id: str, require_acknowledged: bool = False) -> int:
+        with self._approval_stop_lock:
+            return self._stop_room(room_id, cancel_id=cancel_id, require_acknowledged=require_acknowledged)
+
+    def _stop_room(self, room_id: str, *, cancel_id: str, require_acknowledged: bool) -> int:
         gateway_id, epoch = self._owned_authority(room_id)
         fence = hosted_rooms.request_room_stop(
             self.db_path, room_id=room_id, cancel_id=cancel_id, expected_gateway_id=gateway_id,
@@ -710,6 +718,11 @@ class HostedRoomService:
                 # A replay retains its original fence; work admitted afterward belongs to the
                 # next message. Disband must drain all work before deleting the room.
                 if require_acknowledged or int(task["payload"]["source_event_seq"]) < int(fence["seq"])}
+            # A stopped attempt's approval is spent: a concurrent Allow waiting on the lock finds nothing.
+            stopped = {(task["identity"].task_id, task.get("execution_generation")) for task in tasks.values()}
+            for key in [key for key, action in self._pending_actions.items() if key[0] == room_id
+                        and (action.get("task_id"), action.get("execution_generation")) in stopped]:
+                self._pending_actions.pop(key, None)
             for task in tasks.values():
                 own_cancel_id = (
                     task.get("status") == "stopping" and str(task.get("cancel_id") or ""))
@@ -767,33 +780,44 @@ class HostedRoomService:
                 str(pending.get("request_id") or ""), pending.get("task_id"),
                 int(pending.get("execution_generation") or 0),
             ) == (requested_approval_id, task_id, execution_generation)
-        live = _approvable_attempts(driver.list_tasks(self.db_path, room_id=room_id))
-        with self._policy_lock:
-            action = self._pending_actions.get(key)
-            if not requested_approval_id or not matches(action):
-                raise RuntimeError("room approval is no longer pending")
-            if (member_id, task_id, execution_generation) not in live:
-                self._pending_actions.pop(key, None)
-                raise RuntimeError("room approval is no longer pending")
-        if choice not in {"once", "deny"}:
-            raise RuntimeError("room approval choice must be once or deny")
+        # The live check and the dispatch hold the Stop lock, so a Stop either lands first (and this
+        # Allow is refused) or after the member already has its answer.
+        with self._approval_stop_lock:
+            live = _approvable_attempts(driver.list_tasks(self.db_path, room_id=room_id))
+            with self._policy_lock:
+                action = self._pending_actions.get(key)
+                if not requested_approval_id or not matches(action):
+                    raise RuntimeError("room approval is no longer pending")
+                if (member_id, task_id, execution_generation) not in live:
+                    self._pending_actions.pop(key, None)
+                    raise RuntimeError("room approval is no longer pending")
+            if choice not in {"once", "deny"}:
+                raise RuntimeError("room approval choice must be once or deny")
+            result = self._dispatch_approval(
+                route, client, action, task_id=task_id, execution_generation=execution_generation,
+                request_id=requested_approval_id, choice=choice)
+            with self._policy_lock:
+                if matches(self._pending_actions.get(key)):
+                    self._pending_actions.pop(key, None)
+        self.runtime.wakeup()
+        return result
+
+    def _dispatch_approval(
+        self, route, client, action: Mapping[str, Any], *, task_id: str, execution_generation: int,
+        request_id: str, choice: str) -> Mapping[str, Any]:
         approve = _hook(client, "approve_receipt")
         if route is not None and approve is not None:
             result = approve(
                 task_id=task_id, execution_generation=execution_generation,
-                request_id=requested_approval_id, choice=choice, grant=route.grant)
+                request_id=request_id, choice=choice, grant=route.grant)
         else:
             session_id = str(action.get("session_id") or "")
             if not session_id:
                 raise RuntimeError("local room approval identity is unavailable")
             result = self.rpc.approve(
-                session_id=session_id, request_id=requested_approval_id, choice=choice)
+                session_id=session_id, request_id=request_id, choice=choice)
         if result is None:
             raise RuntimeError("room approval target is unavailable")
-        with self._policy_lock:
-            if matches(self._pending_actions.get(key)):
-                self._pending_actions.pop(key, None)
-        self.runtime.wakeup()
         return result
 
     def status(self, room_id: str | None = None) -> dict[str, Any]: