Function enqueue_agent_notification

src/utils/task_notification.py:108–164

Atomically check-and-set the ``notified`` flag, then enqueue. WI-3.2 contract: this is the single duplicate-delivery guard. The ``runtime_tasks.update`` mutator atomically reads ``prev.notified`` and (if False) returns a new state with ``notified=True``. The XML is enqueued only whe

(
    *,
    task_id: str,
    description: str,
    status: NotificationStatus,
    output_file: str,
    registry: "RuntimeTaskRegistry",
    error: str | None = None,
    final_message: str | None = None,
    usage: dict[str, int] | None = None,
    tool_use_id: str | None = None,
)

def enqueue_agent_notification(
    *,
    task_id: str,
    description: str,
    status: NotificationStatus,
    output_file: str,
    registry: "RuntimeTaskRegistry",
    error: str | None = None,
    final_message: str | None = None,
    usage: dict[str, int] | None = None,
    tool_use_id: str | None = None,
) -> bool:
    """Atomically check-and-set the ``notified`` flag, then enqueue.

    WI-3.2 contract: this is the single duplicate-delivery guard. The
    ``runtime_tasks.update`` mutator atomically reads ``prev.notified``
    and (if False) returns a new state with ``notified=True``. The XML
    is enqueued only when the mutator's pre-state had ``notified=False``.

    Returns True iff a notification was actually enqueued.

    Two concurrent callers on the same ``task_id`` race against the
    registry RLock; the second call sees ``notified=True`` and returns
    False without touching the queue.
    """
    # Local import to defer the cycle: ``local_agent`` imports
    # transcript stuff that imports back through the agent module.
    from src.tasks.local_agent import LocalAgentTaskState

    should_enqueue = False

    def _mark_notified(prev: Any) -> Any:
        nonlocal should_enqueue
        if not isinstance(prev, LocalAgentTaskState):
            return prev
        if prev.notified:
            return prev
        should_enqueue = True
        return replace(prev, notified=True)

    registry.update(task_id, _mark_notified)

    if not should_enqueue:
        return False

    xml = build_task_notification_xml(
        task_id=task_id,
        description=description,
        status=status,
        output_file=output_file,
        error=error,
        final_message=final_message,
        usage=usage,
        tool_use_id=tool_use_id,
    )
    enqueue_pending_notification(value=xml, mode="task-notification")
    return True