src/task_progress.py

"""Azure timeline snapshots shared by the dispatcher and queue UI.

Only completed pipeline steps are counted; this is not iteration/sample progress.
Network reads happen in the dispatcher, never in a browser refresh callback.
"""

from html import escape
from threading import Lock
from time import monotonic

import requests

_snapshots = {}
_lock = Lock()
_MAX_AGE = 180


def refresh_progress(run_id, url, headers):
    """Fetch a timeline without allowing an unavailable API to affect dispatch."""
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        records = response.json().get("records", [])
        tasks = [r for r in records if r.get("type") == "Task"]
        if not tasks:
            return
        # A retried task's latest attempt is the authoritative one.
        latest = {}
        for task in tasks:
            key = task.get("identifier") or task.get("id")
            if key not in latest or task.get("attempt", 1) >= latest[key].get("attempt", 1):
                latest[key] = task
        tasks = list(latest.values())
        completed = sum(t.get("state") == "completed" for t in tasks)
        active = [str(t.get("name", "Running")) for t in tasks if t.get("state") == "inProgress"]
        snapshot = (monotonic(), completed, len(tasks), ", ".join(active))
        with _lock:
            for key in list(_snapshots):
                if monotonic() - _snapshots[key][0] > _MAX_AGE:
                    del _snapshots[key]
            _snapshots[str(run_id)] = snapshot
    except (requests.RequestException, ValueError, TypeError, AttributeError):
        return


def render_progress(entry):
    """Render escaped, accessible progress; never invent a percentage."""
    status = entry.get("status", "Pending")
    if status.startswith("Finished"):
        return '<progress max="100" value="100" aria-label="Completed"></progress> Completed'
    if status not in {"Running", "Triggered", "Waiting"}:
        return escape(status)
    with _lock:
        snapshot = _snapshots.get(str(entry.get("ci_run_id", "")))
    if not snapshot or monotonic() - snapshot[0] > _MAX_AGE:
        return '<progress aria-label="Waiting for progress information"></progress> Waiting for progress information'
    _, completed, total, active = snapshot
    label = f"Pipeline steps: {completed}/{total} completed"
    detail = active or ("Waiting for results" if completed == total else "Waiting for next step")
    return (
        f'<progress max="{total}" value="{completed}" aria-label="{label}"></progress>'
        f' <span>{label} — {escape(detail)}</span>'
    )