From 7b95d77d9051e3240a6cb0a73dc064116796091c Mon Sep 17 00:00:00 2001 From: Renn F Date: Mon, 13 Jul 2026 04:53:34 +0200 Subject: [PATCH] [77719d3f] Fix dependency-revival notification event loop mismatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 4 + roboco/services/notification.py | 192 ++++++++++++++++++-------------- roboco/services/task.py | 5 +- 3 files changed, 116 insertions(+), 85 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 02839280..95caabda 100644 --- a/pyproject.toml +++ b/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 diff --git a/roboco/services/notification.py b/roboco/services/notification.py index 3e3bfa7c..a841beb3 100644 --- a/roboco/services/notification.py +++ b/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: - """Create a notification via the database and deliver it.""" - async with get_db_context() as db: - from_agent_uuid = await _resolve_agent_uuid(db, params.from_agent) - if from_agent_uuid is None: - # notifications.from_agent is NOT NULL + FK to agents.id, so - # we cannot insert. Skip-with-warn rather than crash the - # upstream request. - logger.warning( - "Skipping notification: from_agent unresolvable", - from_agent_input=str(params.from_agent), - type=self._notification_type_label(params), - subject=params.subject[:80], - to_agents=[str(a) for a in params.to_agents], - ) - return - to_agents_uuids = await self._resolve_recipients(db, params) - if not to_agents_uuids: - logger.warning( - "Skipping notification: no resolvable recipients", - to_agents_input=[str(a) for a in params.to_agents], - type=self._notification_type_label(params), - subject=params.subject[:80], - ) - return - # Re-fire guard for loop-prone types: a 60s Redis SET-NX window - # coalesces the per-tick re-notify storm the DB dedup below skips - # (these types are requires_ack=False). Fail-open on Redis down. - 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( - "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), + async def _create_notification( + self, + params: CreateNotificationParams, + db_session: AsyncSession | None = None, + ) -> None: + """Create a notification via the database and deliver it. + + When ``db_session`` is provided, use it directly and skip the + internal commit — the caller owns the transaction. This is required + when the caller runs on a different event loop than the singleton + ``_DbHolder`` engine (e.g. ``TaskService`` called outside the FastAPI + request loop); opening ``get_db_context()`` there reuses an engine + whose asyncpg connections are bound to the server's loop, raising + ``Future attached to a different loop``. + """ + if db_session is not None: + await self._create_notification_with_session(params, db_session) + else: + async with get_db_context() as db: + await self._create_notification_with_session(params, db) + await db.commit() + + async def _create_notification_with_session( + self, params: CreateNotificationParams, db: AsyncSession + ) -> None: + from_agent_uuid = await _resolve_agent_uuid(db, params.from_agent) + if from_agent_uuid is None: + # notifications.from_agent is NOT NULL + FK to agents.id, so + # we cannot insert. Skip-with-warn rather than crash the + # upstream request. + logger.warning( + "Skipping notification: from_agent unresolvable", + from_agent_input=str(params.from_agent), + type=self._notification_type_label(params), + subject=params.subject[:80], + to_agents=[str(a) for a in params.to_agents], ) - db.add(notification) - await db.flush() - - # Deliver via Redis Streams for real-time push - from roboco.services.notification_delivery import ( - get_notification_delivery_service, + return + to_agents_uuids = await self._resolve_recipients(db, params) + if not to_agents_uuids: + logger.warning( + "Skipping notification: no resolvable recipients", + to_agents_input=[str(a) for a in params.to_agents], + type=self._notification_type_label(params), + subject=params.subject[:80], ) - - delivery_service = get_notification_delivery_service(db) - await delivery_service.deliver(require_uuid(notification.id)) - - await db.commit() - + return + # Re-fire guard for loop-prone types: a 60s Redis SET-NX window + # coalesces the per-tick re-notify storm the DB dedup below skips + # (these types are requires_ack=False). Fail-open on Redis down. + 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", - notification_id=str(notification.id), + "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) + 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, + ) diff --git a/roboco/services/task.py b/roboco/services/task.py index baaf86a8..c0c1486e 100644 --- a/roboco/services/task.py +++ b/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(