[chore] logical-gaps: a2a service hierarchy gate (typed, unconditional) + persist skill on message row (3 gaps)

create_a2a_notification gated A2A hierarchy only when both ends resolved
(`if from_agent and target_agent:`), so an unattributed (from_agent falsy)
or unresolvable-target request slipped past the hierarchy matrix and
dispatched with from_agent='unknown' / to_agent='' — and a denial came back
as a bare ValueError indistinguishable from the missing-task_id ValueError.
Require both ends present, then validate via the shared typed
validate_a2a_access path (A2AAccessDeniedError + route_hint) so the legacy
notification surface enforces the same who-may-talk-to-whom invariant as the
conversation path.

send() accepts skill= and the gateway callers (qa/doc/pr_gate) pass it
expecting the receiver to learn which capability the message is about, but
send_chat_message never read it from options — silently dropped. Persist a
nullable skill column (migration 054) on a2a_messages, wire it through
send_chat_message + _msg_to_model + the A2AChatMessage model, and fix the
send() docstring (it claimed 'recorded in message metadata').

TDD: 4 red→green (skill recorded on message + surfaces in inbox; permission
denied raises typed A2AAccessDeniedError with route_hint; self-A2A raises
typed; missing from_agent raises instead of silent dispatch). 103 a2a
integration tests green; ruff/mypy clean; migration 054 verified
upgrade/downgrade on throwaway PG.

alembic/versions/054_a2a_message_skill.py

@@ -0,0 +1,32 @@
+"""Persist ``skill`` on a2a_messages.
+
+Directed A2A ``send()`` accepts a ``skill=`` (the capability the sender is
+exercising/requests of the receiver, e.g. ``code_review``) and the gateway
+callers (qa / doc / pr_gate) pass it expecting the receiver to learn which
+capability the message is about. Until now ``send_chat_message`` never read it
+from options, so it was silently dropped — the receiver's inbox showed a bare
+message with no skill context. This adds a nullable ``skill`` column so the
+capability rides on the row and surfaces in the inbox model.
+
+Revision ID: 054_a2a_message_skill
+Revises: 053_playbook_archived_attr
+Create Date: 2026-06-30
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "054_a2a_message_skill"
+down_revision = "053_playbook_archived_attr"
+branch_labels: dict[str, str] | None = None
+depends_on: dict[str, str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("a2a_messages", sa.Column("skill", sa.String(100), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("a2a_messages", "skill")

roboco/db/tables.py

@@ -1821,6 +1821,9 @@ class A2AMessageTable(Base):
         nullable=False,
         default=A2AMessageKind.MESSAGE,
     )
+    # Capability this A2A concerns (e.g. ``code_review``); nullable for legacy /
+    # unspecified messages. Migration 054.
+    skill: Mapped[str | None] = mapped_column(String(100), nullable=True)
 
     # Threading
     response_to_id: Mapped[UUID | None] = mapped_column(

roboco/models/a2a.py

@@ -527,6 +527,13 @@ class A2AChatMessage(RobocoBase):
     message_kind: A2AMessageKind = Field(
         default=A2AMessageKind.MESSAGE, description="Type of message"
     )
+    # The capability the sender is exercising/requests of the receiver (e.g.
+    # ``code_review``). Recorded on the row so the receiver (and the inbox)
+    # learns which skill a directed A2A is about — callers pass ``skill=`` on
+    # ``send()`` expecting it to reach the receiver, not be dropped.
+    skill: str | None = Field(
+        default=None, description="Capability this A2A concerns, if any"
+    )
 
     # Threading
     response_to_id: str | None = Field(

roboco/services/a2a.py

@@ -605,14 +605,29 @@ class A2AService:
         target_agent = self.resolve_target_agent(metadata)
         skill = metadata.get("skill", "general")
 
-        # Enforce A2A hierarchy permissions
-        if from_agent and target_agent:
-            from roboco.agents_config import can_a2a_direct, get_a2a_route_hint
-
-            allowed, error_msg = can_a2a_direct(from_agent, target_agent)
-            if not allowed:
-                hint = get_a2a_route_hint(from_agent, target_agent)
-                raise ValueError(f"{error_msg} Hint: {hint}")
+        # Enforce A2A hierarchy permissions — UNCONDITIONALLY. The conversation
+        # path (validate_a2a_access, below) requires both ends present and
+        # rejects self-A2A with a typed A2AAccessDeniedError + route_hint. This
+        # legacy notification path used to gate on `if from_agent and
+        # target_agent:`, so an unattributed (from_agent falsy) or untargeted
+        # (target unresolvable) request slipped past the hierarchy matrix and
+        # dispatched with from_agent='unknown' / to_agent='' — and a hierarchy
+        # denial came back as a bare ValueError indistinguishable from the
+        # missing-task_id ValueError above. Require both resolved, then validate
+        # via the shared typed path so both A2A surfaces enforce the same
+        # who-may-talk-to-whom invariant.
+        if not from_agent:
+            raise ValueError(
+                "A2A notification requires a 'from_agent' in metadata — an "
+                "unattributed request would bypass the hierarchy gate"
+            )
+        if not target_agent:
+            raise ValueError(
+                "A2A notification could not resolve a target agent — provide an "
+                "explicit 'target_agent' (a known agent slug) or a 'skill' that "
+                "matches an agent's capability"
+            )
+        validate_a2a_access(from_agent, target_agent)
         # Priority parsing: full tristate (NORMAL/HIGH/URGENT) survives
         # end-to-end. Resolution rules live in
         # foundation.policy.communications.parse_priority.
@@ -1024,6 +1039,7 @@ class A2AService:
         message_kind = opts.get("message_kind", A2AMessageKind.MESSAGE)
         response_to_id = opts.get("response_to_id")
         requires_response = opts.get("requires_response", False)
+        skill = opts.get("skill")
 
         result = await self.session.execute(
             select(A2AConversationTable).where(
@@ -1072,6 +1088,7 @@ class A2AService:
             message_kind=message_kind,
             response_to_id=response_to_id,
             requires_response=requires_response,
+            skill=skill,
         )
         self.session.add(msg)
 
@@ -1354,6 +1371,7 @@ class A2AService:
             from_agent=msg.from_agent,
             content=msg.content,
             message_kind=msg.message_kind,
+            skill=msg.skill,
             response_to_id=str(msg.response_to_id) if msg.response_to_id else None,
             requires_response=msg.requires_response,
             read_at=msg.read_at,
@@ -1396,8 +1414,8 @@ class A2AService:
         1. `get_or_create_conversation(sender_slug, recipient_slug, task_id=...)`
         2. `send_chat_message(conversation.id, sender_slug, content=body, ...)`
 
-        `skill` is recorded in message metadata so the receiver knows which
-        capability is being requested.
+        `skill` is persisted on the message row so the receiver (and the
+        inbox) learns which capability is being requested.
         """
         from_slug = await self._resolve_slug_from_id(from_agent)
         to_slug = (

tests/integration/test_a2a_service.py

@@ -417,6 +417,30 @@ async def test_send_a2a_returns_handler_result(a2a_setup: dict) -> None:
         pass
 
 
+@pytest.mark.asyncio
+async def test_send_records_skill_on_message_for_receiver(a2a_setup: dict) -> None:
+    """#1416: a2a.send(skill=...) must record the requested capability on the
+    persisted message so the receiver learns it — not silently drop it. The
+    gateway adapter's docstring promised exactly this, but send_chat_message
+    never read the ``skill`` opt and the table had no skill column, so every
+    gateway A2A send lost the capability signal the caller passed."""
+    svc = a2a_setup["svc"]
+    dev = a2a_setup["dev"]
+    task_id = a2a_setup["task_id"]
+    sent = await svc.send(
+        from_agent=dev.id,
+        to_agent="be-qa",
+        task_id=task_id,
+        body="please review my PR",
+        skill="code_review",
+    )
+    assert sent.skill == "code_review"
+    # The receiver reads the capability back via get_messages.
+    inbox = await svc.get_messages(UUID(sent.conversation_id), "be-qa")
+    assert inbox
+    assert inbox[-1].skill == "code_review"
+
+
 # ---------------------------------------------------------------------------
 # Conversation creation happy path with allowed pair
 # ---------------------------------------------------------------------------
@@ -992,7 +1016,10 @@ async def test_create_a2a_notification_with_target_calls_notification_service(
 async def test_create_a2a_notification_permission_denied(
     a2a_setup: dict,
 ) -> None:
-    """can_a2a_direct returns False → ValueError with hint."""
+    """Hierarchy denial → the typed A2AAccessDeniedError (with route_hint), routed
+    through validate_a2a_access so the legacy notification path matches the
+    conversation path — not a bare ValueError a caller can't distinguish from a
+    malformed request."""
     svc = a2a_setup["svc"]
     task_id = str(a2a_setup["task_id"])
     msg = A2AMessage(role="user", parts=[TextPart(text="hi")], task_id=task_id)
@@ -1002,16 +1029,64 @@ async def test_create_a2a_notification_permission_denied(
     )
     with (
         patch(
-            "roboco.agents_config.can_a2a_direct",
+            "roboco.enforcement.a2a_access.can_a2a_direct",
             return_value=(False, "denied"),
         ),
         patch(
-            "roboco.agents_config.get_a2a_route_hint",
+            "roboco.enforcement.a2a_access.get_a2a_route_hint",
             return_value="use channel",
         ),
-        pytest.raises(ValueError, match="Hint:"),
+        pytest.raises(A2AAccessDeniedError) as exc,
     ):
         await svc.create_a2a_notification(req)
+    assert exc.value.route_hint == "use channel"
+
+
+@pytest.mark.asyncio
+async def test_create_a2a_notification_self_a2a_raises_typed_error(
+    a2a_setup: dict,
+) -> None:
+    """#612: a self-directed A2A notification (from_agent == target) must raise
+    the typed A2AAccessDeniedError — the self-check the conversation path
+    (validate_a2a_access) enforces — not a bare ValueError and not a silently
+    sent notification."""
+    svc = a2a_setup["svc"]
+    task_id = str(a2a_setup["task_id"])
+    msg = A2AMessage(role="user", parts=[TextPart(text="self note")], task_id=task_id)
+    req = SendMessageRequest(
+        message=msg,
+        metadata={"from_agent": "be-dev-1", "target_agent": "be-dev-1"},
+    )
+    mock_ns = AsyncMock()
+    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
+    with (
+        patch("roboco.services.notification.NotificationService", return_value=mock_ns),
+        pytest.raises(A2AAccessDeniedError),
+    ):
+        await svc.create_a2a_notification(req)
+    mock_ns.send_a2a_notification.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_create_a2a_notification_missing_from_agent_raises_not_silent(
+    a2a_setup: dict,
+) -> None:
+    """#609: an unattributed A2A request (no from_agent, unresolvable target)
+    must NOT slip past the hierarchy gate and dispatch with
+    from_agent='unknown'. The gate is unconditional — both ends must be present
+    and resolvable before any notification is created."""
+    svc = a2a_setup["svc"]
+    task_id = str(a2a_setup["task_id"])
+    msg = A2AMessage(role="user", parts=[TextPart(text="anon")], task_id=task_id)
+    req = SendMessageRequest(message=msg, metadata={})
+    mock_ns = AsyncMock()
+    mock_ns.send_a2a_notification = AsyncMock(return_value=None)
+    with (
+        patch("roboco.services.notification.NotificationService", return_value=mock_ns),
+        pytest.raises(ValueError, match="from_agent"),
+    ):
+        await svc.create_a2a_notification(req)
+    mock_ns.send_a2a_notification.assert_not_called()
 
 
 # ---------------------------------------------------------------------------