fix(gate): make the work-session invariant fix pass the full gate

Two failures the full make-quality flagged after 06adf978 landed on master:
- mypy: _second_agent (test helper) now returns the agent's UUID, so
  WorkSessionCreate(agent_id=...) receives a real uuid.UUID rather than the
  ORM column type.
- xenon: the single-active-per-task supersede is extracted out of
  _create_work_session_if_needed into _supersede_other_active_sessions,
  bringing the former back under complexity rank B.

No behavior change; 1566 work-session/gateway/service tests green.

roboco/services/task.py

@@ -2095,6 +2095,31 @@ class TaskService(BaseService):
     # GIT WORK SESSION INTEGRATION
     # =========================================================================
 
+    async def _supersede_other_active_sessions(
+        self, task_id: UUID, keep_agent_id: UUID
+    ) -> None:
+        """Abandon any ACTIVE work session for a task owned by a different agent.
+
+        Enforces the single-active-per-task invariant at claim time: a re-claim
+        by a different agent (pool release, reaper unclaim, escalation redirect)
+        otherwise left the prior holder's ACTIVE row open, and the duplicate
+        ACTIVE rows crashed ``WorkSessionService.get_active_for_task`` with
+        ``MultipleResultsFound`` — the i_will_plan respawn-loop wedge.
+        """
+        stale = await self.session.execute(
+            select(WorkSessionTable).where(
+                and_(
+                    WorkSessionTable.task_id == task_id,
+                    WorkSessionTable.agent_id != keep_agent_id,
+                    WorkSessionTable.status == WorkSessionStatus.ACTIVE,
+                )
+            )
+        )
+        for prior in stale.scalars().all():
+            prior.status = WorkSessionStatus.ABANDONED
+            prior.ended_at = datetime.now(UTC)
+        await self.session.flush()
+
     async def _create_work_session_if_needed(
         self,
         task: TaskTable,
@@ -2170,23 +2195,10 @@ class TaskService(BaseService):
             return None
 
         # Single-active-per-task invariant: close any OTHER agent's stale ACTIVE
-        # session for this task before opening a new one. A re-claim by a
+        # session for this task before opening a new one (a re-claim by a
-        # different agent (pool release, reaper unclaim, escalation redirect)
+        # different agent otherwise left duplicate ACTIVE rows that crashed
-        # otherwise left duplicate ACTIVE rows that crashed get_active_for_task
+        # get_active_for_task with MultipleResultsFound — the respawn-loop wedge).
-        # with MultipleResultsFound — the i_will_plan respawn-loop wedge.
+        await self._supersede_other_active_sessions(cast("UUID", task.id), agent_id)
-        stale = await self.session.execute(
-            select(WorkSessionTable).where(
-                and_(
-                    WorkSessionTable.task_id == task.id,
-                    WorkSessionTable.agent_id != agent_id,
-                    WorkSessionTable.status == WorkSessionStatus.ACTIVE,
-                )
-            )
-        )
-        for prior in stale.scalars().all():
-            prior.status = WorkSessionStatus.ABANDONED
-            prior.ended_at = datetime.now(UTC)
-        await self.session.flush()
 
         # Determine target branch:
         # - For subtasks: merge into parent task's branch

tests/integration/test_work_session_service.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
 import pytest_asyncio
@@ -469,10 +469,11 @@ async def test_has_unpushed_commits_false_after_pr(ws_setup: dict) -> None:
 # ---------------------------------------------------------------------------
 
 
-async def _second_agent(db_session: AsyncSession) -> AgentTable:
+async def _second_agent(db_session: AsyncSession) -> UUID:
-    """A distinct agent to simulate a re-claim by someone else."""
+    """Insert a distinct agent (re-claim by someone else); return its id."""
+    aid = uuid4()
     agent = AgentTable(
-        id=uuid4(),
+        id=aid,
         name="Dev2",
         slug=f"be-dev-{uuid4().hex[:8]}",
         role=AgentRole.DEVELOPER,
@@ -486,7 +487,7 @@ async def _second_agent(db_session: AsyncSession) -> AgentTable:
     )
     db_session.add(agent)
     await db_session.flush()
-    return agent
+    return aid
 
 
 @pytest.mark.asyncio
@@ -496,12 +497,12 @@ async def test_create_supersedes_other_agents_active_session(
     """A re-claim by a different agent abandons the stale session (no dup ACTIVE)."""
     svc = ws_setup["svc"]
     first = await svc.create(_payload(ws_setup))  # agent A
-    agent_b = await _second_agent(db_session)
+    agent_b_id = await _second_agent(db_session)
     second = await svc.create(
         WorkSessionCreate(
             project_id=ws_setup["project_id"],
             task_id=ws_setup["task_id"],
-            agent_id=agent_b.id,
+            agent_id=agent_b_id,
             branch_name=f"feature/x-{uuid4().hex[:6]}",
             base_branch="main",
             target_branch="main",
@@ -536,12 +537,12 @@ async def test_partial_unique_index_blocks_two_active_for_task(
     """The DB backstop: a second ACTIVE row for one task violates the index."""
     svc = ws_setup["svc"]
     await svc.create(_payload(ws_setup))  # one ACTIVE session
-    agent_b = await _second_agent(db_session)
+    agent_b_id = await _second_agent(db_session)
     # Bypass the service supersede and force a raw duplicate ACTIVE row.
     dup = WorkSessionTable(
         project_id=ws_setup["project_id"],
         task_id=ws_setup["task_id"],
-        agent_id=agent_b.id,
+        agent_id=agent_b_id,
         branch_name="feature/dup",
         base_branch="main",
         target_branch="main",