A2A switchboard (pair cards), Secretary/PM task access + closed over-permission hole, MegaTask conventions fix (#298)

* feat(tasks): Secretary full task access; PM lighter editing — and a closed over-permission hole

Secretary: the CEO-gated edit directive covers the full content surface
(title/description/AC/priority/team/complexity/nature + claim-aware
reassignment through the real reassign paths, enum coercion, slug or
UUID assignees), and read_task returns full detail (notes, plan,
bounded progress, PR refs). The submit_directive tool docs never
mentioned edit at all — fixed, it was undiscoverable.

PMs: scouted the PATCH route and found has_higher_perms gave PM
identities UNRESTRICTED admin (ASSIGN is not team-scoped) — wider than
'not that much'. Now: cell PMs hard-403 outside their team, and both PM
roles are capped to the content allowlist (title/description/AC/
priority) with zero status changes via this surface. CEO/Board/Auditor
keep full admin. Built subagent-driven (Sonnet 5), reviewed.

* feat(a2a): the switchboard — org-chart pair cards with live activity

70 permission-matrix-derived pair cards (cells/pm-chain/board/cross),
lighting on either direction's a2a.message frames with a 45s fade —
A2A only, never verbs, per CEO ruling. Click-through reuses the v1
transcript + chime-in drawer; v1 list stays as the mobile fallback.
One CEO-gated /a2a/chat/admin/pairs route joins the static matrix
against conversations in a single bulk query. Built subagent-driven
(Sonnet 5), reviewed; pre-existing agent-utils slug-map gap flagged.

* fix(runtime): conventions ambient covers the MegaTask project_ids scope

_resolve_intake_ambient forwarded project_ids only to the history-digest
resolver — a MegaTask intake got no architectural-conventions block even
with the flag on. The conventions resolver now takes project_ids first
(mirroring the history resolver), both share one order-preserving
_projects_by_ids helper, and a regression test pins the threading to
both sub-resolvers. Built subagent-driven (Sonnet 5), reviewed.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

tests/integration/test_a2a_routes.py

@@ -19,7 +19,7 @@ from roboco.api.routes.a2a import wellknown_router
 from roboco.db.tables import AgentTable, ProjectTable, TaskTable
 from roboco.enforcement import A2AAccessDeniedError
 from roboco.models import AgentRole, AgentStatus, Team
-from roboco.models.a2a import A2ATask, A2ATaskState, A2ATaskStatus
+from roboco.models.a2a import A2AAdminPairSummary, A2ATask, A2ATaskState, A2ATaskStatus
 from roboco.models.base import (
     TaskNature,
     TaskStatus,
@@ -35,6 +35,8 @@ if TYPE_CHECKING:
 
 _PAGE_TOKEN_OFFSET = 20
 _MIN_STREAM_CHUNKS = 2
+_EXPECTED_PAIR_LIST_TOTAL = 2
+_EXPECTED_PAIR_MESSAGE_COUNT = 4
 
 
 @pytest_asyncio.fixture
@@ -1067,6 +1069,71 @@ async def test_admin_list_conversations_as_ceo(a2a_route_client: dict) -> None:
     assert body["items"][0]["agent_b"] == "fe-dev-1"
 
 
+@pytest.mark.asyncio
+async def test_admin_list_pairs_forbidden_for_non_ceo(a2a_route_client: dict) -> None:
+    client = a2a_route_client["client"]
+    response = await client.get("/api/a2a/chat/admin/pairs", headers=_HDR)
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_admin_list_pairs_as_ceo(a2a_route_client: dict) -> None:
+    """CEO gets the switchboard's pair cards — the static matrix joined with
+    each pair's representative conversation stats."""
+    app = a2a_route_client["app"]
+    dev = a2a_route_client["dev"]
+    client = a2a_route_client["client"]
+    _set_ceo_context(app, dev)
+
+    conv_id = uuid4()
+    pair_with_history = A2AAdminPairSummary(
+        agent_a="be-dev-1",
+        role_a="developer",
+        team_a="backend",
+        agent_b="be-qa",
+        role_b="qa",
+        team_b="backend",
+        group_key="cell-backend",
+        conversation_id=str(conv_id),
+        last_message_at=datetime.now(UTC),
+        message_count=4,
+    )
+    pair_never_talked = A2AAdminPairSummary(
+        agent_a="auditor",
+        role_a="auditor",
+        team_a="board",
+        agent_b="product-owner",
+        role_b="product_owner",
+        team_b="board",
+        group_key="board",
+        conversation_id=None,
+        last_message_at=None,
+        message_count=0,
+    )
+    with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
+        instance = AsyncMock()
+        instance.list_admin_pairs = AsyncMock(
+            return_value=[pair_with_history, pair_never_talked]
+        )
+        mock_service_cls.return_value = instance
+        response = await client.get("/api/a2a/chat/admin/pairs", headers=_HDR)
+
+    assert response.status_code == HTTPStatus.OK
+    instance.list_admin_pairs.assert_awaited_once_with()
+    body = response.json()
+    assert body["total"] == _EXPECTED_PAIR_LIST_TOTAL
+    first = body["items"][0]
+    assert first["agent_a"] == "be-dev-1"
+    assert first["agent_b"] == "be-qa"
+    assert first["group_key"] == "cell-backend"
+    assert first["conversation_id"] == str(conv_id)
+    assert first["message_count"] == _EXPECTED_PAIR_MESSAGE_COUNT
+    second = body["items"][1]
+    assert second["group_key"] == "board"
+    assert second["conversation_id"] is None
+    assert second["message_count"] == 0
+
+
 @pytest.mark.asyncio
 async def test_admin_get_messages_as_ceo_returns_full_transcript(
     a2a_route_client: dict,

tests/integration/test_a2a_service.py

@@ -12,6 +12,7 @@ from uuid import uuid4 as _u
 
 import pytest
 import pytest_asyncio
+from roboco.agents_config import A2A_ALLOWED_PAIRS
 from roboco.db.tables import (
     A2AConversationTable,
     A2AMessageTable,
@@ -637,6 +638,86 @@ async def test_get_conversation_admin_returns_none_for_unknown(
     assert await svc.get_conversation_admin(uuid4()) is None
 
 
+# ---------------------------------------------------------------------------
+# list_admin_pairs — the A2A switchboard's static-matrix + DB join
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_list_admin_pairs_bounded_by_static_matrix(a2a_setup: dict) -> None:
+    """With no conversations at all, every pair from the static matrix is
+    still returned (conversation-less), sized exactly to the matrix."""
+    svc = a2a_setup["svc"]
+    pairs = await svc.list_admin_pairs()
+
+    assert len(pairs) == len(A2A_ALLOWED_PAIRS)
+    assert all(p.conversation_id is None for p in pairs)
+    assert all(p.message_count == 0 for p in pairs)
+    assert all(p.last_message_at is None for p in pairs)
+
+
+@pytest.mark.asyncio
+async def test_list_admin_pairs_joins_representative_conversation(
+    a2a_setup: dict,
+) -> None:
+    svc = a2a_setup["svc"]
+    conv = await svc.get_or_create_conversation("be-dev-1", "be-qa")
+    await svc.send_chat_message(UUID(conv.id), "be-dev-1", "hello")
+
+    pairs = await svc.list_admin_pairs()
+
+    match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"})
+    assert match.conversation_id == conv.id
+    assert match.message_count == 1
+    assert match.last_message_at is not None
+    assert match.group_key == "cell-backend"
+    assert match.role_a == "developer"
+    assert match.role_b == "qa"
+
+
+@pytest.mark.asyncio
+async def test_list_admin_pairs_picks_most_recently_updated_conversation(
+    a2a_setup: dict,
+) -> None:
+    """A pair with two conversations (distinct topics) surfaces the more
+    recently active one as its representative conversation."""
+    svc = a2a_setup["svc"]
+    db = a2a_setup["db"]
+    conv_old = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t1")
+    conv_new = await svc.get_or_create_conversation("be-dev-1", "be-qa", topic="t2")
+
+    now = datetime.now(UTC)
+    row_old = await db.get(A2AConversationTable, UUID(conv_old.id))
+    row_new = await db.get(A2AConversationTable, UUID(conv_new.id))
+    assert row_old is not None
+    assert row_new is not None
+    row_old.updated_at = now - timedelta(minutes=10)
+    row_new.updated_at = now
+    await db.flush()
+
+    pairs = await svc.list_admin_pairs()
+
+    match = next(p for p in pairs if {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"})
+    assert match.conversation_id == conv_new.id
+
+
+@pytest.mark.asyncio
+async def test_list_admin_pairs_excludes_disallowed_pairs(a2a_setup: dict) -> None:
+    """A conversation row between two agents the matrix does NOT allow (dev
+    A2A is same-cell only — this should never legitimately exist, but the
+    join must be robust against it) never surfaces as a pair card: the
+    service iterates the static matrix, not "any conversation row"."""
+    svc = a2a_setup["svc"]
+    db = a2a_setup["db"]
+    stray = A2AConversationTable(agent_a="be-dev-1", agent_b="fe-dev-1")
+    db.add(stray)
+    await db.flush()
+
+    pairs = await svc.list_admin_pairs()
+
+    assert not any({p.agent_a, p.agent_b} == {"be-dev-1", "fe-dev-1"} for p in pairs)
+
+
 # ---------------------------------------------------------------------------
 # CEO reply-only budget — an agent may only reply to the CEO inside a
 # conversation the CEO itself opened, and only up to the CEO's own message

tests/integration/test_tasks_route_pm_lighter_patch.py

@@ -0,0 +1,241 @@
+"""PM-lighter PATCH surface on PATCH /api/tasks/{id}.
+
+The CEO's spec: PMs get a *lighter* content-only slice of the same REST PATCH
+path the Secretary's edit directive uses — title/description/
+acceptance_criteria/priority — scoped to tasks in the PM's remit (cell_pm:
+own team only; main_pm: any team). No status changes, no structural/
+ownership fields, no git fields — those stay on the lifecycle-verb surface.
+
+cell_pm/main_pm already hold TaskAction.ASSIGN (see TASK_PERMISSIONS), which
+is *not* team-scoped in ``can_perform_task_action`` — so absent this gate a
+PM would already ride the CEO/Board/Auditor "full admin" bypass on this
+route (any field, any team). These tests pin the narrower behavior down.
+"""
+
+from __future__ import annotations
+
+from http import HTTPStatus
+from typing import TYPE_CHECKING, Any, cast
+from uuid import UUID, uuid4
+
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from httpx import ASGITransport, AsyncClient
+from roboco.api.deps import get_agent_context, get_db
+from roboco.api.routes.tasks import router as tasks_router
+from roboco.db.tables import AgentTable, ProjectTable, TaskTable
+from roboco.models import AgentRole, AgentStatus, Team
+from roboco.models.base import TaskNature, TaskStatus, TaskType
+from roboco.models.permissions import AgentContext
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator
+
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _make_client(
+    db_session: AsyncSession, *, role: AgentRole, team: Team | None
+) -> dict[str, Any]:
+    pm = AgentTable(
+        id=uuid4(),
+        name="PM",
+        slug=f"pm-{uuid4().hex[:8]}",
+        role=role,
+        team=team,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt="pm",
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db_session.add(pm)
+    await db_session.flush()
+    project = ProjectTable(
+        id=uuid4(),
+        name="PL-Proj",
+        slug=f"pl-proj-{uuid4().hex[:6]}",
+        git_url="https://example.com/pl.git",
+        assigned_cell=Team.BACKEND,
+        created_by=pm.id,
+    )
+    db_session.add(project)
+    await db_session.flush()
+
+    app = FastAPI()
+    app.include_router(tasks_router, prefix="/api/tasks")
+
+    async def _override_db() -> AsyncIterator[AsyncSession]:
+        yield db_session
+
+    async def _override_agent() -> AgentContext:
+        return AgentContext(agent_id=cast("UUID", pm.id), role=role, team=team)
+
+    app.dependency_overrides[get_db] = _override_db
+    app.dependency_overrides[get_agent_context] = _override_agent
+
+    transport = ASGITransport(app=app)
+    client = AsyncClient(transport=transport, base_url="http://test")
+    return {
+        "client": client,
+        "app": app,
+        "agent": pm,
+        "project": project,
+        "db": db_session,
+    }
+
+
+@pytest_asyncio.fixture
+async def cell_pm_client(db_session: AsyncSession) -> AsyncIterator[dict]:
+    """A backend cell PM (ASSIGN, no UPDATE_OWN)."""
+    setup = await _make_client(db_session, role=AgentRole.CELL_PM, team=Team.BACKEND)
+    async with setup["client"]:
+        yield setup
+    setup["app"].dependency_overrides.clear()
+
+
+@pytest_asyncio.fixture
+async def main_pm_client(db_session: AsyncSession) -> AsyncIterator[dict]:
+    """The Main PM (ASSIGN, no team restriction)."""
+    setup = await _make_client(db_session, role=AgentRole.MAIN_PM, team=None)
+    async with setup["client"]:
+        yield setup
+    setup["app"].dependency_overrides.clear()
+
+
+def _seed_task(setup: dict, **kw: Any) -> TaskTable:
+    team = kw.pop("team", Team.BACKEND)
+    task = TaskTable(
+        id=uuid4(),
+        title=kw.pop("title", "t"),
+        description=kw.pop("description", "d"),
+        acceptance_criteria=["ac"],
+        status=kw.pop("status", TaskStatus.IN_PROGRESS),
+        priority=2,
+        task_type=TaskType.CODE,
+        nature=TaskNature.TECHNICAL,
+        project_id=setup["project"].id,
+        created_by=setup["agent"].id,
+        assigned_to=None,
+        team=team,
+    )
+    setup["db"].add(task)
+    return task
+
+
+def _hdr(agent: AgentTable, role: AgentRole) -> dict[str, str]:
+    return {"X-Agent-ID": str(agent.id), "X-Agent-Role": role.value}
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_can_patch_content_field_on_own_team_task(
+    cell_pm_client: dict,
+) -> None:
+    setup = cell_pm_client
+    task = _seed_task(setup, team=Team.BACKEND)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"title": "Sharper title from the cell PM"},
+        headers=_hdr(setup["agent"], AgentRole.CELL_PM),
+    )
+    assert response.status_code == HTTPStatus.OK
+    assert response.json()["title"] == "Sharper title from the cell PM"
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_cannot_patch_task_outside_own_team(
+    cell_pm_client: dict,
+) -> None:
+    """ASSIGN is not team-scoped — without an explicit check a cell PM would
+    ride the same admin bypass CEO/Board get on ANY team's task."""
+    setup = cell_pm_client
+    task = _seed_task(setup, team=Team.FRONTEND)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"title": "Should not land"},
+        headers=_hdr(setup["agent"], AgentRole.CELL_PM),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("dev_notes", "trying to sneak a note in"),
+        ("team", "frontend"),
+        ("assigned_to", str(uuid4())),
+    ],
+)
+@pytest.mark.asyncio
+async def test_cell_pm_cannot_patch_fields_outside_lighter_allowlist(
+    cell_pm_client: dict, field: str, value: object
+) -> None:
+    setup = cell_pm_client
+    task = _seed_task(setup, team=Team.BACKEND)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={field: value},
+        headers=_hdr(setup["agent"], AgentRole.CELL_PM),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_cannot_change_status_via_patch(cell_pm_client: dict) -> None:
+    """No status changes beyond what the lifecycle verbs already grant — the
+    PM-lighter PATCH surface must not become a side-door status override."""
+    setup = cell_pm_client
+    task = _seed_task(setup, team=Team.BACKEND, status=TaskStatus.IN_PROGRESS)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"status": "completed", "force": True},
+        headers=_hdr(setup["agent"], AgentRole.CELL_PM),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_main_pm_can_patch_content_field_on_any_team_task(
+    main_pm_client: dict,
+) -> None:
+    setup = main_pm_client
+    task = _seed_task(setup, team=Team.UX_UI)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"description": "A clarified description of at least twenty chars."},
+        headers=_hdr(setup["agent"], AgentRole.MAIN_PM),
+    )
+    assert response.status_code == HTTPStatus.OK
+
+
+@pytest.mark.asyncio
+async def test_main_pm_cannot_patch_privileged_field(main_pm_client: dict) -> None:
+    setup = main_pm_client
+    task = _seed_task(setup, team=Team.BACKEND)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"parent_task_id": str(uuid4())},
+        headers=_hdr(setup["agent"], AgentRole.MAIN_PM),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_main_pm_cannot_change_status_via_patch(main_pm_client: dict) -> None:
+    setup = main_pm_client
+    task = _seed_task(setup, status=TaskStatus.AWAITING_PM_REVIEW)
+    await setup["db"].flush()
+    response = await setup["client"].patch(
+        f"/api/tasks/{task.id}",
+        json={"status": "completed"},
+        headers=_hdr(setup["agent"], AgentRole.MAIN_PM),
+    )
+    assert response.status_code == HTTPStatus.FORBIDDEN

tests/integration/test_tasks_routes.py

@@ -102,6 +102,7 @@ async def task_client(
             "agent": main_pm,
             "project": project,
             "db": db_session,
+            "app": app,
         }
     app.dependency_overrides.clear()
 
@@ -109,6 +110,24 @@ async def task_client(
 _HDR = {"X-Agent-ID": str(uuid4()), "X-Agent-Role": "main_pm"}
 
 
+def _as_ceo(setup: dict) -> None:
+    """Re-override this client's agent identity to CEO for the rest of the
+    test — the general PATCH admin surface (status overrides, structural
+    fields, force hatches, ...) is CEO/Board/Auditor-only now that
+    cell_pm/main_pm get the narrower "PM lighter" content-only slice (see
+    test_tasks_route_pm_lighter_patch.py). ``task_client``'s default agent
+    stays main_pm so the many other tests that specifically exercise
+    PM-role behavior (or "not CEO" refusals) are unaffected.
+    """
+
+    async def _override_ceo() -> AgentContext:
+        return AgentContext(
+            agent_id=cast("UUID", setup["agent"].id), role=AgentRole.CEO, team=None
+        )
+
+    setup["app"].dependency_overrides[get_agent_context] = _override_ceo
+
+
 def _seed_task(
     setup: dict, *, status: TaskStatus = TaskStatus.PENDING, **kw: Any
 ) -> TaskTable:
@@ -278,6 +297,7 @@ async def test_update_task_status_override_recovers_blocked(task_client: dict) -
     override, so an operator can recover a task wedged in ``blocked`` (which
     ``/complete`` refuses) instead of the status being silently dropped. The
     ``force`` flag acknowledges the bypass past the lifecycle gate (#13)."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.BLOCKED)
     await task_client["db"].flush()
@@ -297,6 +317,7 @@ async def test_update_task_status_override_refused_without_force(
     """#13: pasting over the lifecycle gate into a terminal/final hatch state
     (completed / awaiting_qa / awaiting_pm_review) without ``force`` is refused
     with 400 — the bypass must be an explicit, acknowledged forced override."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS)
     await task_client["db"].flush()
@@ -316,6 +337,7 @@ async def test_update_task_status_override_non_hatch_needs_no_force(
 ) -> None:
     """#13: a non-terminal recovery override (blocked -> pending) does NOT require
     ``force`` — only the terminal/final hatch states do."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.BLOCKED)
     await task_client["db"].flush()
@@ -336,6 +358,7 @@ async def test_update_task_override_gate_states_require_force(
     """The hatch set covers the CEO gate and the terminal cancel too (not just
     completed/awaiting_qa/awaiting_pm_review): a privileged PATCH into either
     without ``force`` is refused 400."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS)
     await task_client["db"].flush()
@@ -353,6 +376,7 @@ async def test_update_task_override_gate_states_require_force(
 async def test_update_task_override_gate_states_with_force_succeeds(
     task_client: dict, hatch: str
 ) -> None:
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.IN_PROGRESS)
     await task_client["db"].flush()
@@ -370,6 +394,7 @@ async def test_update_task_resurrect_terminal_requires_force(task_client: dict)
     """Resurrecting a COMPLETED task back to in_progress is a bypass of the merge
     decision; the target (in_progress) is not itself a hatch state, so the
     target-only gate would miss it — the source-terminal check requires force."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.COMPLETED)
     await task_client["db"].flush()
@@ -416,6 +441,7 @@ async def test_admin_complete_with_open_pr_names_the_pr(task_client: dict) -> No
     """Admin status→completed on a task whose PR is still OPEN strands its
     commits (bit the CEO twice live, 2026-07-02). The refusal must name the
     PR and the stranding — not just the generic lifecycle-gate text."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL)
     await task_client["db"].flush()
@@ -438,6 +464,7 @@ async def test_admin_complete_with_open_pr_force_still_escapes(
 ) -> None:
     """``force`` remains the deliberate, audited escape — an operator who
     KNOWS the PR should be stranded can still complete."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL)
     await task_client["db"].flush()
@@ -457,6 +484,7 @@ async def test_admin_complete_with_merged_pr_gets_generic_gate_only(
 ) -> None:
     """A merged PR strands nothing — the refusal stays the generic hatch
     text (no PR callout), and force completes as before."""
+    _as_ceo(task_client)
     client = task_client["client"]
     task = _seed_task(task_client, status=TaskStatus.AWAITING_CEO_APPROVAL)
     await task_client["db"].flush()
@@ -3163,6 +3191,7 @@ async def test_get_sessions_for_task_not_found(task_client: dict) -> None:
 @pytest.mark.asyncio
 async def test_patch_nature_persists(task_client: dict) -> None:
     """PATCH with nature=non_technical persists; GET returns updated value."""
+    _as_ceo(task_client)
     task = _seed_task(task_client, nature=TaskNature.TECHNICAL)
     await task_client["db"].flush()
     response = await task_client["client"].patch(
@@ -3178,6 +3207,7 @@ async def test_patch_nature_persists(task_client: dict) -> None:
 @pytest.mark.asyncio
 async def test_patch_task_type_persists(task_client: dict) -> None:
     """PATCH with task_type=research persists; GET returns updated value."""
+    _as_ceo(task_client)
     task = _seed_task(task_client, task_type=TaskType.CODE)
     await task_client["db"].flush()
     response = await task_client["client"].patch(
@@ -3193,6 +3223,7 @@ async def test_patch_task_type_persists(task_client: dict) -> None:
 @pytest.mark.asyncio
 async def test_patch_project_id_persists(task_client: dict) -> None:
     """PATCH with project_id=<valid-uuid> persists; GET returns updated value."""
+    _as_ceo(task_client)
     task = _seed_task(task_client)
     # Create a second project to switch to
     second_project = ProjectTable(
@@ -3250,6 +3281,7 @@ async def test_patch_title_only_changes_title(task_client: dict) -> None:
 @pytest.mark.asyncio
 async def test_patch_assigned_to_slug_resolves_to_uuid(task_client: dict) -> None:
     """PATCH assigned_to with agent slug resolves to agent UUID."""
+    _as_ceo(task_client)
     dev = await _seed_agent(task_client)
     task = _seed_task(task_client)
     await task_client["db"].flush()
@@ -3267,6 +3299,7 @@ async def test_patch_assigned_to_slug_resolves_to_uuid(task_client: dict) -> Non
 @pytest.mark.asyncio
 async def test_patch_assigned_to_null_unassigns(task_client: dict) -> None:
     """PATCH assigned_to: null sets assigned_to to null (unassign)."""
+    _as_ceo(task_client)
     dev = await _seed_agent(task_client)
     task = _seed_task(task_client, assigned_to=dev.id)
     await task_client["db"].flush()

tests/unit/runtime/test_intake_spawn.py

@@ -17,6 +17,7 @@ from unittest.mock import patch
 from uuid import UUID, uuid4
 
 import pytest
+from roboco.config import settings
 from roboco.runtime.orchestrator import (
     INTAKE_AGENT_ID,
     AgentInstance,
@@ -265,7 +266,7 @@ class TestIntakeScopeSlugs:
 
 # ---------------------------------------------------------------------------
 # _resolve_history_digest_projects — the prompter-memory ambient's project scope
-# (covers all three intake scopes, unlike the conventions ambient resolver).
+# (covers all three intake scopes: project_slug, product_id, project_ids).
 # ---------------------------------------------------------------------------
 
 
@@ -375,6 +376,284 @@ class TestResolveHistoryDigestAmbient:
         assert result is None
 
 
+# ---------------------------------------------------------------------------
+# _resolve_ambient_projects — the conventions ambient's project scope. Mirrors
+# _resolve_history_digest_projects's shape, including the MegaTask project_ids
+# scope (the pre-existing gap: this resolver used to stop at project_slug /
+# product_id / task_id and never saw a MegaTask's explicit project_ids).
+# ---------------------------------------------------------------------------
+
+
+class TestResolveAmbientProjects:
+    @pytest.mark.asyncio
+    async def test_project_slug_branch_resolves_single_project(self) -> None:
+        orch = _make_minimal_orchestrator()
+
+        class _FakeProjectSvc:
+            async def get_by_slug(self, slug: str) -> Any:
+                return SimpleNamespace(slug=slug, id=uuid4())
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await orch._resolve_ambient_projects(
+                object(),
+                project_slug="roboco",
+                task_id=None,
+                product_id=None,
+            )
+        assert [p.slug for p in projects] == ["roboco"]
+
+    @pytest.mark.asyncio
+    async def test_project_slug_missing_returns_empty(self) -> None:
+        orch = _make_minimal_orchestrator()
+
+        class _FakeProjectSvc:
+            async def get_by_slug(self, _slug: str) -> Any:
+                return None
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await orch._resolve_ambient_projects(
+                object(),
+                project_slug="ghost",
+                task_id=None,
+                product_id=None,
+            )
+        assert projects == []
+
+    @pytest.mark.asyncio
+    async def test_product_id_branch_delegates_to_ambient_product_projects(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        orch = _make_minimal_orchestrator()
+        sentinel = [SimpleNamespace(slug="p1", id=uuid4())]
+
+        async def _fake_product_projects(_db: Any, product_id: str) -> list[Any]:
+            assert product_id == "prod-1"
+            return sentinel
+
+        # Patched on the instance (not the class): _ambient_product_projects is
+        # a staticmethod, so accessing it via `self.` never binds `self` — but a
+        # plain function patched onto the *class* would, since it's no longer
+        # wrapped in `staticmethod`. Patching the instance attribute sidesteps
+        # the descriptor lookup entirely.
+        monkeypatch.setattr(orch, "_ambient_product_projects", _fake_product_projects)
+        projects = await orch._resolve_ambient_projects(
+            object(),
+            project_slug=None,
+            task_id=None,
+            product_id="prod-1",
+        )
+        assert projects is sentinel
+
+    @pytest.mark.asyncio
+    async def test_project_ids_branch_preserves_order_and_skips_missing(
+        self,
+    ) -> None:
+        """The MegaTask scope: an explicit project_ids list must resolve into
+        projects (order preserved, unresolvable ids skipped) — this is the gap
+        fix, mirroring the history digest resolver's own project_ids branch."""
+        orch = _make_minimal_orchestrator()
+        good1 = "11111111-1111-1111-1111-111111111111"
+        missing = "22222222-2222-2222-2222-222222222222"
+        good2 = "33333333-3333-3333-3333-333333333333"
+
+        class _FakeProjectSvc:
+            async def get(self, pid: Any) -> Any:
+                if str(pid) == missing:
+                    return None
+                return SimpleNamespace(slug=f"proj-{str(pid)[0]}", id=pid)
+
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await orch._resolve_ambient_projects(
+                object(),
+                project_slug=None,
+                task_id=None,
+                product_id=None,
+                project_ids=[good1, missing, good2],
+            )
+        assert [p.slug for p in projects] == ["proj-1", "proj-3"]
+
+    @pytest.mark.asyncio
+    async def test_project_ids_takes_priority_over_other_scopes(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """A MegaTask spawn passes project_slug/product_id as None in practice,
+        but the resolver must still prefer the explicit project_ids set over a
+        stray product_id/task_id if both were somehow present."""
+        orch = _make_minimal_orchestrator()
+        pid = "11111111-1111-1111-1111-111111111111"
+
+        class _FakeProjectSvc:
+            async def get(self, _pid: Any) -> Any:
+                return SimpleNamespace(slug="from-ids", id=_pid)
+
+        async def _fail_product_projects(*_a: Any, **_k: Any) -> list[Any]:
+            raise AssertionError("product_id branch must not run")
+
+        monkeypatch.setattr(orch, "_ambient_product_projects", _fail_product_projects)
+        with patch(
+            "roboco.services.project.get_project_service",
+            lambda _db: _FakeProjectSvc(),
+        ):
+            projects = await orch._resolve_ambient_projects(
+                object(),
+                project_slug=None,
+                task_id=None,
+                product_id="prod-should-be-ignored",
+                project_ids=[pid],
+            )
+        assert [p.slug for p in projects] == ["from-ids"]
+
+    @pytest.mark.asyncio
+    async def test_no_scope_given_returns_empty(self) -> None:
+        orch = _make_minimal_orchestrator()
+        projects = await orch._resolve_ambient_projects(
+            object(),
+            project_slug=None,
+            task_id=None,
+            product_id=None,
+        )
+        assert projects == []
+
+
+# ---------------------------------------------------------------------------
+# _resolve_conventions_ambient — flag-gated + best-effort, now MegaTask-aware.
+# ---------------------------------------------------------------------------
+
+
+class TestResolveConventionsAmbient:
+    @pytest.mark.asyncio
+    async def test_flag_off_returns_none(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        orch = _make_minimal_orchestrator()
+        monkeypatch.setattr(settings, "conventions_enabled", False)
+
+        result = await orch._resolve_conventions_ambient(
+            "roboco", project_ids=["11111111-1111-1111-1111-111111111111"]
+        )
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_failure_returns_none_not_raises(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        orch = _make_minimal_orchestrator()
+        monkeypatch.setattr(settings, "conventions_enabled", True)
+
+        def _boom() -> Any:
+            raise RuntimeError("db unavailable")
+
+        monkeypatch.setattr("roboco.db.base.get_session_factory", _boom)
+
+        result = await orch._resolve_conventions_ambient(
+            None, project_ids=["11111111-1111-1111-1111-111111111111"]
+        )
+        assert result is None
+
+    @pytest.mark.asyncio
+    async def test_project_ids_scope_reaches_conventions_layer(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        """The MegaTask project_ids scope must flow all the way through to
+        conventions_ambient_layer — the actual gap this fix closes."""
+        orch = _make_minimal_orchestrator()
+        monkeypatch.setattr(settings, "conventions_enabled", True)
+
+        class _FakeFactory:
+            def __call__(self) -> Any:
+                return self
+
+            async def __aenter__(self) -> Any:
+                return "fake-db"
+
+            async def __aexit__(self, *_a: Any) -> None:
+                return None
+
+        monkeypatch.setattr("roboco.db.base.get_session_factory", _FakeFactory)
+
+        sentinel_projects = [SimpleNamespace(slug="proj-1")]
+        captured: dict[str, Any] = {}
+
+        async def _fake_resolve_projects(_self: Any, _db: Any, **kwargs: Any) -> Any:
+            captured["kwargs"] = kwargs
+            return sentinel_projects
+
+        async def _fake_layer(_db: Any, projects: Any) -> str:
+            captured["projects"] = projects
+            return "RENDERED BLOCK"
+
+        monkeypatch.setattr(
+            AgentOrchestrator, "_resolve_ambient_projects", _fake_resolve_projects
+        )
+        monkeypatch.setattr(
+            "roboco.agents.factories._base.conventions_ambient_layer", _fake_layer
+        )
+
+        result = await orch._resolve_conventions_ambient(
+            None, project_ids=["11111111-1111-1111-1111-111111111111"]
+        )
+        assert result == "RENDERED BLOCK"
+        assert captured["kwargs"]["project_ids"] == [
+            "11111111-1111-1111-1111-111111111111"
+        ]
+        assert captured["projects"] is sentinel_projects
+
+
+# ---------------------------------------------------------------------------
+# _resolve_intake_ambient — must forward project_ids to BOTH sub-resolvers.
+# Regression test for the gap: it used to thread project_ids only to the
+# history-digest resolver, leaving a MegaTask intake with no conventions block.
+# ---------------------------------------------------------------------------
+
+
+class TestResolveIntakeAmbientThreadsProjectIds:
+    @pytest.mark.asyncio
+    async def test_project_ids_forwarded_to_conventions_and_history(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        orch = _make_minimal_orchestrator()
+        conventions_calls: list[dict[str, Any]] = []
+        history_calls: list[dict[str, Any]] = []
+
+        async def _conventions(_slug: Any, **kwargs: Any) -> str | None:
+            conventions_calls.append(kwargs)
+            return "CONVENTIONS"
+
+        async def _history(_slug: Any, **kwargs: Any) -> str | None:
+            history_calls.append(kwargs)
+            return "HISTORY"
+
+        monkeypatch.setattr(orch, "_resolve_conventions_ambient", _conventions)
+        monkeypatch.setattr(orch, "_resolve_history_digest_ambient", _history)
+
+        result = await orch._resolve_intake_ambient(
+            None,
+            product_id=None,
+            project_ids=["11111111-1111-1111-1111-111111111111"],
+        )
+
+        assert result == "CONVENTIONS\n\n---\n\nHISTORY"
+        assert conventions_calls == [
+            {
+                "product_id": None,
+                "project_ids": ["11111111-1111-1111-1111-111111111111"],
+            }
+        ]
+        assert history_calls == [
+            {
+                "product_id": None,
+                "project_ids": ["11111111-1111-1111-1111-111111111111"],
+            }
+        ]
+
+
 # ---------------------------------------------------------------------------
 # spawn_intake_session / reap_intake_session — orchestration (docker mocked).
 # ---------------------------------------------------------------------------

tests/unit/services/test_secretary_service.py

@@ -2,12 +2,14 @@
 
 from __future__ import annotations
 
+from types import SimpleNamespace
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 from uuid import uuid4
 
 import pytest
 from roboco.db.tables import SecretaryDirectiveTable
+from roboco.models.base import Complexity, TaskNature, TaskStatus, Team
 from roboco.models.secretary import DirectiveKind, DirectiveStatus
 from roboco.services import secretary as sec_module
 from roboco.services.base import ValidationError
@@ -35,7 +37,12 @@ def _patch(monkeypatch: pytest.MonkeyPatch) -> dict[str, MagicMock]:
     task.approve_and_start = AsyncMock()
     task.admin_set_status = AsyncMock()
     task.update = AsyncMock()
+    task.get = AsyncMock(return_value=None)
+    task.reassign = AsyncMock()
+    task.reassign_active_claim = AsyncMock()
     monkeypatch.setattr(sec_module, "get_task_service", lambda _s: task)
+    agent_lookup = AsyncMock(return_value=None)
+    monkeypatch.setattr(sec_module, "get_agent_by_slug", agent_lookup)
     notifier = MagicMock()
     notifier.send_ack_notification = AsyncMock()
     monkeypatch.setattr(
@@ -47,6 +54,7 @@ def _patch(monkeypatch: pytest.MonkeyPatch) -> dict[str, MagicMock]:
         "pitch": pitch,
         "task": task,
         "notifier": notifier,
+        "agent_lookup": agent_lookup,
     }
 
 
@@ -224,3 +232,241 @@ async def test_control_task_edit_rejects_non_allowlisted_fields(
     out = await svc.confirm_directive(row.id, uuid4())
     assert out.status == DirectiveStatus.FAILED.value
     svcs["task"].update.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_extended_content_fields(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Secretary FULL: team/estimated_complexity/nature ride the edit too,
+    coerced into their proper enums before hitting TaskService.update."""
+    svcs = _patch(monkeypatch)
+    svc = SecretaryService(_session())
+    tid = uuid4()
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(tid),
+            "action": "edit",
+            "fields": {
+                "team": "frontend",
+                "estimated_complexity": "high",
+                "nature": "technical",
+            },
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.EXECUTED.value
+    svcs["task"].update.assert_awaited_once()
+    _, kwargs = svcs["task"].update.await_args
+    assert kwargs["team"] == Team.FRONTEND
+    assert kwargs["estimated_complexity"] == Complexity.HIGH
+    assert kwargs["nature"] == TaskNature.TECHNICAL
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_bad_enum_value_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    svcs = _patch(monkeypatch)
+    svc = SecretaryService(_session())
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(uuid4()),
+            "action": "edit",
+            "fields": {"team": "not-a-real-team"},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.FAILED.value
+    svcs["task"].update.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_reassigns_active_claim(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Reassigning an active (claimed/in_progress) task goes through
+    ``reassign_active_claim`` so the new assignee reseeds its heartbeat and
+    isn't immediately stale to the reaper — not a naive setattr."""
+    svcs = _patch(monkeypatch)
+    new_assignee = uuid4()
+    svcs["task"].get = AsyncMock(
+        return_value=SimpleNamespace(status=TaskStatus.IN_PROGRESS)
+    )
+    svcs["task"].reassign_active_claim = AsyncMock(
+        return_value=SimpleNamespace(status=TaskStatus.IN_PROGRESS)
+    )
+    svc = SecretaryService(_session())
+    tid = uuid4()
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(tid),
+            "action": "edit",
+            "fields": {"assigned_to": str(new_assignee)},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.EXECUTED.value
+    svcs["task"].reassign_active_claim.assert_awaited_once_with(tid, new_assignee)
+    svcs["task"].reassign.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_reassigns_non_active_task(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A non-active task (e.g. pending) reassigns through the general
+    ``reassign`` path — no heartbeat to reseed."""
+    svcs = _patch(monkeypatch)
+    new_assignee = uuid4()
+    svcs["task"].get = AsyncMock(
+        return_value=SimpleNamespace(status=TaskStatus.PENDING)
+    )
+    svc = SecretaryService(_session())
+    tid = uuid4()
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(tid),
+            "action": "edit",
+            "fields": {"assigned_to": str(new_assignee)},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.EXECUTED.value
+    svcs["task"].reassign.assert_awaited_once_with(tid, new_assignee)
+    svcs["task"].reassign_active_claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_reassigns_by_slug(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The CEO refers to agents by slug (e.g. 'be-dev-1'); the edit resolves
+    it to a UUID the same way the REST PATCH path does."""
+    svcs = _patch(monkeypatch)
+    agent_row_id = uuid4()
+    svcs["agent_lookup"].return_value = SimpleNamespace(id=agent_row_id)
+    svcs["task"].get = AsyncMock(
+        return_value=SimpleNamespace(status=TaskStatus.PENDING)
+    )
+    svc = SecretaryService(_session())
+    tid = uuid4()
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(tid),
+            "action": "edit",
+            "fields": {"assigned_to": "be-dev-1"},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.EXECUTED.value
+    svcs["task"].reassign.assert_awaited_once_with(tid, agent_row_id)
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_unknown_assignee_slug_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    svcs = _patch(monkeypatch)
+    svc = SecretaryService(_session())
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(uuid4()),
+            "action": "edit",
+            "fields": {"assigned_to": "no-such-agent"},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.FAILED.value
+    svcs["task"].reassign.assert_not_awaited()
+    svcs["task"].reassign_active_claim.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_confirm_control_task_edit_combines_fields_and_reassign(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A single edit directive may both update content fields and reassign."""
+    svcs = _patch(monkeypatch)
+    new_assignee = uuid4()
+    svcs["task"].get = AsyncMock(
+        return_value=SimpleNamespace(status=TaskStatus.PENDING)
+    )
+    svc = SecretaryService(_session())
+    tid = uuid4()
+    row = _pending(
+        DirectiveKind.CONTROL_TASK,
+        {
+            "task_id": str(tid),
+            "action": "edit",
+            "fields": {"title": "Renamed", "assigned_to": str(new_assignee)},
+        },
+    )
+    monkeypatch.setattr(svc, "get_directive", AsyncMock(return_value=row))
+    out = await svc.confirm_directive(row.id, uuid4())
+    assert out.status == DirectiveStatus.EXECUTED.value
+    svcs["task"].update.assert_awaited_once()
+    _, kwargs = svcs["task"].update.await_args
+    assert kwargs == {"title": "Renamed"}
+    svcs["task"].reassign.assert_awaited_once_with(tid, new_assignee)
+
+
+_SEEDED_PR_NUMBER = 42
+_SEEDED_PROGRESS_UPDATE_COUNT = sec_module._MAX_PROGRESS_UPDATES + 10
+
+
+@pytest.mark.asyncio
+async def test_read_task_includes_full_detail(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Secretary FULL read breadth: notes/progress/plan/pr fields join the
+    brief identity fields that read_task already carried."""
+    svcs = _patch(monkeypatch)
+    tid = uuid4()
+    fake_task = SimpleNamespace(
+        id=tid,
+        title="Ship the thing",
+        status=TaskStatus.IN_PROGRESS,
+        team="backend",
+        assigned_to=uuid4(),
+        description="A real description.",
+        acceptance_criteria=["works"],
+        priority=1,
+        estimated_complexity="high",
+        nature="technical",
+        plan={"approach": "do it"},
+        progress_updates=[
+            {"message": f"update {i}"} for i in range(_SEEDED_PROGRESS_UPDATE_COUNT)
+        ],
+        dev_notes="dev note",
+        qa_notes="qa note",
+        auditor_notes="audit note",
+        pr_reviewer_notes="pr note",
+        doc_notes="doc note",
+        quick_context="ctx",
+        branch_name="feature/x",
+        pr_number=_SEEDED_PR_NUMBER,
+        pr_url="https://example.com/pr/42",
+    )
+    svcs["task"].get = AsyncMock(return_value=fake_task)
+    svc = SecretaryService(_session())
+    out = await svc.read_task(tid)
+    assert out["title"] == "Ship the thing"
+    assert out["plan"] == {"approach": "do it"}
+    assert out["dev_notes"] == "dev note"
+    assert out["pr_number"] == _SEEDED_PR_NUMBER
+    assert out["branch_name"] == "feature/x"
+    # Bounded: only the most recent entries survive.
+    assert len(out["progress_updates"]) == sec_module._MAX_PROGRESS_UPDATES
+    last_index = _SEEDED_PROGRESS_UPDATE_COUNT - 1
+    assert out["progress_updates"][-1]["message"] == f"update {last_index}"

tests/unit/test_agents_config.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+from collections import Counter
 from typing import TYPE_CHECKING
 
 from roboco.agents_config import (
+    A2A_ALLOWED_PAIRS,
     can_a2a_direct,
     can_assign_tasks,
     can_cancel_tasks,
@@ -27,6 +29,7 @@ from roboco.agents_config import (
     issue_panel_token,
     verify_agent_token,
 )
+from roboco.foundation import identity as foundation
 from roboco.seeds.initial_data import CEO_AGENT_ID
 
 if TYPE_CHECKING:
@@ -441,3 +444,85 @@ def test_get_a2a_route_hint_unknown_from_agent_falls_through() -> None:
     """from_agent with no team falls through to escalate fallback (line 774)."""
     hint = get_a2a_route_hint("ghost", "be-dev-1")
     assert "escalate" in hint.lower()
+
+
+# ---------------------------------------------------------------------------
+# A2A_ALLOWED_PAIRS — the switchboard's static org-chart pair matrix
+# ---------------------------------------------------------------------------
+
+_EXPECTED_PAIR_COUNT = 70
+_EXPECTED_GROUP_COUNTS = {
+    "board": 3,
+    "cell-backend": 15,
+    "cell-frontend": 15,
+    "cell-ux_ui": 15,
+    "cross": 16,
+    "pm-chain": 6,
+}
+
+
+def test_a2a_allowed_pairs_total_count() -> None:
+    assert len(A2A_ALLOWED_PAIRS) == _EXPECTED_PAIR_COUNT
+
+
+def test_a2a_allowed_pairs_canonical_lexical_order() -> None:
+    """agent_a < agent_b always — matches A2AConversationTable's canonical
+    ordering, so the service's DB join keys line up."""
+    for p in A2A_ALLOWED_PAIRS:
+        assert p.agent_a < p.agent_b
+
+
+def test_a2a_allowed_pairs_no_duplicates() -> None:
+    keys = [(p.agent_a, p.agent_b) for p in A2A_ALLOWED_PAIRS]
+    assert len(keys) == len(set(keys))
+
+
+def test_a2a_allowed_pairs_excludes_human_only_and_sentinel_roles() -> None:
+    """CEO, the intake interviewer, the secretary, and the system sentinel
+    are not real A2A participants in the org chart."""
+    slugs = {p.agent_a for p in A2A_ALLOWED_PAIRS} | {
+        p.agent_b for p in A2A_ALLOWED_PAIRS
+    }
+    for excluded in ("ceo", "intake-1", "secretary-1", "system"):
+        assert excluded not in slugs
+
+
+def test_a2a_allowed_pairs_group_key_counts() -> None:
+    counts = Counter(p.group_key for p in A2A_ALLOWED_PAIRS)
+    assert dict(counts) == _EXPECTED_GROUP_COUNTS
+
+
+def test_a2a_allowed_pairs_contains_same_cell_pair() -> None:
+    assert any(
+        {p.agent_a, p.agent_b} == {"be-dev-1", "be-qa"} for p in A2A_ALLOWED_PAIRS
+    )
+
+
+def test_a2a_allowed_pairs_contains_pm_chain_pair() -> None:
+    assert any(
+        {p.agent_a, p.agent_b} == {"be-pm", "main-pm"} for p in A2A_ALLOWED_PAIRS
+    )
+
+
+def test_a2a_allowed_pairs_contains_board_pair() -> None:
+    assert any(
+        {p.agent_a, p.agent_b} == {"auditor", "product-owner"}
+        for p in A2A_ALLOWED_PAIRS
+    )
+
+
+def test_a2a_allowed_pairs_reflects_can_a2a_direct_matrix() -> None:
+    """Every listed pair allows >=1 direction per the live matrix — catches
+    drift if can_a2a_direct changes without regenerating the static list."""
+    for p in A2A_ALLOWED_PAIRS:
+        allowed_ab, _ = can_a2a_direct(p.agent_a, p.agent_b)
+        allowed_ba, _ = can_a2a_direct(p.agent_b, p.agent_a)
+        assert allowed_ab or allowed_ba
+
+
+def test_a2a_allowed_pairs_role_team_fields_match_registry() -> None:
+    for p in A2A_ALLOWED_PAIRS:
+        assert p.role_a == foundation.AGENTS[p.agent_a].role.value
+        assert p.team_a == foundation.AGENTS[p.agent_a].team.value
+        assert p.role_b == foundation.AGENTS[p.agent_b].role.value
+        assert p.team_b == foundation.AGENTS[p.agent_b].team.value