[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.
This commit is contained in:
Renn F
2026-07-13 04:53:34 +02:00
parent 617b7890e6
commit 7b95d77d90
3 changed files with 116 additions and 85 deletions
+4
View File
@@ -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
+30 -6
View File
@@ -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,9 +777,31 @@ class NotificationService:
return True
return False
async def _create_notification(self, params: CreateNotificationParams) -> None:
"""Create a notification via the database and deliver it."""
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
@@ -853,8 +879,6 @@ class NotificationService:
delivery_service = get_notification_delivery_service(db)
await delivery_service.deliver(require_uuid(notification.id))
await db.commit()
logger.info(
"Notification created and delivered",
notification_id=str(notification.id),
+4 -1
View File
@@ -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(