fix(messaging): reuse a group's live session instead of recreating per open

create_session and create_session_with_access_check closed the group's active
session and opened a new one on every call. A group is meant to have ONE live
session (groups.active_session_id) that all participants post into, so this
churned a single conversation across many sessions — the smoke run showed ~one
session per message, and the CEO could not hold a conversation in a channel.

Both now reuse the live session when one is active and only open a fresh one
when none is (closed via timeout / boundary / merge), matching the existing
get_or_create_active_session contract.

roboco/services/messaging.py

@@ -389,10 +389,16 @@ class MessagingService(BaseService):
         if not group:
             raise ValueError(f"Group {req.group_id} not found")
 
-        # Close existing active session if any
+        # A group has ONE live session that all participants post into. Reuse it
+        # instead of closing it and opening a new one on every call — the
+        # close-and-recreate pattern churned a single conversation across many
+        # sessions (the smoke run showed ~one session per message, and the CEO
+        # could not hold a conversation). Only open a fresh session when none is
+        # currently active (the prior one closed via timeout / boundary / merge).
         if group.active_session_id:
-            session_id = cast("UUID", group.active_session_id)
+            existing = await self.get_session(cast("UUID", group.active_session_id))
-            await self.close_session(session_id, "New session started")
+            if existing is not None and existing.status == SessionStatus.ACTIVE:
+                return existing
 
         session = SessionTable(
             group_id=req.group_id,
@@ -619,9 +625,12 @@ class MessagingService(BaseService):
             )
         )
         active = active_result.scalar_one_or_none()
-        if active:
+        if active is not None:
-            active.status = SessionStatus.CLOSED
+            # Reuse the group's live session (see create_session). The CEO and
-            active.closed_at = datetime.now(UTC)
+            # agents post into one session per group, not a fresh one per open —
+            # closing + recreating here is what fragmented one conversation
+            # across many sessions.
+            return active
 
         new_session = SessionTable(
             group_id=request.group_id,

tests/integration/test_messaging_service.py

@@ -1655,15 +1655,15 @@ async def test_close_session_handles_bus_failure(msg_setup: dict) -> None:
 
 
 # ---------------------------------------------------------------------------
-# create_session_with_access_check — closes prior active + privileged path
+# create_session_with_access_check — reuses the group's live session
 # ---------------------------------------------------------------------------
 
 
 @pytest.mark.asyncio
-async def test_create_session_with_access_check_closes_prior(
+async def test_create_session_with_access_check_reuses_active(
     msg_setup: dict,
 ) -> None:
-    """Existing ACTIVE session is closed before creating new one."""
+    """A group has ONE live session; opening again reuses it (no churn)."""
     svc = msg_setup["svc"]
     aid = msg_setup["agent_id"]
     ch = await svc.create_channel(_channel_req(uuid4().hex[:6]))
@@ -1683,7 +1683,11 @@ async def test_create_session_with_access_check_closes_prior(
             timeout_seconds=300,
         ),
     )
-    assert new_sess.id != prior.id
+    # Same live session is returned, not a fresh one — and it stays active.
+    assert new_sess.id == prior.id
+    refetched = await svc.get_session(prior.id)
+    assert refetched is not None
+    assert refetched.status == SessionStatus.ACTIVE
 
 
 # ---------------------------------------------------------------------------