feat(gateway): restore Gate Set F completion-time guards

cell_pm_complete, main_pm_complete, and submit_up already had subtask-
terminality gates inherited from the pre-gateway closure check at
roboco/services/task.py. This commit verifies the gate is preserved
and improves the remediation hint to actually NAME the non-terminal
subtasks instead of telling the PM to "call triage()".

The improvement uses a new private helper
``Choreographer._non_terminal_subtask_ids`` that queries get_subtasks
and filters to non-terminal statuses, returning a comma-separated
list of "<id> (<status>)" pairs. The PM now sees exactly which
subtasks are blocking the parent's completion.

Pre-gateway reference: roboco/services/task.py closure check
(documented in PRE_GATEWAY_LIFECYCLE.md §4.3).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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