fix(groups): a Stop older than a late cap notice still retires its pause

The cap's room.activity is written from a snapshot, so a Stop can land just
before it. That Stop fences the discussion (Continue could never reopen it),
yet the pause still read as actionable on the server and in Desktop. Both
readers now treat any Stop after the discussion's own message as final.

apps/desktop/src/plugins/hermes-bots/canonical-group-routed.test.tsx

@@ -113,6 +113,13 @@ it('offers no Continue once the discussion used every continue', () => {
   expect(screen.queryByTestId('room-continue')).toBeNull()
 })
 
+it('a Stop that landed before a late cap notice still retires the pause', () => {
+  const source = { ...userMsg(1), event_id: D }
+  renderHistory([source, memberMsg(2), memberMsg(3), ev(4, 'room.stop_requested'), bounded(5)], { maxContinues: 5, onContinue: vi.fn(), onStop: vi.fn() })
+  expect(screen.getByTestId('room-paused')).toBeTruthy()
+  expect(screen.queryByTestId('room-continue')).toBeNull()
+})
+
 it('an unrelated thread\'s user message does not retire the pause', () => {
   renderHistory([...pausedLog, userMsg(6, 'thread-2')], { maxContinues: 5, onContinue: vi.fn(), onStop: vi.fn() })
   expect(screen.getByTestId('room-continue')).toBeTruthy()

apps/desktop/src/plugins/hermes-bots/canonical-group-routed.tsx

@@ -108,13 +108,17 @@ export function pausedDiscussions(events: readonly CanonicalGroupEvent[], maxCon
   const ordered = [...events].sort((a, b) => a.seq - b.seq)
   const { counts, grants } = pauseCounts(ordered)
   const moved = movedOnPauses(ordered)
+  // A Stop after the discussion's own message fences it for good, even one older than the cap notice.
+  const lastStop = ordered.reduce((seq, event) => event.kind === 'room.stop_requested' ? event.seq : seq, 0)
+  const sourceSeq = new Map(ordered.flatMap(event => event.kind === 'message.user' && event.event_id ? [[event.event_id, event.seq] as const] : []))
   const result = new Map<number, CanonicalPausedDiscussion>()
 
   for (const event of ordered) {
     if (!isBoundedActivity(event)) {continue}
 
     const discussion = discussionOf(event)
-    result.set(event.seq, { count: counts.get(event.seq) ?? 0, actionable: !!discussion && !moved.has(event.seq) && (grants.get(discussion) ?? 0) < maxContinues })
+    const fenced = lastStop > (sourceSeq.get(discussion) ?? event.seq)
+    result.set(event.seq, { count: counts.get(event.seq) ?? 0, actionable: !!discussion && !fenced && !moved.has(event.seq) && (grants.get(discussion) ?? 0) < maxContinues })
   }
 
   return result

gateway/hosted_room_router.py

@@ -117,7 +117,8 @@ def validate_continued_event(kind: str, payload: Mapping[str, Any], actor: Mappi
 
 def paused_discussion(events: Sequence[Mapping[str, Any]], discussion_event_id: str) -> dict[str, Any] | None:
     """``{"thread_id", "grant"}`` when the discussion's latest outcome is the cap (``room.activity``
-    ``bounded`` after its last Continue) and neither a Stop nor a newer message in its thread followed;
+    ``bounded`` after its last Continue), no Stop followed its message and no newer message in its thread
+    followed the cap;
     else None. ``grant`` is the number the next Continue takes (1-based)."""
     bounded = None
     grants = 0
@@ -131,9 +132,13 @@ def paused_discussion(events: Sequence[Mapping[str, Any]], discussion_event_id:
     if bounded is None:
         return None
     thread_id = bounded["payload"].get("thread_id")
-    if any(int(event["seq"]) > int(bounded["seq"]) and (
-            event.get("kind") == "room.stop_requested"
-            or (event.get("kind") == "message.user" and (event.get("payload") or {}).get("thread_id") == thread_id))
+    # A Stop anywhere after the discussion's own message fences it for good, even one that landed
+    # before a cap notice computed from an older snapshot: Continue could not reopen it.
+    source_seq = next((int(event["seq"]) for event in events if event.get("event_id") == discussion_event_id),
+                      int(bounded["seq"]))
+    if any((event.get("kind") == "room.stop_requested" and int(event["seq"]) > source_seq) or (
+            int(event["seq"]) > int(bounded["seq"]) and event.get("kind") == "message.user"
+            and (event.get("payload") or {}).get("thread_id") == thread_id)
            for event in events):
         return None
     return {"thread_id": thread_id, "grant": grants + 1}

tests/tui_gateway/test_hosted_room_router.py

@@ -341,6 +341,30 @@ def test_an_unrelated_event_in_the_continue_gap_still_continues(tmp_path, monkey
     assert event["kind"] == "room.continued" and event["seq"] == max(e["seq"] for e in service._events(ROOM_ID))
 
 
+def test_a_stop_before_a_late_cap_notice_refuses_continue(tmp_path, monkeypatch):
+    """The cap notice is written from a snapshot; a Stop can land just before it. That Stop still fences
+    the discussion, so Continue (which could never reopen it) is refused rather than doing nothing."""
+    (tmp_path / "config.yaml").write_text("hosted_rooms:\n  discussion:\n    max_rounds: 3\n    max_messages: 2\n")
+    service = _service(tmp_path)
+    _room(service, "everyone")
+    _send(service, "m1", "Discuss.")
+    _settle_queued(service, "@everyone next", 0)
+    gateway_id, epoch = service._owned_authority(ROOM_ID)
+    status = service._append_room_status
+
+    def stop_first(room, decision):
+        hosted_rooms.request_room_stop(service.db_path, room_id=ROOM_ID, cancel_id="late", expected_gateway_id=gateway_id,
+                                       expected_epoch=epoch)
+        status(room, decision)
+    monkeypatch.setattr(service, "_append_room_status", stop_first)
+    _settle_queued(service, "@everyone next", 1)
+    kinds = [event["kind"] for event in service._events(ROOM_ID)]
+    assert kinds.index("room.stop_requested") < kinds.index("room.activity")
+    discussion_id = _kinds(service, "room.activity")[0]["payload"]["discussion_event_id"]
+    with pytest.raises(hosted_rooms.HostedRoomError, match="not paused"):
+        service.continue_room(ROOM_ID, discussion_event_id=discussion_id)
+
+
 def test_continue_stops_at_the_continue_limit(tmp_path, monkeypatch):
     monkeypatch.setattr(router, "MAX_DISCUSSION_CONTINUES", 1)
     service = _bounded_room(tmp_path)