diff --git a/roboco/api/routes/v2/flow_cell_pm.py b/roboco/api/routes/v2/flow_cell_pm.py index 2d0a1aa6..d0748b71 100644 --- a/roboco/api/routes/v2/flow_cell_pm.py +++ b/roboco/api/routes/v2/flow_cell_pm.py @@ -8,13 +8,16 @@ from fastapi import APIRouter, Depends, Header from roboco.api.deps import get_choreographer from roboco.api.schemas.v2.flow import ( CompleteRequest, + DelegateRequest, EscalateUpRequest, GiveMeWorkRequest, IAmIdleRequest, + IWillPlanRequest, + SubmitUpRequest, TriageRequest, UnblockRequest, ) -from roboco.services.gateway.choreographer import Choreographer +from roboco.services.gateway.choreographer import Choreographer, DelegateInputs router = APIRouter(prefix="/api/v2/flow/cell_pm", tags=["v2-flow-cell-pm"]) @@ -29,7 +32,46 @@ async def give_me_work( x_agent_id: _AgentIdHeader, choreographer: _ChoreographerDep, ) -> dict: - env = await choreographer.give_me_work(x_agent_id) + env = await choreographer.pm_give_me_work(x_agent_id) + return env.as_dict() + + +@router.post("/i_will_plan") +async def i_will_plan( + body: IWillPlanRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.i_will_plan(x_agent_id, body.task_id, body.plan) + return env.as_dict() + + +@router.post("/delegate") +async def delegate( + body: DelegateRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + inputs = DelegateInputs( + title=body.title, + description=body.description, + assigned_to=body.assigned_to, + team=body.team, + task_type=body.task_type, + acceptance_criteria=body.acceptance_criteria, + estimated_complexity=body.estimated_complexity, + ) + env = await choreographer.delegate(x_agent_id, body.parent_task_id, inputs) + return env.as_dict() + + +@router.post("/submit_up") +async def submit_up( + body: SubmitUpRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.submit_up(x_agent_id, body.task_id, body.notes) return env.as_dict() diff --git a/roboco/api/routes/v2/flow_main_pm.py b/roboco/api/routes/v2/flow_main_pm.py index 831c0183..25d29b09 100644 --- a/roboco/api/routes/v2/flow_main_pm.py +++ b/roboco/api/routes/v2/flow_main_pm.py @@ -8,13 +8,16 @@ from fastapi import APIRouter, Depends, Header from roboco.api.deps import get_choreographer from roboco.api.schemas.v2.flow import ( CompleteRequest, + DelegateRequest, EscalateToCeoRequest, EscalateUpRequest, + GiveMeWorkRequest, IAmIdleRequest, + IWillPlanRequest, TriageRequest, UnblockRequest, ) -from roboco.services.gateway.choreographer import Choreographer +from roboco.services.gateway.choreographer import Choreographer, DelegateInputs router = APIRouter(prefix="/api/v2/flow/main_pm", tags=["v2-flow-main-pm"]) @@ -23,6 +26,45 @@ _AgentIdHeader = Annotated[UUID, Header(alias="X-Agent-ID")] _ChoreographerDep = Annotated[Choreographer, Depends(get_choreographer)] +@router.post("/give_me_work") +async def give_me_work( + _body: GiveMeWorkRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.pm_give_me_work(x_agent_id) + return env.as_dict() + + +@router.post("/i_will_plan") +async def i_will_plan( + body: IWillPlanRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + env = await choreographer.i_will_plan(x_agent_id, body.task_id, body.plan) + return env.as_dict() + + +@router.post("/delegate") +async def delegate( + body: DelegateRequest, + x_agent_id: _AgentIdHeader, + choreographer: _ChoreographerDep, +) -> dict: + inputs = DelegateInputs( + title=body.title, + description=body.description, + assigned_to=body.assigned_to, + team=body.team, + task_type=body.task_type, + acceptance_criteria=body.acceptance_criteria, + estimated_complexity=body.estimated_complexity, + ) + env = await choreographer.delegate(x_agent_id, body.parent_task_id, inputs) + return env.as_dict() + + @router.post("/triage_all") async def triage_all( _body: TriageRequest, diff --git a/roboco/api/schemas/v2/flow.py b/roboco/api/schemas/v2/flow.py index fa89ebac..e9162d1a 100644 --- a/roboco/api/schemas/v2/flow.py +++ b/roboco/api/schemas/v2/flow.py @@ -78,3 +78,24 @@ class EscalateUpRequest(BaseModel): class EscalateToCeoRequest(BaseModel): task_id: UUID reason: str = Field(..., min_length=1) + + +class IWillPlanRequest(BaseModel): + task_id: UUID + plan: str = Field(..., min_length=1) + + +class DelegateRequest(BaseModel): + parent_task_id: UUID + title: str = Field(..., min_length=1) + description: str = Field(..., min_length=1) + assigned_to: str = Field(..., min_length=1) + team: str = Field(..., min_length=1) + task_type: str = "code" + acceptance_criteria: list[str] | None = None + estimated_complexity: str = "medium" + + +class SubmitUpRequest(BaseModel): + task_id: UUID + notes: str = Field(..., min_length=1) diff --git a/roboco/mcp/flow_server.py b/roboco/mcp/flow_server.py index a8a77eb7..105aaff9 100644 --- a/roboco/mcp/flow_server.py +++ b/roboco/mcp/flow_server.py @@ -161,6 +161,38 @@ def escalate_to_ceo(task_id: str, reason: str) -> dict[str, Any]: return _post(_role_path("escalate_to_ceo"), {"task_id": task_id, "reason": reason}) +# ---------- Cell PM + Main PM extras ---------- +# i_will_plan, delegate, submit_up, give_me_work — restore the pre-Phase-4 +# PM lifecycle so PMs can drive parent tasks instead of stalling. + + +def i_will_plan(task_id: str, plan: str) -> dict[str, Any]: + """PM: claim+start a pending parent task with a one-paragraph plan.""" + return _post(_role_path("i_will_plan"), {"task_id": task_id, "plan": plan}) + + +def delegate( + parent_task_id: str, title: str, description: str, body: dict +) -> dict[str, Any]: + """PM: create a subtask of parent_task_id. + + Required body keys: ``assigned_to``, ``team``. Optional: ``task_type``, + ``acceptance_criteria``, ``estimated_complexity``. + """ + payload: dict[str, Any] = { + "parent_task_id": parent_task_id, + "title": title, + "description": description, + } + payload.update(body) + return _post(_role_path("delegate"), payload) + + +def submit_up(task_id: str, notes: str) -> dict[str, Any]: + """Cell PM: bubble a finished cell-scope task up to the Main PM.""" + return _post(_role_path("submit_up"), {"task_id": task_id, "notes": notes}) + + # ---------- Tool registry ---------- # # Maps the verb name an agent calls (matches manifest entries and the @@ -189,6 +221,9 @@ _TOOLS: dict[str, Any] = { "unblock": unblock, "complete": complete, "escalate_up": escalate_up, + "i_will_plan": i_will_plan, + "delegate": delegate, + "submit_up": submit_up, # board / main pm "escalate_to_ceo": escalate_to_ceo, } diff --git a/tests/unit/api/routes/v2/test_flow_cell_pm.py b/tests/unit/api/routes/v2/test_flow_cell_pm.py index 66e6ee1a..6e6e80d2 100644 --- a/tests/unit/api/routes/v2/test_flow_cell_pm.py +++ b/tests/unit/api/routes/v2/test_flow_cell_pm.py @@ -44,9 +44,13 @@ def _build_app(mock_choreographer: MagicMock) -> FastAPI: @pytest.mark.asyncio async def test_give_me_work_returns_envelope() -> None: - """POST /api/v2/flow/cell_pm/give_me_work returns 200 with envelope shape.""" + """POST /api/v2/flow/cell_pm/give_me_work returns 200 with envelope shape. + + Cell PM's give_me_work routes to ``pm_give_me_work`` so the response + surfaces non-pending PM tasks (paused, awaiting_pm_review) too. + """ mock_chore = MagicMock() - mock_chore.give_me_work = AsyncMock(return_value=_make_envelope(status="idle")) + mock_chore.pm_give_me_work = AsyncMock(return_value=_make_envelope(status="idle")) client = TestClient(_build_app(mock_chore)) resp = client.post( @@ -58,7 +62,7 @@ async def test_give_me_work_returns_envelope() -> None: assert resp.status_code == _HTTP_200 body = resp.json() assert body["status"] == "idle" - mock_chore.give_me_work.assert_awaited_once() + mock_chore.pm_give_me_work.assert_awaited_once() @pytest.mark.asyncio @@ -216,3 +220,83 @@ def test_escalate_up_rejects_empty_reason() -> None: ) assert resp.status_code == _HTTP_422 + + +@pytest.mark.asyncio +async def test_i_will_plan_dispatches_to_choreographer() -> None: + """POST /api/v2/flow/cell_pm/i_will_plan forwards task_id and plan.""" + mock_chore = MagicMock() + mock_chore.i_will_plan = AsyncMock( + return_value=_make_envelope(status="in_progress", task_id=_TASK_ID) + ) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/cell_pm/i_will_plan", + json={"task_id": _TASK_ID, "plan": "break into 3 subtasks for backend"}, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.i_will_plan.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delegate_dispatches_inputs_bundle() -> None: + """POST /api/v2/flow/cell_pm/delegate forwards body via DelegateInputs.""" + mock_chore = MagicMock() + mock_chore.delegate = AsyncMock( + return_value=_make_envelope(status="created", task_id=_TASK_ID) + ) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/cell_pm/delegate", + json={ + "parent_task_id": _TASK_ID, + "title": "Implement /v1/foo", + "description": "Add the foo endpoint with tests.", + "assigned_to": "be-dev-1", + "team": "backend", + }, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.delegate.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_submit_up_dispatches_notes() -> None: + """POST /api/v2/flow/cell_pm/submit_up forwards task_id and notes.""" + mock_chore = MagicMock() + mock_chore.submit_up = AsyncMock( + return_value=_make_envelope(status="awaiting_pm_review", task_id=_TASK_ID) + ) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/cell_pm/submit_up", + json={ + "task_id": _TASK_ID, + "notes": "cell finished all subtasks, ready for main pm review", + }, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.submit_up.assert_awaited_once() + + +def test_submit_up_rejects_empty_notes() -> None: + """POST /api/v2/flow/cell_pm/submit_up rejects empty notes.""" + mock_chore = MagicMock() + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/cell_pm/submit_up", + json={"task_id": _TASK_ID, "notes": ""}, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_422 diff --git a/tests/unit/api/routes/v2/test_flow_main_pm.py b/tests/unit/api/routes/v2/test_flow_main_pm.py index 906c3de4..727904ef 100644 --- a/tests/unit/api/routes/v2/test_flow_main_pm.py +++ b/tests/unit/api/routes/v2/test_flow_main_pm.py @@ -174,3 +174,64 @@ def test_escalate_up_rejects_empty_reason() -> None: ) assert resp.status_code == _HTTP_422 + + +@pytest.mark.asyncio +async def test_give_me_work_routes_to_pm_give_me_work() -> None: + """POST /api/v2/flow/main_pm/give_me_work delegates to pm_give_me_work.""" + mock_chore = MagicMock() + mock_chore.pm_give_me_work = AsyncMock(return_value=_make_envelope(status="idle")) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/main_pm/give_me_work", + json={}, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.pm_give_me_work.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_i_will_plan_dispatches_to_choreographer() -> None: + """POST /api/v2/flow/main_pm/i_will_plan forwards task_id and plan.""" + mock_chore = MagicMock() + mock_chore.i_will_plan = AsyncMock( + return_value=_make_envelope(status="in_progress", task_id=_TASK_ID) + ) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/main_pm/i_will_plan", + json={"task_id": _TASK_ID, "plan": "split into backend, frontend, ux cells"}, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.i_will_plan.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_delegate_to_cell_pm_dispatches_inputs_bundle() -> None: + """POST /api/v2/flow/main_pm/delegate forwards body via DelegateInputs.""" + mock_chore = MagicMock() + mock_chore.delegate = AsyncMock( + return_value=_make_envelope(status="created", task_id=_TASK_ID) + ) + client = TestClient(_build_app(mock_chore)) + + resp = client.post( + "/api/v2/flow/main_pm/delegate", + json={ + "parent_task_id": _TASK_ID, + "title": "Backend slice", + "description": "Plan + drive backend work for feature X.", + "assigned_to": "be-pm", + "team": "backend", + }, + headers=_HEADERS, + ) + + assert resp.status_code == _HTTP_200 + mock_chore.delegate.assert_awaited_once() diff --git a/tests/unit/gateway/test_choreographer_pm_extras.py b/tests/unit/gateway/test_choreographer_pm_extras.py new file mode 100644 index 00000000..53539e6b --- /dev/null +++ b/tests/unit/gateway/test_choreographer_pm_extras.py @@ -0,0 +1,509 @@ +"""Tests for the restored PM lifecycle verbs. + +Covers: i_will_plan, delegate, submit_up, pm_give_me_work, and the +auto-pause behavior of i_am_idle for PMs. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock +from uuid import UUID, uuid4 + +import pytest +from roboco.seeds.initial_data import AGENT_UUIDS +from roboco.services.gateway.choreographer import ( + Choreographer, + ChoreographerDeps, + DelegateInputs, +) + + +def _make_deps(**overrides: Any) -> ChoreographerDeps: + base = { + "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) + + +# --------------------------------------------------------------------------- +# i_will_plan +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_i_will_plan_claims_starts_and_sets_plan() -> None: + pm_id = uuid4() + task_id = uuid4() + pending = MagicMock(id=task_id, status="pending", plan=None, assigned_to=None) + claimed = MagicMock(id=task_id, status="claimed", plan=None, assigned_to=pm_id) + started = MagicMock( + id=task_id, status="in_progress", plan={"text": "x"}, assigned_to=pm_id + ) + task_svc = AsyncMock() + task_svc.get.return_value = pending + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.claim.return_value = claimed + task_svc.set_plan.return_value = claimed + task_svc.start.return_value = started + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="break the work into 3 subtasks") + assert env.error is None + assert env.status == "in_progress" + task_svc.claim.assert_awaited_once_with(pm_id, task_id) + task_svc.set_plan.assert_awaited_once() + task_svc.start.assert_awaited_once_with(pm_id, task_id) + + +@pytest.mark.asyncio +async def test_i_will_plan_rejects_non_pm_role() -> None: + pm_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock(id=task_id, status="pending") + task_svc.agent_for.return_value = MagicMock(role="developer", team="backend") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="x") + body = env.as_dict() + assert body["error"] == "not_authorized" + + +@pytest.mark.asyncio +async def test_i_will_plan_rejects_non_pending_state() -> None: + pm_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock(id=task_id, status="in_progress") + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="x") + body = env.as_dict() + assert body["error"] == "invalid_state" + + +@pytest.mark.asyncio +async def test_i_will_plan_returns_tracing_gap_without_plan() -> None: + pm_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = MagicMock( + id=task_id, status="pending", plan=None, assigned_to=None + ) + task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="") + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert "plan" in body["missing"] + + +@pytest.mark.asyncio +async def test_i_will_plan_task_not_found() -> None: + pm_id = uuid4() + task_id = uuid4() + task_svc = AsyncMock() + task_svc.get.return_value = None + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_will_plan(pm_id, task_id, plan="x") + assert env.as_dict()["error"] == "not_found" + + +# --------------------------------------------------------------------------- +# delegate +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_delegate_main_pm_to_cell_pm_creates_subtask() -> None: + main_pm_id = uuid4() + parent_id = uuid4() + project_id = uuid4() + parent = MagicMock(id=parent_id, project_id=project_id) + new_task = MagicMock(id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm") + task_svc.create_subtask.return_value = new_task + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + main_pm_id, + parent_id, + DelegateInputs( + title="Backend planning", + description="Plan backend work for feature X", + assigned_to="be-pm", + team="backend", + ), + ) + assert env.error is None + assert env.status == "created" + task_svc.create_subtask.assert_awaited_once() + req = task_svc.create_subtask.call_args.args[0] + assert req.parent_task_id == parent_id + assert req.assigned_to == UUID(AGENT_UUIDS["be-pm"]) + + +@pytest.mark.asyncio +async def test_delegate_cell_pm_to_team_dev_creates_subtask() -> None: + cell_pm_id = uuid4() + parent_id = uuid4() + project_id = uuid4() + parent = MagicMock(id=parent_id, project_id=project_id) + new_task = MagicMock(id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.create_subtask.return_value = new_task + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + cell_pm_id, + parent_id, + DelegateInputs( + title="Implement endpoint", + description="Add /v1/foo endpoint with tests", + assigned_to="be-dev-1", + team="backend", + ), + ) + assert env.error is None + assert env.status == "created" + + +@pytest.mark.asyncio +async def test_delegate_main_pm_to_dev_is_rejected() -> None: + main_pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock(id=parent_id, project_id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + main_pm_id, + parent_id, + DelegateInputs( + title="x", description="y", assigned_to="be-dev-1", team="backend" + ), + ) + body = env.as_dict() + assert body["error"] == "not_authorized" + assert "be-pm" in body["remediate"] + + +@pytest.mark.asyncio +async def test_delegate_cell_pm_to_other_pm_rejected() -> None: + cell_pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock(id=parent_id, project_id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + cell_pm_id, + parent_id, + DelegateInputs(title="x", description="y", assigned_to="be-pm", team="backend"), + ) + body = env.as_dict() + assert body["error"] == "not_authorized" + + +@pytest.mark.asyncio +async def test_delegate_unknown_assignee_returns_invalid_state() -> None: + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock(id=parent_id, project_id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + pm_id, + parent_id, + DelegateInputs( + title="x", description="y", assigned_to="nope-pm", team="backend" + ), + ) + body = env.as_dict() + assert body["error"] == "not_authorized" + + +@pytest.mark.asyncio +async def test_delegate_invalid_team_enum_rejected() -> None: + pm_id = uuid4() + parent_id = uuid4() + parent = MagicMock(id=parent_id, project_id=uuid4()) + task_svc = AsyncMock() + task_svc.get.return_value = parent + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.delegate( + pm_id, + parent_id, + DelegateInputs( + title="x", description="y", assigned_to="be-dev-1", team="not-a-team" + ), + ) + assert env.as_dict()["error"] == "invalid_state" + + +# --------------------------------------------------------------------------- +# submit_up +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_submit_up_opens_pr_and_reassigns_to_main_pm() -> None: + pm_id = uuid4() + task_id = uuid4() + main_pm_id = uuid4() + t = MagicMock( + id=task_id, + status="in_progress", + assigned_to=pm_id, + branch_name="feature/backend/abc123", + team="backend", + ) + after = MagicMock( + id=task_id, + status="awaiting_pm_review", + assigned_to=pm_id, + branch_name="feature/backend/abc123", + team="backend", + ) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.all_subtasks_terminal.return_value = True + task_svc.submit_pm_review.return_value = after + task_svc.main_pm_agent.return_value = MagicMock(id=main_pm_id) + git_svc = AsyncMock() + git_svc.create_pr.return_value = {"pr_number": 12, "pr_url": "x"} + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = True + deps = _make_deps(task=task_svc, git=git_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.submit_up( + pm_id, task_id, notes="cell completed all subtasks; ready for main pm" + ) + assert env.error is None + assert env.status == "awaiting_pm_review" + git_svc.create_pr.assert_awaited_once() + task_svc.reassign.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_submit_up_blocks_when_subtasks_not_terminal() -> None: + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + status="in_progress", + assigned_to=pm_id, + branch_name="feature/backend/abc123", + team="backend", + ) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + task_svc.all_subtasks_terminal.return_value = False + 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, task_id, notes="ready for main pm please review") + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert "subtasks" in str(body["missing"]).lower() + + +@pytest.mark.asyncio +async def test_submit_up_rejects_main_pm_role() -> None: + main_pm_id = uuid4() + task_id = uuid4() + t = MagicMock(id=task_id, status="in_progress", assigned_to=main_pm_id) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="main_pm", team="main_pm") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.submit_up(main_pm_id, task_id, notes="enough words to pass min len") + body = env.as_dict() + assert body["error"] == "not_authorized" + + +@pytest.mark.asyncio +async def test_submit_up_blocks_without_journal_decision() -> None: + pm_id = uuid4() + task_id = uuid4() + t = MagicMock( + id=task_id, + status="in_progress", + assigned_to=pm_id, + branch_name="feature/backend/abc", + team="backend", + ) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + journal_svc = AsyncMock() + journal_svc.has_decision_for_task.return_value = False + deps = _make_deps(task=task_svc, journal=journal_svc) + c = Choreographer(deps) + + env = await c.submit_up(pm_id, task_id, notes="enough words to pass min len") + body = env.as_dict() + assert body["error"] == "tracing_gap" + assert "journal:decision" in body["missing"] + + +@pytest.mark.asyncio +async def test_submit_up_short_notes_rejected() -> None: + pm_id = uuid4() + task_id = uuid4() + t = MagicMock(id=task_id, status="in_progress", assigned_to=pm_id) + task_svc = AsyncMock() + task_svc.get.return_value = t + task_svc.agent_for.return_value = MagicMock(role="cell_pm", team="backend") + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.submit_up(pm_id, task_id, notes="short") + body = env.as_dict() + assert body["error"] == "tracing_gap" + + +# --------------------------------------------------------------------------- +# pm_give_me_work +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_pm_give_me_work_returns_first_assigned() -> None: + pm_id = uuid4() + t = MagicMock(id=uuid4(), status="pending", title="x", team="backend") + task_svc = AsyncMock() + task_svc.list_assigned_for_agent.return_value = [t] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.pm_give_me_work(pm_id) + assert env.error is None + assert env.task_id == str(t.id) + assert "i_will_plan" in env.next + + +@pytest.mark.asyncio +async def test_pm_give_me_work_returns_idle_when_empty() -> None: + pm_id = uuid4() + task_svc = AsyncMock() + task_svc.list_assigned_for_agent.return_value = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.pm_give_me_work(pm_id) + assert env.status == "idle" + + +@pytest.mark.asyncio +async def test_pm_give_me_work_paused_hint_mentions_subtasks() -> None: + pm_id = uuid4() + t = MagicMock(id=uuid4(), status="paused", title="x", team="backend") + task_svc = AsyncMock() + task_svc.list_assigned_for_agent.return_value = [t] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.pm_give_me_work(pm_id) + assert "subtasks" in env.next or "complete" in env.next + + +# --------------------------------------------------------------------------- +# i_am_idle auto-pause +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_i_am_idle_auto_pauses_in_progress_tasks() -> None: + agent_id = uuid4() + t = MagicMock(id=uuid4(), status="in_progress") + task_svc = AsyncMock() + task_svc.list_in_progress_for_agent.return_value = [t] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_am_idle(agent_id) + assert env.status == "idle" + task_svc.pause_for_agent.assert_awaited_once_with(agent_id, t.id) + task_svc.mark_agent_idle.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_i_am_idle_no_in_progress_skips_pause() -> None: + agent_id = uuid4() + task_svc = AsyncMock() + task_svc.list_in_progress_for_agent.return_value = [] + deps = _make_deps(task=task_svc) + c = Choreographer(deps) + + env = await c.i_am_idle(agent_id) + assert env.status == "idle" + task_svc.pause_for_agent.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_i_am_idle_with_unread_skips_pause_and_idle() -> None: + agent_id = uuid4() + task_svc = AsyncMock() + deps = _make_deps(task=task_svc) + # Override the default empty list AFTER _make_deps has zeroed it out. + deps.evidence_repo.list_unread_a2a.return_value = ["something"] + c = Choreographer(deps) + + env = await c.i_am_idle(agent_id) + assert env.status == "idle_with_unread" + task_svc.list_in_progress_for_agent.assert_not_awaited() + task_svc.mark_agent_idle.assert_not_awaited()