fix(security): scope A2A task routes to their agent, and stop the ruff pin drifting

.pre-commit-config.yaml

@@ -14,12 +14,27 @@ repos:
       - id: debug-statements
       - id: check-merge-conflict
 
-  - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.6
+  # Ruff runs from the project's own lockfile rather than a separately
+  # pinned hook environment. The two used to drift: the hook pinned v0.8.6
+  # while the dev dependency resolved to 0.15, and they disagreed on 34 lint
+  # findings and 9 files' formatting. Every PR that touched one of those
+  # files failed the gate on code it had not written. Sourcing ruff from
+  # `uv run` — as the mypy hook below already does — makes that impossible:
+  # there is one ruff, the one in uv.lock.
+  - repo: local
     hooks:
       - id: ruff
-        args: [--fix]
+        name: ruff
+        entry: uv run ruff check --fix
+        language: system
+        types_or: [python, pyi]
+        require_serial: true
       - id: ruff-format
+        name: ruff-format
+        entry: uv run ruff format
+        language: system
+        types_or: [python, pyi]
+        require_serial: true
 
   - repo: local
     hooks:

docs/architecture/api-reference.md

@@ -478,6 +478,11 @@ curl -X POST http://localhost:8000/api/v1/my_agent/a2a/tasks \
 
 Get A2A task status.
 
+These per-agent A2A routes — status, `events` and `cancel` — only accept tasks
+submitted to *this* agent. A task belonging to another agent answers `404`, the
+same as an id that does not exist. The shared task board at
+`/api/v1/tasks/{id}` is cross-agent by design and is unaffected.
+
 ```bash
 curl http://localhost:8000/api/v1/my_agent/a2a/tasks/task_a1b2c3d4e5f6
 ```

pyproject.toml

@@ -69,7 +69,7 @@ dev = [
     "pytest-asyncio>=0.24",
     "pytest-cov>=6.0",
     "pytest-mock>=3.14",
-    "ruff>=0.8",
+    "ruff>=0.15",
     "mypy>=1.13",
     "pre-commit>=4.0",
     "build>=1.0",
@@ -104,7 +104,7 @@ dev = [
     "pytest-asyncio>=0.24",
     "pytest-cov>=6.0",
     "pytest-mock>=3.14",
-    "ruff>=0.8",
+    "ruff>=0.15",
     "mypy>=1.13",
     "pre-commit>=4.0",
     "build>=1.0",

src/agentomatic/core/router_factory.py

@@ -1306,6 +1306,33 @@ def _a2a_view(record: Any) -> dict[str, Any]:
             view["error"] = record.error
         return view
 
+    async def _owned_a2a_task(task_id: str) -> Any:
+        """Fetch a task, insisting it belongs to *this* agent.
+
+        These routes are namespaced per agent but used to accept any task id,
+        so one agent's route would read — and cancel — another agent's task.
+        Zero-trust derives its policy from the agent segment of the path, so
+        that let a caller reach work the policy engine had denied them.
+
+        A task owned by another agent answers 404, exactly as a nonexistent
+        one does: a distinct status would confirm the id exists.
+
+        Args:
+            task_id: Task being addressed.
+
+        Returns:
+            The task record.
+
+        Raises:
+            HTTPException: 404 when the task is missing or owned elsewhere.
+        """
+        if task_manager is None:
+            raise HTTPException(501, "A2A task tracking requires a task manager")
+        record = await task_manager.get(task_id)
+        if record is None or record.target != agent_name:
+            raise HTTPException(404, f"Task '{task_id}' not found")
+        return record
+
     def _a2a_event_view(evt: Any, record: Any) -> dict[str, Any]:
         """Render one recorded event as an A2A-shaped task object.
 
@@ -1404,16 +1431,13 @@ async def submit_a2a_task(request: A2ATaskRequest) -> dict[str, Any]:
     @router.get("/a2a/tasks/{task_id}")
     async def get_a2a_task(task_id: str) -> dict[str, Any]:
         """Get A2A task status."""
-        if task_manager is not None:
-            record = await task_manager.get(task_id)
-            if record is None:
-                raise HTTPException(404, f"Task '{task_id}' not found")
-            return _a2a_view(record)
-        return {
-            "task_id": task_id,
-            "status": "completed",
-            "message": "Task tracking requires a task manager",
-        }
+        if task_manager is None:
+            return {
+                "task_id": task_id,
+                "status": "completed",
+                "message": "Task tracking requires a task manager",
+            }
+        return _a2a_view(await _owned_a2a_task(task_id))
 
     @router.get("/a2a/tasks/{task_id}/events")
     async def stream_a2a_task(task_id: str, request: Request, since: int = Query(default=0, ge=0)):
@@ -1429,9 +1453,7 @@ async def stream_a2a_task(task_id: str, request: Request, since: int = Query(def
 
         if task_manager is None:
             raise HTTPException(501, "A2A task streaming requires a task manager")
-        record = await task_manager.get(task_id)
-        if record is None:
-            raise HTTPException(404, f"Task '{task_id}' not found")
+        await _owned_a2a_task(task_id)
 
         resume_from = since or parse_last_event_id(request.headers.get("last-event-id"))
 
@@ -1504,6 +1526,9 @@ async def cancel_a2a_task(task_id: str) -> dict[str, Any]:
         """Cancel an in-flight A2A task."""
         if task_manager is None:
             raise HTTPException(501, "Task cancellation requires a task manager")
+        # Cancelling is a write, so the ownership check matters more here than
+        # on the read routes: without it any agent could stop any task.
+        await _owned_a2a_task(task_id)
         cancelled = await task_manager.cancel(task_id)
         if not cancelled:
             raise HTTPException(409, f"Task '{task_id}' not found or already terminal")

tests/test_a2a_streaming.py

@@ -10,7 +10,9 @@
 
 from __future__ import annotations
 
+import asyncio
 import json
+import time
 from dataclasses import dataclass, field
 from typing import Any
 
@@ -244,3 +246,99 @@ async def counting_get(self, tid: str):
         # One lookup to 404-check, one for the snapshot, at most one for the
         # terminal payload — never one per frame.
         assert len(calls) <= 3, f"{len(calls)} store reads for {frames} frames"
+
+
+class TestA2ATasksAreScopedToTheirAgent:
+    """The per-agent A2A routes are namespaced by agent but used to accept any
+    task id. Zero-trust derives its policy from that agent segment, so one
+    agent's route could read — and cancel — work the policy engine had denied
+    the caller on the owning agent.
+
+    The shared task board at ``/api/v1/tasks/{id}`` is deliberately untouched:
+    it is cross-agent by design.
+    """
+
+    @pytest.fixture
+    def two_agents(self) -> Any:
+        async def secret(state: dict[str, Any]) -> dict[str, Any]:
+            return {"response": "CONFIDENTIAL-PAYROLL"}
+
+        async def slow(state: dict[str, Any]) -> dict[str, Any]:
+            await asyncio.sleep(5)
+            return {"response": "done"}
+
+        platform = AgentPlatform(
+            agents_dir="/tmp/agentomatic_a2a_scope_test",
+            title="A2A Scope",
+            version="0.0.1",
+            enable_tasks=True,
+        )
+        platform.register_agent(
+            manifest=AgentManifest(name="payroll", slug="p", description="p"), node_fn=secret
+        )
+        platform.register_agent(
+            manifest=AgentManifest(name="slowpoke", slug="s", description="s"), node_fn=slow
+        )
+        platform.register_agent(
+            manifest=AgentManifest(name="faq", slug="f", description="f"), node_fn=_echo_fn
+        )
+        with TestClient(platform.build()) as test_client:
+            yield test_client
+
+    def _submit(self, client: Any, agent: str) -> str:
+        return str(
+            client.post(f"{BASE}/{agent}/a2a/tasks", json={"message": {"content": "x"}}).json()[
+                "task_id"
+            ]
+        )
+
+    def test_another_agent_cannot_read_the_task(self, two_agents: Any) -> None:
+        task_id = self._submit(two_agents, "payroll")
+
+        leaked = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}")
+
+        assert leaked.status_code == 404
+        assert "CONFIDENTIAL-PAYROLL" not in leaked.text
+
+    def test_another_agent_cannot_stream_the_task(self, two_agents: Any) -> None:
+        task_id = self._submit(two_agents, "payroll")
+
+        leaked = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}/events")
+
+        assert leaked.status_code == 404
+        assert "CONFIDENTIAL-PAYROLL" not in leaked.text
+
+    def test_another_agent_cannot_cancel_a_running_task(self, two_agents: Any) -> None:
+        """The write is the sharper end of this: without the check any agent
+        could stop any task."""
+        task_id = self._submit(two_agents, "slowpoke")
+        time.sleep(0.3)  # let it reach running
+
+        assert two_agents.post(f"{BASE}/faq/a2a/tasks/{task_id}/cancel").status_code == 404
+
+    def test_the_owning_agent_still_reads_and_streams(self, two_agents: Any) -> None:
+        task_id = self._submit(two_agents, "payroll")
+
+        assert two_agents.get(f"{BASE}/payroll/a2a/tasks/{task_id}").status_code == 200
+        assert two_agents.get(f"{BASE}/payroll/a2a/tasks/{task_id}/events").status_code == 200
+
+    def test_the_owning_agent_can_still_cancel(self, two_agents: Any) -> None:
+        task_id = self._submit(two_agents, "slowpoke")
+        time.sleep(0.3)
+
+        assert two_agents.post(f"{BASE}/slowpoke/a2a/tasks/{task_id}/cancel").status_code == 200
+
+    def test_a_foreign_task_looks_exactly_like_a_missing_one(self, two_agents: Any) -> None:
+        """A distinct status would confirm the id exists."""
+        task_id = self._submit(two_agents, "payroll")
+
+        foreign = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}")
+        missing = two_agents.get(f"{BASE}/faq/a2a/tasks/task_deadbeefdeadbeef")
+
+        assert foreign.status_code == missing.status_code == 404
+
+    def test_the_shared_task_board_is_unchanged(self, two_agents: Any) -> None:
+        """This fix is deliberately local to the per-agent A2A routes."""
+        task_id = self._submit(two_agents, "payroll")
+
+        assert two_agents.get(f"{BASE}/tasks/{task_id}").status_code == 200