Function enqueue_workflow_notification

src/tasks/local_workflow.py:210–265

Deliver a workflow's terminal result to the model via the shared `` `` queue (mirrors ``enqueue_agent_notification``). The ``notified`` flag is check-and-set atomically, so exactly one envelope is delivered even if complete/fail/kill race.

(
    task_id: str,
    registry: "RuntimeTaskRegistry",
    *,
    status: str,
    error: str | None = None,
)

def enqueue_workflow_notification(
    task_id: str,
    registry: "RuntimeTaskRegistry",
    *,
    status: str,
    error: str | None = None,
) -> bool:
    """Deliver a workflow's terminal result to the model via the shared
    ``<task-notification>`` queue (mirrors ``enqueue_agent_notification``).

    The ``notified`` flag is check-and-set atomically, so exactly one envelope
    is delivered even if complete/fail/kill race."""
    from src.utils.message_queue_manager import enqueue_pending_notification
    from src.utils.task_notification import build_task_notification_xml

    # Read the token total OUTSIDE the registry mutator: progress is mutated by
    # the engine on its own daemon thread, so iterating it under the registry
    # lock could raise "list changed size during iteration". _safe_token_total
    # guards it; the mutator itself only flips a scalar flag (the stock-agent
    # pattern in enqueue_agent_notification).
    snapshot = registry.get(task_id)
    tokens = _safe_token_total(getattr(snapshot, "progress", None)) if snapshot is not None else 0

    captured: dict[str, Any] = {}
    should_enqueue = False

    def _mark(prev: TaskStateBase) -> TaskStateBase:
        nonlocal should_enqueue
        if not isinstance(prev, LocalWorkflowTaskState) or prev.notified:
            return prev
        should_enqueue = True
        captured.update(
            name=prev.workflow_name or "workflow",
            output_file=prev.output_file,
            result=prev.result,
            tool_use_id=prev.tool_use_id,
        )
        return replace(prev, notified=True)

    registry.update(task_id, _mark)
    if not should_enqueue:
        return False

    final_message = _render_result(captured["result"]) if status == "completed" else None
    xml = build_task_notification_xml(
        task_id=task_id,
        description=captured["name"],
        status=status,  # type: ignore[arg-type]
        output_file=captured["output_file"],
        error=error,
        final_message=final_message,
        usage={"total_tokens": tokens, "tool_uses": 0, "duration_ms": 0},
        tool_use_id=captured["tool_use_id"],
    )
    enqueue_pending_notification(value=xml, mode="task-notification")
    return True