Add batch and job progress aggregate to the test gatherer (#24774)

* Add batch and job progress aggregate to the test gatherer

The gatherer now publishes a DispatcherProgress -> BatchProgress ->
JobProgress -> JobAttemptProgress snapshot on every UpdatePRComment, so
the upcoming PR updater renders a complete immutable view of the run
without touching GitHub API models or pipeline internals.

The gatherer is initialized with the complete batch plan and emits
revision 0 through build_initial_update(), and TestBatch/BatchFinished
now carry an explicit batch_id instead of overloading BaseMessage.id as
the logical batch identity.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Add changelog entry

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Apply ruff formatting

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

* Address review themes from the gatherer PR

Derive done from the aggregate instead of a received-batch counter, so a
batch that reports no jobs no longer blocks completion, and key every
registry and the duplicate guard on batch_id rather than the message id.

Merge a finished batch into its registered plan instead of rebuilding it
from the message: jobs are preserved and executions are appended to each
job's history, which is what a failed-job rerun needs.

Model conclusion as WorkflowJobConclusion and errors as a ProgressError
enum, make the attempt status non-optional, drop the invented "timed out"
step name, and share one batch_status rule between the flat view and the
aggregate.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Carry only the snapshot on UpdatePRComment

The message shipped done and four counter properties that the progress
snapshot it also shipped already answered. Keep revision, which is
ordering metadata the snapshot deliberately does not carry, and progress,
which is everything else.

WorkflowStatus and JobResult stay: they are built from the same sources as
the aggregate rather than from it, and remain the gatherer's local registry
of what each batch reported. They are just no longer published.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Shorten the comments and docstrings

Same content, fewer words: multi-paragraph docstrings collapse to one or
two sentences and the three-line inline comments to one. No logic change.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Address the review comments

Take the batch status from the workflow instead of rolling it up from the
tracked jobs: a workflow also runs setup and finalization steps that can
fail while every job passes. That makes BatchFinished.status the source of
truth and leaves batch_status without a caller, so it goes and
WorkflowStatus.status returns to counting.

Validate the batch before gathering, so an unplanned or already-gathered
message cannot organize artifacts into the shared output tree, where the
names carry no batch and could overwrite a planned job's coverage.

Keep jobs the plan never mentioned out of the totals, log them instead.
Rename jobs to jobs_progress so it cannot be read as a tuple of BatchJob,
collapse the workflow_job conditionals, and cut the comments back.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: HadhemiDD

ddev/changelog.d/24774.added

@@ -0,0 +1 @@
+Adds the batch and job progress aggregate (`DispatcherProgress`) to the test gatherer.

ddev/src/ddev/cli/ci/tests/messages.py

@@ -15,6 +15,7 @@
 if TYPE_CHECKING:
     from pathlib import Path
 
+    from ddev.cli.ci.tests.progress import DispatcherProgress
     from ddev.utils.github_async.models import WorkflowJob
     from ddev.utils.junit import JUnitReport, JUnitTestCase
 
@@ -163,17 +164,27 @@ def status(self) -> Status:
 
 @dataclass
 class TestBatch(BaseMessage):
-    """Dispatched to trigger a matrix of test jobs."""
+    """Dispatched to trigger a matrix of test jobs.
 
+    ``batch_id`` is the logical batch identity (e.g. ``batch-01``): assigned during planning, stable
+    across workflow attempts, and distinct from ``BaseMessage.id``, which identifies one message.
+    """
+
+    batch_id: str
     job_list: list[BatchJob]
     jobs_count: int
     integrations: list[str]
 
 
 @dataclass
 class BatchFinished(BaseMessage):
-    """Emitted when a GitHub Actions test workflow has completed."""
+    """Emitted when a GitHub Actions test workflow has completed.
 
+    ``batch_id`` is the identity of the ``TestBatch`` this run came from, so the gatherer can resolve
+    it in the plan.
+    """
+
+    batch_id: str
     status: Status
     run_id: int
     workflow_url: str
@@ -186,28 +197,10 @@ class BatchFinished(BaseMessage):
 class UpdatePRComment(BaseMessage):
     """Emitted per finished batch to request a PR comment update.
 
-    ``revision`` is the gatherer's monotonic counter (one per consumed ``BatchFinished``); the
-    PR-updater renders the latest and rejects stale revisions. ``done`` is ``True`` only on the
-    revision that completes the final expected batch.
+    ``revision`` is ordering metadata: revision ``0`` is the initial plan, then one per consumed
+    ``BatchFinished``. The updater renders the latest and rejects stale revisions. ``progress`` is
+    the whole payload, including whether the run is done and every count the comment needs.
     """
 
     revision: int
-    done: bool
-    workflows: list[WorkflowStatus]
-
-    @property
-    def passed(self) -> int:
-        return sum(workflow.success_count for workflow in self.workflows)
-
-    @property
-    def failed(self) -> int:
-        return sum(workflow.failed_count for workflow in self.workflows)
-
-    @property
-    def skipped(self) -> int:
-        return sum(workflow.skipped_count for workflow in self.workflows)
-
-    @property
-    def complete(self) -> int:
-        """Total jobs finished so far (passed + failed + skipped across all gathered batches)."""
-        return sum(len(workflow.results) for workflow in self.workflows)
+    progress: DispatcherProgress

ddev/src/ddev/cli/ci/tests/progress.py

@@ -0,0 +1,161 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Aggregate state the gatherer owns and the PR updater renders.
+
+``DispatcherProgress`` -> ``BatchProgress`` -> ``JobProgress`` -> ``JobAttemptProgress``: batches,
+their planned jobs, and each job's executions in attempt order.
+
+``conclusion`` is the one GitHub value kept, as its enum, because it distinguishes outcomes
+(cancelled, timed out, action required) that ``Status`` collapses.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum, StrEnum, auto
+from typing import TYPE_CHECKING
+
+from ddev.cli.ci.tests.status import Status
+from ddev.utils.junit import TestStatus
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+    from ddev.cli.ci.tests.messages import BatchJob
+    from ddev.utils.github_async.models.workflow import WorkflowJobConclusion
+    from ddev.utils.junit import JUnitReport, JUnitTestCase
+
+
+class ProgressError(StrEnum):
+    """Why a batch or execution is unavailable, as a closed set to branch on rather than prose."""
+
+    TIMED_OUT = auto()
+    NO_JOB_RESULTS = auto()
+    NO_ARTIFACTS = auto()
+
+
+class ExecutionState(Enum):
+    """Where an execution is in its lifecycle, orthogonal to its outcome (``Status``).
+
+    ``RUNNING`` and ``RETRYING`` become reachable with the retry work.
+    """
+
+    PLANNED = "planned"
+    RUNNING = "running"
+    RETRYING = "retrying"
+    FINISHED = "finished"
+
+
+@dataclass(frozen=True)
+class JobAttemptProgress:
+    """One observed execution of one planned job.
+
+    An attempt exists only because the job ran, so ``status`` is always known: an undetermined job
+    has no attempt. ``attempt`` is its 1-based position in the job's history. ``job_id``,
+    ``conclusion`` and ``job_url`` are ``None`` when GitHub never reported the job.
+    """
+
+    attempt: int
+    job_id: int | None
+    status: Status
+    conclusion: WorkflowJobConclusion | None
+    failed_steps: tuple[str, ...]
+    job_url: str | None
+    reports: tuple[JUnitReport, ...]
+    error: ProgressError | None = None
+
+    @property
+    def failed_tests(self) -> list[JUnitTestCase]:
+        """Every failed/errored test case across this execution's reports."""
+        return [
+            case
+            for report in self.reports
+            for suite in report.test_suites
+            for case in suite.test_cases
+            if case.status in (TestStatus.FAILED, TestStatus.ERROR)
+        ]
+
+
+@dataclass(frozen=True)
+class JobProgress:
+    """One logical planned job throughout execution.
+
+    ``attempts`` can be sparse: a job that already succeeded does not run again in a failed-job rerun.
+    """
+
+    job: BatchJob
+    attempts: tuple[JobAttemptProgress, ...]
+
+    @property
+    def latest(self) -> JobAttemptProgress | None:
+        """The most recent execution: the only attempt that counts toward totals."""
+        return self.attempts[-1] if self.attempts else None
+
+    @property
+    def retry_count(self) -> int:
+        """Executions minus one. Not ``run_attempt - 1``: histories can be sparse."""
+        return max(0, len(self.attempts) - 1)
+
+
+@dataclass(frozen=True)
+class BatchProgress:
+    """One logical batch, from planning through its terminal outcome.
+
+    ``status`` is the workflow's own, not a roll-up of ``jobs_progress``: a workflow can fail in a
+    step no tracked job covers. It stays ``None`` until ``FINISHED``.
+    """
+
+    batch_id: str
+    run_id: int | None
+    workflow_url: str | None
+    state: ExecutionState
+    status: Status | None
+    current_attempt: int | None
+    max_attempts: int
+    retries_remaining: int
+    retrying_jobs: tuple[BatchJob, ...]
+    jobs_progress: tuple[JobProgress, ...]
+    error: ProgressError | None = None
+
+
+@dataclass(frozen=True)
+class DispatcherProgress:
+    """The complete point-in-time aggregate across all batches.
+
+    Carries all state known at its revision, so the PR updater renders the newest snapshot and
+    discards the rest. The counters derive from ``JobProgress.latest`` only, so a job retried to
+    success counts once.
+    """
+
+    batches: tuple[BatchProgress, ...]
+    done: bool
+
+    @property
+    def passed(self) -> int:
+        return self._count(Status.SUCCESS)
+
+    @property
+    def failed(self) -> int:
+        return self._count(Status.FAILURE)
+
+    @property
+    def skipped(self) -> int:
+        return self._count(Status.SKIPPED)
+
+    @property
+    def complete(self) -> int:
+        """Planned jobs that have run."""
+        return sum(1 for job in self._jobs_progress if job.latest is not None)
+
+    @property
+    def total(self) -> int:
+        """Every planned job, run or not."""
+        return sum(1 for _ in self._jobs_progress)
+
+    @property
+    def _jobs_progress(self) -> Iterator[JobProgress]:
+        return (job for batch in self.batches for job in batch.jobs_progress)
+
+    def _count(self, status: Status) -> int:
+        return sum(1 for job in self._jobs_progress if job.latest is not None and job.latest.status == status)