diff --git a/roboco/services/gateway/choreographer.py b/roboco/services/gateway/choreographer.py index 7978e9dc..8a0eedbd 100644 --- a/roboco/services/gateway/choreographer.py +++ b/roboco/services/gateway/choreographer.py @@ -227,6 +227,20 @@ class Choreographer: siblings: list[Any] = await self.task.get_subtasks(parent_id) return siblings + async def _non_terminal_subtask_ids(self, parent_task_id: UUID) -> str: + """Return a human-readable comma-separated list of non-terminal subtasks. + + Used by Gate Set F closure-time guards to name exactly which + subtasks are blocking parent completion. + """ + terminal = {"completed", "cancelled"} + subtasks: list[Any] = await self.task.get_subtasks(parent_task_id) + non_terminal = [s for s in subtasks if str(s.status) not in terminal] + if not non_terminal: + return "(none — query out-of-sync, retry)" + # Format: " ()" + return ", ".join(f"{s.id} ({s.status})" for s in non_terminal) + async def i_will_work_on( self, agent_id: UUID, task_id: UUID, plan: str | None = None ) -> Envelope: @@ -1383,11 +1397,13 @@ class Choreographer: context_briefing=await self._briefing_for(pm_agent_id, task_id), ) if not await self.task.all_subtasks_terminal(task_id): + non_terminal = await self._non_terminal_subtask_ids(task_id) return Envelope.tracing_gap( missing=["subtasks not all terminal"], remediate=( "all subtasks must be in completed/cancelled before" - " bubbling up. Call triage() to find pending subtasks." + " bubbling up. Non-terminal subtasks: " + + non_terminal ), context_briefing=await self._briefing_for(pm_agent_id, task_id), ) @@ -1583,11 +1599,13 @@ class Choreographer: ) all_terminal = await self.task.all_subtasks_terminal(task_id) if not all_terminal: + non_terminal = await self._non_terminal_subtask_ids(task_id) return Envelope.tracing_gap( missing=["subtasks not all terminal"], remediate=( "all subtasks must be in completed/cancelled before" - " completing parent. Call triage() to find pending subtasks." + " completing parent. Non-terminal subtasks: " + + non_terminal ), context_briefing=await self._briefing_for(pm_agent_id, task_id), ) @@ -1712,9 +1730,13 @@ class Choreographer: ) all_terminal = await self.task.all_subtasks_terminal(root_task_id) if not all_terminal: + non_terminal = await self._non_terminal_subtask_ids(root_task_id) return Envelope.tracing_gap( missing=["subtasks not all terminal"], - remediate="all subtasks must be in completed/cancelled state", + remediate=( + "all subtasks must be in completed/cancelled state. " + "Non-terminal subtasks: " + non_terminal + ), context_briefing=await self._briefing_for( main_pm_agent_id, root_task_id ), diff --git a/tests/unit/gateway/test_choreographer_completion_guards.py b/tests/unit/gateway/test_choreographer_completion_guards.py new file mode 100644 index 00000000..cc2796c5 --- /dev/null +++ b/tests/unit/gateway/test_choreographer_completion_guards.py @@ -0,0 +1,187 @@ +"""Gate Set F: completion-time guards. + +cell_pm_complete / main_pm_complete / submit_up must refuse to advance +the parent past awaiting_pm_review when any subtask is still non- +terminal. Pre-gateway location: roboco/services/task.py closure check. + +These tests verify: +1. The non-terminal-subtask refusal fires. +2. The remediation NAMES the non-terminal subtasks (improvement over + the previous generic "find pending subtasks" hint). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base: dict[str, Any] = { + "task": AsyncMock(), + "work_session": AsyncMock(), + "git": AsyncMock(), + "a2a": AsyncMock(), + "journal": AsyncMock(), + "audit": AsyncMock(), + "evidence_repo": AsyncMock(), + } + base.update(overrides) + repo = base["evidence_repo"] + 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(repo, method).return_value = [] + return ChoreographerDeps(**base) + + +# --------------------------------------------------------------------------- +# cell_pm_complete subtask gate +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_cell_pm_complete_blocks_when_subtask_pending() -> None: + pm_id = uuid4() + parent_id = uuid4() + sub_id = uuid4() + t = MagicMock( + id=parent_id, + status="awaiting_pm_review", + assigned_to=pm_id, + pr_number=10, + team="backend", + branch_name="feature/backend/abc", + ) + sub = MagicMock(id=sub_id, status="pending", title="Half-done subtask") + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.all_subtasks_terminal.return_value = False + task_svc.get_subtasks.return_value = [sub] + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.cell_pm_complete(pm_id, parent_id, "done") + body = env.as_dict() + assert body["error"] == "tracing_gap" + # Improvement: non-terminal subtask must be named. + assert str(sub_id) in body["remediate"] + task_svc.cell_pm_complete.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cell_pm_complete_allows_when_all_terminal() -> None: + pm_id = uuid4() + parent_id = uuid4() + t = MagicMock( + id=parent_id, + status="awaiting_pm_review", + assigned_to=pm_id, + pr_number=10, + team="backend", + branch_name="feature/backend/abc", + parent_task_id=None, + ) + after = MagicMock(**{**t.__dict__, "status": "completed"}) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.all_subtasks_terminal.return_value = True + task_svc.get_subtasks.return_value = [] + task_svc.cell_pm_complete.return_value = after + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + git_svc = AsyncMock() + git_svc.pr_merge.return_value = {"merge_commit_sha": "abc"} + deps = _make_deps(task=task_svc, journal=journal_svc, git=git_svc) + c = Choreographer(deps) + + env = await c.cell_pm_complete(pm_id, parent_id, "done") + assert env.error is None + task_svc.cell_pm_complete.assert_awaited_once() + + +# --------------------------------------------------------------------------- +# main_pm_complete subtask gate (root-task case) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_main_pm_complete_blocks_when_subtask_pending() -> None: + pm_id = uuid4() + root_id = uuid4() + sub_id = uuid4() + t = MagicMock( + id=root_id, + status="awaiting_pm_review", + assigned_to=pm_id, + parent_task_id=None, + pr_number=10, + team="backend", + branch_name="feature/backend/abc", + ) + sub = MagicMock(id=sub_id, status="in_progress", title="Subtask still active") + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.all_subtasks_terminal.return_value = False + task_svc.get_subtasks.return_value = [sub] + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.main_pm_complete(pm_id, root_id, "ship it") + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert str(sub_id) in body["remediate"] + task_svc.escalate_to_ceo.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# submit_up subtask gate (cell PM bubbling up to main PM) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_submit_up_blocks_when_subtask_pending() -> None: + pm_id = uuid4() + parent_id = uuid4() + sub_id = uuid4() + t = MagicMock( + id=parent_id, + status="in_progress", + assigned_to=pm_id, + branch_name="feature/backend/abc", + team="backend", + ) + sub = MagicMock(id=sub_id, status="paused", title="Paused subtask") + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="cell_pm") + task_svc.all_subtasks_terminal.return_value = False + task_svc.get_subtasks.return_value = [sub] + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.submit_up( + pm_id, + parent_id, + "ready for main PM review and merge into master branch", + ) + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert str(sub_id) in body["remediate"] + task_svc.submit_pm_review.assert_not_awaited()