fix(groups): fence Stop replays and drain all work on Disband

Keep explicit Stop retries behind their original admission fence. Mint a
fresh identity when callers omit one, and drain every stoppable task on
Disband retries before allowing deletion.

(cherry picked from commit 92bed47fd5b8b6a8ce7071d67fb75515741a3139)

gateway/session_group_controls.py

@@ -3,6 +3,7 @@ import asyncio
 import logging
 import os
 from pathlib import Path
+import uuid
 import weakref
 
 from hermes_state_runtime import RuntimeStoreError
@@ -283,7 +284,7 @@ def _execution_control(service, method, params):
     handlers = {
         'groups.send': send,
         'groups.stop': lambda: {'cancelled': service.stop_room(
-            params.get('room_id'), cancel_id=params.get('cancel_id') or 'desktop-stop')},
+            params.get('room_id'), cancel_id=params.get('cancel_id') or uuid.uuid4().hex)},
         'groups.continue': lambda: {'continued': True, 'event': service.continue_room(
             params.get('room_id'), discussion_event_id=str(params.get('discussion_event_id') or ''))},
         'groups.retry': attempt_control,

tests/tui_gateway/test_groups_stop_lifetime.py

@@ -0,0 +1,83 @@
+"""Both room-control transports must stop newer work on a fresh Stop or Disband."""
+
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from gateway import hosted_room_driver as driver, hosted_rooms
+from gateway.session_group_controls import _group
+from hermes_state import SessionDB
+from tests.tui_gateway.test_hosted_room_service import _FakeRPC
+from tests.tui_gateway.test_hosted_room_stop_replay import _service
+
+
+@pytest.fixture(params=["canonical", "standalone"])
+def controls(request, tmp_path, monkeypatch):
+    import tui_gateway.server as srv
+    from tui_gateway import methods_groups_forward
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    with SessionDB(tmp_path / "state.db") as db:
+        service = _service(db.db_path)
+        service.create_room(room_id="room", name="Stop lifetime", members=[
+            {"member_id": "default", "profile": "default", "handle": "hermes"},
+            {"member_id": "ops", "profile": "ops", "handle": "ops"}])
+        monkeypatch.setattr(service.runtime, "status", lambda: {"running": True})
+        monkeypatch.setattr(srv, "get_hosted_room_service", lambda: service)
+        monkeypatch.setattr(methods_groups_forward, "owner_endpoint", lambda _home: None)
+        # Ownership is supplied by the authenticated connection in production.
+        service.authorize_room = lambda *_args, **_kwargs: None
+        authority = SimpleNamespace(db=db, hosted_room_service=service)
+
+        def call(method, **params):
+            params = {"room_id": "room", **params}
+            if request.param == "canonical":
+                return _group(authority, SimpleNamespace(subject="alice"), method, params)
+            result = srv._methods[method](1, params)
+            if "error" in result:
+                raise RuntimeError(result["error"]["message"])
+            return result["result"]
+
+        yield service, call
+
+
+@pytest.mark.parametrize("cancel_id", [None, "explicit-stop"])
+def test_successive_stops_use_a_fresh_fence_unless_explicitly_replayed(controls, cancel_id):
+    service, call = controls
+    params = {} if cancel_id is None else {"cancel_id": cancel_id}
+    call("groups.send", event_id="A", payload={"text": "@ops A", "thread_id": "A"})
+    assert call("groups.stop", **params) == {"cancelled": 1}
+    call("groups.send", event_id="B", payload={"text": "@ops B", "thread_id": "B"})
+    assert call("groups.stop", **params) == {"cancelled": 1 if cancel_id is None else 0}
+    queued = driver.list_tasks(service.db_path, room_id="room", status="queued")
+    assert len(queued) == (0 if cancel_id is None else 1)
+
+
+def test_retried_disband_waits_for_work_admitted_after_its_first_fence(controls):
+    service, call = controls
+
+    class PendingStopRPC(_FakeRPC):
+        def info(self, **_kwargs):
+            return {"active": True, "task_id": task["identity"].task_id}
+
+        def interrupt(self, **_kwargs):
+            return None
+
+    service.runtime.rpc = rpc = PendingStopRPC()
+    rpc.sessions[("ops", "Group: room")] = {"session_id": "ops-session"}
+    binding, = service.bindings()
+    lease = service.runtime._ensure_lease(binding)
+    for event_id in ("A", "B"):
+        call("groups.send", event_id=event_id, payload={"text": f"@ops {event_id}", "thread_id": event_id})
+        task, = driver.list_tasks(service.db_path, room_id="room", status="queued")
+        driver.start_task(service.db_path, task["identity"], lease, expected_cancel_generation=0, clock=time.time)
+        with pytest.raises(RuntimeError, match="still stopping"):
+            call("groups.disband", cancel_id="same-disband")
+        assert hosted_rooms.room_state(service.db_path, room_id="room").get("disbanded_at") is None
+        task = driver.get_task(service.db_path, task["identity"])
+        assert task["status"] == "stopping"
+        driver.complete_task_cancel(
+            service.db_path, task["identity"], cancel_id=task["cancel_id"],
+            expected_cancel_generation=task["cancel_generation"], clock=time.time)
+    assert call("groups.disband", cancel_id="same-disband")["tombstone"]

tests/tui_gateway/test_hosted_room_service.py

@@ -321,7 +321,7 @@ def test_stop_room_snapshots_tasks_before_status_transitions(monkeypatch, tmp_pa
     """One running task must not be counted again after it becomes stopping."""
 
     identity = driver.TaskIdentity("room-1", "task-1", "thread-1", "turn-1")
-    task = {"identity": identity, "status": "running", "cancel_id": None}
+    task = {"identity": identity, "status": "running", "cancel_id": None, "payload": {"source_event_seq": 1}}
     calls = []
 
     def listed(_db, *, room_id, status):
@@ -340,7 +340,7 @@ def test_stop_room_snapshots_tasks_before_status_transitions(monkeypatch, tmp_pa
         "request_room_stop",
         lambda _db, *, room_id, cancel_id, **_authority: {
             "room_id": room_id,
-            "cancel_id": cancel_id,
+            "cancel_id": cancel_id, "seq": 2,
         },
     )
     service = HostedRoomService(_server(), db_path=tmp_path / "state.db")

tests/tui_gateway/test_hosted_room_stop_replay.py

@@ -0,0 +1,54 @@
+"""An old Stop must never cancel a user message admitted after its durable fence."""
+
+from types import SimpleNamespace
+import threading
+
+import pytest
+
+from gateway import hosted_room_driver as driver, hosted_rooms
+from tui_gateway.hosted_room_service import HostedRoomService
+
+
+def _service(db):
+    service = HostedRoomService(
+        SimpleNamespace(_methods={}, _sessions={}, _sessions_lock=threading.Lock()), db_path=db)
+    service.local_profiles = lambda: ("default", "ops")
+    return service
+
+
+@pytest.mark.parametrize("restart", [False, True])
+def test_replayed_stop_leaves_a_new_message_queued(tmp_path, restart):
+    db = tmp_path / "state.db"
+    service = _service(db)
+    service.create_room(room_id="room", name="Stop replay", members=[
+        {"member_id": "default", "profile": "default", "handle": "hermes"},
+        {"member_id": "ops", "profile": "ops", "handle": "ops"}])
+    service.send(room_id="room", event_id="first", payload={"text": "@hermes First", "thread_id": "first"})
+    assert service.stop_room("room", cancel_id="old-stop") == 1
+    service.send(room_id="room", event_id="next", payload={"text": "@hermes Next", "thread_id": "next"})
+    queued = driver.list_tasks(db, room_id="room", status="queued")
+    assert len(queued) == 1
+    if restart:
+        service = _service(db)
+    assert service.stop_room("room", cancel_id="old-stop") == 0
+    assert driver.list_tasks(db, room_id="room", status="queued")[0]["identity"] == queued[0]["identity"]
+    assert service.stop_room("room", cancel_id="new-stop") == 1
+    assert driver.list_tasks(db, room_id="room", status="queued") == []
+
+
+def test_message_admitted_between_stop_fence_and_task_scan_survives(tmp_path, monkeypatch):
+    db = tmp_path / "state.db"
+    service = _service(db)
+    service.create_room(room_id="room", name="Concurrent Stop", members=[
+        {"member_id": "default", "profile": "default", "handle": "hermes"},
+        {"member_id": "ops", "profile": "ops", "handle": "ops"}])
+    append_stop = hosted_rooms.request_room_stop
+
+    def append_then_send(*args, **kwargs):
+        fence = append_stop(*args, **kwargs)
+        service.send(room_id="room", event_id="next", payload={"text": "@hermes Next", "thread_id": "next"})
+        return fence
+
+    monkeypatch.setattr(hosted_rooms, "request_room_stop", append_then_send)
+    assert service.stop_room("room", cancel_id="stop") == 0
+    assert len(driver.list_tasks(db, room_id="room", status="queued")) == 1

tui_gateway/hosted_room_service.py

@@ -693,14 +693,17 @@ class HostedRoomService:
     def stop_room(
         self, room_id: str, *, cancel_id: str, require_acknowledged: bool = False) -> int:
         gateway_id, epoch = self._owned_authority(room_id)
-        hosted_rooms.request_room_stop(
+        fence = hosted_rooms.request_room_stop(
             self.db_path, room_id=room_id, cancel_id=cancel_id, expected_gateway_id=gateway_id,
             expected_epoch=epoch)
         pending = 0
         with self._policy_lock:
             tasks = {
                 (task["identity"].room_id, task["identity"].task_id): task
-                for task in self._list_tasks(room_id, _STOPPABLE_STATUSES)}
+                for task in self._list_tasks(room_id, _STOPPABLE_STATUSES)
+                # 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"])}
             for task in tasks.values():
                 own_cancel_id = (
                     task.get("status") == "stopping" and str(task.get("cancel_id") or ""))

tui_gateway/methods_groups.py

@@ -10,6 +10,7 @@ import contextlib
 import importlib
 import os
 import threading
+import uuid
 
 # Imported, so never rebound: it reads the bound server from this module, not server.py's globals.
 from .methods_groups_forward import forward_bound as _forward_to_owner
@@ -470,7 +471,7 @@ def _(rid, params: dict, service) -> dict:
 def _(rid, params: dict, service) -> dict:
     """Durably cancel queued or running work for one hosted room."""
     count = service.stop_room(
-        str(params.get("room_id") or ""), cancel_id=str(params.get("cancel_id") or "desktop-stop"))
+        str(params.get("room_id") or ""), cancel_id=str(params.get("cancel_id") or uuid.uuid4().hex))
     return _ok(rid, {"cancelled": count})