feat(gateway): wire PM lifecycle verbs through API + MCP + tests

* api/schemas/v2/flow.py: IWillPlanRequest, DelegateRequest,
  SubmitUpRequest with min_length=1 validators where appropriate.

* api/routes/v2/flow_cell_pm.py: give_me_work routes to
  pm_give_me_work; new endpoints i_will_plan, delegate, submit_up.
* api/routes/v2/flow_main_pm.py: new endpoints give_me_work,
  i_will_plan, delegate.

* mcp/flow_server.py: Python wrappers for i_will_plan, delegate,
  submit_up registered in _TOOLS so manifest-scoped agents can call
  them.

* tests/unit/gateway/test_choreographer_pm_extras.py: 22 tests
  covering happy + reject paths for each new verb plus i_am_idle's
  auto-pause behavior.
* tests/unit/api/routes/v2/test_flow_cell_pm.py +
  test_flow_main_pm.py: route-level tests for the new endpoints.

Test count: 352 → 381 (+29). make quality-fast green.

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()
 
 

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,

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)

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,
 }

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

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()

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()