mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
[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
@@ -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
|
||||
|
||||
+108
-84
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user