[77719d3f] Fix dependency-revival notification event loop mismatch

The dependency-revival test calls _unblock_dependents directly via
stack.run_db, which creates a new asyncio event loop. Inside,
_notify_dependency_revival -> NotificationService._create_notification
opened its own session via get_db_context(), which reuses the singleton
_DbHolder engine — bound to the FastAPI server's event loop. The
asyncpg connection raised 'Future attached to a different loop' and the
exception was silently caught + logged as a warning, so the notification
never persisted and the test saw 0 rows.

Fix: add an optional db_session parameter to _create_notification and
the two send methods. When provided, use the caller's session directly
and skip the internal commit (the caller owns the transaction). The
TaskService's _notify_unblock and _notify_dependency_revival now pass
self.session, keeping the notification in the same event loop + session
as the task transition.

pyproject.toml

@@ -154,6 +154,10 @@ select = [
 # agent_id, project_ids, route, session_id) — same >5-kwarg rationale as the
 # gateway verb surfaces below.
 "roboco/services/prompter.py" = ["PLR0913"]
+# send_dependency_revival_notification carries the coordination-event contract
+# (task_id, assignee, completed_dependency_id, from_agent, to_ceo, db_session) —
+# db_session is the caller's session for event-loop-safe notification creation.
+"roboco/services/notification.py" = ["PLR0913"]
 # open_video_task's kwargs (occasion, script, platforms, brief,
 # suggested_input_props, project_id) are the authoring-task contract shared
 # by the release/spotlight/on-demand callers — same "bundling would just

roboco/services/notification.py

@@ -442,6 +442,7 @@ class NotificationService:
         restored_owner: str | None,
         from_agent: str | None = None,
         to_ceo: str = "ceo",
+        db_session: AsyncSession | None = None,
     ) -> None:
         """Tell the restored owner (+ CEO) a blocked task is workable again.
 
@@ -471,7 +472,8 @@ class NotificationService:
                 subject=f"Task {task_id} unblocked",
                 body=body,
                 related_task_id=task_id,
-            )
+            ),
+            db_session=db_session,
         )
 
     async def send_dependency_revival_notification(
@@ -481,6 +483,7 @@ class NotificationService:
         completed_dependency_id: str,
         from_agent: str | None = None,
         to_ceo: str = "ceo",
+        db_session: AsyncSession | None = None,
     ) -> None:
         """Tell the revived task's owner (+ CEO) its last dependency landed.
 
@@ -517,7 +520,8 @@ class NotificationService:
                 subject=f"Task {task_id} revived by dependency completion",
                 body=body,
                 related_task_id=task_id,
-            )
+            ),
+            db_session=db_session,
         )
 
     async def send_stale_claim_reaped_notification(
@@ -773,90 +777,110 @@ class NotificationService:
                 return True
         return False
 
-    async def _create_notification(self, params: CreateNotificationParams) -> None:
+    async def _create_notification(
-        """Create a notification via the database and deliver it."""
+        self,
-        async with get_db_context() as db:
+        params: CreateNotificationParams,
-            from_agent_uuid = await _resolve_agent_uuid(db, params.from_agent)
+        db_session: AsyncSession | None = None,
-            if from_agent_uuid is None:
+    ) -> None:
-                # notifications.from_agent is NOT NULL + FK to agents.id, so
+        """Create a notification via the database and deliver it.
-                # we cannot insert. Skip-with-warn rather than crash the
+
-                # upstream request.
+        When ``db_session`` is provided, use it directly and skip the
-                logger.warning(
+        internal commit — the caller owns the transaction. This is required
-                    "Skipping notification: from_agent unresolvable",
+        when the caller runs on a different event loop than the singleton
-                    from_agent_input=str(params.from_agent),
+        ``_DbHolder`` engine (e.g. ``TaskService`` called outside the FastAPI
-                    type=self._notification_type_label(params),
+        request loop); opening ``get_db_context()`` there reuses an engine
-                    subject=params.subject[:80],
+        whose asyncpg connections are bound to the server's loop, raising
-                    to_agents=[str(a) for a in params.to_agents],
+        ``Future attached to a different loop``.
-                )
+        """
-                return
+        if db_session is not None:
-            to_agents_uuids = await self._resolve_recipients(db, params)
+            await self._create_notification_with_session(params, db_session)
-            if not to_agents_uuids:
+        else:
-                logger.warning(
+            async with get_db_context() as db:
-                    "Skipping notification: no resolvable recipients",
+                await self._create_notification_with_session(params, db)
-                    to_agents_input=[str(a) for a in params.to_agents],
+                await db.commit()
-                    type=self._notification_type_label(params),
+
-                    subject=params.subject[:80],
+    async def _create_notification_with_session(
-                )
+        self, params: CreateNotificationParams, db: AsyncSession
-                return
+    ) -> None:
-            # Re-fire guard for loop-prone types: a 60s Redis SET-NX window
+        from_agent_uuid = await _resolve_agent_uuid(db, params.from_agent)
-            # coalesces the per-tick re-notify storm the DB dedup below skips
+        if from_agent_uuid is None:
-            # (these types are requires_ack=False). Fail-open on Redis down.
+            # notifications.from_agent is NOT NULL + FK to agents.id, so
-            if await all_recipients_recently_notified(
+            # we cannot insert. Skip-with-warn rather than crash the
-                ntype=params.notification_type,
+            # upstream request.
-                from_agent=from_agent_uuid,
+            logger.warning(
-                recipients=to_agents_uuids,
+                "Skipping notification: from_agent unresolvable",
-                related_task_id=params.related_task_id,
+                from_agent_input=str(params.from_agent),
-                subject=params.subject,
+                type=self._notification_type_label(params),
-            ):
+                subject=params.subject[:80],
-                logger.info(
+                to_agents=[str(a) for a in params.to_agents],
-                    "Suppressed re-fire notification (loop-prone, recent window)",
-                    from_agent=str(from_agent_uuid),
-                    type=params.notification_type.value,
-                    related_task_id=str(params.related_task_id)
-                    if params.related_task_id is not None
-                    else None,
-                    to_agents=[str(a) for a in to_agents_uuids],
-                )
-                return
-            # Purpose-based dedup (CEO directive, 2026-06-10): suppress a second
-            # notification for the SAME purpose while a prior one is unacked. See
-            # ``_duplicate_unacked_exists`` for the rationale + the action-only
-            # scope (informational types carry distinct content per send).
-            if await self._duplicate_unacked_exists(
-                db,
-                from_agent_uuid=from_agent_uuid,
-                params=params,
-                to_agents_uuids=to_agents_uuids,
-            ):
-                return
-            notification = NotificationTable(
-                type=params.notification_type,
-                priority=params.priority,
-                from_agent=from_agent_uuid,
-                to_agents=to_agents_uuids,
-                subject=params.subject,
-                body=params.body,
-                related_task_id=params.related_task_id,
-                # requires_ack follows ACK_REQUIRED_BY_TYPE (action-required vs
-                # informational), not the column's True default; default True
-                # for an unmapped type preserves the safe action-required bias.
-                requires_ack=ACK_REQUIRED_BY_TYPE.get(params.notification_type, True),
             )
-            db.add(notification)
+            return
-            await db.flush()
+        to_agents_uuids = await self._resolve_recipients(db, params)
-
+        if not to_agents_uuids:
-            # Deliver via Redis Streams for real-time push
+            logger.warning(
-            from roboco.services.notification_delivery import (
+                "Skipping notification: no resolvable recipients",
-                get_notification_delivery_service,
+                to_agents_input=[str(a) for a in params.to_agents],
+                type=self._notification_type_label(params),
+                subject=params.subject[:80],
             )
-
+            return
-            delivery_service = get_notification_delivery_service(db)
+        # Re-fire guard for loop-prone types: a 60s Redis SET-NX window
-            await delivery_service.deliver(require_uuid(notification.id))
+        # coalesces the per-tick re-notify storm the DB dedup below skips
-
+        # (these types are requires_ack=False). Fail-open on Redis down.
-            await db.commit()
+        if await all_recipients_recently_notified(
-
+            ntype=params.notification_type,
+            from_agent=from_agent_uuid,
+            recipients=to_agents_uuids,
+            related_task_id=params.related_task_id,
+            subject=params.subject,
+        ):
             logger.info(
-                "Notification created and delivered",
+                "Suppressed re-fire notification (loop-prone, recent window)",
-                notification_id=str(notification.id),
+                from_agent=str(from_agent_uuid),
                 type=params.notification_type.value,
+                related_task_id=str(params.related_task_id)
+                if params.related_task_id is not None
+                else None,
+                to_agents=[str(a) for a in to_agents_uuids],
             )
+            return
+        # Purpose-based dedup (CEO directive, 2026-06-10): suppress a second
+        # notification for the SAME purpose while a prior one is unacked. See
+        # ``_duplicate_unacked_exists`` for the rationale + the action-only
+        # scope (informational types carry distinct content per send).
+        if await self._duplicate_unacked_exists(
+            db,
+            from_agent_uuid=from_agent_uuid,
+            params=params,
+            to_agents_uuids=to_agents_uuids,
+        ):
+            return
+        notification = NotificationTable(
+            type=params.notification_type,
+            priority=params.priority,
+            from_agent=from_agent_uuid,
+            to_agents=to_agents_uuids,
+            subject=params.subject,
+            body=params.body,
+            related_task_id=params.related_task_id,
+            # requires_ack follows ACK_REQUIRED_BY_TYPE (action-required vs
+            # informational), not the column's True default; default True
+            # for an unmapped type preserves the safe action-required bias.
+            requires_ack=ACK_REQUIRED_BY_TYPE.get(params.notification_type, True),
+        )
+        db.add(notification)
+        await db.flush()
+
+        # Deliver via Redis Streams for real-time push
+        from roboco.services.notification_delivery import (
+            get_notification_delivery_service,
+        )
+
+        delivery_service = get_notification_delivery_service(db)
+        await delivery_service.deliver(require_uuid(notification.id))
+
+        logger.info(
+            "Notification created and delivered",
+            notification_id=str(notification.id),
+            type=params.notification_type.value,
+        )

roboco/services/task.py

@@ -4757,7 +4757,9 @@ class TaskService(BaseService):
             from roboco.services.notification import NotificationService
 
             await NotificationService().send_unblock_notification(
-                task_id=str(task_id), restored_owner=str(restored_owner)
+                task_id=str(task_id),
+                restored_owner=str(restored_owner),
+                db_session=self.session,
             )
         except Exception as e:
             self.log.warning(
@@ -7011,6 +7013,7 @@ class TaskService(BaseService):
                 task_id=str(task.id),
                 assignee=str(owner),
                 completed_dependency_id=str(completed_dependency_id),
+                db_session=self.session,
             )
         except Exception as e:
             self.log.warning(