fix(bot-mode): cancel re-routes on worker races instead of failing disband

Follow-ups on the salvaged #97744 runner:

- tui_gateway/hosted_room_driver.py: HostedRoomRuntime.cancel() treated its
  initial status read as truth, so a task transitioning queued->running (or
  settling) between the read and the state call surfaced a transient
  'running work requires acknowledged two-phase cancellation' /
  StaleTaskError to the caller and failed groups.disband. Deterministic
  repro on the PR head: test_client_event_id_cannot_squat_disband_receipt
  failed 5/5 locally. cancel() now re-reads and re-routes on every
  race-shaped failure (bounded retries), returns already-cancelled tasks
  idempotently, and rejects truly terminal states honestly.
- methods_groups conflict resolution keeps both method sets: the replication
  surface from #99047 (groups.replicate/replica_state/promote/demote) and
  the runner surface from this layer (groups.stop/retry/approve).
- test_groups_replication_methods.py updated to the runner's stricter
  create contract (2-6 profile-backed members, live worker service).

tests/tui_gateway/test_groups_replication_methods.py

@@ -6,6 +6,7 @@ from __future__ import annotations
 import pytest
 
 import tui_gateway.server as srv
+from tui_gateway import methods_groups
 
 MEMBERS = [{"kind": "bot", "id": "planner"}]
 
@@ -14,8 +15,12 @@ MEMBERS = [{"kind": "bot", "id": "planner"}]
 def home(tmp_path, monkeypatch):
     path = tmp_path / ".hermes"
     path.mkdir()
+    (path / "profiles" / "ops").mkdir(parents=True)
     monkeypatch.setenv("HERMES_HOME", str(path))
-    return path
+    methods_groups.stop_hosted_room_service(timeout=1.0)
+    methods_groups.start_hosted_room_service()
+    yield path
+    methods_groups.stop_hosted_room_service(timeout=1.0)
 
 
 def _result(envelope):
@@ -124,7 +129,18 @@ def test_demote_fences_local_room_against_newer_epoch(home):
     _result(
         srv._methods["groups.create"](
             1,
-            {"room_id": "room-1", "name": "Local room", "members": MEMBERS},
+            {
+                "room_id": "room-1",
+                "name": "Local room",
+                "members": [
+                    {
+                        "member_id": "default",
+                        "profile": "default",
+                        "handle": "hermes",
+                    },
+                    {"member_id": "ops", "profile": "ops", "handle": "ops"},
+                ],
+            },
         )
     )
     observed_gateway = "install:" + "b" * 32

tui_gateway/hosted_room_driver.py

@@ -29,6 +29,8 @@ from typing import Any, ContextManager, Protocol, cast
 
 from gateway import hosted_room_driver as state
 
+_CANCEL_ROUTE_RETRIES = 8
+
 
 ROOM_SESSION_SOURCE = "bot_room"
 MAX_TERMINAL_TEXT_BYTES = 64 * 1024
@@ -252,41 +254,73 @@ class HostedRoomRuntime:
         *,
         cancel_id: str,
     ) -> dict[str, Any]:
-        """Persist a stop intent, then commit cancellation after acknowledgement."""
-        before = state.get_task(self.db_path, identity)
-        if before["status"] in {"queued", "deferred"}:
-            cancelled = state.cancel_task(
-                self.db_path,
-                identity,
-                cancel_id=cancel_id,
-                expected_cancel_generation=before["cancel_generation"],
-                clock=self.clock,
-            )
-            self.wakeup()
-            return cancelled
+        """Persist a stop intent, then commit cancellation after acknowledgement.
 
-        stopping = state.begin_task_cancel(
-            self.db_path,
-            identity,
-            cancel_id=cancel_id,
-            expected_cancel_generation=before["cancel_generation"],
-            clock=self.clock,
-        )
-        binding = self._binding_for_room(identity.room_id)
-        try:
-            if binding is not None and self._interrupt_stopping_task(binding, stopping):
-                stopping = state.complete_task_cancel(
+        The worker thread transitions tasks concurrently with cancellation
+        (queued -> running -> terminal), so the status read below is only a
+        routing hint. Every fast-path failure caused by a concurrent
+        transition re-reads and re-routes instead of surfacing a transient
+        `InvalidTaskTransitionError`/`StaleTaskError` to the caller.
+        """
+        for _ in range(_CANCEL_ROUTE_RETRIES):
+            before = state.get_task(self.db_path, identity)
+            if before["status"] == "cancelled":
+                return before
+            if before["status"] in state.TERMINAL_STATUSES:
+                raise state.InvalidTaskTransitionError(
+                    f"cannot cancel task in state '{before['status']}'"
+                )
+            if before["status"] in {"queued", "deferred"}:
+                try:
+                    cancelled = state.cancel_task(
+                        self.db_path,
+                        identity,
+                        cancel_id=cancel_id,
+                        expected_cancel_generation=before["cancel_generation"],
+                        clock=self.clock,
+                    )
+                except (state.InvalidTaskTransitionError, state.StaleTaskError):
+                    # Lost the race with the worker; re-read and re-route.
+                    continue
+                self.wakeup()
+                return cancelled
+            try:
+                stopping = state.begin_task_cancel(
                     self.db_path,
                     identity,
                     cancel_id=cancel_id,
-                    expected_cancel_generation=stopping["cancel_generation"],
+                    expected_cancel_generation=before["cancel_generation"],
                     clock=self.clock,
                 )
-        except Exception as exc:
-            self._record_error(f"stop remains pending: {exc}")
-        stopping = state.get_task(self.db_path, identity)
-        self.wakeup()
-        return stopping
+            except (state.InvalidTaskTransitionError, state.StaleTaskError):
+                # Task settled or re-queued mid-flight; re-read and re-route.
+                continue
+            binding = self._binding_for_room(identity.room_id)
+            try:
+                if binding is not None and self._interrupt_stopping_task(
+                    binding, stopping
+                ):
+                    stopping = state.complete_task_cancel(
+                        self.db_path,
+                        identity,
+                        cancel_id=cancel_id,
+                        expected_cancel_generation=stopping["cancel_generation"],
+                        clock=self.clock,
+                    )
+            except Exception as exc:
+                self._record_error(f"stop remains pending: {exc}")
+            stopping = state.get_task(self.db_path, identity)
+            self.wakeup()
+            return stopping
+        # Exhausted routing retries under sustained contention: surface the
+        # live status honestly rather than a transient transition error.
+        final = state.get_task(self.db_path, identity)
+        if final["status"] == "cancelled":
+            return final
+        raise state.InvalidTaskTransitionError(
+            f"cancel kept losing races with task transitions "
+            f"(last observed state '{final['status']}')"
+        )
 
     def retry_indeterminate(self, identity: state.TaskIdentity) -> dict[str, Any]:
         """Explicitly retry one uncertain attempt under the current room lease."""