From dba4a378ad6f698c88e3be33f2fb943cf88503f0 Mon Sep 17 00:00:00 2001 From: Renn F Date: Wed, 24 Jun 2026 05:44:59 +0200 Subject: [PATCH] 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 | 46 ++++++++++++------- .../integration/test_work_session_service.py | 19 ++++---- 2 files changed, 39 insertions(+), 26 deletions(-) diff --git a/roboco/services/task.py b/roboco/services/task.py index d96fbd04..fe26bd50 100644 --- a/roboco/services/task.py +++ b/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 - # different agent (pool release, reaper unclaim, escalation redirect) - # otherwise left duplicate ACTIVE rows that crashed 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 != 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() + # session for this task before opening a new one (a re-claim by a + # different agent otherwise left duplicate ACTIVE rows that crashed + # get_active_for_task with MultipleResultsFound — the respawn-loop wedge). + await self._supersede_other_active_sessions(cast("UUID", task.id), agent_id) # Determine target branch: # - For subtasks: merge into parent task's branch diff --git a/tests/integration/test_work_session_service.py b/tests/integration/test_work_session_service.py index e4e71a35..e567aec3 100644 --- a/tests/integration/test_work_session_service.py +++ b/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: - """A distinct agent to simulate a re-claim by someone else.""" +async def _second_agent(db_session: AsyncSession) -> UUID: + """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",