From ed48ae01b9320cfd7ae08493f42227623a5899c3 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 22:37:56 +0200 Subject: [PATCH] [F113] collapse WorkSession creation to the validated service path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _create_work_session_if_needed constructed WorkSessionTable directly, duplicating WorkSessionService.create's validation (existing-active check, single-active-per-task supersede, project/task existence). The two sites had drifted. Route through WorkSessionService.create instead, mapping ConflictError to the idempotent 'if needed' None. Remove the now-dead _supersede_other_active_sessions (create's supersede_active_sessions_for_task replaces it). Fix three pre-existing RED tests surfaced by the sweep (all confirmed failing on the F110 commit before this change): - test_fail_qa_work_session_fallback_excludes_qa_session: inserted two ACTIVE work_sessions per task, violating uq_work_sessions_one_active _per_task (migration 047). The QA session is now ABANDONED — still in the fallback query's result set (the query filters by task_id + agent_id, not status), so the exclude filter (agent_id != qa_id) is still exercised and the dev is resolved. - test_ceo_reject_routes_coordination_task_to_main_pm / test_ceo_reject_routes_batch_umbrella_to_main_pm: ceo_reject emits an audit row keyed to CEO_AGENT_ID, but the tests never seeded the CEO agent row (fk_audit_log_agent_id_agents). Seed the CEO agent (get-or- create, mirroring test_ceo_reject_writes_handoff_journal). --- roboco/services/task.py | 102 ++++++------------ .../test_task_service_lifecycle_misc.py | 50 ++++++++- .../test_task_service_transitions.py | 50 ++++++++- 3 files changed, 131 insertions(+), 71 deletions(-) diff --git a/roboco/services/task.py b/roboco/services/task.py index 8b9e93ce..06d26018 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -56,16 +56,18 @@ from roboco.models.base import ( ) from roboco.models.permissions import AgentContext, TaskAction from roboco.models.task import TaskCreateRequest -from roboco.models.work_session import WorkSessionStatus +from roboco.models.work_session import WorkSessionCreate from roboco.seeds.initial_data import AGENT_UUIDS from roboco.services.base import ( BaseService, + ConflictError, NotFoundError, ServiceError, UnauthorizedError, ValidationError, ) from roboco.services.content_notes import apply_structured_note +from roboco.services.work_session import WorkSessionService from roboco.utils.converters import require_uuid, to_python_uuid if TYPE_CHECKING: @@ -2362,7 +2364,9 @@ class TaskService(BaseService): false negative — so it is not a correctness concern. """ await self.session.execute( - text("SELECT pg_advisory_xact_lock(hashtextextended(CAST(:aid AS text), 0))"), + text( + "SELECT pg_advisory_xact_lock(hashtextextended(CAST(:aid AS text), 0))" + ), {"aid": str(agent_id)}, ) @@ -2486,31 +2490,6 @@ 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, @@ -2563,34 +2542,6 @@ class TaskService(BaseService): ) return None - # Check if session already exists for this task+agent (resilient to any - # legacy duplicate ACTIVE rows — see - # WorkSessionService.get_active_for_task). - existing = await self.session.execute( - select(WorkSessionTable) - .where( - and_( - WorkSessionTable.task_id == task.id, - WorkSessionTable.agent_id == agent_id, - WorkSessionTable.status == WorkSessionStatus.ACTIVE, - ) - ) - .order_by(WorkSessionTable.started_at.desc()) - ) - if existing.scalars().first(): - self.log.debug( - "Work session already exists", - task_id=str(task.id), - agent_id=str(agent_id), - ) - 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 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 # - For parent tasks: merge into default branch (master) @@ -2604,19 +2555,34 @@ class TaskService(BaseService): if parent_branch: target_branch = str(parent_branch) - # Create the work session - work_session = WorkSessionTable( - project_id=project_id, - task_id=task.id, - agent_id=agent_id, - branch_name=branch_name, - base_branch=target_branch, # Created from target - target_branch=target_branch, # Will merge back to target - status=WorkSessionStatus.ACTIVE, - ) - - self.session.add(work_session) - await self.session.flush() + # Create the work session through the validated service path (F113): the + # service-layer ``WorkSessionService.create`` is the single source of + # truth — it enforces the existing-active check (ConflictError on a + # duplicate), the single-active-per-task supersede invariant, and + # project/task existence. Constructing ``WorkSessionTable`` directly + # here duplicated that validation and the two sites had drifted. The + # ``ConflictError`` maps to the "if needed" idempotent semantics — an + # agent re-claiming its own already-active session is a no-op (None), + # not an error. The supersede still runs inside ``create`` before the + # insert (closing any OTHER agent's stale ACTIVE row first). + try: + work_session = await WorkSessionService(self.session).create( + WorkSessionCreate( + project_id=cast("UUID", project_id), + task_id=cast("UUID", task.id), + agent_id=agent_id, + branch_name=branch_name, + base_branch=target_branch, # Created from target + target_branch=target_branch, # Will merge back to target + ) + ) + except ConflictError: + self.log.debug( + "Work session already exists", + task_id=str(task.id), + agent_id=str(agent_id), + ) + return None # Link session to task task.work_session_id = cast("Any", work_session.id) diff --git a/tests/integration/test_task_service_lifecycle_misc.py b/tests/integration/test_task_service_lifecycle_misc.py index 9781b610..b9d427f7 100644 --- a/tests/integration/test_task_service_lifecycle_misc.py +++ b/tests/integration/test_task_service_lifecycle_misc.py @@ -36,12 +36,13 @@ from roboco.models.base import ( ) from roboco.models.permissions import AgentContext from roboco.models.task import TaskCreateRequest -from roboco.models.work_session import WorkSessionStatus +from roboco.models.work_session import WorkSessionCreate, WorkSessionStatus from roboco.services.task import ( SoftBlockInfo, SoftBlockInput, TaskService, ) +from roboco.services.work_session import WorkSessionService from sqlalchemy import select if TYPE_CHECKING: @@ -448,6 +449,53 @@ async def test_create_work_session_no_project_returns_none( assert out is None +@pytest.mark.asyncio +async def test_create_work_session_delegates_to_service_create( + task_setup: dict, db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """F113: the claim path must create the WorkSession through the validated + ``WorkSessionService.create`` (the single source of truth), not construct a + ``WorkSessionTable`` directly. Two divergent creation sites had drifted and + bypassed the service-layer validation (existing-active check, supersede + invariant, project/task existence). Routing through ``create`` collapses + them to one validated path. The derived target_branch (parent branch for + subtasks, project default for roots) is passed in via ``WorkSessionCreate``. + """ + svc = task_setup["svc"] + task = await svc.create(_req(task_setup)) + task.branch_name = "feature/backend/delegate" + await db_session.flush() + + captured: dict[str, Any] = {} + original_create = WorkSessionService.create + + async def _spy_create(_self: Any, data: WorkSessionCreate) -> Any: + # Record the WorkSessionCreate the claim path handed to the service, + # then run the real create so the row persists (the FK on + # tasks.work_session_id requires a real work_sessions row). + captured["data"] = data + return await original_create(_self, data) + + monkeypatch.setattr(WorkSessionService, "create", _spy_create) + + out = await svc._create_work_session_if_needed( + task, task_setup["agent_id"], "developer" + ) + + assert out is not None + assert "data" in captured + sent = captured["data"] + assert isinstance(sent, WorkSessionCreate) + assert sent.project_id == task_setup["project_id"] + assert sent.task_id == task.id + assert sent.agent_id == task_setup["agent_id"] + assert sent.branch_name == "feature/backend/delegate" + # Root task targets the project default branch. + assert sent.target_branch == sent.base_branch + # The claim path links the session back onto the task. + assert task.work_session_id == out.id + + # --------------------------------------------------------------------------- # unclaim_for_reaper / unclaim_for_agent — work-session abandon paths # --------------------------------------------------------------------------- diff --git a/tests/integration/test_task_service_transitions.py b/tests/integration/test_task_service_transitions.py index 2f68f066..7cc0ed17 100644 --- a/tests/integration/test_task_service_transitions.py +++ b/tests/integration/test_task_service_transitions.py @@ -952,7 +952,13 @@ async def test_fail_qa_work_session_fallback_excludes_qa_session( db_session.add(qa) await db_session.flush() - # A QA-attributed work session (older) and a dev-attributed one (newer). + # A QA-attributed work session (older, abandoned) and a dev-attributed one + # (newer, active). Only one ACTIVE session may exist per task + # (``uq_work_sessions_one_active_per_task``), so the QA session is + # ABANDONED — realistic for a stale QA review session and still in the + # fallback query's result set (the query filters by task_id + agent_id, + # not by status). The exclude filter (``agent_id != qa_agent_id``) is + # what must keep the QA session from being misread as the revision dev. qa_ws = WorkSessionTable( id=uuid4(), project_id=task_setup["project_id"], @@ -961,7 +967,7 @@ async def test_fail_qa_work_session_fallback_excludes_qa_session( branch_name="feature/backend/abc", base_branch="main", target_branch="main", - status=WorkSessionStatus.ACTIVE, + status=WorkSessionStatus.ABANDONED, ) dev_ws = WorkSessionTable( id=uuid4(), @@ -1083,6 +1089,25 @@ async def test_ceo_reject_routes_coordination_task_to_main_pm( metrics={}, ) ) + # The CEO rejects the task; ceo_reject emits an audit row keyed to the CEO + # agent, so the CEO row must exist (fk_audit_log_agent_id_agents). + ceo_id = UUID(AGENT_UUIDS["ceo"]) + if await db_session.get(AgentTable, ceo_id) is None: + db_session.add( + AgentTable( + id=ceo_id, + name="CEO", + slug="ceo", + role=AgentRole.CEO, + team=Team.MAIN_PM, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="ceo", + capabilities=[], + permissions=[], + metrics={}, + ) + ) product = ProductTable( name="P", slug=f"p-{uuid4().hex[:8]}", created_by=task_setup["agent_id"] ) @@ -1133,6 +1158,27 @@ async def test_ceo_reject_routes_batch_umbrella_to_main_pm( ) await db_session.flush() + # The CEO rejects the umbrella; ceo_reject emits an audit row keyed to the + # CEO agent, so the CEO row must exist (fk_audit_log_agent_id_agents). + ceo_id = UUID(AGENT_UUIDS["ceo"]) + if await db_session.get(AgentTable, ceo_id) is None: + db_session.add( + AgentTable( + id=ceo_id, + name="CEO", + slug="ceo", + role=AgentRole.CEO, + team=Team.MAIN_PM, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="ceo", + capabilities=[], + permissions=[], + metrics={}, + ) + ) + await db_session.flush() + task = await svc.create(_req(task_setup)) task.status = TaskStatus.AWAITING_CEO_APPROVAL task.project_id = None # umbrella: no project, no product — carries a batch_id