Function update_progress_from_message

src/tasks/progress.py:123–190

Fold one assistant message into the tracker. Mirrors ``updateProgressFromMessage`` in ``LocalAgentTask.tsx:68-96``. Updates token counts (per the cumulative-vs-per-turn split), pushes one ``ToolActivity`` per non-blacklisted ``tool_use`` block, and drops the oldest activity on cap-5

(
    tracker: ProgressTracker,
    message: "AssistantMessage",
    *,
    resolve_activity_description: ActivityDescriptionResolver | None = None,
)

def update_progress_from_message(
    tracker: ProgressTracker,
    message: "AssistantMessage",
    *,
    resolve_activity_description: ActivityDescriptionResolver | None = None,
) -> None:
    """Fold one assistant message into the tracker.

    Mirrors ``updateProgressFromMessage`` in ``LocalAgentTask.tsx:68-96``.
    Updates token counts (per the cumulative-vs-per-turn split), pushes
    one ``ToolActivity`` per non-blacklisted ``tool_use`` block, and
    drops the oldest activity on cap-5 overflow.

    Tolerant of missing/partial usage payloads — ``message.usage`` is
    ``dict[str, Any] | None`` per ``src.types.messages``, and a stub
    test message that omits it is a no-op for token accounting.
    """
    # Local imports defer the cycle: ``progress`` doesn't need to know
    # about message types at module load.
    from src.types.content_blocks import ToolUseBlock
    from src.types.messages import AssistantMessage

    if not isinstance(message, AssistantMessage):
        return

    usage = message.usage if isinstance(message.usage, dict) else None
    if usage is not None:
        # Cumulative-per-call inputs: include cache read/creation token
        # counts so the total matches what the API billed for.
        input_tokens = int(usage.get("input_tokens", 0) or 0)
        cache_creation = int(usage.get("cache_creation_input_tokens", 0) or 0)
        cache_read = int(usage.get("cache_read_input_tokens", 0) or 0)
        tracker.latest_input_tokens = input_tokens + cache_creation + cache_read

        tracker.cumulative_output_tokens += int(usage.get("output_tokens", 0) or 0)

    content = message.content
    if not isinstance(content, list):
        return

    for block in content:
        if not isinstance(block, ToolUseBlock):
            continue
        tracker.tool_use_count += 1
        if block.name in _PREVIEW_BLACKLIST:
            # Counted in tool_use_count (chapter does the same) but
            # omitted from the preview ring.
            continue
        block_input = getattr(block, "input", None)
        if not isinstance(block_input, dict):
            block_input = {}
        description: str | None = None
        if resolve_activity_description is not None:
            try:
                description = resolve_activity_description(block.name, block_input)
            except Exception:
                # A resolver bug must not poison the tracker.
                description = None