Function build_task_notification_xml

src/utils/task_notification.py:53–105

Produce the `` `` envelope for a terminal task. Pure function — no registry mutation, no queue push. ``enqueue_agent_notification`` composes this with the WI-3.2 check-and-set; tests use it directly for snapshot comparisons. Format matches TS LocalAgentTask.tsx:25

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

def build_task_notification_xml(
    *,
    task_id: str,
    description: str,
    status: NotificationStatus,
    output_file: str,
    error: str | None = None,
    final_message: str | None = None,
    usage: dict[str, int] | None = None,
    tool_use_id: str | None = None,
) -> str:
    """Produce the ``<task-notification>`` envelope for a terminal task.

    Pure function — no registry mutation, no queue push. ``enqueue_agent_notification``
    composes this with the WI-3.2 check-and-set; tests use it directly
    for snapshot comparisons.

    Format matches TS LocalAgentTask.tsx:252-257 byte-for-byte (modulo
    optional sections that disappear when their inputs are absent). All
    values are rendered as-is (no XML escaping) — the chapter shape
    treats these as model-facing user content, and TS does the same
    raw concatenation. Callers are responsible for not embedding
    closing tags inside summary/result text.
    """
    summary = _build_summary(description, status, error)
    tool_use_line = (
        f"\n<{TOOL_USE_ID_TAG}>{tool_use_id}</{TOOL_USE_ID_TAG}>"
        if tool_use_id
        else ""
    )
    result_section = (
        f"\n<{RESULT_TAG}>{final_message}</{RESULT_TAG}>"
        if final_message
        else ""
    )
    if usage is not None:
        usage_section = (
            f"\n<{USAGE_TAG}>"
            f"<{TOTAL_TOKENS_TAG}>{usage.get('total_tokens', 0)}</{TOTAL_TOKENS_TAG}>"
            f"<{TOOL_USES_TAG}>{usage.get('tool_uses', 0)}</{TOOL_USES_TAG}>"
            f"<{DURATION_MS_TAG}>{usage.get('duration_ms', 0)}</{DURATION_MS_TAG}>"
            f"</{USAGE_TAG}>"
        )
    else:
        usage_section = ""
    return (
        f"<{TASK_NOTIFICATION_TAG}>\n"
        f"<{TASK_ID_TAG}>{task_id}</{TASK_ID_TAG}>{tool_use_line}\n"
        f"<{OUTPUT_FILE_TAG}>{output_file}</{OUTPUT_FILE_TAG}>\n"
        f"<{STATUS_TAG}>{status}</{STATUS_TAG}>\n"
        f"<{SUMMARY_TAG}>{summary}</{SUMMARY_TAG}>{result_section}{usage_section}\n"
        f"</{TASK_NOTIFICATION_TAG}>"
    )


def enqueue_agent_notification(