[pull] master from DataDog:master

ddev/changelog.d/23722.added

@@ -0,0 +1 @@
+Add `ddev release test-agent` command that dispatches the Linux and Windows Agent test workflows against a release branch or tag.

ddev/changelog.d/23860.changed

@@ -0,0 +1 @@
+`ddev release branch tag` now accepts `--release/-r`, `--ref`, `--rc N`, and `--yes/-y`, and prompts to confirm when run on a release branch without `--release`. Existing non-interactive callers that piped a single `y` need to pass `--yes` or one extra confirmation.

ddev/changelog.d/24253.fixed

@@ -0,0 +1 @@
+Bundle per-job correlation into the CI dispatcher's BatchFinished via BatchJobResult.

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

@@ -3,11 +3,23 @@
 # Licensed under a 3-clause BSD style license (see LICENSE)
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import Literal
+import re
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Literal
 
 from ddev.event_bus.orchestrator import BaseMessage
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from ddev.utils.github_async.models import WorkflowJob
+
+# Characters GitHub disallows in an artifact name (plus CR/LF).
+ARTIFACT_NAME_DISALLOWED = re.compile(r'["\:<>|*?\\/\r\n]')
+# Separator between the artifact name's fields. Names are matched by reconstruction, not by
+# splitting, so the separator does not need to be absent from the field values.
+ARTIFACT_NAME_SEPARATOR = "_"
+
 
 @dataclass
 class BatchJob:
@@ -21,6 +33,16 @@ class BatchJob:
     unit_tests: bool
     e2e_tests: bool
 
+    def artifact_name(self) -> str:
+        """Deterministic artifact name built from the job's target, environment, and platform.
+
+        Pure and deterministic. Each field is sanitized to GitHub's artifact-name constraints and
+        joined by the separator. Uniqueness within a batch relies on those three fields being
+        distinct per job.
+        """
+        fields = (self.target, self.environment, self.platform)
+        return ARTIFACT_NAME_SEPARATOR.join(ARTIFACT_NAME_DISALLOWED.sub("_", field) for field in fields)
+
 
 @dataclass
 class FailedCheck:
@@ -30,6 +52,57 @@ class FailedCheck:
     url: str
 
 
+@dataclass
+class BatchJobResult:
+    """Everything known about a single job in a finished batch, correlated by the producer.
+
+    ``artifact_name_path`` is the single downloaded folder for the job (named after the job's
+    ``artifact_name``); the three ``*_artifact_name`` fields are the expected per-facet file names
+    inside that folder.
+    """
+
+    job: BatchJob
+    workflow_job: WorkflowJob | None
+    artifact_name_path: str | None
+    unit_artifact_name: str
+    e2e_artifact_name: str
+    coverage_artifact_name: str
+
+    @staticmethod
+    def correlate(
+        job_list: list[BatchJob],
+        jobs: list[WorkflowJob],
+        artifact_dirs: dict[str, Path],
+    ) -> list[BatchJobResult]:
+        """Correlate each job's spec, its workflow-run result, and its artifact directory.
+
+        The workflow-job join is by name (tolerant of misses). Each job's artifact folder is matched
+        by reconstructing its name from the job's fields (``artifact_name``) and looking it up among
+        the downloaded folders; the path is recorded only when it exists on disk. That single folder
+        holds the three per-facet files, whose names (``unit-``/``e2e-``/``coverage-`` prefixed on
+        the base name) are recorded for the gatherer. A job missing from the API or from disk still
+        yields a well-formed result.
+        """
+        jobs_by_name = {job.name: job for job in jobs}
+
+        results: list[BatchJobResult] = []
+        for batch_job in job_list:
+            base = batch_job.artifact_name()
+            artifact_dir = artifact_dirs.get(base)
+            artifact_name_path = str(artifact_dir) if artifact_dir is not None and artifact_dir.exists() else None
+            results.append(
+                BatchJobResult(
+                    job=batch_job,
+                    workflow_job=jobs_by_name.get(batch_job.name),
+                    artifact_name_path=artifact_name_path,
+                    unit_artifact_name=f"unit-{base}",
+                    e2e_artifact_name=f"e2e-{base}",
+                    coverage_artifact_name=f"coverage-{base}",
+                )
+            )
+        return results
+
+
 @dataclass
 class WorkflowStatus:
     """Status of a single GitHub Actions workflow run."""
@@ -58,6 +131,8 @@ class BatchFinished(BaseMessage):
     run_id: int
     workflow_url: str
     artifacts_path: str
+    timed_out: bool = False
+    batch_jobs: list[BatchJobResult] = field(default_factory=list)
 
 
 @dataclass

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

@@ -11,10 +11,10 @@
 from pathlib import Path
 from typing import Any, Literal
 
-from ddev.cli.ci.tests.messages import BatchFinished, TestBatch
+from ddev.cli.ci.tests.messages import BatchFinished, BatchJob, BatchJobResult, TestBatch
 from ddev.event_bus.orchestrator import AsyncProcessor
 from ddev.utils.github_async import AsyncGitHubClient, GitHubResponse
-from ddev.utils.github_async.models import WorkflowRun
+from ddev.utils.github_async.models import WorkflowJob, WorkflowRun
 
 
 def _conclusion_to_status(conclusion: str | None) -> Literal["success", "failure", "skipped"]:
@@ -63,7 +63,12 @@ async def process_message(self, message: TestBatch) -> None:
         log_extra: dict[str, Any] = {"batch_id": message.id}
 
         dispatch = await self._client.create_workflow_dispatch(
-            self._options.owner, self._options.repo, self._options.workflow_id, ref=self._options.ref, inputs=inputs
+            self._options.owner,
+            self._options.repo,
+            self._options.workflow_id,
+            ref=self._options.ref,
+            inputs=inputs,
+            return_run_details=True,
         )
         run_id = dispatch.data.workflow_run_id
         log_extra["run_id"] = run_id
@@ -98,15 +103,19 @@ async def process_message(self, message: TestBatch) -> None:
                 self._logger.warning("Workflow completed with null conclusion", extra=log_extra)
             final_conclusion = raw or "neutral"
 
-            artifacts_path = await self._download_artifacts(run_id, log_extra)
+            artifact_dirs = await self._download_artifacts(run_id, log_extra)
             self._logger.info("Artifacts downloaded", extra=log_extra)
 
+            jobs = await self._list_jobs(run_id, log_extra)
+            batch_jobs = BatchJobResult.correlate(message.job_list, jobs, artifact_dirs)
+
             finished = BatchFinished(
                 id=message.id,
                 status=_conclusion_to_status(raw),
                 run_id=run_id,
                 workflow_url=workflow_url,
-                artifacts_path=str(artifacts_path),
+                artifacts_path=str(self._options.artifacts_base_path),
+                batch_jobs=batch_jobs,
             )
         finally:
             try:
@@ -134,16 +143,38 @@ async def _poll_until_complete(self, run_id: int, log_extra: dict[str, Any]) ->
                 self._logger.info("Workflow completed", extra=log_extra)
                 return run
 
+    async def _list_jobs(self, run_id: int, log_extra: dict[str, Any]) -> list[WorkflowJob]:
+        """Fetch the workflow run's jobs; on failure log a warning and return an empty list."""
+        jobs: list[WorkflowJob] = []
+        try:
+            async for page in self._client.list_workflow_jobs(self._options.owner, self._options.repo, run_id):
+                jobs.extend(page.data.jobs)
+        except Exception:
+            self._logger.warning("Failed to list workflow jobs", extra=log_extra, exc_info=True)
+        return jobs
+
     def _build_inputs(self, message: TestBatch) -> dict[str, str]:
         return {
             "batch_id": message.id,
             "checkout_sha": self._options.checkout_sha,
             "integrations": json.dumps(message.integrations),
-            "job_list": json.dumps([dataclasses.asdict(job) for job in message.job_list]),
+            "job_list": json.dumps([self._job_input(job) for job in message.job_list]),
         }
 
-    async def _download_artifacts(self, run_id: int, log_extra: dict[str, Any]) -> Path:
-        run_path = self._options.artifacts_base_path / str(run_id)
+    @staticmethod
+    def _job_input(job: BatchJob) -> dict[str, Any]:
+        """Serialize a job for the workflow, carrying the artifact name so all its files upload under
+        a single folder/zip named after it (matched later via ``BatchJob.artifact_name``)."""
+        return {**dataclasses.asdict(job), "artifact_name": job.artifact_name()}
+
+    async def _download_artifacts(self, run_id: int, log_extra: dict[str, Any]) -> dict[str, Path]:
+        """Download the run's artifacts and return an artifact-name -> path map.
+
+        The map keys on the GitHub artifact name (the contract a ``BatchJob`` reproduces via
+        ``artifact_name``), letting the producer resolve each job's directory deterministically.
+        """
+        base_path = self._options.artifacts_base_path
+        artifact_dirs: dict[str, Path] = {}
         failures: list[tuple[int, str]] = []
         try:
             async for page in self._client.list_workflow_run_artifacts(self._options.owner, self._options.repo, run_id):
@@ -164,9 +195,10 @@ async def _download_artifacts(self, run_id: int, log_extra: dict[str, Any]) -> P
                             extra=log_extra,
                         )
                         continue
-                    target = run_path / f"{artifact.id}-{artifact.name}"
+                    target = base_path / artifact.name
                     try:
                         await self._client.download_artifact(artifact.archive_download_url, target)
+                        artifact_dirs[artifact.name] = target
                         self._logger.info("Downloaded artifact %s -> %s", artifact.id, target, extra=log_extra)
                     except Exception as exc:
                         self._logger.warning(
@@ -186,4 +218,4 @@ async def _download_artifacts(self, run_id: int, log_extra: dict[str, Any]) -> P
                 failures,
                 extra=log_extra,
             )
-        return run_path
+        return artifact_dirs

ddev/src/ddev/cli/release/__init__.py

@@ -14,6 +14,7 @@
 from ddev.cli.release.port_commit import port_commit
 from ddev.cli.release.show import show
 from ddev.cli.release.stats import stats
+from ddev.cli.release.test_agent import test_agent
 
 
 @click.group(short_help='Manage the release of integrations')
@@ -33,4 +34,5 @@ def release():
 release.add_command(show)
 release.add_command(stats)
 release.add_command(tag)
+release.add_command(test_agent)
 release.add_command(upload)