fix(security): active guard enforcement, CEO A2A target check, notification expiry (#595)

* fix(security): guard goes active; CEO A2A respects no-comms roles; ack notifications expire

ROBOCO_GUARD_PASSIVE_MODE defaults to false in both compose files — the
deferred post-calibration flip; fail_secure stays off and the env override
remains the rollback. can_a2a_direct no longer short-circuits the CEO past
the no-comms set (auditor/pr_reviewer/prompter/secretary), now canonical
in foundation.policy.communications.NO_COMMS_ROLES and shared with the
content-actions gate; the A2A service refuses at conversation creation
instead of silently suppressing the wake. Ack-required notifications get
expires_at stamped from ROBOCO_NOTIFICATION_ACK_TTL_HOURS (default 48,
0 disables), so the re-escalation sweeper's expires_at query matches rows
for the first time.

* refactor(notification): extract _ack_and_expiry — xenon rank back under B

The expires_at stamping pushed _create_notification_with_session to
rank C; the requires_ack + expiry derivation moves into a helper with
the same semantics and comments.

* test(conftest): dispose the global DB engine after every test

Production code reaching get_db_context()/get_engine() lazily creates the
process-global engine bound to the current event loop; with per-test
function-scoped loops, any later test touching the global path inherits a
dead-loop engine and dies with 'Future attached to a different loop' —
the order-dependent class that has been wandering the suite (cloud_auth
login, metrics, tasks-routes, full-lifecycle) whenever collection order
shifts. An autouse fixture now close_db()s after every test, keeping the
global path loop-local; no-op when untouched.

---------

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

tests/conftest.py

@@ -36,6 +36,7 @@ Redis isolation:
 
 from __future__ import annotations
 
+import contextlib
 import json
 import os
 import socket
@@ -48,7 +49,7 @@ import pytest
 import pytest_asyncio
 from roboco.config import settings as _settings
 from roboco.db import tables as roboco_tables
-from roboco.db.base import Base
+from roboco.db.base import Base, close_db
 from roboco.db.tables import (
     AgentTable,
     AuditLogTable,
@@ -78,6 +79,23 @@ if TYPE_CHECKING:
     from collections.abc import AsyncIterator
 
 
+@pytest_asyncio.fixture(autouse=True)
+async def _dispose_global_db_engine() -> AsyncIterator[None]:
+    """Never let the lazy global engine outlive the test that created it.
+
+    Production code reaching ``get_db_context()``/``get_engine()`` creates
+    the process-global ``_DbHolder`` engine bound to the CURRENT event loop.
+    With function-scoped test loops, any later test touching that global
+    path inherits an engine from a dead loop and crashes with ``Future
+    attached to a different loop`` — an order-dependent failure class that
+    moves around whenever test collection shifts. Disposing after every
+    test keeps the global path loop-local; a no-op when nothing touched it.
+    """
+    yield
+    with contextlib.suppress(Exception):
+        await close_db()
+
+
 @pytest.fixture(autouse=True)
 def _no_live_redis(monkeypatch: pytest.MonkeyPatch) -> None:
     """Keep every test off the real localhost Redis (see module docstring).

tests/integration/test_a2a_service.py

@@ -2494,21 +2494,36 @@ async def test_agent_reply_to_ceo_creates_no_wake(a2a_setup: dict) -> None:
 
 
 @pytest.mark.asyncio
-async def test_ceo_dm_to_non_a2a_role_creates_no_wake(a2a_setup: dict) -> None:
-    """A CEO DM to a role with no read_a2a on its manifest (pr_reviewer,
-    auditor) must NOT create a wake row — the recipient could never ack it,
-    so it would be immortal, permanently suppress future wakes via the dedup
-    pre-check, and drive futile respawns."""
+async def test_ceo_dm_to_non_a2a_role_denied_at_conversation_creation(
+    a2a_setup: dict,
+) -> None:
+    """A CEO DM to a role with no dm/read_a2a on its manifest (pr_reviewer,
+    auditor) must be refused outright at conversation creation — the root-
+    cause fix (can_a2a_direct's CEO branch now excludes NO_COMMS_ROLES)
+    supersedes the old symptom-level fix of letting the conversation exist
+    and only suppressing the wake notification (the recipient could never
+    ack it, so it would be immortal, permanently suppress future wakes via
+    the dedup pre-check, and drive futile respawns)."""
     svc: A2AService = a2a_setup["svc"]
-    conv = await svc.get_or_create_conversation(agent_a="ceo", agent_b="pr-reviewer-1")
-    conv_id = UUID(conv.id)
+    with pytest.raises(A2AAccessDeniedError, match="no agent-comms surface"):
+        await svc.get_or_create_conversation(agent_a="ceo", agent_b="pr-reviewer-1")
 
+
+@pytest.mark.asyncio
+async def test_maybe_wake_ceo_recipient_still_noops_for_no_comms_role(
+    a2a_setup: dict,
+) -> None:
+    """Defense-in-depth: _maybe_wake_ceo_recipient's own read_a2a manifest
+    check independently no-ops for a no-comms role — unreachable through the
+    normal send path now that conversation creation refuses it first, but
+    it must stay safe if ever called directly (e.g. on a pre-fix row)."""
+    svc: A2AService = a2a_setup["svc"]
     mock_ns = AsyncMock()
     mock_ns.send_a2a_notification = AsyncMock(return_value=None)
     with patch(
         "roboco.services.notification.NotificationService", return_value=mock_ns
     ):
-        await svc.send_chat_message(conv_id, "ceo", "review status?")
+        await svc._maybe_wake_ceo_recipient("ceo", "pr-reviewer-1", None)
 
     mock_ns.send_a2a_notification.assert_not_awaited()
 

tests/integration/test_notification_expiry_sweep.py

@@ -0,0 +1,169 @@
+"""expires_at is now stamped at creation (NotificationService) and actually
+matched by NotificationDeliveryService.sweep_expired_notifications' SQL
+WHERE clause — before the fix the column was never written, so this query
+always matched zero rows regardless of how stale a notification was.
+
+Integration tests against the migrated Postgres DB: `sweep_expired_notifications`
+issues a real `expires_at < now()` query, so a mocked session (as
+`tests/unit/services/test_notification_delivery.py` uses) can't exercise it.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING, cast
+from uuid import UUID, uuid4
+
+import pytest
+from roboco.config import settings
+from roboco.db.tables import AgentTable, NotificationTable
+from roboco.models import AgentRole, AgentStatus, NotificationPriority, NotificationType
+from roboco.models.base import Team
+from roboco.models.notification import CreateNotificationParams
+from roboco.services.notification import NotificationService
+from roboco.services.notification_delivery import get_notification_delivery_service
+from sqlalchemy import select
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+async def _seed_agent(db: AsyncSession, *, role: AgentRole, slug: str) -> UUID:
+    agent = AgentTable(
+        id=uuid4(),
+        name=slug,
+        slug=slug,
+        role=role,
+        team=Team.BACKEND,
+        status=AgentStatus.ACTIVE,
+        model_config={},
+        system_prompt=slug,
+        capabilities=[],
+        permissions={},
+        metrics={},
+    )
+    db.add(agent)
+    await db.flush()
+    return cast("UUID", agent.id)
+
+
+@pytest.mark.asyncio
+async def test_created_notification_expires_at_is_stamped_and_matched_by_sweep(
+    db_session: AsyncSession,
+) -> None:
+    """End-to-end: NotificationService._create_notification stamps expires_at
+    for an ack-required row, and once that deadline is in the past,
+    sweep_expired_notifications' real Postgres query finds it (count 1) —
+    the exact round trip that was a dead no-op before this fix, since
+    expires_at was always NULL and `expires_at < now()` never matched."""
+    unique = uuid4().hex[:8]
+    sender = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, slug=f"sndr-{unique}"
+    )
+    recipient = await _seed_agent(
+        db_session, role=AgentRole.CELL_PM, slug=f"pm-{unique}"
+    )
+
+    svc = NotificationService()
+    await svc._create_notification(
+        CreateNotificationParams(
+            notification_type=NotificationType.BLOCKER_ESCALATION,
+            priority=NotificationPriority.HIGH,
+            from_agent=str(sender),
+            to_agents=[str(recipient)],
+            subject="blocked",
+            body="external dependency",
+        ),
+        db_session=db_session,
+    )
+    await db_session.flush()
+
+    row = (
+        await db_session.execute(
+            select(NotificationTable).where(
+                NotificationTable.type == NotificationType.BLOCKER_ESCALATION,
+                NotificationTable.from_agent == sender,
+            )
+        )
+    ).scalar_one()
+    assert row.expires_at is not None
+    assert row.requires_ack is True
+
+    # Backdate it past the deadline (no real clock wait) and confirm the
+    # sweep's `expires_at < now()` predicate now actually matches.
+    row.expires_at = datetime.now(UTC) - timedelta(minutes=1)
+    await db_session.flush()
+
+    deliv = get_notification_delivery_service(db_session)
+    count = await deliv.sweep_expired_notifications()
+    assert count >= 1
+
+
+@pytest.mark.asyncio
+async def test_directly_stamped_expired_row_is_matched_by_sweep_query(
+    db_session: AsyncSession,
+) -> None:
+    """Isolates the sweep query mechanics from creation: a hand-built
+    ack-required, unacked row with expires_at in the past must be counted."""
+    unique = uuid4().hex[:8]
+    sender = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, slug=f"s2-{unique}"
+    )
+    recipient = await _seed_agent(db_session, role=AgentRole.QA, slug=f"r2-{unique}")
+
+    notification = NotificationTable(
+        type=NotificationType.ALERT,
+        priority=NotificationPriority.HIGH,
+        from_agent=sender,
+        to_agents=[recipient],
+        subject="stale alert",
+        body="body",
+        requires_ack=True,
+        expires_at=datetime.now(UTC) - timedelta(hours=1),
+    )
+    db_session.add(notification)
+    await db_session.flush()
+
+    deliv = get_notification_delivery_service(db_session)
+    count = await deliv.sweep_expired_notifications()
+    assert count >= 1
+
+
+@pytest.mark.asyncio
+async def test_zero_ttl_disables_expires_at_stamping_end_to_end(
+    db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """notification_ack_ttl_hours=0 leaves expires_at NULL even for an
+    ack-required notification created through the real service."""
+    monkeypatch.setattr(settings, "notification_ack_ttl_hours", 0)
+    unique = uuid4().hex[:8]
+    sender = await _seed_agent(
+        db_session, role=AgentRole.DEVELOPER, slug=f"s3-{unique}"
+    )
+    recipient = await _seed_agent(
+        db_session, role=AgentRole.CELL_PM, slug=f"pm3-{unique}"
+    )
+
+    svc = NotificationService()
+    await svc._create_notification(
+        CreateNotificationParams(
+            notification_type=NotificationType.BLOCKER_ESCALATION,
+            priority=NotificationPriority.HIGH,
+            from_agent=str(sender),
+            to_agents=[str(recipient)],
+            subject="blocked",
+            body="external dependency",
+        ),
+        db_session=db_session,
+    )
+    await db_session.flush()
+
+    row = (
+        await db_session.execute(
+            select(NotificationTable).where(
+                NotificationTable.type == NotificationType.BLOCKER_ESCALATION,
+                NotificationTable.from_agent == sender,
+            )
+        )
+    ).scalar_one()
+    assert row.expires_at is None

tests/unit/enforcement/test_a2a_access.py

@@ -9,6 +9,7 @@ from roboco.enforcement.a2a_access import (
     get_a2a_allowed_targets,
     validate_a2a_access,
 )
+from roboco.foundation.policy.communications import NO_COMMS_ROLES
 
 
 def test_validate_a2a_self_a2a_denied() -> None:
@@ -94,3 +95,27 @@ def test_can_a2a_direct_to_ceo_message_explains_reply_only() -> None:
     assert allowed is False
     assert reason is not None
     assert "reply" in reason.lower()
+
+
+@pytest.mark.parametrize(
+    "target_slug",
+    ["auditor", "pr-reviewer-1", "intake-1", "secretary-1"],
+)
+def test_can_a2a_direct_ceo_to_no_comms_role_denied(target_slug: str) -> None:
+    """The CEO's asymmetric reach still can't target a role with no dm/
+    read_a2a on its manifest (auditor, pr_reviewer, prompter, secretary) —
+    nothing on the other end could ever read or answer the DM. The panel's
+    New-DM dialog already filters these client-side (EXCLUDE_NON_DM_ROLES);
+    this is the server-side backstop for a direct API/A2A-service call."""
+    allowed, reason = can_a2a_direct("ceo", target_slug)
+    assert allowed is False
+    assert reason is not None
+    assert "comms" in reason.lower()
+
+
+def test_can_a2a_direct_ceo_to_no_comms_role_reuses_canonical_set() -> None:
+    """The refusal set must be exactly foundation.policy.communications'
+    NO_COMMS_ROLES — the same set services.gateway.content_actions uses to
+    gate the dm() sender side — so the two never drift apart."""
+    expected = {"auditor", "pr_reviewer", "prompter", "secretary"}
+    assert {role.value for role in NO_COMMS_ROLES} == expected

tests/unit/services/test_notification.py

@@ -9,6 +9,7 @@ without spinning up a Postgres + Redis stack.
 from __future__ import annotations
 
 from contextlib import asynccontextmanager
+from datetime import UTC, datetime, timedelta
 from typing import TYPE_CHECKING, Any, cast
 
 if TYPE_CHECKING:
@@ -17,6 +18,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 from uuid import UUID, uuid4
 
 import pytest
+from roboco.config import settings
 from roboco.foundation.policy.communications import ACK_REQUIRED_BY_TYPE
 from roboco.models import NotificationPriority, NotificationType
 from roboco.models.notification import CreateNotificationParams
@@ -414,6 +416,74 @@ async def test_create_notification_requires_ack_derives_from_type(
     )
 
 
+# ---------------------------------------------------------------------------
+# expires_at stamping (notification_ack_ttl_hours) — feeds
+# NotificationDeliveryService.sweep_expired_notifications' re-escalation.
+# Column existed but was never written, so the sweep query always matched
+# zero rows.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_ack_required_notification_gets_expires_at(
+    svc: NotificationService,
+) -> None:
+    """An ack-required row (BLOCKER_ESCALATION) is stamped expires_at ~=
+    now + notification_ack_ttl_hours."""
+    aid = uuid4()
+    db = _FakeDb(agent_uuid=aid)
+    before = datetime.now(UTC)
+    with _patch_db_context(db):
+        await svc.send_blocker_notification(
+            task_id="t1", blocker_reason="r", from_agent="system", to_pm="cell-pm"
+        )
+    after = datetime.now(UTC)
+    rows = [r for r in db.added if r.type == NotificationType.BLOCKER_ESCALATION]
+    assert rows
+    expires_at = rows[0].expires_at
+    assert expires_at is not None
+    ttl = timedelta(hours=settings.notification_ack_ttl_hours)
+    assert before + ttl <= expires_at <= after + ttl
+
+
+@pytest.mark.asyncio
+async def test_informational_notification_gets_no_expires_at(
+    svc: NotificationService,
+) -> None:
+    """A non-ack-required row (REVIEW_REQUEST) never gets a deadline — the
+    sweep only ever re-escalates ack-required rows, so stamping one would be
+    dead weight."""
+    aid = uuid4()
+    db = _FakeDb(agent_uuid=aid)
+    with _patch_db_context(db):
+        await svc.send_qa_ready_notification(
+            task_id="t1", from_agent="be-dev-1", to_qa="be-qa"
+        )
+    rows = [r for r in db.added if r.type == NotificationType.REVIEW_REQUEST]
+    assert rows
+    assert rows[0].expires_at is None
+
+
+@pytest.mark.asyncio
+async def test_ack_required_notification_expires_at_disabled_by_zero_ttl(
+    svc: NotificationService,
+) -> None:
+    """notification_ack_ttl_hours=0 disables stamping entirely (legacy: NULL,
+    never expires) even for an ack-required type."""
+    aid = uuid4()
+    db = _FakeDb(agent_uuid=aid)
+    with (
+        patch("roboco.services.notification.settings.notification_ack_ttl_hours", 0),
+        _patch_db_context(db),
+    ):
+        await svc.send_blocker_notification(
+            task_id="t1", blocker_reason="r", from_agent="system", to_pm="cell-pm"
+        )
+    rows = [r for r in db.added if r.type == NotificationType.BLOCKER_ESCALATION]
+    assert rows
+    assert rows[0].expires_at is None
+
+
 # ---------------------------------------------------------------------------
 # Coordination-event producers (reassignment / collision / unblock /
 # dependency-revival / stale-claim-reaped)

tests/unit/test_agents_config.py

@@ -243,7 +243,6 @@ def test_get_agent_skills_unknown_agent() -> None:
 def test_issue_agent_token_returns_unsigned_when_secret_missing(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-
     monkeypatch.delenv("ROBOCO_AGENT_AUTH_SECRET", raising=False)
     assert issue_agent_token("be-dev-1", "developer", "backend") == "UNSIGNED"
 
@@ -262,7 +261,6 @@ def test_issue_agent_token_signs_when_secret_present(
 
 
 def test_verify_agent_token_round_trips(monkeypatch: pytest.MonkeyPatch) -> None:
-
     monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "rt-secret")
     tok = issue_agent_token("be-dev-1", "developer", "backend")
     assert verify_agent_token(tok, "be-dev-1", "developer", "backend") is True
@@ -271,7 +269,6 @@ def test_verify_agent_token_round_trips(monkeypatch: pytest.MonkeyPatch) -> None
 def test_verify_agent_token_rejects_when_secret_missing(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-
     monkeypatch.delenv("ROBOCO_AGENT_AUTH_SECRET", raising=False)
     assert verify_agent_token("anything", "be-dev-1", "developer", "backend") is False
 
@@ -279,7 +276,6 @@ def test_verify_agent_token_rejects_when_secret_missing(
 def test_verify_agent_token_rejects_unsigned_sentinel(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-
     monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "any-secret")
     assert verify_agent_token("UNSIGNED", "be-dev-1", "developer", "backend") is False
 
@@ -287,7 +283,6 @@ def test_verify_agent_token_rejects_unsigned_sentinel(
 def test_verify_agent_token_rejects_empty_token(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-
     monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "any-secret")
     assert verify_agent_token("", "be-dev-1", "developer", "backend") is False
 
@@ -295,7 +290,6 @@ def test_verify_agent_token_rejects_empty_token(
 def test_verify_agent_token_rejects_mismatched_signature(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-
     monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", "real-secret")
     tok = issue_agent_token("be-dev-1", "developer", "backend")
     # Verify with different role → mismatch.
@@ -539,10 +533,10 @@ def test_get_a2a_route_hint_unknown_from_agent_falls_through() -> None:
 # A2A_ALLOWED_PAIRS — the switchboard's static org-chart pair matrix
 # ---------------------------------------------------------------------------
 
-_EXPECTED_PAIR_COUNT = 93
+_EXPECTED_PAIR_COUNT = 88
 _EXPECTED_GROUP_COUNTS = {
     "board": 3,
-    "ceo": 23,
+    "ceo": 18,
     "cell-backend": 15,
     "cell-frontend": 15,
     "cell-ux_ui": 15,
@@ -579,15 +573,23 @@ def test_a2a_allowed_pairs_excludes_non_participants_keeps_ceo() -> None:
     assert "ceo" in slugs
 
 
-def test_a2a_allowed_pairs_ceo_paired_with_every_agent() -> None:
-    """CEO → anyone is always allowed, so every non-CEO switchboard slug
+def test_a2a_allowed_pairs_ceo_paired_with_every_dm_capable_agent() -> None:
+    """CEO → anyone with an agent-comms surface is allowed, so every non-CEO
+    switchboard slug EXCEPT the no-comms roles (auditor, pr_reviewer — no
+    dm/read_a2a on their manifest, so a CEO DM to them is a black hole)
     appears in exactly one ``ceo``-group pair."""
     ceo_pairs = [p for p in A2A_ALLOWED_PAIRS if "ceo" in (p.agent_a, p.agent_b)]
     non_ceo_slugs = (
         {p.agent_a for p in A2A_ALLOWED_PAIRS} | {p.agent_b for p in A2A_ALLOWED_PAIRS}
     ) - {"ceo"}
+    dm_capable_slugs = {
+        s for s in non_ceo_slugs if get_agent_role(s) not in ("auditor", "pr_reviewer")
+    }
     assert all(p.group_key == "ceo" for p in ceo_pairs)
-    assert len(ceo_pairs) == len(non_ceo_slugs)
+    assert len(ceo_pairs) == len(dm_capable_slugs)
+    # And the no-comms roles are confirmed absent from any ceo-group pair.
+    ceo_slugs = {p.agent_a for p in ceo_pairs} | {p.agent_b for p in ceo_pairs}
+    assert ceo_slugs.isdisjoint(non_ceo_slugs - dm_capable_slugs)
 
 
 def test_a2a_allowed_pairs_group_key_counts() -> None: