From b53d8fe194d7be9d1605bc8f51e1998f9a8b3143 Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 12 May 2026 05:22:44 +0200 Subject: [PATCH] feat(gateway): C4 auto-create WorkSession on claim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pre-gateway parity. Smoke run 3 showed task.work_session_id null on every task — the choreographer's claim/plan/start path didn't create the row that downstream subsystems (panel, PR tracking, merge chain) need to track agent-per-task git activity. Add TaskService.ensure_work_session(task_id, agent_id) as a public wrapper around the existing _create_work_session_if_needed logic. Role restriction lifted to None so both developers and PMs get a session (pre-gateway always created sessions for all claimants). Built-in re-entry guard prevents duplicate rows on re-claim. Wire the call into both _claim_plan_start_run and _resume_from_claimed immediately before _touch, so every successful in_progress transition (including the stuck-claimed recovery path) creates the row. Spec ref: Wave C task C4 (2026-05-12). --- .../services/gateway/choreographer/_impl.py | 10 + roboco/services/task.py | 22 ++ .../gateway/test_work_session_auto_create.py | 248 ++++++++++++++++++ 3 files changed, 280 insertions(+) create mode 100644 tests/unit/gateway/test_work_session_auto_create.py diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index 22d17c84..e1df128e 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -738,6 +738,10 @@ class Choreographer: task_id=task_id, verb=verb_name, ) + # Wave C4 (2026-05-12) — pre-gateway parity. Ensure WorkSession row + # exists on the stuck-claimed recovery path too (same guarantee as + # _claim_plan_start_run). Re-entry guard inside ensure_work_session. + await self.task.ensure_work_session(task_id, agent_id) await self._touch(task_id) return Envelope.ok( status=str(t.status), @@ -840,6 +844,12 @@ class Choreographer: task_id=ctx.task_id, verb=verb_name, ) + # Wave C4 (2026-05-12) — pre-gateway parity. Create the WorkSession + # row so downstream subsystems (panel, PR, merge chain) can track + # this agent's per-task git activity. work_session_id stored on the + # task; one WorkSession per (agent, task) claim cycle; re-entry + # guard inside ensure_work_session prevents duplicate rows. + await self.task.ensure_work_session(ctx.task_id, ctx.agent_id) await self._touch(ctx.task_id) return Envelope.ok( status=str(t.status), diff --git a/roboco/services/task.py b/roboco/services/task.py index 51daf62d..d988c623 100644 --- a/roboco/services/task.py +++ b/roboco/services/task.py @@ -4712,6 +4712,28 @@ class TaskService(BaseService): await self.session.flush() return task + async def ensure_work_session( + self, + task_id: UUID, + agent_id: UUID, + ) -> None: + """Create a WorkSession row if one does not already exist for this claim. + + Wave C4 (2026-05-12) — pre-gateway parity. The gateway's claim/plan/ + start path calls this after the task reaches in_progress so every + (agent, task) claim cycle has a WorkSession row that downstream + subsystems (panel, PR tracking, merge chain) can use. Delegates to + _create_work_session_if_needed with role=None so both developers and + PMs get a session (pre-gateway created sessions for all claimants). + No-ops if work_session_id is already set (re-entry guard). + """ + task = await self.get(task_id) + if not task: + return + if task.work_session_id: + return + await self._create_work_session_if_needed(task, agent_id, agent_role=None) + async def mark_evidence_inspected(self, task_id: UUID) -> None: """Set qa_evidence_inspected=True on the task.""" task = await self.get(task_id) diff --git a/tests/unit/gateway/test_work_session_auto_create.py b/tests/unit/gateway/test_work_session_auto_create.py new file mode 100644 index 00000000..60464d44 --- /dev/null +++ b/tests/unit/gateway/test_work_session_auto_create.py @@ -0,0 +1,248 @@ +"""Wave C4 (2026-05-12): claim auto-creates a WorkSession row. + +Smoke run 3 showed task.work_session_id null on every task. Pre-gateway +created the row at claim time so the panel/PR/merge subsystems could +track agent-per-task git activity (branch, commits, PR number/url, +merge status). Restoring that side-effect. + +The choreographer's _claim_plan_start_run (and _resume_from_claimed) +calls TaskService.ensure_work_session(task_id, agent_id) after the +task reaches in_progress, which creates the WorkSession and stores +its id on the task. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + + +def _make_task_svc(agent_id, task_id, *, status: str): + """Build a TaskService AsyncMock that completes the (claim, set_plan, start) + sequence and returns a task with branch_name set (as the real service does + after auto-creating the branch during claim side-effects). + """ + in_progress_task = MagicMock( + id=task_id, + status="in_progress", + plan={"text": "plan text"}, + assigned_to=agent_id, + branch_name="feature/backend/abc", + work_session_id=None, + commits=[], + pr_number=None, + quick_context=None, + team="backend", + task_type="code", + parent_task_id=None, + sequence=0, + project_id=uuid4(), + ) + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock( + id=task_id, + status=status, + plan=None, + assigned_to=None, + branch_name="feature/backend/abc", + work_session_id=None, + commits=[], + pr_number=None, + quick_context=None, + team="backend", + task_type="code", + parent_task_id=None, + sequence=0, + project_id=uuid4(), + ) + task_svc.agent_for.return_value = MagicMock( + id=agent_id, role="developer", team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.claim.return_value = MagicMock( + id=task_id, status="claimed", plan=None, assigned_to=agent_id + ) + task_svc.set_plan.return_value = MagicMock( + id=task_id, status="claimed", plan={"text": "plan text"}, assigned_to=agent_id + ) + task_svc.start.return_value = in_progress_task + task_svc.ensure_work_session.return_value = None + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + return task_svc + + +def _make_deps(task_svc) -> ChoreographerDeps: + evidence_repo = AsyncMock() + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(evidence_repo, method).return_value = [] + return ChoreographerDeps( + task=task_svc, + work_session=AsyncMock(), + git=AsyncMock(), + a2a=AsyncMock(), + journal=AsyncMock(), + audit=AsyncMock(), + evidence_repo=evidence_repo, + ) + + +@pytest.mark.asyncio +async def test_i_will_work_on_calls_ensure_work_session() -> None: + """After a successful i_will_work_on, TaskService.ensure_work_session is + called once with (task_id, agent_id) so a WorkSession row is created and + task.work_session_id is populated. + + Wave C4 (2026-05-12): pre-gateway parity. Smoke run 3 showed + task.work_session_id null on every in_progress task. + """ + agent_id = uuid4() + task_id = uuid4() + task_svc = _make_task_svc(agent_id, task_id, status="pending") + deps = _make_deps(task_svc) + c = Choreographer(deps) + + env = await c.i_will_work_on(agent_id, task_id, plan="do x then y") + + assert env.error is None, f"Expected ok, got error={env.error} msg={env.message}" + assert env.status == "in_progress" + task_svc.ensure_work_session.assert_awaited_once_with(task_id, agent_id) + + +@pytest.mark.asyncio +async def test_i_will_plan_calls_ensure_work_session() -> None: + """PMs also get a WorkSession via ensure_work_session (cell_pm role, planning + task). Both i_will_work_on and i_will_plan share _claim_plan_start_run so + the same hook fires for both. + + Wave C4 (2026-05-12): pre-gateway parity. + """ + pm_agent_id = uuid4() + task_id = uuid4() + + in_progress_task = MagicMock( + id=task_id, + status="in_progress", + plan={"approach": "plan text", "sub_tasks": ["t1"]}, + assigned_to=pm_agent_id, + branch_name="feature/main_pm/abc", + work_session_id=None, + commits=[], + pr_number=None, + quick_context=None, + team="main_pm", + task_type="planning", + parent_task_id=None, + sequence=0, + project_id=uuid4(), + ) + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock( + id=task_id, + status="pending", + plan=None, + assigned_to=None, + branch_name="feature/main_pm/abc", + work_session_id=None, + commits=[], + pr_number=None, + quick_context=None, + team="main_pm", + task_type="planning", + parent_task_id=None, + sequence=0, + project_id=uuid4(), + ) + task_svc.agent_for.return_value = MagicMock( + id=pm_agent_id, role="cell_pm", team="backend", slug=None + ) + task_svc.list_in_progress_for_agent.return_value = [] + task_svc.list_paused_for_agent.return_value = [] + task_svc.get_subtasks.return_value = [] + task_svc.claim.return_value = MagicMock( + id=task_id, status="claimed", plan=None, assigned_to=pm_agent_id + ) + task_svc.set_plan.return_value = MagicMock( + id=task_id, + status="claimed", + plan={"approach": "plan text"}, + assigned_to=pm_agent_id, + ) + task_svc.start.return_value = in_progress_task + task_svc.ensure_work_session.return_value = None + task_svc.session = MagicMock() + task_svc.session.begin_nested = MagicMock( + return_value=MagicMock( + __aenter__=AsyncMock(return_value=None), + __aexit__=AsyncMock(return_value=False), + ) + ) + evidence_repo = AsyncMock() + for method in ( + "list_unread_a2a", + "list_unread_mentions", + "list_pending_notifications", + "task_metadata_gaps", + "recent_team_activity", + "blockers_in_lane", + "journal_highlights_for_task", + ): + getattr(evidence_repo, method).return_value = [] + deps = ChoreographerDeps( + task=task_svc, + work_session=AsyncMock(), + git=AsyncMock(), + a2a=AsyncMock(), + journal=AsyncMock(), + audit=AsyncMock(), + evidence_repo=evidence_repo, + ) + c = Choreographer(deps) + + rich_plan = { + "approach": "plan text with enough detail to pass the 20 char gate", + "sub_tasks": [{"title": "t1", "description": "desc1"}], + } + env = await c.i_will_plan( + pm_agent_id, task_id, plan="plan text", rich_plan=rich_plan + ) + + assert env.error is None, f"Expected ok, got error={env.error} msg={env.message}" + assert env.status == "in_progress" + task_svc.ensure_work_session.assert_awaited_once_with(task_id, pm_agent_id) + + +@pytest.mark.asyncio +async def test_ensure_work_session_not_called_when_start_fails() -> None: + """If start() returns None (task in wrong state), ensure_work_session must + NOT be called — the WorkSession must not be created for a failed transition. + """ + agent_id = uuid4() + task_id = uuid4() + task_svc = _make_task_svc(agent_id, task_id, status="pending") + task_svc.start.return_value = None # start fails + deps = _make_deps(task_svc) + c = Choreographer(deps) + + env = await c.i_will_work_on(agent_id, task_id, plan="do x then y") + + assert env.error == "invalid_state" + task_svc.ensure_work_session.assert_not_awaited()