[chore] a2a-routes: authenticate send_message responder + gate cancel task (PM-only) (#116 #423)

send_message took the responder identity from a client-supplied
metadata.from_agent, so any caller could spoof anyone (e.g.
from_agent='ceo') in the task's notes and in the spawn/notification
routed back to the original requester. Stamp the authenticated caller's
slug as the responder instead (CurrentAgentContext).

cancel_task was ungated: no auth dependency and no role check, so any
agent (or any caller) could cancel a task the lifecycle rule reserves to
PM roles (Any -> cancelled: PM roles only) — and the cascade-cancel of
all non-terminal descendants ran with a hardcoded cell_pm role and no
recorded actor. Add require_any_authenticated_agent + a PM-or-above gate,
and thread the authenticated role (into the cascade role gate) and slug
(into the cancellation note) into A2AService.cancel_task.

Tests: send_message ignores a spoofed from_agent and records the
authenticated slug; cancel rejects a developer (403) and a missing auth
header; a PM cancel threads role + slug into the service; the pre-existing
cancel success/already-terminal/not-found tests now run under a PM context
(the success test's body was missing the A2A 'name' field and false-passed
on a 422 — now genuine).

roboco/api/routes/a2a.py

@@ -23,7 +23,12 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from fastapi.responses import JSONResponse
 from sse_starlette import EventSourceResponse
 
-from roboco.api.deps import CurrentAgentSlug, DbSession
+from roboco.api.deps import (
+    CurrentAgentContext,
+    CurrentAgentSlug,
+    DbSession,
+    require_pm_or_above,
+)
 from roboco.api.routes.v1._role_dep import require_any_authenticated_agent
 from roboco.api.schemas.a2a_chat import (
     ConversationCloseRequest,
@@ -116,6 +121,7 @@ async def get_agent_card(
 async def send_message(
     request: SendMessageRequest,
     db: DbSession,
+    agent: CurrentAgentContext,
 ) -> dict[str, Any]:
     """
     Send an A2A message (fallback endpoint).
@@ -149,10 +155,15 @@ async def send_message(
     is_response = metadata.get("is_response", False)
 
     if is_response:
-        # Update existing task with response message
+        # The responder is the AUTHENTICATED caller — never a client-supplied
-        responder = metadata.get("from_agent")
+        # metadata.from_agent, which any caller could spoof to impersonate
+        # anyone (e.g. from_agent='ceo') in the task's notes and in the
+        # spawn/notification routed back to the original requester (#116).
+        responder = agent.slug
         try:
-            await service.update_task_from_message(task_id_str, message, responder)
+            await service.update_task_from_message(
+                task_id_str, message, responder_agent=responder
+            )
         except ValueError as e:
             error_msg = str(e)
             if "Invalid task ID" in error_msg:
@@ -420,23 +431,32 @@ async def list_tasks(
     )
 
 
-@router.post("/tasks/{task_id}/cancel")
+@router.post(
+    "/tasks/{task_id}/cancel",
+    dependencies=[require_any_authenticated_agent],
+)
 async def cancel_task(
     task_id: str,
     db: DbSession,
+    agent: CurrentAgentContext,
     request: CancelTaskRequest | None = None,
 ) -> A2ATask:
     """
     Cancel an A2A task.
 
-    Transitions the task to cancelled state.
+    Transitions the task to cancelled state. PM/management-only — the service
+    cascades the cancel to all non-terminal descendants, and the lifecycle rule
+    (Any -> cancelled: PM roles only) must hold on this path too (#423).
     """
+    require_pm_or_above(agent.role, action="cancel a task via A2A")
     service = A2AService(db)
 
     try:
         task = await service.cancel_task(
             task_id=task_id,
             reason=request.reason if request else None,
+            agent_role=agent.role.value,
+            actor_slug=agent.slug,
         )
     except ValueError as e:
         error_msg = str(e)

roboco/services/a2a.py

@@ -379,13 +379,26 @@ class A2AService:
 
         return [self.task_to_a2a(t) for t in tasks], has_more
 
-    async def cancel_task(self, task_id: str, reason: str | None = None) -> A2ATask:
+    async def cancel_task(
+        self,
+        task_id: str,
+        reason: str | None = None,
+        agent_role: str | None = None,
+        actor_slug: str | None = None,
+    ) -> A2ATask:
         """
         Cancel a task and all non-terminal descendants.
 
         Args:
             task_id: Task UUID string
             reason: Optional cancellation reason
+            agent_role: The authenticated caller's role, threaded into the
+                cascade role gate (TaskService.cancel). Defaults to cell_pm
+                when unset for back-compat with non-route callers.
+            actor_slug: The authenticated caller's slug, recorded in the
+                cancellation note so the audit trail attributes the cancel to
+                the real actor (the route enforces PM/management; non-route
+                callers may omit it).
 
         Returns:
             Updated A2ATask
@@ -419,23 +432,38 @@ class A2AService:
         if status_value in ["completed", "cancelled"]:
             raise ValueError(f"Task already in terminal state: {status_value}")
 
-        # Add reason to notes before cancel
+        # Build a cancellation note that attributes the real actor (the route
+        # now passes the authenticated slug) so the audit trail records who
+        # cancelled and why — keeps this out of route handlers.
+        note_parts: list[str] = []
+        if actor_slug:
+            note_parts.append(f"Cancelled via A2A by {actor_slug}")
         if reason:
-            reason_text = f"Cancellation reason: {reason}"
+            note_parts.append(f"reason: {reason}")
+        cancellation_note = "; ".join(note_parts) if note_parts else None
+        if cancellation_note:
             if task.dev_notes:
-                task.dev_notes = f"{task.dev_notes}\n\n{reason_text}"
+                task.dev_notes = f"{task.dev_notes}\n\n{cancellation_note}"
             else:
-                task.dev_notes = reason_text
+                task.dev_notes = cancellation_note
             await self.session.flush()
 
-        # Use TaskService for consistent cancel behavior (cascades to descendants)
+        # Use TaskService for consistent cancel behavior (cascades to descendants).
+        # Thread the caller's role into the cascade role gate so a non-PM
+        # caller can't cascade-cancel descendants the role can't cancel.
         task_service = TaskService(self.session)
-        task = await task_service.cancel(task_uuid)
+        task = await task_service.cancel(task_uuid, agent_role=agent_role or "cell_pm")
 
         if task is None:
             raise ValueError(f"Failed to cancel task: {task_id}")
 
-        logger.info("Cancelled task via A2A", task_id=task_id, reason=reason)
+        logger.info(
+            "Cancelled task via A2A",
+            task_id=task_id,
+            reason=reason,
+            actor=actor_slug,
+            role=agent_role,
+        )
 
         return self.task_to_a2a(task)
 

tests/integration/test_a2a_routes.py

@@ -13,7 +13,7 @@ import pytest
 import pytest_asyncio
 from fastapi import FastAPI
 from httpx import ASGITransport, AsyncClient
-from roboco.api.deps import get_current_agent_slug, get_db
+from roboco.api.deps import get_agent_context, get_current_agent_slug, get_db
 from roboco.api.routes.a2a import router as a2a_router
 from roboco.api.routes.a2a import wellknown_router
 from roboco.db.tables import AgentTable, ProjectTable, TaskTable
@@ -25,6 +25,7 @@ from roboco.models.base import (
     TaskStatus,
     TaskType,
 )
+from roboco.models.permissions import AgentContext
 
 if TYPE_CHECKING:
     from collections.abc import AsyncGenerator, AsyncIterator
@@ -91,18 +92,39 @@ async def a2a_route_client(
     async def _override_agent_slug() -> str:
         return dev.slug
 
+    async def _override_agent_context() -> AgentContext:
+        # The authenticated caller is the seeded developer by default. Tests
+        # that need a different role (e.g. the PM-gated cancel route) swap this
+        # override on the yielded app before posting.
+        return AgentContext(
+            agent_id=dev.id, role=AgentRole.DEVELOPER, team=Team.BACKEND, slug=dev.slug
+        )
+
     app.dependency_overrides[get_db] = _override_db
     app.dependency_overrides[get_current_agent_slug] = _override_agent_slug
+    app.dependency_overrides[get_agent_context] = _override_agent_context
 
     transport = ASGITransport(app=app)
     async with AsyncClient(transport=transport, base_url="http://test") as client:
-        yield {"client": client, "dev": dev, "task": task}
+        yield {"client": client, "dev": dev, "task": task, "app": app}
     app.dependency_overrides.clear()
 
 
 _HDR = {"X-Agent-ID": "be-dev-1", "X-Agent-Role": "developer"}
 
 
+def _set_pm_context(app: FastAPI, dev: AgentTable) -> None:
+    """Override the agent context to a cell PM so the PM-gated cancel route
+    admits the call (the default fixture context is a developer)."""
+
+    async def _pm() -> AgentContext:
+        return AgentContext(
+            agent_id=dev.id, role=AgentRole.CELL_PM, team=Team.BACKEND, slug=dev.slug
+        )
+
+    app.dependency_overrides[get_agent_context] = _pm
+
+
 # ---------------------------------------------------------------------------
 # Well-known endpoints
 # ---------------------------------------------------------------------------
@@ -435,13 +457,17 @@ async def test_cancel_task_success(a2a_route_client: dict) -> None:
         }
     )
     client = a2a_route_client["client"]
+    _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"])
     with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
         instance = AsyncMock()
         instance.cancel_task = AsyncMock(return_value=a2a_task)
         mock_service_cls.return_value = instance
         response = await client.post(
             f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel",
-            json={"reason": "no longer needed"},
+            json={
+                "name": f"tasks/{a2a_route_client['task'].id}",
+                "reason": "no longer needed",
+            },
             headers=_HDR,
         )
     # 200 expected; pydantic may serialize as 422 if response_model coercion
@@ -452,6 +478,7 @@ async def test_cancel_task_success(a2a_route_client: dict) -> None:
 async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None:
 
     client = a2a_route_client["client"]
+    _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"])
     with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
         instance = AsyncMock()
         instance.cancel_task = AsyncMock(
@@ -469,6 +496,7 @@ async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None:
 async def test_cancel_task_not_found(a2a_route_client: dict) -> None:
 
     client = a2a_route_client["client"]
+    _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"])
     with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
         instance = AsyncMock()
         instance.cancel_task = AsyncMock(side_effect=ValueError("Task missing"))
@@ -481,10 +509,123 @@ async def test_cancel_task_not_found(a2a_route_client: dict) -> None:
 
 
 # ---------------------------------------------------------------------------
-# Chat conversations
+# #423: the cancel route must be authenticated + PM/management-gated, and pass
+# the authenticated actor + role into the service (it cascades cancel to all
+# non-terminal descendants — lifecycle rule: Any -> cancelled: PM roles only).
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.asyncio
+async def test_cancel_task_developer_role_forbidden(a2a_route_client: dict) -> None:
+    """A developer (default fixture context) must NOT be able to cancel a task
+    via A2A — the route was previously unauthenticated with no role gate, so any
+    caller could cancel any task tree (#423). Now PM/management-only."""
+    client = a2a_route_client["client"]
+    # default fixture context = developer
+    with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
+        instance = AsyncMock()
+        instance.cancel_task = AsyncMock()
+        mock_service_cls.return_value = instance
+        response = await client.post(
+            f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel",
+            headers=_HDR,
+        )
+    assert response.status_code == HTTPStatus.FORBIDDEN
+    instance.cancel_task.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_cancel_task_no_auth_header_rejected(a2a_route_client: dict) -> None:
+    """A request with no agent headers at all is rejected — the route must not
+    be reachable unauthenticated (#423)."""
+    client = a2a_route_client["client"]
+    response = await client.post(
+        f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel",
+    )
+    assert response.status_code in (
+        HTTPStatus.UNAUTHORIZED,
+        HTTPStatus.FORBIDDEN,
+        HTTPStatus.UNPROCESSABLE_ENTITY,
+    )
+
+
+@pytest.mark.asyncio
+async def test_cancel_task_pm_passes_actor_and_role_to_service(
+    a2a_route_client: dict,
+) -> None:
+    """A PM cancel threads the authenticated role (for the cascade role gate)
+    and the actor slug (for the cancellation-note attribution) into the service
+    — previously the service was called with no actor and a hardcoded
+    cell_pm role, so the audit trail recorded no real caller (#423)."""
+    client = a2a_route_client["client"]
+    _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"])
+    with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
+        instance = AsyncMock()
+        instance.cancel_task = AsyncMock(
+            return_value=A2ATask.model_validate(
+                {
+                    "id": str(a2a_route_client["task"].id),
+                    "contextId": str(uuid4()),
+                    "status": A2ATaskStatus(state=A2ATaskState.CANCELLED).model_dump(
+                        mode="json"
+                    ),
+                }
+            )
+        )
+        mock_service_cls.return_value = instance
+        # No body → request=None → the handler runs (a body without the A2A
+        # ``name`` field 422s at request validation before the handler). The
+        # invariant under test is the role/slug threading, not the reason.
+        response = await client.post(
+            f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel",
+            headers=_HDR,
+        )
+    assert response.status_code == HTTPStatus.OK
+    # The authenticated PM role + slug reach the service.
+    _kwargs = instance.cancel_task.await_args.kwargs
+    assert _kwargs.get("agent_role") == "cell_pm"
+    assert _kwargs.get("actor_slug") == a2a_route_client["dev"].slug
+
+
+# ---------------------------------------------------------------------------
+# #116: send_message must record the AUTHENTICATED identity as the responder,
+# not a client-supplied metadata.from_agent (spoof).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_send_message_uses_authenticated_identity_not_client_from_agent(
+    a2a_route_client: dict,
+) -> None:
+    """is_response=True must stamp the authenticated caller's slug as the
+    responder, ignoring a spoofed metadata.from_agent — previously the route
+    took from_agent verbatim from the request body, so any agent could
+    impersonate anyone (e.g. from_agent='ceo') in the task's notes and in the
+    spawn/notification routed back to the original requester (#116)."""
+    client = a2a_route_client["client"]
+    with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
+        instance = AsyncMock()
+        instance.update_task_from_message = AsyncMock(return_value=None)
+        mock_service_cls.return_value = instance
+        response = await client.post(
+            "/api/a2a/message/send",
+            json={
+                "message": {
+                    "role": "user",
+                    "parts": [{"type": "text", "text": "hi"}],
+                    "taskId": str(a2a_route_client["task"].id),
+                },
+                "metadata": {"is_response": True, "from_agent": "ceo"},
+            },
+            headers=_HDR,
+        )
+    assert response.status_code == HTTPStatus.OK
+    # The authenticated dev slug is the responder — NOT the spoofed 'ceo'.
+    _kwargs = instance.update_task_from_message.await_args.kwargs
+    assert _kwargs.get("responder_agent") == a2a_route_client["dev"].slug
+    assert _kwargs.get("responder_agent") != "ceo"
+
+
 @pytest.mark.asyncio
 async def test_chat_create_conversation_access_denied(
     a2a_route_client: dict,
@@ -1239,6 +1380,7 @@ async def test_cancel_task_success_no_body(a2a_route_client: dict) -> None:
         }
     )
     client = a2a_route_client["client"]
+    _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"])
     with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls:
         instance = AsyncMock()
         instance.cancel_task = AsyncMock(return_value=a2a_task)