diff --git a/ddev/changelog.d/25317.added b/ddev/changelog.d/25317.added
new file mode 100644
index 0000000000000..0c78a89556af2
--- /dev/null
+++ b/ddev/changelog.d/25317.added
@@ -0,0 +1 @@
+Emit Dispatcher execution and component-operation metrics, including workflow and artifact-download durations, launched-job totals, and incomplete results.
diff --git a/ddev/src/ddev/cli/ci/dispatch_run.py b/ddev/src/ddev/cli/ci/dispatch_run.py
index d43eabbbd962b..ab0e96f5aea90 100644
--- a/ddev/src/ddev/cli/ci/dispatch_run.py
+++ b/ddev/src/ddev/cli/ci/dispatch_run.py
@@ -136,6 +136,19 @@ class ResolvedRun(BaseModel):
     base_sha: str | None = None
     is_fork: bool = False
 
+    @property
+    def owner(self) -> str:
+        return self.repository.partition('/')[0]
+
+    @property
+    def repo(self) -> str:
+        return self.repository.partition('/')[2]
+
+    @property
+    def concurrency_key(self) -> str:
+        """New PR revisions must cancel old batches, so they key on the PR, not the merge SHA."""
+        return f'pr-{self.pr_number}' if self.pr_number is not None else self.head_sha
+
 
 def write_run_manifest(base_path: Path, *, run: ResolvedRun) -> None:
     """Write the resolved run's manifest as machine-readable JSON under its output directory.
diff --git a/ddev/src/ddev/cli/ci/dispatch_tests.py b/ddev/src/ddev/cli/ci/dispatch_tests.py
index 111643f06b32b..05d804190f291 100644
--- a/ddev/src/ddev/cli/ci/dispatch_tests.py
+++ b/ddev/src/ddev/cli/ci/dispatch_tests.py
@@ -5,6 +5,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import time
 from pathlib import Path
@@ -20,11 +21,15 @@
     resolve_run,
     write_run_manifest,
 )
+from ddev.cli.ci.tests.execution_metrics import ExecutionOutcome
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
     from ddev.cli.application import Application
+    from ddev.cli.ci.dispatch_run import ResolvedRun
     from ddev.cli.ci.tests.batching.units import EnvironmentProvider
-    from ddev.cli.ci.tests.dispatcher import Dispatcher, DispatcherContext
+    from ddev.cli.ci.tests.dispatcher import Dispatcher
     from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig
     from ddev.cli.ci.tests.messages import TestBatch
     from ddev.monitoring import ComponentMonitor, MonitoringRuntime
@@ -80,7 +85,7 @@
     '--tags',
     default=None,
     metavar='"KEY:VALUE ..."',
-    help='Tags the run reports itself under, separated by spaces. Their meaning is the caller\'s to decide.',
+    help='Additional run tags, separated by spaces. Service ownership and resolved identity cannot be overridden.',
 )
 @click.option(
     '--pytest-args',
@@ -178,10 +183,13 @@ def dispatch_tests(
     )
 
     from ddev.cli.application import AppLoggingHandler
+    from ddev.cli.ci.tests.batching.exceptions import PlanningError
     from ddev.cli.ci.tests.batching.hatch_environments import HatchEnvironmentProvider
-    from ddev.cli.ci.tests.dispatcher import DispatcherContext, build_dispatcher
+    from ddev.cli.ci.tests.dispatcher import build_dispatcher
     from ddev.cli.ci.tests.dispatcher_attributes import (
+        BASE_FIELDS,
         PROTECTED_RUN_FIELDS,
+        UNRESOLVED_RUN_FIELDS,
         console_hidden_fields,
         metric_tag_mapping,
         repository_fields,
@@ -208,15 +216,23 @@ def dispatch_tests(
         metrics_sink=metrics_sink,
         metrics_tag_projector=metric_tag_mapping,
         protected_fields=PROTECTED_RUN_FIELDS,
+        base_fields={
+            **{name: value for name, value in tag_fields(caller_tags).items() if name not in PROTECTED_RUN_FIELDS},
+            **BASE_FIELDS,
+            **repository_fields(owner, repo),
+            **UNRESOLVED_RUN_FIELDS,
+        },
     )
-    monitoring.set_run_fields(**{**tag_fields(caller_tags), **repository_fields(owner, repo)})
     datadog_handler = attach_datadog_log_handler(app, monitoring, level=output_level)
     started = time.monotonic()
     monitor = monitoring.component('dispatcher')
+    outcome: ExecutionOutcome | None = None
+    dispatcher: Dispatcher | None = None
+    summary_error: str | None = None
+    batch_count = job_count = 0
     try:
         monitor.logger.info(
             'Dispatcher invocation started',
-            context=tag_fields(caller_tags).get('context'),
             all_targets=all_targets,
             dry_run=dry_run,
             minimum_base_package=minimum_base_package,
@@ -228,13 +244,14 @@ def dispatch_tests(
 
         base_path = app.repo.path / output_dir
 
+        run: ResolvedRun | None
         if run_manifest is not None:
             # The manifest is the whole run: its identity is read, not recalculated, and a
             # manifest the caller explicitly supplied is not rewritten either.
             run = load_run_manifest(app, Path(run_manifest), repository=tested_repository)
             all_targets = run.all_targets
         else:
-            resolved = resolve_run(
+            run = resolve_run(
                 app,
                 repository=tested_repository,
                 pr_resolver=pr_resolver,
@@ -243,14 +260,17 @@ def dispatch_tests(
                 all_targets=all_targets,
                 monitor=monitoring.component('resolution'),
             )
-            if resolved is None:
-                run_summary(monitor, started, outcome='no-op')
+            if run is None:
+                outcome = ExecutionOutcome.NO_OP
                 return
 
-            run = resolved
+        # Bind before setup can fail, so terminal records retain the resolved identity.
+        monitoring.set_run_fields(**run_fields(run, tags=caller_tags))
+
+        if run_manifest is None:
             write_run_manifest(base_path, run=run)
             if resolve_only:
-                run_summary(monitor, started, outcome='resolved')
+                outcome = ExecutionOutcome.RESOLVED
                 app.display_success(f'Resolved run written to {base_path / RUN_MANIFEST_NAME}.')
                 return
 
@@ -258,85 +278,83 @@ def dispatch_tests(
 
         # Read after resolution: `--resolve-only` stops before the run needs planning configuration.
         config = DispatcherConfig.from_repo_config(app.repo.config)
+        overrides = {'workflow': workflow, 'workflow_ref': workflow_ref}
+        config = config.model_copy(update={name: value for name, value in overrides.items() if value})
 
-        context = DispatcherContext(
-            owner=owner,
-            repo=repo,
-            tags=caller_tags,
-            pytest_args=pytest_args or '',
-            checkout_sha=run.checkout_sha,
-            head_sha=run.head_sha,
-            head_branch=run.head_branch,
-            is_fork=run.is_fork,
-            workflow=workflow or config.workflow,
-            workflow_ref=workflow_ref or config.workflow_ref,
-            base_branch=run.base_branch,
-            base_sha=run.base_sha,
-            pr_number=run.pr_number,
-        )
-
-        # Resolved identity binds before planning, so a bad plan is still reported on its own run.
-        monitoring.set_run_fields(**run_fields(context))
+        try:
+            batches = build_plan(
+                app,
+                config=config,
+                changed_files=changed_files,
+                all_targets=all_targets,
+                minimum_base_package=minimum_base_package,
+                environment_provider=HatchEnvironmentProvider(default_python_version=config.default_python_version),
+                monitor=monitoring.component('planner'),
+            )
+        except PlanningError as error:
+            outcome = ExecutionOutcome.PLANNING_FAILED
+            summary_error = str(error)
+            app.abort(f'Could not build a test plan: {error}')
 
-        batches = build_plan(
-            app,
-            config=config,
-            changed_files=changed_files,
-            all_targets=all_targets,
-            minimum_base_package=minimum_base_package,
-            environment_provider=HatchEnvironmentProvider(default_python_version=config.default_python_version),
-            monitor=monitoring.component('planner'),
-        )
         if not batches:
             monitor.logger.info('Nothing to test', reason='the plan covers no target')
-            run_summary(monitor, started, outcome='no-op')
+            outcome = ExecutionOutcome.NO_OP
             return
 
-        display_plan(app, context, batches)
+        display_plan(app, run, config, batches, tags=caller_tags, pytest_args=pytest_args or '')
         if dry_run:
             monitor.logger.info('Dry run: nothing was dispatched')
-            run_summary(
-                monitor,
-                started,
-                outcome='dry-run',
-                batch_count=len(batches),
-                job_count=sum(batch.jobs_count for batch in batches),
-            )
+            outcome = ExecutionOutcome.DRY_RUN
+            batch_count = len(batches)
+            job_count = sum(batch.jobs_count for batch in batches)
             return
 
         dispatcher = build_dispatcher(
             batches=batches,
-            context=context,
+            run=run,
             config=config,
             token=token,
             artifacts_path=base_path / 'artifacts',
             output_path=base_path / 'results',
             monitoring=monitoring,
+            tags=caller_tags,
+            pytest_args=pytest_args or '',
         )
         # A fatal processor or hook failure leaves the bus by raising out of `run`. `on_finalize` has
         # already published whatever it knew by then, so a message is more use here than a traceback.
         try:
             dispatcher.run()
         except Exception as error:
-            run_summary(monitor, started, outcome='failed', dispatcher=dispatcher, error=str(error))
+            outcome = ExecutionOutcome.FAILED
+            summary_error = str(error)
             app.abort(f'Dispatcher execution failed: {error}')
 
-        outcome = dispatcher.outcome
-        if outcome is None or not outcome.successful:
+        dispatcher_outcome = dispatcher.outcome
+        if dispatcher_outcome is None or not dispatcher_outcome.successful:
+            outcome = ExecutionOutcome.CANCELLED if dispatcher.cancelled else ExecutionOutcome.FAILED
+            app.abort('Dispatcher tests failed.')
+
+        outcome = ExecutionOutcome.PASSED
+    except (KeyboardInterrupt, asyncio.CancelledError):
+        if outcome is None:
+            outcome = ExecutionOutcome.CANCELLED
+        raise
+    finally:
+        try:
             run_summary(
                 monitor,
-                started,
-                outcome='cancelled' if dispatcher.cancelled else 'failed',
+                started=started,
+                outcome=outcome or ExecutionOutcome.FAILED,
                 dispatcher=dispatcher,
+                batch_count=batch_count,
+                job_count=job_count,
+                error=summary_error,
             )
-            app.abort('Dispatcher tests failed.')
-
-        run_summary(monitor, started, outcome='passed', dispatcher=dispatcher)
-    finally:
-        monitoring.close()
-        if datadog_handler is not None:
-            # Close the runtime first so no event arrives while delivery drains.
-            datadog_handler.close(EXPORT_DRAIN_TIMEOUT)
+        finally:
+            monitoring.close()
+            if datadog_handler is not None:
+                # Close the runtime first so no event arrives while delivery drains.
+                datadog_handler.close(EXPORT_DRAIN_TIMEOUT)
 
 
 def build_datadog_metrics_sink(app: Application) -> DatadogMetricsSink | None:
@@ -385,26 +403,42 @@ def attach_datadog_log_handler(
 
 def run_summary(
     monitor: ComponentMonitor,
-    started: float,
     *,
-    outcome: str,
+    started: float,
+    outcome: ExecutionOutcome,
     dispatcher: Dispatcher | None = None,
     batch_count: int = 0,
     job_count: int = 0,
     error: str | None = None,
 ) -> None:
     """Emit one terminal record for executed and valid no-op runs."""
-    progress = dispatcher.outcome.progress if dispatcher is not None and dispatcher.outcome is not None else None
+    dispatcher_outcome = dispatcher.outcome if dispatcher is not None else None
+    progress = dispatcher_outcome.progress if dispatcher_outcome is not None else None
     if progress is not None:
         batch_count = len(progress.batches)
         job_count = sum(len(batch.jobs_progress) for batch in progress.batches)
-    cancelled = dispatcher.cancelled if dispatcher is not None else False
-    final_report_published = (
-        dispatcher.outcome.final_report_published
-        if dispatcher is not None and dispatcher.outcome is not None
-        else False
+    cancelled = outcome is ExecutionOutcome.CANCELLED or (
+        dispatcher_outcome is not None and dispatcher_outcome.cancelled
+    )
+    final_report_published = dispatcher_outcome.final_report_published if dispatcher_outcome is not None else False
+    timed_out = dispatcher_outcome is not None and dispatcher_outcome.timed_out
+    elapsed = time.monotonic() - started
+    if outcome not in (ExecutionOutcome.RESOLVED, ExecutionOutcome.DRY_RUN):
+        metrics = monitor.metrics
+        metrics.count('runs.count', 1)
+        metrics.count('runs.failed', int(outcome in (ExecutionOutcome.FAILED, ExecutionOutcome.PLANNING_FAILED)))
+        metrics.count('runs.planning_failed', int(outcome is ExecutionOutcome.PLANNING_FAILED))
+        metrics.count('runs.cancelled', int(cancelled))
+        metrics.count('runs.timed_out', int(timed_out))
+        metrics.count('runs.no_op', int(outcome is ExecutionOutcome.NO_OP))
+        metrics.distribution('run.duration', elapsed)
+    log = (
+        monitor.logger.error
+        if outcome in (ExecutionOutcome.FAILED, ExecutionOutcome.PLANNING_FAILED)
+        else monitor.logger.warning
+        if cancelled
+        else monitor.logger.info
     )
-    log = monitor.logger.error if outcome == 'failed' else monitor.logger.warning if cancelled else monitor.logger.info
     log(
         'Dispatcher run finished',
         outcome=outcome,
@@ -412,7 +446,7 @@ def run_summary(
         batch_count=batch_count,
         job_count=job_count,
         final_report_published=final_report_published,
-        elapsed_seconds=round(time.monotonic() - started, 3),
+        elapsed_seconds=round(elapsed, 3),
         error=error,
     )
 
@@ -427,7 +461,7 @@ def build_plan(
     environment_provider: EnvironmentProvider,
     monitor: ComponentMonitor,
 ) -> list[TestBatch]:
-    """Build the batches this run must execute, aborting with a readable message on a bad plan.
+    """Build the batches this run must execute, raising `PlanningError` on a bad plan.
 
     `--all` plans every eligible target, so it needs no comparison and `changed_files` is None.
     """
@@ -456,7 +490,7 @@ def build_plan(
         )
     except PlanningError as error:
         monitor.logger.error('Planning failed', error=str(error))
-        app.abort(f'Could not build a test plan: {error}')
+        raise
 
     monitor.logger.info(
         'Planning completed',
@@ -476,23 +510,31 @@ def build_plan(
     return batches
 
 
-def display_plan(app: Application, context: DispatcherContext, batches: list[TestBatch]) -> None:
+def display_plan(
+    app: Application,
+    run: ResolvedRun,
+    config: DispatcherConfig,
+    batches: list[TestBatch],
+    *,
+    tags: Sequence[str] = (),
+    pytest_args: str = '',
+) -> None:
     app.display_header('Dispatcher plan')
-    app.display_pair('Repository', f'{context.owner}/{context.repo}')
-    app.display_pair('Head branch', context.head_branch)
-    app.display_pair('Head SHA', context.head_sha)
-    app.display_pair('Checkout SHA', context.checkout_sha)
-    if context.pr_number is not None:
-        app.display_pair('Pull request', str(context.pr_number))
-    if context.base_branch is not None:
-        app.display_pair('Base branch', context.base_branch)
-    if context.base_sha is not None:
-        app.display_pair('Base SHA', context.base_sha)
-    if context.tags:
-        app.display_pair('Tags', ' '.join(context.tags))
-    if context.pytest_args:
-        app.display_pair('Pytest args', context.pytest_args)
-    app.display_pair('Workflow', f'{context.workflow} @ {context.workflow_ref}')
+    app.display_pair('Repository', run.repository)
+    app.display_pair('Head branch', run.head_branch)
+    app.display_pair('Head SHA', run.head_sha)
+    app.display_pair('Checkout SHA', run.checkout_sha)
+    if run.pr_number is not None:
+        app.display_pair('Pull request', str(run.pr_number))
+    if run.base_branch is not None:
+        app.display_pair('Base branch', run.base_branch)
+    if run.base_sha is not None:
+        app.display_pair('Base SHA', run.base_sha)
+    if tags:
+        app.display_pair('Tags', ' '.join(tags))
+    if pytest_args:
+        app.display_pair('Pytest args', pytest_args)
+    app.display_pair('Workflow', f'{config.workflow} @ {config.workflow_ref}')
 
     total = sum(batch.jobs_count for batch in batches)
     app.display_pair('Batches', f'{len(batches)} ({total} jobs)')
diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher.py b/ddev/src/ddev/cli/ci/tests/dispatcher.py
index a6e2897043a57..bea612078a7cd 100644
--- a/ddev/src/ddev/cli/ci/tests/dispatcher.py
+++ b/ddev/src/ddev/cli/ci/tests/dispatcher.py
@@ -10,7 +10,7 @@
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from ddev.cli.ci.tests.dispatcher_attributes import batch_fields, message_fields, run_fields
+from ddev.cli.ci.tests.dispatcher_attributes import batch_fields, job_fields, message_fields, run_fields
 from ddev.cli.ci.tests.messages import BatchFinished, BatchProgressUpdate, TestBatch, UpdatePRComment
 from ddev.cli.ci.tests.pr_comment import render_run_summary, summary_line
 from ddev.cli.ci.tests.rate_limiting import RateLimiterFactory
@@ -31,6 +31,7 @@
     from collections.abc import Sequence
     from pathlib import Path
 
+    from ddev.cli.ci.dispatch_run import ResolvedRun
     from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig
     from ddev.cli.ci.tests.progress import DispatcherProgress
     from ddev.utils.github_async import AsyncGitHubClient
@@ -40,35 +41,6 @@
 CANCELLED_RATE_LIMITS = RelaxedRateLimits(max_wait_seconds=2.0, max_rate=10_000.0)
 
 
-@dataclass(frozen=True)
-class DispatcherContext:
-    """The run being tested. `build_dispatcher` consumes part of it; the rest describes the run
-    for the plan header and for the monitoring run context (see `run_fields`).
-
-    `checkout_sha` is the tree tested; `head_sha` is the revision that receives its results.
-    They differ for a pull request and are the same for a branch run.
-    """
-
-    owner: str
-    repo: str
-    checkout_sha: str
-    head_sha: str
-    head_branch: str
-    workflow: str
-    workflow_ref: str
-    base_branch: str | None = None
-    base_sha: str | None = None
-    pr_number: int | None = None
-    tags: tuple[str, ...] = ()
-    pytest_args: str = ''
-    is_fork: bool = False
-
-    @property
-    def concurrency_key(self) -> str:
-        """New PR revisions must cancel old batches, so they key on the PR, not the merge SHA."""
-        return f'pr-{self.pr_number}' if self.pr_number is not None else self.head_sha
-
-
 @dataclass(frozen=True)
 class DispatcherOutcome:
     """What a finished Dispatcher execution amounts to, for the caller to exit on."""
@@ -90,6 +62,16 @@ def successful(self) -> bool:
             and all(batch.status is not Status.FAILURE for batch in self.progress.batches)
         )
 
+    @property
+    def cancelled(self) -> bool:
+        """Whether the run ended by cancellation rather than by finishing or failing."""
+        return self.shutdown is not None and self.shutdown.kind is ShutdownKind.CANCELLED
+
+    @property
+    def timed_out(self) -> bool:
+        """Whether the run ended because its timeout elapsed."""
+        return self.shutdown is not None and self.shutdown.kind is ShutdownKind.TIMED_OUT
+
 
 def message_scope(context: MonitorContext, batches: Sequence[TestBatch]) -> MessageScope:
     """Scope each message against the plan, so batch-scoped events resolve one canonical batch by id."""
@@ -187,6 +169,8 @@ async def on_message_received(self, message: BaseMessage):
     async def on_finalize(self, exception: Exception | None):
         request = self.shutdown_request
         try:
+            # Processors have drained; report their results even if remote cleanup fails.
+            self._report_incomplete_jobs()
             if request is not None:
                 await self._shutdown_cleanup(request)
             progress = self._gatherer.progress
@@ -201,6 +185,25 @@ async def on_finalize(self, exception: Exception | None):
         finally:
             await self._client.aclose()
 
+    def _report_incomplete_jobs(self) -> None:
+        """Best-effort accounting: reporting errors must not prevent shutdown cleanup."""
+        try:
+            progress_by_batch = {batch.batch_id: batch for batch in self._gatherer.progress.batches}
+            metrics = self._monitor.metrics
+            for batch in self._batches:
+                if batch.run_id is None:
+                    continue
+                final = progress_by_batch[batch.batch_id]
+                collected = {
+                    job_progress.job.name
+                    for job_progress in final.jobs_progress
+                    if job_progress.collected_result is not None
+                }
+                for job in batch.job_list:
+                    metrics.count('jobs.incomplete', int(job.name not in collected), **job_fields(job))
+        except Exception:
+            self._logger.exception('Failed to report incomplete jobs')
+
     async def _shutdown_cleanup(self, request: ShutdownRequest) -> None:
         """Attempt terminal reporting and remote cancellation without either abandoning the other."""
         outcomes = await asyncio.gather(
@@ -222,17 +225,20 @@ async def _shutdown_cleanup(self, request: ShutdownRequest) -> None:
 def build_dispatcher(
     *,
     batches: list[TestBatch],
-    context: DispatcherContext,
+    run: ResolvedRun,
     config: DispatcherConfig,
     token: str,
     artifacts_path: Path,
     output_path: Path,
     monitoring: MonitoringRuntime,
+    tags: Sequence[str] = (),
+    pytest_args: str = '',
 ) -> Dispatcher:
-    """Assemble the client, monitored tasks and Dispatcher from a plan and its run context.
+    """Assemble the client, monitored tasks and Dispatcher from a plan and its resolved run.
 
     One HTTP pool is shared by every task. Artifact collection uses its own local bucket;
-    all buckets share the provider's budget and pauses. The caller owns `monitoring`.
+    all buckets share the provider's budget and pauses. The caller owns `monitoring`; each
+    processor reports its own metrics through the monitor it is given.
     """
     from ddev.utils.github_async import AsyncGitHubClient
 
@@ -241,31 +247,36 @@ def build_dispatcher(
     rate_limiters = RateLimiterFactory(config.github_rate_limits, client_logger)
     client = AsyncGitHubClient(token, rate_limiter=rate_limiters.get_limiter(integrations), logger=client_logger)
 
-    canonical_run_fields = run_fields(context)
+    canonical_run_fields = run_fields(run, tags=tags)
     runner = TaskTestRunner(
         "test-runner",
         client,
         TestRunnerOptions(
-            owner=context.owner,
-            repo=context.repo,
-            workflow_id=context.workflow,
-            ref=context.workflow_ref,
+            owner=run.owner,
+            repo=run.repo,
+            workflow_id=config.workflow,
+            ref=config.workflow_ref,
             run_fields=canonical_run_fields,
-            concurrency_key=context.concurrency_key,
+            concurrency_key=run.concurrency_key,
             artifacts_base_path=artifacts_path,
             poll_interval_seconds=config.poll_interval_seconds,
-            pytest_args=context.pytest_args,
+            pytest_args=pytest_args,
             origin_run_url=get_workflow_run_url(),
-            pr_number=context.pr_number,
+            pr_number=run.pr_number,
         ),
         artifact_client=client.with_rate_limit(rate_limiters.artifacts),
         monitor=monitoring.component('test-runner'),
     )
-    gatherer = TaskTestGatherer("test-gatherer", output_path, batches, monitor=monitoring.component('test-gatherer'))
+    gatherer = TaskTestGatherer(
+        "test-gatherer",
+        output_path,
+        batches,
+        monitor=monitoring.component('test-gatherer'),
+    )
     reporter = TaskRunReporter(
         "run-reporter",
         client,
-        RunReporterOptions(owner=context.owner, repo=context.repo, pr_number=context.pr_number),
+        RunReporterOptions(owner=run.owner, repo=run.repo, pr_number=run.pr_number),
         monitor=monitoring.component('run-reporter'),
     )
 
diff --git a/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py b/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py
index cc7bf2c87f05c..79097656e7bf3 100644
--- a/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py
+++ b/ddev/src/ddev/cli/ci/tests/dispatcher_attributes.py
@@ -14,10 +14,18 @@
 from ddev.event_bus.orchestrator import BaseMessage
 
 if TYPE_CHECKING:
-    from ddev.cli.ci.tests.dispatcher import DispatcherContext
+    from ddev.cli.ci.dispatch_run import ResolvedRun
 
 
 DEFAULT_TEAM = 'agent-integrations'
+BASE_FIELDS = {'team': DEFAULT_TEAM}
+# Keep pre-resolution failures in the same metric grouping dimensions as resolved runs.
+UNRESOLVED_RUN_FIELDS = {
+    'context': 'unresolved',
+    'head_branch': 'unresolved',
+    'base_branch': 'unresolved',
+    'is_fork': 'unresolved',
+}
 
 
 @dataclass(frozen=True)
@@ -52,10 +60,6 @@ class AttributeSpec:
         'git.branch',
         metric_tag=True,
     ),
-    'is_default_branch': AttributeSpec(
-        'git.is_default_branch',
-        metric_tag=True,
-    ),
     'checkout_sha': AttributeSpec(
         'dispatcher.checkout_sha',
         test_tag=True,
@@ -154,7 +158,6 @@ class AttributeSpec:
     'done': AttributeSpec(
         'dispatcher.report.done',
         console_tag=True,
-        metric_tag=True,
     ),
     'comment_id': AttributeSpec(
         'dispatcher.report.comment_id',
@@ -316,11 +319,11 @@ class AttributeSpec:
 
 PROTECTED_RUN_FIELDS = frozenset(
     {
+        'team',
         'repo',
         'repository_url',
         'head_branch',
         'head_sha',
-        'is_default_branch',
         'checkout_sha',
         'context',
         'pr_number',
@@ -425,27 +428,27 @@ def repository_fields(owner: str, repo: str) -> dict[str, str]:
     }
 
 
-def run_fields(context: DispatcherContext) -> dict[str, Any]:
-    """Resolve run identity while retaining an explicit non-PR caller context."""
-    fields: dict[str, Any] = {'team': DEFAULT_TEAM, **tag_fields(context.tags)}
+def run_fields(run: ResolvedRun, *, tags: Sequence[str] = ()) -> dict[str, Any]:
+    """Project run identity, retaining caller context only for non-PR runs."""
+    fields: dict[str, Any] = tag_fields(tags)
     for name in PROTECTED_RUN_FIELDS:
         if name != 'context':
             fields.pop(name, None)
 
     fields.update(
         {
-            **repository_fields(context.owner, context.repo),
-            'head_sha': context.head_sha,
-            'head_branch': context.head_branch,
-            'is_default_branch': context.pr_number is None and context.head_branch == 'master',
-            'checkout_sha': context.checkout_sha,
-            'pr_number': context.pr_number,
-            'base_branch': context.base_branch,
-            'base_sha': context.base_sha,
-            'is_fork': context.is_fork,
+            **BASE_FIELDS,
+            **repository_fields(run.owner, run.repo),
+            'head_sha': run.head_sha,
+            'head_branch': run.head_branch,
+            'checkout_sha': run.checkout_sha,
+            'pr_number': run.pr_number,
+            'base_branch': run.base_branch,
+            'base_sha': run.base_sha,
+            'is_fork': run.is_fork,
         }
     )
-    if context.pr_number is not None:
+    if run.pr_number is not None:
         fields['context'] = 'pr'
     elif 'context' not in fields:
         fields['context'] = 'master'
diff --git a/ddev/src/ddev/cli/ci/tests/execution_metrics.py b/ddev/src/ddev/cli/ci/tests/execution_metrics.py
new file mode 100644
index 0000000000000..10a43c642dfa3
--- /dev/null
+++ b/ddev/src/ddev/cli/ci/tests/execution_metrics.py
@@ -0,0 +1,80 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Reusable metrics helpers for the Dispatcher's processors and its command."""
+
+from __future__ import annotations
+
+import time
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass
+from enum import StrEnum, auto
+
+from ddev.monitoring.metrics import Metrics
+
+
+class ExecutionOutcome(StrEnum):
+    """Terminal command outcomes, including modes that skip execution."""
+
+    NO_OP = 'no-op'
+    PASSED = auto()
+    FAILED = auto()
+    CANCELLED = auto()
+    PLANNING_FAILED = 'planning-failed'
+    RESOLVED = auto()
+    DRY_RUN = 'dry-run'
+
+
+class Operation(StrEnum):
+    """One logical component operation, settled only after retries and fallbacks."""
+
+    DISPATCH_BATCH = auto()
+    FETCH_WORKFLOW = auto()
+    REFRESH_JOBS = auto()
+    COLLECT_ARTIFACTS = auto()
+    GATHER_BATCH_RESULTS = auto()
+    PUBLISH_REPORT = auto()
+
+
+@dataclass
+class OperationResult:
+    """The outcome of one logical operation, including cancellation."""
+
+    failed: bool = False
+    cancelled: bool = False
+
+
+class MetricsHelper:
+    """Operation metrics for one component, built from that component's own monitor."""
+
+    def __init__(self, metrics: Metrics, *, clock: Callable[[], float] = time.monotonic) -> None:
+        self._metrics = metrics
+        self._clock = clock
+
+    def record_operation(self, operation: Operation, *, failed: bool) -> None:
+        """Emit one logical operation's outcome after retries and fallbacks settle."""
+        self._metrics.count('operations.count', 1, operation=operation)
+        self._metrics.count('operations.failed', int(failed), operation=operation)
+
+    @contextmanager
+    def time_operation(self, operation: Operation, *, duration_metric: str) -> Iterator[OperationResult]:
+        """Mark recovered failures through `result.failed`; escaped exceptions propagate.
+
+        Cancellation reports elapsed time but not a settled operation outcome.
+        """
+        result = OperationResult()
+        start = self._clock()
+        try:
+            yield result
+        except Exception:
+            result.failed = True
+            raise
+        except BaseException:
+            result.cancelled = True
+            raise
+        finally:
+            elapsed = self._clock() - start
+            if not result.cancelled:
+                self.record_operation(operation, failed=result.failed)
+            self._metrics.distribution(duration_metric, elapsed, operation=operation)
diff --git a/ddev/src/ddev/cli/ci/tests/messages.py b/ddev/src/ddev/cli/ci/tests/messages.py
index ff9f81197eaca..4b6098a15981d 100644
--- a/ddev/src/ddev/cli/ci/tests/messages.py
+++ b/ddev/src/ddev/cli/ci/tests/messages.py
@@ -179,6 +179,8 @@ class TestBatch(BaseMessage):
     job_list: list[BatchJob]
     jobs_count: int
     integrations: list[str]
+    # Retain accepted launches even when shutdown discards their queued progress updates.
+    run_id: int | None = None
 
 
 @dataclass
diff --git a/ddev/src/ddev/cli/ci/tests/progress.py b/ddev/src/ddev/cli/ci/tests/progress.py
index dd4d45a5a88b3..567d6c0282fcb 100644
--- a/ddev/src/ddev/cli/ci/tests/progress.py
+++ b/ddev/src/ddev/cli/ci/tests/progress.py
@@ -99,6 +99,14 @@ def complete(self) -> bool:
         """Whether the latest execution finished, independently of result collection."""
         return self.latest is not None and self.latest.state is ExecutionState.FINISHED
 
+    @property
+    def collected_result(self) -> JobAttemptProgress | None:
+        """An empty report tuple is collected; `None` means the attempt was only observed."""
+        latest = self.latest
+        if latest is None or latest.state is not ExecutionState.FINISHED or latest.reports is None:
+            return None
+        return latest
+
     @property
     def retry_count(self) -> int:
         """Executions minus one. Not ``run_attempt - 1``: histories can be sparse."""
diff --git a/ddev/src/ddev/cli/ci/tests/task_run_reporter.py b/ddev/src/ddev/cli/ci/tests/task_run_reporter.py
index 265c95f113cc0..db5f97a0cf8c6 100644
--- a/ddev/src/ddev/cli/ci/tests/task_run_reporter.py
+++ b/ddev/src/ddev/cli/ci/tests/task_run_reporter.py
@@ -11,6 +11,7 @@
 
 import httpx
 
+from ddev.cli.ci.tests.execution_metrics import MetricsHelper, Operation
 from ddev.cli.ci.tests.pr_comment import (
     COMMENT_MARKER,
     render_comment,
@@ -99,6 +100,7 @@ def __init__(
         self._lock = asyncio.Lock()
         self._logger = monitor.logger
         self.monitor = monitor
+        self._metrics = MetricsHelper(monitor.metrics)
 
     @property
     def latest_body(self) -> str | None:
@@ -141,8 +143,13 @@ async def process_message(self, message: UpdatePRComment):
                 published = True
             else:
                 self._pr_comment_failed = True
-                published = await self._write(pr_number, body, message.progress, revision=message.revision, now=now)
+                try:
+                    published = await self._write(pr_number, body, message.progress, revision=message.revision, now=now)
+                except Exception:
+                    self._metrics.record_operation(Operation.PUBLISH_REPORT, failed=True)
+                    raise
                 self._pr_comment_failed = not published
+                self._metrics.record_operation(Operation.PUBLISH_REPORT, failed=not published)
 
             if message.progress.done and published:
                 self._final_report_published = True
@@ -172,8 +179,13 @@ async def publish_shutdown(self, request: ShutdownRequest) -> None:
                     return
 
                 self._pr_comment_failed = True
-                published = await self._write(pr_number, body, progress, shutdown=request, now=now)
+                try:
+                    published = await self._write(pr_number, body, progress, shutdown=request, now=now)
+                except Exception:
+                    self._metrics.record_operation(Operation.PUBLISH_REPORT, failed=True)
+                    raise
                 self._pr_comment_failed = not published
+                self._metrics.record_operation(Operation.PUBLISH_REPORT, failed=not published)
                 if published:
                     self._logger.info("Run reported as %s", request.kind.value, published=True)
 
diff --git a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py
index 99d092651f408..73741b85b06e2 100644
--- a/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py
+++ b/ddev/src/ddev/cli/ci/tests/task_test_gatherer.py
@@ -9,6 +9,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from ddev.cli.ci.tests.dispatcher_attributes import job_fields
+from ddev.cli.ci.tests.execution_metrics import MetricsHelper, Operation
 from ddev.cli.ci.tests.messages import (
     BatchFinished,
     BatchJob,
@@ -59,7 +61,12 @@ class TaskTestGatherer(SyncProcessor[BatchFinished | BatchProgressUpdate]):
     """
 
     def __init__(
-        self, name: str, output_base_path: Path, batches: list[TestBatch], *, monitor: ComponentMonitor
+        self,
+        name: str,
+        output_base_path: Path,
+        batches: list[TestBatch],
+        *,
+        monitor: ComponentMonitor,
     ) -> None:
         super().__init__(name)
         self._output_base_path = output_base_path
@@ -74,6 +81,7 @@ def __init__(
         self._lock = threading.Lock()
         self._logger = monitor.logger
         self.monitor = monitor
+        self._metrics = MetricsHelper(monitor.metrics)
 
     def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
         if isinstance(message, BatchProgressUpdate):
@@ -96,9 +104,14 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
             len(message.batch_jobs),
             batch_job_count=len(message.batch_jobs),
         )
-        gathered = self._gather_results(message)
-        if gathered is not None:
-            self._publish_results(message, gathered)
+        try:
+            gathered = self._gather_results(message)
+            published = gathered is not None and self._publish_results(message, gathered)
+        except Exception:
+            self._metrics.record_operation(Operation.GATHER_BATCH_RESULTS, failed=True)
+            raise
+        if published:
+            self._metrics.record_operation(Operation.GATHER_BATCH_RESULTS, failed=False)
 
     def _gather_results(
         self,
@@ -121,22 +134,25 @@ def _publish_results(
         self,
         message: BatchFinished,
         gathered: list[tuple[JobResult, JobAttemptProgress]],
-    ) -> None:
+    ) -> bool:
+        """Commit gathered results; return whether the batch was actually registered."""
         results = [result for result, _ in gathered]
         status = self._build_workflow_status(message, results)
         with self._lock:
             # Cancellation or another collector may have won while these results were parsed.
             if self.stopping:
                 self._logger.warning("Batch gathered but left unregistered: the bus is shutting down")
-                return
+                return False
             if not self._accepts(message.batch_id):
-                return
+                return False
             planned = self._progress_by_batch[message.batch_id]
             if results:
                 self._results_by_batch[message.batch_id] = results
                 self._status_by_batch[message.batch_id] = status
-            self._progress_by_batch[message.batch_id] = self._finished_batch_progress(planned, message, gathered)
+            finished = self._finished_batch_progress(planned, message, gathered)
+            self._progress_by_batch[message.batch_id] = finished
             update = self._publish_update(message.id)
+            self._report_final_result(finished)
 
         self._logger.info(
             "Batch %s results gathered: report revision %s (done=%s)",
@@ -144,6 +160,18 @@ def _publish_results(
             update.revision,
             update.progress.done,
         )
+        return True
+
+    def _report_final_result(self, progress: BatchProgress) -> None:
+        """Count collected results, not attempts only observed through polling."""
+        metrics = self.monitor.metrics
+        metrics.count('batches.failed', int(progress.status == Status.FAILURE))
+        for job_progress in progress.jobs_progress:
+            latest = job_progress.collected_result
+            if latest is None:
+                continue
+            metrics.count('jobs.failed', int(latest.status is Status.FAILURE), **job_fields(job_progress.job))
+            metrics.count('jobs.skipped', int(latest.status is Status.SKIPPED), **job_fields(job_progress.job))
 
     def _observe_progress(self, message: BatchProgressUpdate) -> None:
         with self._lock:
diff --git a/ddev/src/ddev/cli/ci/tests/task_test_runner.py b/ddev/src/ddev/cli/ci/tests/task_test_runner.py
index a2ccbfe543aaf..a62ba9a8f89a5 100644
--- a/ddev/src/ddev/cli/ci/tests/task_test_runner.py
+++ b/ddev/src/ddev/cli/ci/tests/task_test_runner.py
@@ -10,6 +10,7 @@
 import json
 from collections.abc import Mapping
 from dataclasses import dataclass
+from datetime import datetime
 from itertools import count
 from pathlib import Path
 from typing import Any
@@ -17,6 +18,7 @@
 from pydantic import ValidationError
 
 from ddev.cli.ci.tests.dispatcher_attributes import batch_fields, job_fields, test_tag_mapping
+from ddev.cli.ci.tests.execution_metrics import MetricsHelper, Operation
 from ddev.cli.ci.tests.messages import BatchFinished, BatchJob, BatchJobResult, BatchProgressUpdate, TestBatch
 from ddev.cli.ci.tests.progress import ExecutionState
 from ddev.cli.ci.tests.status import conclusion_to_status
@@ -61,6 +63,20 @@ def _sanitize_test_tag_value(value: str) -> str:
     return value.replace(',', '_').replace('\n', '_').replace('\r', '_')
 
 
+def workflow_duration_seconds(run: WorkflowRun) -> float | None:
+    """Use gh's completed-run timing convention, omitting unavailable or invalid timestamps.
+
+    https://github.com/cli/cli/blob/trunk/pkg/cmd/run/shared/shared.go
+    """
+    if run.run_started_at is None or run.updated_at is None:
+        return None
+    try:
+        duration = (datetime.fromisoformat(run.updated_at) - datetime.fromisoformat(run.run_started_at)).total_seconds()
+    except (ValueError, TypeError):
+        return None
+    return duration if duration >= 0 else None
+
+
 def encode_job_list(jobs: list[dict[str, Any]]) -> str:
     """Encode a batch's jobs for a workflow input, as gzip then base64.
 
@@ -114,6 +130,7 @@ def __init__(
         self._runs_in_flight: dict[str, int] = {}
         self._logger = monitor.logger
         self.monitor = monitor
+        self._metrics = MetricsHelper(monitor.metrics)
 
     def _response_failure(
         self, operation: str, batch_id: str, run_id: int | None, error: ValidationError
@@ -167,9 +184,16 @@ async def _dispatch_batch(self, message: TestBatch) -> tuple[int, str]:
                 return_run_details=True,
             )
         except ValidationError as error:
+            self._metrics.record_operation(Operation.DISPATCH_BATCH, failed=True)
             raise self._response_failure("dispatching the batch", message.batch_id, None, error) from error
+        except Exception:
+            self._metrics.record_operation(Operation.DISPATCH_BATCH, failed=True)
+            raise
+        self._metrics.record_operation(Operation.DISPATCH_BATCH, failed=False)
         run_id = dispatch.data.workflow_run_id
         self._runs_in_flight[message.batch_id] = run_id
+        message.run_id = run_id
+        self._report_launch(message)
         self.submit_message(
             BatchProgressUpdate(
                 id=f"{message.id}-progress-0",
@@ -217,6 +241,15 @@ async def _collect_results(
         )
         self._logger.info("Batch %s workflow results ready: %s", message.batch_id, status.value)
 
+    def _report_launch(self, batch: TestBatch) -> None:
+        """Count the batch and its jobs only after GitHub accepts its dispatch."""
+        metrics = self.monitor.metrics
+        metrics.count('batches.count', 1)
+        # One sample per launch supports batch-size averages without per-batch tags.
+        metrics.distribution('batch.jobs.count', batch.jobs_count)
+        for job in batch.job_list:
+            metrics.count('jobs.count', 1, **job_fields(job))
+
     async def cancel_dispatched_runs(self) -> None:
         """Concurrently cancel all tracked unfinished runs."""
         if not self._runs_in_flight:
@@ -251,11 +284,18 @@ async def _poll_until_complete(
             try:
                 run = await self._client.get_workflow_run(self._options.owner, self._options.repo, run_id)
             except ValidationError as error:
+                self._metrics.record_operation(Operation.FETCH_WORKFLOW, failed=True)
                 raise self._response_failure("polling workflow status", message.batch_id, run_id, error) from error
+            except Exception:
+                self._metrics.record_operation(Operation.FETCH_WORKFLOW, failed=True)
+                raise
+            self._metrics.record_operation(Operation.FETCH_WORKFLOW, failed=False)
             completed = run.data.is_completed
             # Shutdown must not try to cancel a completed run while its artifacts are still being collected.
             if completed:
                 self._runs_in_flight.pop(message.batch_id, None)
+                if (duration := workflow_duration_seconds(run.data)) is not None:
+                    self.monitor.metrics.distribution('batch.duration', duration)
 
             # Report workflow progress first. The jobs request may be delayed by the API rate limit.
             progress = self._publish_workflow_progress(message, run_id, run.data, known_jobs, next(sequences))
@@ -358,9 +398,13 @@ async def _list_jobs(self, run_id: int, batch_id: str, operation: str) -> list[W
             ):
                 jobs.extend(page.data.jobs)
         except ValidationError as error:
+            self._metrics.record_operation(Operation.REFRESH_JOBS, failed=True)
             raise self._response_failure(operation, batch_id, run_id, error) from error
         except Exception:
             self._logger.warning("Failed to list workflow jobs", exc_info=True)
+            self._metrics.record_operation(Operation.REFRESH_JOBS, failed=True)
+        else:
+            self._metrics.record_operation(Operation.REFRESH_JOBS, failed=False)
         return jobs
 
     def _build_inputs(self, message: TestBatch) -> dict[str, str]:
@@ -368,7 +412,7 @@ def _build_inputs(self, message: TestBatch) -> dict[str, str]:
         inputs = {
             "batch_id": message.batch_id,
             "checkout_sha": str(run["checkout_sha"]),
-            # Keys the workflow's cancellation group; see `DispatcherContext.concurrency_key`.
+            # Keys the workflow's cancellation group; see `ResolvedRun.concurrency_key`.
             "concurrency_key": self._options.concurrency_key,
             # The batch is dispatched at the default branch, so its own context describes master.
             # These two say which commit the results belong to, for CI Visibility and the check run.
@@ -411,23 +455,31 @@ async def _download_artifacts(self, run_id: int, batch_id: str) -> dict[str, Pat
         """
         artifact_dirs: dict[str, Path] = {}
         failures: list[tuple[int, str]] = []
-        try:
-            async for page in self._artifact_client.list_workflow_run_artifacts(
-                self._options.owner, self._options.repo, run_id, per_page=100
-            ):
-                for artifact in page.data.artifacts:
-                    url = self._artifact_download_url(artifact)
-                    if url is None:
-                        continue
-                    target = await self._download_artifact(artifact, url)
-                    if target is None:
-                        failures.append((artifact.id, artifact.name))
-                    else:
-                        artifact_dirs[artifact.name] = target
-        except ValidationError as error:
-            raise self._response_failure("listing workflow artifacts", batch_id, run_id, error) from error
-        except Exception:
-            self._logger.warning("Failed to list workflow run artifacts", exc_info=True)
+        with self._metrics.time_operation(
+            Operation.COLLECT_ARTIFACTS, duration_metric='artifacts.download.duration'
+        ) as result:
+            degraded = False
+            try:
+                async for page in self._artifact_client.list_workflow_run_artifacts(
+                    self._options.owner, self._options.repo, run_id, per_page=100
+                ):
+                    for artifact in page.data.artifacts:
+                        url = self._artifact_download_url(artifact)
+                        if url is None:
+                            degraded = True
+                            continue
+                        target = await self._download_artifact(artifact, url)
+                        if target is None:
+                            failures.append((artifact.id, artifact.name))
+                        else:
+                            artifact_dirs[artifact.name] = target
+            except ValidationError as error:
+                raise self._response_failure("listing workflow artifacts", batch_id, run_id, error) from error
+            except Exception:
+                self._logger.warning("Failed to list workflow run artifacts", exc_info=True)
+                degraded = True
+            # An incomplete listing degrades the collection like a failed download does.
+            result.failed = degraded or bool(failures)
         if failures:
             self._logger.warning(
                 "Failed to download %s %s for workflow run %s",
diff --git a/ddev/src/ddev/monitoring/runtime.py b/ddev/src/ddev/monitoring/runtime.py
index 439f29b82517c..eb1454e842819 100644
--- a/ddev/src/ddev/monitoring/runtime.py
+++ b/ddev/src/ddev/monitoring/runtime.py
@@ -15,7 +15,7 @@
 import structlog
 from structlog.stdlib import BoundLogger
 
-from ddev.monitoring.context import MonitorContext
+from ddev.monitoring.context import EMPTY_FIELDS, MonitorContext
 from ddev.monitoring.diagnostics import DiagnosticCategory
 from ddev.monitoring.logger import logger_processors
 from ddev.monitoring.metrics import Metrics, MetricsSink, TagProjector
@@ -60,11 +60,13 @@ def __init__(
         metrics_sink: MetricsSink | None = None,
         metrics_tag_projector: TagProjector | None = None,
         protected_fields: Collection[str] = (),
+        base_fields: Mapping[str, Any] = EMPTY_FIELDS,
     ) -> None:
         self._closed = False
         self._metrics_closed = False
         self._close_lock = threading.Lock()
         self._context = MonitorContext(protected_fields)
+        self._context.set_fields(**base_fields)
         self._metrics_sink = metrics_sink
         self._stdlib_logger = logging.Logger(f'ddev.monitoring.{next(_runtime_names)}', level=logging.DEBUG)
         self._stdlib_logger.propagate = False
diff --git a/ddev/src/ddev/utils/github_async/models/workflow.py b/ddev/src/ddev/utils/github_async/models/workflow.py
index 17fa610ec328c..1461200233734 100644
--- a/ddev/src/ddev/utils/github_async/models/workflow.py
+++ b/ddev/src/ddev/utils/github_async/models/workflow.py
@@ -83,6 +83,8 @@ class WorkflowRun(BaseModel):
     html_url: str
     created_at: str | None = None
     updated_at: str | None = None
+    # Start of the latest attempt, reset on rerun (see the workflow-run schema linked above).
+    run_started_at: str | None = None
 
     @property
     def is_completed(self) -> bool:
diff --git a/ddev/tests/cli/ci/helpers.py b/ddev/tests/cli/ci/helpers.py
index 7e81d341b8873..8be8a29bb4297 100644
--- a/ddev/tests/cli/ci/helpers.py
+++ b/ddev/tests/cli/ci/helpers.py
@@ -1,15 +1,26 @@
 # (C) Datadog, Inc. 2026-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
-"""Pull request list responses shared by Dispatcher CLI and resolution tests."""
+"""GitHub payloads shared by Dispatcher CLI and component tests."""
 
+import base64
+import gzip
+import json
+from typing import Any
+
+from ddev.cli.ci.tests.messages import BatchJob
 from ddev.utils.github_async import GitHubResponse
-from ddev.utils.github_async.models import PullRequestSimple
+from ddev.utils.github_async.models import PullRequestSimple, WorkflowJob, WorkflowJobsList
+from tests.helpers.github_async import FakeAsyncGitHubClient
 
 PR_NUMBER = 4242
 HEAD_SHA = 'head-sha-aaa'
 
 
+def decode_job_list(encoded: str) -> list[dict[str, Any]]:
+    return json.loads(gzip.decompress(base64.b64decode(encoded)).decode())
+
+
 def listed_pull_request(
     number: int = PR_NUMBER,
     head_sha: str = HEAD_SHA,
@@ -34,3 +45,13 @@ def listed_pull_request(
 
 def pulls_page(*pulls: PullRequestSimple) -> GitHubResponse[list[PullRequestSimple]]:
     return GitHubResponse[list[PullRequestSimple]].model_validate({'data': list(pulls), 'headers': {}})
+
+
+def mock_job_result(fake: FakeAsyncGitHubClient, job: BatchJob, conclusion: str) -> None:
+    fake.mock_response(
+        "list_workflow_jobs",
+        WorkflowJobsList(
+            total_count=1,
+            jobs=[WorkflowJob(id=1, run_id=123, name=job.name, status="completed", conclusion=conclusion)],
+        ),
+    )
diff --git a/ddev/tests/cli/ci/test_dispatch_tests.py b/ddev/tests/cli/ci/test_dispatch_tests.py
index fe8944b6d3db0..a925f091eeaf3 100644
--- a/ddev/tests/cli/ci/test_dispatch_tests.py
+++ b/ddev/tests/cli/ci/test_dispatch_tests.py
@@ -8,6 +8,7 @@
 import json
 import logging
 import subprocess
+from collections import Counter
 from functools import partial
 from typing import TYPE_CHECKING, Any
 from unittest.mock import ANY
@@ -17,13 +18,16 @@
 from ddev.cli.application import Application
 from ddev.cli.ci.dispatch_run import resolve_run
 from ddev.cli.ci.dispatch_tests import attach_datadog_log_handler
+from ddev.cli.ci.tests.batching.exceptions import PlanningError
+from ddev.cli.ci.tests.dispatcher_attributes import metric_tag_mapping
+from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig
 from ddev.cli.ci.tests.dispatcher_logging import dispatcher_datadog_formatter
 from ddev.monitoring import MonitoringRuntime
 from ddev.monitoring.datadog import DatadogLogHandler
 from ddev.monitoring.datadog_metrics import DatadogMetricsSink
 from ddev.utils.git import ChangedFile, ChangeType, GitCommit
-from ddev.utils.github_async.models import PullRequest
-from tests.cli.ci.helpers import HEAD_SHA, PR_NUMBER, listed_pull_request, pulls_page
+from ddev.utils.github_async.models import PullRequest, WorkflowRun
+from tests.cli.ci.helpers import HEAD_SHA, PR_NUMBER, decode_job_list, listed_pull_request, mock_job_result, pulls_page
 from tests.cli.ci.tests.helpers import make_batch, make_job
 from tests.helpers.datadog import FakeLogSubmitter, FakeMetricsSubmitter
 from tests.helpers.monitoring import RecordingJsonHandler, RecordingSink, projector_for
@@ -450,6 +454,50 @@ def test_pytest_args_are_shown_in_the_plan(ddev, github, planned):
     assert '-m "not flaky"' in result.output
 
 
+def test_invocation_options_reach_the_plan_and_batch_workflow(
+    ddev: CliRunner,
+    github: FakeAsyncGitHubClient,
+    planned: MagicMock,
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    step_summary: Path,
+):
+    job = planned.return_value[0].job_list[0]
+    mock_job_result(github, job, 'success')
+    fast_dispatcher_config(mocker)
+    monkeypatch.setattr('ddev.utils.github_async.AsyncGitHubClient', lambda token, rate_limiter=None, **kwargs: github)
+
+    result = ddev(
+        'ci',
+        'dispatch-tests',
+        '--pr',
+        str(PR_NUMBER),
+        '--workflow',
+        'custom-batch.yml',
+        '--workflow-ref',
+        '7.62.x',
+        '--tags',
+        'team:platform',
+        '--pytest-args',
+        '-m "not flaky"',
+        '--output-dir',
+        str(tmp_path),
+    )
+
+    assert result.exit_code == 0, result.output
+    assert 'custom-batch.yml @ 7.62.x' in result.output
+    assert 'team:platform' in result.output
+    assert '-m "not flaky"' in result.output
+    [dispatch] = github.calls_to('create_workflow_dispatch')
+    assert dispatch.kwargs['workflow_id'] == 'custom-batch.yml'
+    assert dispatch.kwargs['ref'] == '7.62.x'
+    inputs = dispatch.kwargs['inputs']
+    assert inputs['pytest_args'] == '-m "not flaky"'
+    [job_input] = decode_job_list(inputs['job_list'])
+    assert 'team:agent-integrations' in job_input['additional_tags'].split(',')
+
+
 @pytest.mark.parametrize(
     'mode_options',
     [[], ['--dry-run'], ['--resolve-only']],
@@ -483,8 +531,15 @@ def resolve(*args: Any, monitor: ComponentMonitor, **kwargs: Any) -> ResolvedRun
     assert result.exit_code == 0, result.output
     if mode_options:
         assert submitter.series == []
+        assert submitter.distributions == []
     else:
-        assert [series['metric'] for series in submitter.series] == ['agent_integrations.test_dispatcher.probe']
+        assert {
+            'agent_integrations.test_dispatcher.probe',
+            'agent_integrations.test_dispatcher.runs.count',
+        } <= {series['metric'] for series in submitter.series}
+        assert [series['metric'] for series in submitter.distributions] == [
+            'agent_integrations.test_dispatcher.run.duration'
+        ]
 
 
 def test_early_exit_disables_monitoring(
@@ -513,8 +568,10 @@ def make_runtime(**kwargs: Any) -> MonitoringRuntime:
     assert result.exit_code == 0, result.output
     assert 'No open pull request matches the requested revision' in result.output
     [monitor] = monitors
+    records_at_exit = list(sink.records)
+    assert sink.records_named('before-exit')
     monitor.metrics.count('after-exit')
-    assert [record.name for record in sink.records] == ['before-exit']
+    assert sink.records == records_at_exit
 
 
 def test_resolved_identity_reaches_planning_even_when_there_are_no_targets(
@@ -549,11 +606,11 @@ def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any)
     assert 'Nothing to test' in result.output
     # An empty plan is a valid outcome, so the run it belongs to is still identified on disk.
     assert (tmp_path / 'run.json').exists()
-    [record] = sink.records
+    [record] = sink.records_named('plan')
     assert record.tags == {
         'repo': 'DataDog/integrations-core',
         'head_sha': 'a-sha',
-        'team': 'platform',
+        'team': 'agent-integrations',
         'component': 'planner',
     }
 
@@ -762,8 +819,7 @@ def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any)
     result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER))
 
     assert result.exit_code == 0, result.output
-    [series] = submitter.series
-    assert series['metric'] == 'agent_integrations.test_dispatcher.planned'
+    [series] = [item for item in submitter.series if item['metric'] == 'agent_integrations.test_dispatcher.planned']
     assert series['type'] == 1
     tags = series['tags']
     assert 'dispatcher.batch.job.environment:py3.13' in tags
@@ -774,6 +830,314 @@ def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any)
     assert not any('blob' in tag for tag in tags)
 
 
+def recording_runtime(mocker: MockerFixture, handler: logging.Handler | None = None) -> RecordingSink:
+    """Deliver the command's metrics into a recording sink, under the production tag policy."""
+    sink = RecordingSink()
+
+    def make_runtime(**kwargs: Any) -> MonitoringRuntime:
+        kwargs['metrics_sink'] = sink
+        kwargs['metrics_tag_projector'] = metric_tag_mapping
+        runtime = MonitoringRuntime(**kwargs)
+        if handler is not None:
+            runtime.add_log_handler(handler)
+        return runtime
+
+    mocker.patch('ddev.monitoring.MonitoringRuntime', make_runtime)
+    return sink
+
+
+def fast_dispatcher_config(mocker: MockerFixture) -> None:
+    """Run the real bus without its production tail: the grace period is a wait for late messages."""
+    config = DispatcherConfig(grace_period_seconds=0.1, global_timeout_seconds=5, poll_interval_seconds=0.01)
+    mocker.patch.object(DispatcherConfig, 'from_repo_config', return_value=config)
+
+
+@pytest.mark.parametrize('global_options', [(), ('-qq',)], ids=['normal', 'quiet'])
+def test_an_executed_run_reports_its_execution_metrics(
+    ddev: CliRunner,
+    github: FakeAsyncGitHubClient,
+    planned: MagicMock,
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    step_summary: Path,
+    global_options: tuple[str, ...],
+):
+    job = planned.return_value[0].job_list[0]
+    mock_job_result(github, job, 'success')
+    sink = recording_runtime(mocker)
+    fast_dispatcher_config(mocker)
+    monkeypatch.setattr('ddev.utils.github_async.AsyncGitHubClient', lambda token, rate_limiter=None, **kwargs: github)
+
+    result = ddev(*global_options, 'ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--output-dir', str(tmp_path))
+
+    assert result.exit_code == 0, result.output
+    cohort = Counter(record.name for record in sink.records if not record.name.startswith('operations.'))
+    assert cohort == Counter(
+        {
+            'batches.count': 1,
+            'batch.jobs.count': 1,
+            'jobs.count': 1,
+            'batch.duration': 1,
+            'artifacts.download.duration': 1,
+            'runs.count': 1,
+            'runs.failed': 1,
+            'runs.planning_failed': 1,
+            'runs.cancelled': 1,
+            'runs.timed_out': 1,
+            'runs.no_op': 1,
+            'run.duration': 1,
+            'jobs.incomplete': 1,
+            'batches.failed': 1,
+            'jobs.failed': 1,
+            'jobs.skipped': 1,
+        }
+    )
+    runs = sink.records_named('runs.count')[0]
+    assert runs.tags['dispatcher.context'] == 'pr'
+    assert runs.tags['git.repository.id_v2'] == 'github.com/datadog/integrations-core'
+    assert runs.tags['dispatcher.component'] == 'dispatcher'
+    # Run-level records carry no job dimensions: a mixed batch must not split them per integration.
+    assert not any(tag.startswith('dispatcher.batch.job') for tag in runs.tags)
+    counted = sink.records_named('jobs.count')[0]
+    assert counted.tags['dispatcher.batch.job.integration'] == 'ntp'
+    assert counted.tags['dispatcher.batch.job.environment'] == 'py3.13'
+    operation_failures = {}
+    for record in sink.records_named('operations.failed'):
+        operation = record.tags['dispatcher.operation']
+        operation_failures[operation] = operation_failures.get(operation, 0) + record.value
+    assert operation_failures == {
+        'dispatch_batch': 0,
+        'fetch_workflow': 0,
+        'refresh_jobs': 0,
+        'collect_artifacts': 0,
+        'gather_batch_results': 0,
+        'publish_report': 0,
+    }
+    attempted = {record.tags['dispatcher.operation'] for record in sink.records_named('operations.count')}
+    assert attempted == {
+        'dispatch_batch',
+        'fetch_workflow',
+        'refresh_jobs',
+        'collect_artifacts',
+        'gather_batch_results',
+        'publish_report',
+    }
+
+
+def test_a_failed_run_reports_failure_metrics_and_counts_itself_once(
+    ddev: CliRunner,
+    github: FakeAsyncGitHubClient,
+    planned: MagicMock,
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    step_summary: Path,
+):
+    job = planned.return_value[0].job_list[0]
+    mock_job_result(github, job, 'failure')
+    github.mock_response(
+        'get_workflow_run',
+        WorkflowRun(
+            id=123,
+            name='test-batch',
+            status='completed',
+            conclusion='failure',
+            html_url='https://github.com/DataDog/integrations-core/actions/runs/123',
+        ),
+    )
+    sink = recording_runtime(mocker)
+    fast_dispatcher_config(mocker)
+    monkeypatch.setattr('ddev.utils.github_async.AsyncGitHubClient', lambda token, rate_limiter=None, **kwargs: github)
+
+    result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--output-dir', str(tmp_path))
+
+    assert result.exit_code == 1, result.output
+    assert 'Dispatcher tests failed.' in result.output
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [1]
+    assert [record.value for record in sink.records_named('batches.failed')] == [1]
+    failed = sink.records_named('jobs.failed')
+    assert [record.value for record in failed] == [1]
+    assert failed[0].tags['dispatcher.batch.job.integration'] == 'ntp'
+    assert [record.value for record in sink.records_named('jobs.incomplete')] == [0]
+    assert [record.value for record in sink.records_named('jobs.skipped')] == [0]
+
+
+def test_a_run_whose_final_report_fails_is_counted_failed_without_invented_job_outcomes(
+    ddev: CliRunner,
+    github: FakeAsyncGitHubClient,
+    planned: MagicMock,
+    mocker: MockerFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    step_summary: Path,
+):
+    job = planned.return_value[0].job_list[0]
+    mock_job_result(github, job, 'success')
+    for method in ('create_issue_comment', 'update_issue_comment', 'list_issue_comments'):
+        github.mock_response(method, RuntimeError('the comment API is down'))
+    sink = recording_runtime(mocker)
+    fast_dispatcher_config(mocker)
+    monkeypatch.setattr('ddev.utils.github_async.AsyncGitHubClient', lambda token, rate_limiter=None, **kwargs: github)
+
+    result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--output-dir', str(tmp_path))
+
+    assert result.exit_code == 1, result.output
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [1]
+    assert [record.value for record in sink.records_named('batches.failed')] == [0]
+    assert [record.value for record in sink.records_named('jobs.failed')] == [0]
+    assert [record.value for record in sink.records_named('jobs.skipped')] == [0]
+    assert [record.value for record in sink.records_named('jobs.incomplete')] == [0]
+
+
+def test_a_planning_failure_is_reported_as_its_own_outcome(
+    ddev: CliRunner, github: FakeAsyncGitHubClient, mocker: MockerFixture, tmp_path: Path
+):
+    handler = RecordingJsonHandler()
+    sink = recording_runtime(mocker, handler=handler)
+    mocker.patch('ddev.cli.ci.dispatch_tests.build_plan', side_effect=PlanningError('the plan is not valid'))
+
+    result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--output-dir', str(tmp_path))
+
+    assert result.exit_code == 1, result.output
+    assert 'Could not build a test plan: the plan is not valid' in result.output
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.planning_failed')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [1]
+    assert [record.value for record in sink.records_named('runs.no_op')] == [0]
+    # No batch ran, so no batch outcome exists; there is no aggregate zero to invent.
+    assert sink.records_named('batches.failed') == []
+    assert not any(record.name.startswith('jobs.') for record in sink.records)
+    [finished] = [event for event in handler.events if event['event'] == 'Dispatcher run finished']
+    assert finished['outcome'] == 'planning-failed'
+    assert finished['level'] == 'error'
+
+
+@pytest.mark.parametrize(
+    ('from_manifest', 'failure_point'),
+    [
+        (False, 'ddev.cli.ci.dispatch_tests.write_run_manifest'),
+        (False, 'ddev.cli.ci.dispatch_tests.changes_for_run'),
+        (False, 'ddev.cli.ci.tests.dispatcher_config.DispatcherConfig.from_repo_config'),
+        (True, 'ddev.cli.ci.dispatch_tests.changes_for_run'),
+        (True, 'ddev.cli.ci.tests.dispatcher_config.DispatcherConfig.from_repo_config'),
+    ],
+    ids=['manifest-write', 'resolved-changes', 'resolved-config', 'manifest-changes', 'manifest-config'],
+)
+def test_an_unexpected_failure_keeps_resolved_identity_and_still_propagates(
+    ddev: CliRunner,
+    github: FakeAsyncGitHubClient,
+    mocker: MockerFixture,
+    tmp_path: Path,
+    from_manifest: bool,
+    failure_point: str,
+):
+    handler = RecordingJsonHandler()
+    sink = recording_runtime(mocker, handler=handler)
+    options = ['--pr', str(PR_NUMBER)]
+    if from_manifest:
+        manifest = tmp_path / 'run.json'
+        manifest.write_text(json.dumps(PULL_REQUEST_RUN_MANIFEST), encoding='utf-8')
+        options = ['--run-manifest', str(manifest)]
+    mocker.patch(failure_point, side_effect=OSError('setup failed'))
+
+    with pytest.raises(OSError, match='setup failed'):
+        ddev('ci', 'dispatch-tests', *options, '--output-dir', str(tmp_path))
+
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.cancelled')] == [0]
+    [failed] = sink.records_named('runs.failed')
+    assert failed.value == 1
+    assert failed.tags['dispatcher.context'] == 'pr'
+    assert failed.tags['git.branch'] == 'hs/a-branch'
+    assert failed.tags['dispatcher.base_branch'] == 'a-target-branch'
+    assert failed.tags['dispatcher.run.is_fork'] == 'false'
+    assert failed.tags['team'] == 'agent-integrations'
+    [finished] = [event for event in handler.events if event['event'] == 'Dispatcher run finished']
+    assert finished['context'] == 'pr'
+    assert finished['pr_number'] == PR_NUMBER
+    assert finished['checkout_sha'] == MERGE_SHA
+    assert finished['head_sha'] == HEAD_SHA
+    assert finished['base_sha'] == BASE_SHA
+
+
+def test_an_interrupt_during_resolution_is_counted_as_cancellation(
+    ddev: CliRunner, github: FakeAsyncGitHubClient, mocker: MockerFixture, tmp_path: Path
+):
+    sink = recording_runtime(mocker)
+    mocker.patch('ddev.cli.ci.dispatch_tests.changes_for_run', side_effect=KeyboardInterrupt)
+
+    result = ddev('ci', 'dispatch-tests', '--pr', str(PR_NUMBER), '--output-dir', str(tmp_path))
+
+    assert result.exit_code == 1
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.cancelled')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [0]
+
+
+@pytest.mark.parametrize('from_manifest', [False, True], ids=['resolution', 'manifest'])
+def test_a_resolution_failure_keeps_unresolved_metric_dimensions(
+    ddev: CliRunner,
+    fake_async_github: FakeAsyncGitHubClient,
+    mocker: MockerFixture,
+    tmp_path: Path,
+    from_manifest: bool,
+):
+    mocker.patch('ddev.cli.ci.dispatch_run.MERGE_COMMIT_REFRESH_SECONDS', 0.0)
+    fake_async_github.mock_response('get_pull_request', pull_request(merge_commit_sha=None))
+    handler = RecordingJsonHandler()
+    sink = recording_runtime(mocker, handler=handler)
+    options = ['--run-manifest', str(tmp_path / 'missing.json')] if from_manifest else ['--pr', str(PR_NUMBER)]
+
+    result = ddev(
+        'ci',
+        'dispatch-tests',
+        *options,
+        '--output-dir',
+        str(tmp_path),
+        '--tags',
+        'team:platform context:caller head_branch:caller base_branch:caller is_fork:false',
+    )
+
+    assert result.exit_code == 1, result.output
+    assert ('Could not read run manifest' if from_manifest else 'reports no merge commit') in result.output
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [1]
+    assert [record.value for record in sink.records_named('runs.no_op')] == [0]
+    assert not any(record.name.startswith('jobs.') for record in sink.records)
+    [failed] = sink.records_named('runs.failed')
+    assert failed.tags == {
+        'team': 'agent-integrations',
+        'git.repository.id_v2': 'github.com/datadog/integrations-core',
+        'dispatcher.context': 'unresolved',
+        'git.branch': 'unresolved',
+        'dispatcher.base_branch': 'unresolved',
+        'dispatcher.run.is_fork': 'unresolved',
+        'dispatcher.component': 'dispatcher',
+    }
+    [started] = [event for event in handler.events if event['event'] == 'Dispatcher invocation started']
+    assert started['team'] == 'agent-integrations'
+    assert all(started[name] == 'unresolved' for name in ('context', 'head_branch', 'base_branch', 'is_fork'))
+
+
+def test_a_superseded_revision_is_a_no_op_run_not_a_failure(
+    ddev: CliRunner, github: FakeAsyncGitHubClient, planned: MagicMock, mocker: MockerFixture
+):
+    github.mock_response('list_pull_requests', pulls_page())
+    sink = recording_runtime(mocker)
+
+    result = ddev('ci', 'dispatch-tests', *HEAD_LOOKUP_OPTIONS)
+
+    assert result.exit_code == 0, result.output
+    assert 'No open pull request matches the requested revision' in result.output
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.no_op')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [0]
+    assert not any(record.name.startswith('jobs.') for record in sink.records)
+
+
 @pytest.mark.usefixtures('resolved_changes')
 @pytest.mark.parametrize('global_options', [(), ('-qq',)], ids=['normal', 'quiet'])
 def test_console_visibility_does_not_change_structured_events(
@@ -812,7 +1176,7 @@ def observe_plan(app: Application, *, monitor: ComponentMonitor, **kwargs: Any)
     [event] = [item for item in json_handler.events if item['event'] == 'planning batches']
     assert event['repo'] == 'DataDog/integrations-core'
     assert event['head_sha'] == 'a-sha'
-    assert event['team'] == 'platform'
+    assert event['team'] == 'agent-integrations'
     assert event['component'] == 'planner'
     [finished] = [item for item in json_handler.events if item['event'] == 'Dispatcher run finished']
     assert finished['outcome'] == 'no-op'
diff --git a/ddev/tests/cli/ci/tests/helpers.py b/ddev/tests/cli/ci/tests/helpers.py
index ff2089ef31c2c..13137c89b3656 100644
--- a/ddev/tests/cli/ci/tests/helpers.py
+++ b/ddev/tests/cli/ci/tests/helpers.py
@@ -12,6 +12,7 @@
 from pydantic import ValidationError
 
 from ddev.cli.ci.tests.batching.units import ResolvedEnvironment, TestUnit
+from ddev.cli.ci.tests.dispatcher_attributes import metric_tag_mapping
 from ddev.cli.ci.tests.messages import BatchJob, TestBatch
 from ddev.cli.ci.tests.progress import (
     BatchProgress,
@@ -23,12 +24,14 @@
 )
 from ddev.cli.ci.tests.status import Status
 from ddev.event_bus.orchestrator import BaseMessage
+from ddev.monitoring import MonitoringRuntime
 from ddev.utils.git import ChangedFile, ChangeType
 from ddev.utils.github_async import GitHubResponse
 from ddev.utils.github_async.models import IssueComment
 from ddev.utils.github_async.models.workflow import WorkflowJobConclusion
 from ddev.utils.junit import JUnitCounts, JUnitReport, JUnitResult, JUnitResultKind, JUnitTestCase, JUnitTestSuite
 from ddev.utils.platform import PlatformName
+from tests.helpers.monitoring import RecordingSink
 
 DEFAULT_PYTHON_VERSION = "3.13"
 DEFAULT_RUNNER_LABELS = ("ubuntu-22.04",)
@@ -110,6 +113,13 @@ def make_batch(*batch_jobs: BatchJob, batch_id: str = "batch-01") -> TestBatch:
     )
 
 
+def recording_runtime() -> tuple[MonitoringRuntime, RecordingSink]:
+    """A monitoring runtime delivering every component's metrics into one recording sink."""
+    sink = RecordingSink()
+    monitoring = MonitoringRuntime(metrics_sink=sink, metrics_tag_projector=metric_tag_mapping)
+    return monitoring, sink
+
+
 def invalid_response_error() -> ValidationError:
     """A jobs-response validation error for an unsupported step status."""
     return ValidationError.from_exception_data(
@@ -263,13 +273,14 @@ def failing_report(*test_names: str) -> JUnitReport:
 
 
 def attempt(
-    status: Status = Status.SUCCESS,
+    status: Status | None = Status.SUCCESS,
     *,
     number: int = 1,
     failed_steps: tuple[str, ...] = (),
-    reports: tuple[JUnitReport, ...] = (),
+    reports: tuple[JUnitReport, ...] | None = (),
     job_url: str | None = JOB_URL,
     error: ProgressError | None = None,
+    state: ExecutionState = ExecutionState.FINISHED,
 ) -> JobAttemptProgress:
     return JobAttemptProgress(
         attempt=number,
@@ -280,6 +291,7 @@ def attempt(
         job_url=job_url,
         reports=reports,
         error=error,
+        state=state,
     )
 
 
diff --git a/ddev/tests/cli/ci/tests/test_dispatcher.py b/ddev/tests/cli/ci/tests/test_dispatcher.py
index b28d2b4a55023..d7096f182abe0 100644
--- a/ddev/tests/cli/ci/tests/test_dispatcher.py
+++ b/ddev/tests/cli/ci/tests/test_dispatcher.py
@@ -6,7 +6,6 @@
 from __future__ import annotations
 
 import asyncio
-import dataclasses
 import logging
 import os
 import signal
@@ -20,15 +19,11 @@
 import httpx
 import pytest
 
+from ddev.cli.ci.dispatch_run import ResolvedRun
+from ddev.cli.ci.dispatch_tests import run_summary
 from ddev.cli.ci.tests import dispatcher as dispatcher_module
 from ddev.cli.ci.tests import rate_limiting
-from ddev.cli.ci.tests.dispatcher import (
-    CANCELLED_RATE_LIMITS,
-    Dispatcher,
-    DispatcherContext,
-    build_dispatcher,
-    message_scope,
-)
+from ddev.cli.ci.tests.dispatcher import CANCELLED_RATE_LIMITS, Dispatcher, build_dispatcher, message_scope
 from ddev.cli.ci.tests.dispatcher_attributes import (
     PROTECTED_RUN_FIELDS,
     console_hidden_fields,
@@ -37,6 +32,7 @@
     run_fields,
 )
 from ddev.cli.ci.tests.dispatcher_config import DispatcherConfig
+from ddev.cli.ci.tests.execution_metrics import ExecutionOutcome
 from ddev.cli.ci.tests.messages import (
     BatchFinished,
     BatchJob,
@@ -72,11 +68,13 @@
     WorkflowRun,
 )
 from ddev.utils.rate_limiting import BucketEvent, InstrumentedAsyncLimiter, RateLimitEvent
+from tests.cli.ci.helpers import mock_job_result
 from tests.cli.ci.tests.helpers import (
     invalid_response_error,
     jobs_reported,
     make_batch,
     make_job,
+    recording_runtime,
 )
 from tests.helpers.github_async import DEFAULT_COMMENT_ID, DEFAULT_DISPATCH_HTML_URL, FakeAsyncGitHubClient
 from tests.helpers.monitoring import RecordingJsonHandler, RecordingSink, projector_for
@@ -85,17 +83,15 @@
 # this the reports land in the real job summary whenever the suite runs inside a workflow.
 pytestmark = pytest.mark.usefixtures("step_summary")
 
-CONTEXT = DispatcherContext(
-    owner="DataDog",
-    repo="integrations-core",
+CONTEXT = ResolvedRun(
+    repository="DataDog/integrations-core",
     checkout_sha="merge-sha-mmm",
     head_sha="head-sha",
     head_branch="a-branch",
-    workflow="test-batch.yml",
-    workflow_ref="master",
+    all_targets=False,
+    pr_number=42,
     base_branch="master",
     base_sha="base-sha",
-    pr_number=42,
 )
 
 
@@ -106,17 +102,19 @@ def build_bus(
     *,
     pr_number: int | None = 42,
     max_timeout: float = 30,
+    monitoring: MonitoringRuntime | None = None,
 ) -> Dispatcher:
     """A Dispatcher over the three real tasks, so the subscriptions under test are production's."""
-    monitoring = MonitoringRuntime()
+    if monitoring is None:
+        monitoring = MonitoringRuntime()
     runner = TaskTestRunner(
         "test-runner",
         client,  # type: ignore[arg-type]
         TestRunnerOptions(
             owner=CONTEXT.owner,
             repo=CONTEXT.repo,
-            workflow_id=CONTEXT.workflow,
-            ref=CONTEXT.workflow_ref,
+            workflow_id="test-batch.yml",
+            ref="master",
             run_fields=run_fields(CONTEXT),
             concurrency_key=CONTEXT.concurrency_key,
             artifacts_base_path=tmp_path / "artifacts",
@@ -126,7 +124,10 @@ def build_bus(
         monitor=monitoring.component('test-runner'),
     )
     gatherer = TaskTestGatherer(
-        "test-gatherer", tmp_path / "results", batches, monitor=monitoring.component('test-gatherer')
+        "test-gatherer",
+        tmp_path / "results",
+        batches,
+        monitor=monitoring.component('test-gatherer'),
     )
     reporter = TaskRunReporter(
         "run-reporter",
@@ -165,16 +166,6 @@ def client(request) -> FakeAsyncGitHubClient:
     return fake
 
 
-def mock_job_result(fake: FakeAsyncGitHubClient, job: BatchJob, conclusion: str) -> None:
-    fake.mock_response(
-        "list_workflow_jobs",
-        WorkflowJobsList(
-            total_count=1,
-            jobs=[WorkflowJob(id=1, run_id=123, name=job.name, status="completed", conclusion=conclusion)],
-        ),
-    )
-
-
 def test_a_batch_travels_from_dispatch_to_the_pull_request_comment(client, tmp_path):
     """The wiring assertion: one batch in, and its result reaches the comment.
 
@@ -249,14 +240,15 @@ def make_client(
 
     monkeypatch.setattr("ddev.utils.github_async.AsyncGitHubClient", make_client)
     monkeypatch.setattr(rate_limiting, "event_logger", lambda _: events.append)
+    monitoring = MonitoringRuntime()
     dispatcher = build_dispatcher(
         batches=[make_batch(job)],
-        context=dataclasses.replace(CONTEXT, pr_number=None),
+        run=CONTEXT.model_copy(update={"pr_number": None}),
         config=DispatcherConfig(grace_period_seconds=0.1, global_timeout_seconds=5),
         token="test-token",
         artifacts_path=tmp_path / "artifacts",
         output_path=tmp_path / "results",
-        monitoring=MonitoringRuntime(),
+        monitoring=monitoring,
     )
 
     dispatcher.run()
@@ -573,6 +565,153 @@ def test_a_comment_write_failure_does_not_prevent_remote_cancellation(
     assert not outcome.successful
 
 
+def test_a_cancelled_run_counts_its_uncollected_jobs_as_incomplete(client: FakeAsyncGitHubClient, tmp_path: Path):
+    monitoring, sink = recording_runtime()
+    dispatcher = build_bus(client, tmp_path, [make_batch(make_job())], monitoring=monitoring)
+    a_run_that_never_finishes(client)
+
+    stop_from_inside_the_run(dispatcher, client, request=ShutdownRequest.cancelled())
+
+    run_summary(
+        monitoring.component('dispatcher'),
+        started=time.monotonic(),
+        outcome=ExecutionOutcome.CANCELLED if dispatcher.cancelled else ExecutionOutcome.FAILED,
+        dispatcher=dispatcher,
+    )
+
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.cancelled')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [0]
+    incomplete = sink.records_named('jobs.incomplete')
+    assert [record.value for record in incomplete] == [1]
+    assert incomplete[0].tags['dispatcher.batch.job.integration'] == 'ntp'
+    assert incomplete[0].tags['dispatcher.component'] == 'dispatcher'
+    assert sink.records_named('jobs.failed') == []
+    assert sink.records_named('batches.failed') == []
+
+
+def test_a_timed_out_run_is_counted_failed_with_its_jobs_incomplete(client: FakeAsyncGitHubClient, tmp_path: Path):
+    monitoring, sink = recording_runtime()
+    dispatcher = build_bus(client, tmp_path, [make_batch(make_job())], max_timeout=0.5, monitoring=monitoring)
+    a_run_that_never_finishes(client)
+
+    dispatcher.run()
+
+    run_summary(
+        monitoring.component('dispatcher'),
+        started=time.monotonic(),
+        outcome=ExecutionOutcome.CANCELLED if dispatcher.cancelled else ExecutionOutcome.FAILED,
+        dispatcher=dispatcher,
+    )
+
+    assert [record.value for record in sink.records_named('runs.count')] == [1]
+    assert [record.value for record in sink.records_named('runs.failed')] == [1]
+    assert [record.value for record in sink.records_named('runs.timed_out')] == [1]
+    assert [record.value for record in sink.records_named('runs.cancelled')] == [0]
+    assert [record.value for record in sink.records_named('jobs.incomplete')] == [1]
+    assert sink.records_named('jobs.failed') == []
+
+
+@pytest.mark.parametrize('cancelled', [False, True], ids=['completed', 'cancelled'])
+def test_accounting_errors_allow_final_reporting_and_remote_cleanup(
+    client: FakeAsyncGitHubClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, cancelled: bool
+):
+    handler = RecordingJsonHandler()
+    monitoring = MonitoringRuntime(console_handler=handler)
+    job = make_job()
+    dispatcher = build_bus(client, tmp_path, [make_batch(job)], monitoring=monitoring)
+
+    def fail_job_fields(job: BatchJob) -> dict[str, Any]:
+        raise ValueError('Cannot build job dimensions')
+
+    monkeypatch.setattr(dispatcher_module, 'job_fields', fail_job_fields)
+    if cancelled:
+        a_run_that_never_finishes(client)
+        stop_from_inside_the_run(dispatcher, client, request=ShutdownRequest.cancelled())
+    else:
+        mock_job_result(client, job, 'success')
+        dispatcher.run()
+
+    outcome = dispatcher.outcome
+    assert outcome is not None
+    assert outcome.cancelled is cancelled
+    assert outcome.successful is not cancelled
+    if cancelled:
+        assert CANCELLED_HEADING in client.last_call('update_issue_comment').kwargs['body']
+    else:
+        assert outcome.final_report_published
+    assert [call.kwargs['run_id'] for call in client.calls_to('cancel_workflow_run')] == ([123] if cancelled else [])
+    [error] = [event for event in handler.events if event['event'] == 'Failed to report incomplete jobs']
+    assert error['level'] == 'error'
+    assert 'Cannot build job dimensions' in error['exception']
+
+
+def test_a_cleanup_failure_keeps_the_collected_accounting(
+    client: FakeAsyncGitHubClient, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+):
+    monitoring, sink = recording_runtime()
+    dispatcher = build_bus(client, tmp_path, [make_batch(make_job())], monitoring=monitoring)
+    a_run_that_never_finishes(client)
+
+    async def fail_cleanup(request: ShutdownRequest) -> None:
+        raise asyncio.CancelledError
+
+    monkeypatch.setattr(dispatcher, '_shutdown_cleanup', fail_cleanup)
+
+    with pytest.raises(asyncio.CancelledError):
+        stop_from_inside_the_run(dispatcher, client, request=ShutdownRequest.cancelled())
+
+    assert dispatcher.outcome is None
+    assert [record.value for record in sink.records_named('jobs.count')] == [1]
+    assert [record.value for record in sink.records_named('jobs.incomplete')] == [1]
+
+
+def test_shutdown_before_progress_counts_only_launched_jobs(client: FakeAsyncGitHubClient, tmp_path: Path):
+    monitoring, sink = recording_runtime()
+    a_run_that_never_finishes(client)
+    batches = [
+        make_batch(make_job('j1')),
+        make_batch(make_job('j2', target='kafka'), batch_id='batch-02'),
+    ]
+    dispatcher = build_bus(client, tmp_path, batches, monitoring=monitoring)
+
+    stop_from_inside_the_run(dispatcher, client, request=ShutdownRequest.cancelled())
+
+    assert dispatcher.outcome is not None
+    assert {batch.state for batch in dispatcher.outcome.progress.batches} == {ExecutionState.PLANNED}
+    assert [record.value for record in sink.records_named('batches.count')] == [1]
+    launched = [record.tags['dispatcher.batch.job.integration'] for record in sink.records_named('jobs.count')]
+    assert launched == ['ntp']
+    incomplete = [
+        (record.value, record.tags['dispatcher.batch.job.integration'])
+        for record in sink.records_named('jobs.incomplete')
+    ]
+    assert incomplete == [(1, 'ntp')]
+
+
+def test_a_jobs_dimensions_stay_identical_from_launch_to_terminal_accounting(
+    client: FakeAsyncGitHubClient, tmp_path: Path
+):
+    monitoring, sink = recording_runtime()
+    job = make_job()
+    mock_job_result(client, job, 'success')
+    dispatcher = build_bus(client, tmp_path, [make_batch(job)], monitoring=monitoring)
+
+    dispatcher.run()
+
+    records = {
+        name: next(record for record in sink.records_named(name))
+        for name in ('jobs.count', 'jobs.failed', 'jobs.skipped', 'jobs.incomplete')
+    }
+    components = {record.tags['dispatcher.component'] for record in records.values()}
+    assert components == {'test-runner', 'test-gatherer', 'dispatcher'}
+    shared = {
+        tuple(sorted((key, value) for key, value in record.tags.items() if key != 'dispatcher.component'))
+        for record in records.values()
+    }
+    assert len(shared) == 1
+
+
 @requires_signals
 def test_a_run_still_winds_down_when_shutdown_mode_cannot_be_entered(client, tmp_path):
     """The signal handler's own failures are invisible: the loop logs them and the next signal, which
@@ -599,17 +738,17 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
 
 
 @pytest.mark.parametrize(
-    ('context', 'fields'),
+    ('run', 'tags', 'fields'),
     [
-        (
+        pytest.param(
             CONTEXT,
+            (),
             {
                 'team': 'agent-integrations',
                 'repo': 'DataDog/integrations-core',
                 'repository_url': 'https://github.com/datadog/integrations-core',
                 'head_sha': 'head-sha',
                 'head_branch': 'a-branch',
-                'is_default_branch': False,
                 'checkout_sha': 'merge-sha-mmm',
                 'is_fork': False,
                 'pr_number': 42,
@@ -617,24 +756,26 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
                 'base_branch': 'master',
                 'base_sha': 'base-sha',
             },
+            id='pull-request',
         ),
-        (
-            dataclasses.replace(
-                CONTEXT,
-                pr_number=None,
-                checkout_sha='a-master-sha',
-                head_sha='a-master-sha',
-                head_branch='master',
-                base_branch=None,
-                base_sha=None,
+        pytest.param(
+            CONTEXT.model_copy(
+                update={
+                    'pr_number': None,
+                    'checkout_sha': 'a-master-sha',
+                    'head_sha': 'a-master-sha',
+                    'head_branch': 'master',
+                    'base_branch': None,
+                    'base_sha': None,
+                }
             ),
+            (),
             {
                 'team': 'agent-integrations',
                 'repo': 'DataDog/integrations-core',
                 'repository_url': 'https://github.com/datadog/integrations-core',
                 'head_sha': 'a-master-sha',
                 'head_branch': 'master',
-                'is_default_branch': True,
                 'checkout_sha': 'a-master-sha',
                 'pr_number': None,
                 'base_branch': None,
@@ -642,30 +783,27 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
                 'is_fork': False,
                 'context': 'master',
             },
+            id='default-branch',
         ),
-        (
-            dataclasses.replace(
-                CONTEXT,
-                tags=(
-                    'repo:sneaky/repo',
-                    'head_sha:sneaky',
-                    'head_branch:sneaky',
-                    'is_default_branch:true',
-                    'checkout_sha:sneaky',
-                    'base_sha:sneaky',
-                    'base_branch:sneaky',
-                    'pr_number:999',
-                    'context:sneaky',
-                    'team:platform',
-                ),
+        pytest.param(
+            CONTEXT,
+            (
+                'repo:sneaky/repo',
+                'head_sha:sneaky',
+                'head_branch:sneaky',
+                'checkout_sha:sneaky',
+                'base_sha:sneaky',
+                'base_branch:sneaky',
+                'pr_number:999',
+                'context:sneaky',
+                'team:platform',
             ),
             {
-                'team': 'platform',
+                'team': 'agent-integrations',
                 'repo': 'DataDog/integrations-core',
                 'repository_url': 'https://github.com/datadog/integrations-core',
                 'head_sha': 'head-sha',
                 'head_branch': 'a-branch',
-                'is_default_branch': False,
                 'checkout_sha': 'merge-sha-mmm',
                 'is_fork': False,
                 'pr_number': 42,
@@ -673,9 +811,11 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
                 'base_branch': 'master',
                 'base_sha': 'base-sha',
             },
+            id='caller-tags',
         ),
-        (
-            dataclasses.replace(CONTEXT, pr_number=None, tags=('context:release',)),
+        pytest.param(
+            CONTEXT.model_copy(update={'pr_number': None}),
+            ('context:release',),
             {
                 'team': 'agent-integrations',
                 'context': 'release',
@@ -683,19 +823,18 @@ def process_message(self, message: BatchFinished | BatchProgressUpdate) -> None:
                 'repository_url': 'https://github.com/datadog/integrations-core',
                 'head_sha': 'head-sha',
                 'head_branch': 'a-branch',
-                'is_default_branch': False,
                 'checkout_sha': 'merge-sha-mmm',
                 'pr_number': None,
                 'is_fork': False,
                 'base_branch': 'master',
                 'base_sha': 'base-sha',
             },
+            id='caller-context',
         ),
     ],
-    ids=['pull-request', 'default-branch', 'caller-tags', 'caller-context'],
 )
-def test_run_fields(context, fields):
-    assert run_fields(context) == fields
+def test_run_fields(run, tags, fields):
+    assert run_fields(run, tags=tags) == fields
 
 
 def test_non_pr_runtime_events_carry_the_resolved_custom_context():
@@ -703,15 +842,9 @@ def test_non_pr_runtime_events_carry_the_resolved_custom_context():
     monitoring = MonitoringRuntime(protected_fields=PROTECTED_RUN_FIELDS)
     monitoring.add_log_handler(handler)
     monitoring.set_run_fields(pr_number='999', base_branch='caller-base', base_sha='caller-sha')
-    context = dataclasses.replace(
-        CONTEXT,
-        pr_number=None,
-        base_branch=None,
-        base_sha=None,
-        tags=('context:test-agent',),
-    )
+    run = CONTEXT.model_copy(update={'pr_number': None, 'base_branch': None, 'base_sha': None})
 
-    monitoring.set_run_fields(**run_fields(context))
+    monitoring.set_run_fields(**run_fields(run, tags=('context:test-agent',)))
     monitoring.component('dispatcher').logger.info('Resolved run')
 
     [event] = handler.events
@@ -771,7 +904,7 @@ def test_the_shared_runtime_is_wired_through_build_dispatcher(client, tmp_path,
 
     dispatcher = build_dispatcher(
         batches=[make_batch(job)],
-        context=CONTEXT,
+        run=CONTEXT,
         config=DispatcherConfig(grace_period_seconds=0.1, global_timeout_seconds=5),
         token="test-token",
         artifacts_path=tmp_path / "artifacts",
@@ -782,7 +915,9 @@ def test_the_shared_runtime_is_wired_through_build_dispatcher(client, tmp_path,
     dispatcher.run()
 
     assert sink.records
-    for record in sink.records:
+    observed = sink.records_named("observed")
+    assert observed
+    for record in observed:
         assert record.tags["batch_id"] == record.tags["tag"]
         assert record.tags["component"] == "test-gatherer"
     queued = [line for line in stream.getvalue().splitlines() if "Queued planned batches" in line]
@@ -801,7 +936,7 @@ def test_a_monitored_run_carries_message_and_workflow_identity_per_event(client,
 
     dispatcher = build_dispatcher(
         batches=[make_batch(job)],
-        context=CONTEXT,
+        run=CONTEXT,
         config=DispatcherConfig(grace_period_seconds=0.1, global_timeout_seconds=5),
         token="test-token",
         artifacts_path=tmp_path / "artifacts",
diff --git a/ddev/tests/cli/ci/tests/test_dispatcher_attributes.py b/ddev/tests/cli/ci/tests/test_dispatcher_attributes.py
index 10f786dafe395..01c0fe6c1f0ea 100644
--- a/ddev/tests/cli/ci/tests/test_dispatcher_attributes.py
+++ b/ddev/tests/cli/ci/tests/test_dispatcher_attributes.py
@@ -27,6 +27,7 @@
     'pr_number': 42,
     'team': 'agent-integrations',
     'component': 'test-runner',
+    'done': False,
     'batch_id': 'batch-01',
     'job_status': 'success',
     'integration': 'postgres',
@@ -41,6 +42,7 @@
         (
             attribute_mapping,
             {
+                'dispatcher.report.done': 'false',
                 'git.repository.id_v2': 'github.com/datadog/integrations-core',
                 'git.repository_url': 'https://github.com/DataDog/Integrations-Core',
                 'git.commit.sha': 'head-sha',
@@ -59,6 +61,7 @@
         (
             log_tag_mapping,
             {
+                'dispatcher.report.done': False,
                 'git.repository.id_v2': 'github.com/datadog/integrations-core',
                 'git.repository_url': 'https://github.com/DataDog/Integrations-Core',
                 'git.commit.sha': 'head-sha',
diff --git a/ddev/tests/cli/ci/tests/test_execution_metrics.py b/ddev/tests/cli/ci/tests/test_execution_metrics.py
new file mode 100644
index 0000000000000..7326dd5347d1b
--- /dev/null
+++ b/ddev/tests/cli/ci/tests/test_execution_metrics.py
@@ -0,0 +1,89 @@
+# (C) Datadog, Inc. 2026-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+"""Tests for the metrics helper: operation outcomes and operation timing."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import pytest
+
+from ddev.cli.ci.tests.execution_metrics import MetricsHelper, Operation
+from ddev.monitoring.metrics import MetricKind
+from tests.cli.ci.tests.helpers import recording_runtime
+from tests.helpers.clock import FakeClock
+from tests.helpers.monitoring import RecordingSink
+
+OPERATION = 'dispatcher.operation'
+
+
+def helper_with_sink(clock: Callable[[], float]) -> tuple[MetricsHelper, RecordingSink]:
+    monitoring, sink = recording_runtime()
+    return MetricsHelper(monitoring.component('test-runner').metrics, clock=clock), sink
+
+
+def values(sink: RecordingSink, name: str) -> list[float]:
+    return [record.value for record in sink.records_named(name)]
+
+
+@pytest.mark.parametrize('failed', [False, True], ids=['settled-healthy', 'settled-failed'])
+def test_record_operation_emits_one_attempt_and_its_settled_outcome(failed: bool):
+    metrics, sink = helper_with_sink(FakeClock())
+
+    metrics.record_operation(Operation.DISPATCH_BATCH, failed=failed)
+
+    assert values(sink, 'operations.count') == [1]
+    assert values(sink, 'operations.failed') == [int(failed)]
+    assert sink.records_named('operations.count')[0].tags[OPERATION] == 'dispatch_batch'
+
+
+@pytest.mark.parametrize('failed', [False, True], ids=['healthy', 'recovered-failure'])
+def test_a_timed_operation_reports_its_elapsed_time_and_outcome_once(failed: bool):
+    clock = FakeClock(100.0)
+    metrics, sink = helper_with_sink(clock)
+
+    with metrics.time_operation(Operation.COLLECT_ARTIFACTS, duration_metric='artifacts.download.duration') as result:
+        result.failed = failed
+        clock.advance(37.5)
+
+    duration = sink.records_named('artifacts.download.duration')
+    assert values(sink, 'artifacts.download.duration') == [37.5]
+    assert duration[0].kind is MetricKind.DISTRIBUTION
+    assert duration[0].tags[OPERATION] == 'collect_artifacts'
+    assert values(sink, 'operations.count') == [1]
+    assert values(sink, 'operations.failed') == [int(failed)]
+
+
+def test_an_escaped_exception_fails_the_operation_and_propagates_unchanged():
+    clock = FakeClock(100.0)
+    metrics, sink = helper_with_sink(clock)
+
+    error = RuntimeError('boom')
+    with (
+        pytest.raises(RuntimeError) as exc_info,
+        metrics.time_operation(Operation.COLLECT_ARTIFACTS, duration_metric='artifacts.download.duration'),
+    ):
+        clock.advance(5.0)
+        raise error
+
+    assert exc_info.value is error
+    assert values(sink, 'operations.count') == [1]
+    assert values(sink, 'operations.failed') == [1]
+    assert values(sink, 'artifacts.download.duration') == [5.0]
+
+
+def test_cancellation_propagates_without_settling_the_operation():
+    clock = FakeClock(100.0)
+    metrics, sink = helper_with_sink(clock)
+
+    with (
+        pytest.raises(KeyboardInterrupt),
+        metrics.time_operation(Operation.COLLECT_ARTIFACTS, duration_metric='artifacts.download.duration'),
+    ):
+        clock.advance(3.0)
+        raise KeyboardInterrupt
+
+    assert sink.records_named('operations.count') == []
+    assert sink.records_named('operations.failed') == []
+    assert values(sink, 'artifacts.download.duration') == [3.0]
diff --git a/ddev/tests/cli/ci/tests/test_task_run_reporter.py b/ddev/tests/cli/ci/tests/test_task_run_reporter.py
index aaa76c49f642c..b9b08f9c71241 100644
--- a/ddev/tests/cli/ci/tests/test_task_run_reporter.py
+++ b/ddev/tests/cli/ci/tests/test_task_run_reporter.py
@@ -24,6 +24,7 @@
 from ddev.cli.ci.tests.status import Status
 from ddev.cli.ci.tests.task_run_reporter import RunReporterOptions, TaskRunReporter
 from ddev.event_bus.shutdown import ShutdownKind, ShutdownRequest
+from ddev.monitoring import ComponentMonitor
 from ddev.utils.github_async import GitHubResponse
 from ddev.utils.github_async.models import IssueComment
 from ddev.utils.github_errors import GitHubAuthenticationError, GitHubBodyTooLongError
@@ -38,6 +39,7 @@
     failing_report,
     job_progress,
     jobs_reported,
+    recording_runtime,
     uniform_progress,
 )
 from tests.helpers.github_async import DEFAULT_COMMENT_ID, FakeAsyncGitHubClient
@@ -71,12 +73,13 @@ def _reporter(
     pr_number: int | None = PR_NUMBER,
     handler: logging.Handler | None = None,
     clock: Callable[[], datetime] | None = None,
+    monitor: ComponentMonitor | None = None,
 ) -> TaskRunReporter:
     return TaskRunReporter(
         "run-reporter",
         client,
         RunReporterOptions(owner=OWNER, repo=REPO, pr_number=pr_number),
-        monitor=make_monitor('run-reporter', handler=handler),
+        monitor=monitor or make_monitor('run-reporter', handler=handler),
         clock=clock,
     )
 
@@ -211,10 +214,14 @@ def test_a_comment_we_cannot_edit_is_replaced_by_one_we_own(error: httpx.HTTPSta
     client = FakeAsyncGitHubClient()
     client.mock_response("list_issue_comments", comment_page(_marked_comment(77, "not ours")))
     client.mock_response("update_issue_comment", error, once=True)
-    reporter = _reporter(client)
+    monitoring, sink = recording_runtime()
+    reporter = _reporter(client, monitor=monitoring.component("run-reporter"))
 
     asyncio.run(reporter.process_message(_update(1)))
 
+    assert [record.value for record in sink.records_named("operations.count")] == [1]
+    assert [record.value for record in sink.records_named("operations.failed")] == [0]
+
     # First it tried to edit the comment it found, then it created its own.
     assert len(client.calls_to("update_issue_comment")) == 1
     assert len(client.calls_to("create_issue_comment")) == 1
@@ -474,6 +481,66 @@ def test_a_token_refused_for_every_comment_gives_up_rather_than_looping():
     assert reporter.pr_comment_failed
 
 
+def test_each_publication_reports_one_operation_outcome():
+    monitoring, sink = recording_runtime()
+    monitor = monitoring.component("run-reporter")
+    reporter = _reporter(FakeAsyncGitHubClient(), monitor=monitor)
+
+    with monitor.scope(done=False, revision=1):
+        asyncio.run(reporter.process_message(_update(1)))
+    with monitor.scope(done=True, revision=2):
+        asyncio.run(reporter.process_message(_update(2, done=True)))
+
+    assert [record.value for record in sink.records_named("operations.count")] == [1, 1]
+    assert [record.value for record in sink.records_named("operations.failed")] == [0, 0]
+    assert all(
+        record.tags["dispatcher.operation"] == "publish_report" for record in sink.records_named("operations.count")
+    )
+
+
+def test_a_write_that_settled_as_failed_reports_one_failed_operation():
+    client = FakeAsyncGitHubClient()
+    client.mock_response("list_issue_comments", comment_page(_marked_comment()))
+    client.mock_response("update_issue_comment", _auth_error(403))
+    client.mock_response("create_issue_comment", _auth_error(403))
+    monitoring, sink = recording_runtime()
+    reporter = _reporter(client, monitor=monitoring.component("run-reporter"))
+
+    asyncio.run(reporter.process_message(_update(1)))
+
+    assert [record.value for record in sink.records_named("operations.count")] == [1]
+    assert [record.value for record in sink.records_named("operations.failed")] == [1]
+
+
+def test_an_escaped_write_error_fails_the_publication_operation_and_propagates():
+    client = FakeAsyncGitHubClient()
+    client.mock_response("create_issue_comment", RuntimeError("the comment API is down"))
+    monitoring, sink = recording_runtime()
+    reporter = _reporter(client, monitor=monitoring.component("run-reporter"))
+
+    with pytest.raises(RuntimeError, match="the comment API is down"):
+        asyncio.run(reporter.process_message(_update(1)))
+
+    assert [record.value for record in sink.records_named("operations.failed")] == [1]
+
+
+@pytest.mark.parametrize(
+    ("pr_number", "revisions", "expected"),
+    [
+        pytest.param(PR_NUMBER, [1, 1], [1], id="stale-revision"),
+        pytest.param(None, [1], [], id="no-pull-request"),
+    ],
+)
+def test_an_ignored_publication_reports_no_operation(pr_number: int | None, revisions: list[int], expected: list[int]):
+    monitoring, sink = recording_runtime()
+    reporter = _reporter(FakeAsyncGitHubClient(), pr_number=pr_number, monitor=monitoring.component("run-reporter"))
+
+    for revision in revisions:
+        asyncio.run(reporter.process_message(_update(revision)))
+
+    assert [record.value for record in sink.records_named("operations.count")] == expected
+
+
 # ---------------------------------------------------------------------------
 # Body too long
 # ---------------------------------------------------------------------------
@@ -893,7 +960,8 @@ async def test_a_shutdown_report_waiting_on_the_report_lock_expires_without_canc
 ):
     """A terminal lock timeout must leave the active writer and subsequent publication usable."""
     client = FakeAsyncGitHubClient()
-    reporter = _reporter(client)
+    monitoring, sink = recording_runtime()
+    reporter = _reporter(client, monitor=monitoring.component("run-reporter"))
     write_started = asyncio.Event()
     release_write = asyncio.Event()
     create_comment = client.create_issue_comment
@@ -917,6 +985,7 @@ async def blocked_create_comment(owner: str, repo: str, pr_number: int, body: st
                     await reporter.publish_shutdown(_shutdown_request(ShutdownKind.CANCELLED))
 
         assert (reporter.latest_body, reporter.pr_comment_failed) == retained
+        assert sink.records_named("operations.count") == []
         release_write.set()
         await asyncio.wait_for(progress_write, timeout=2)
         assert not reporter.pr_comment_failed
@@ -924,6 +993,8 @@ async def blocked_create_comment(owner: str, repo: str, pr_number: int, body: st
         await reporter.publish_shutdown(_shutdown_request(ShutdownKind.CANCELLED))
         body = client.last_call("update_issue_comment").kwargs["body"]
         assert SHUTDOWN_HEADINGS[ShutdownKind.CANCELLED] in body
+        assert [record.value for record in sink.records_named("operations.count")] == [1, 1]
+        assert [record.value for record in sink.records_named("operations.failed")] == [0, 0]
     finally:
         release_write.set()
         await asyncio.gather(progress_write, return_exceptions=True)
diff --git a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py
index a544761b29338..91be86598b458 100644
--- a/ddev/tests/cli/ci/tests/test_task_test_gatherer.py
+++ b/ddev/tests/cli/ci/tests/test_task_test_gatherer.py
@@ -36,10 +36,17 @@
 from ddev.cli.ci.tests.task_run_reporter import RunReporterOptions, TaskRunReporter
 from ddev.cli.ci.tests.task_test_gatherer import INITIAL_UPDATE_MESSAGE_ID, TaskTestGatherer
 from ddev.event_bus.orchestrator import BaseMessage, EventBusOrchestrator
+from ddev.monitoring import ComponentMonitor
 from ddev.utils.github_async.models import JobStep, WorkflowJob, WorkflowJobConclusion, WorkflowJobStatus
 from ddev.utils.junit import TestStatus
 from ddev.utils.platform import PlatformName
-from tests.cli.ci.tests.helpers import RecordingBus, drain_queue, jobs_reported, make_job
+from tests.cli.ci.tests.helpers import (
+    RecordingBus,
+    drain_queue,
+    jobs_reported,
+    make_job,
+    recording_runtime,
+)
 from tests.helpers.github_async import FakeAsyncGitHubClient
 from tests.helpers.monitoring import RecordingJsonHandler, make_monitor
 
@@ -145,6 +152,7 @@ def _make_gatherer(
     plan: dict[str, list[BatchJob]] | None = None,
     *,
     handler: logging.Handler | None = None,
+    monitor: ComponentMonitor | None = None,
 ) -> TaskTestGatherer:
     """Gatherer primed with the complete plan, given as ``{batch_id: planned jobs}``."""
     if plan is None:
@@ -153,7 +161,7 @@ def _make_gatherer(
         "gatherer",
         output_base_path=tmp_path / "out",
         batches=[_test_batch(batch_id, jobs) for batch_id, jobs in plan.items()],
-        monitor=make_monitor('test-gatherer', handler=handler),
+        monitor=monitor or make_monitor('test-gatherer', handler=handler),
     )
     gatherer.bus = RecordingBus()  # type: ignore[assignment]
     return gatherer
@@ -382,7 +390,8 @@ def test_a_shutting_down_bus_abandons_gathering_without_registering_the_batch(
     artifacts = tmp_path / "artifacts" / "100"
     job_dir = _make_job_tree(artifacts, "j1")
 
-    gatherer = _make_gatherer(tmp_path)
+    monitoring, sink = recording_runtime()
+    gatherer = _make_gatherer(tmp_path, monitor=monitoring.component("test-gatherer"))
     bus = RecordingBus()
     gatherer.bus = bus  # type: ignore[assignment]
     start_stopping(gatherer, bus)
@@ -397,6 +406,9 @@ def test_a_shutting_down_bus_abandons_gathering_without_registering_the_batch(
     assert _registry(gatherer) == []
     # Still planned, so nothing downstream can read the batch as one that finished.
     assert gatherer._progress_by_batch["batch-1"].state is ExecutionState.PLANNED
+    assert sink.records_named("jobs.failed") == []
+    assert sink.records_named("batches.failed") == []
+    assert sink.records_named("operations.count") == []
 
 
 def test_happy_path_organizes_artifacts_and_emits_update(tmp_path: Path):
@@ -429,6 +441,95 @@ def test_happy_path_organizes_artifacts_and_emits_update(tmp_path: Path):
     assert (tmp_path / "out" / "test_results" / "ntp_py3.13_linux-test-e2e-py3.13.xml").is_file()
 
 
+def test_an_accepted_final_result_reports_the_batches_outcomes(tmp_path: Path):
+    """An accepted commit reports its batch's outcome and a zero-filled flag per collected job."""
+    monitoring, sink = recording_runtime()
+    j1 = _batch_job("j1")
+    j2 = _batch_job("j2", target="kafka")
+    j3 = _batch_job("j3", target="redis")
+    gatherer = _make_gatherer(tmp_path, {"batch-1": [j1, j2, j3]}, monitor=monitoring.component("test-gatherer"))
+
+    artifacts = tmp_path / "artifacts" / "100"
+    j1_dir = _make_job_tree(artifacts, "j1", junit=JUNIT_FAILING, e2e=False)
+    j2_dir = _make_job_tree(artifacts, "j2", e2e=False)
+    j3_dir = _make_job_tree(artifacts, "j3", e2e=False)
+    gatherer.process_message(
+        _batch_finished(
+            artifacts,
+            status="failure",
+            batch_jobs=[
+                _batch_job_result(j1, _workflow_job("j1", "failure"), j1_dir),
+                _batch_job_result(j2, _workflow_job("j2", "skipped"), j2_dir),
+                _batch_job_result(j3, _workflow_job("j3", "success"), j3_dir),
+            ],
+        )
+    )
+
+    drain_queue(gatherer.bus.queue)
+    assert [record.value for record in sink.records_named("batches.failed")] == [1]
+    failed = [
+        (record.value, record.tags["dispatcher.batch.job.integration"]) for record in sink.records_named("jobs.failed")
+    ]
+    assert failed == [(1, "ntp"), (0, "kafka"), (0, "redis")]
+    skipped = [
+        (record.value, record.tags["dispatcher.batch.job.integration"]) for record in sink.records_named("jobs.skipped")
+    ]
+    assert skipped == [(0, "ntp"), (1, "kafka"), (0, "redis")]
+    assert {record.tags["dispatcher.component"] for record in sink.records_named("jobs.failed")} == {"test-gatherer"}
+    assert sink.records_named("jobs.incomplete") == []
+    gathered_operations = [
+        record
+        for record in sink.records_named("operations.count")
+        if record.tags["dispatcher.operation"] == "gather_batch_results"
+    ]
+    assert [record.value for record in gathered_operations] == [1]
+    assert {record.value for record in sink.records_named("operations.failed")} == {0}
+
+
+def test_an_observed_attempt_the_commit_never_collected_reports_no_outcome(tmp_path: Path):
+    """An observed job is not a collected result, so the commit reports no outcome for it."""
+    monitoring, sink = recording_runtime()
+    j1 = _batch_job("j1")
+    j2 = _batch_job("j2", target="kafka")
+    gatherer = _make_gatherer(tmp_path, {"batch-1": [j1, j2]}, monitor=monitoring.component("test-gatherer"))
+
+    gatherer.process_message(_progress_update(WorkflowJob(id=2, run_id=100, name="j2", status="in_progress")))
+    drain_queue(gatherer.bus.queue)
+    artifacts = tmp_path / "artifacts" / "100"
+    j1_dir = _make_job_tree(artifacts, "j1", e2e=False)
+    gatherer.process_message(
+        _batch_finished(artifacts, batch_jobs=[_batch_job_result(j1, _workflow_job("j1", "success"), j1_dir)])
+    )
+    drain_queue(gatherer.bus.queue)
+
+    reported = [record.tags["dispatcher.batch.job.integration"] for record in sink.records_named("jobs.failed")]
+    assert reported == ["ntp"]
+    assert [record.value for record in sink.records_named("jobs.failed")] == [0]
+    assert [
+        (record.value, record.tags["dispatcher.batch.job.integration"]) for record in sink.records_named("jobs.skipped")
+    ] == [(0, "ntp")]
+
+
+def test_a_duplicate_final_message_does_not_report_the_batch_twice(tmp_path: Path):
+    monitoring, sink = recording_runtime()
+    job = _batch_job("j1")
+    gatherer = _make_gatherer(tmp_path, monitor=monitoring.component("test-gatherer"))
+
+    artifacts = tmp_path / "artifacts" / "100"
+    job_dir = _make_job_tree(artifacts, "j1", junit=JUNIT_FAILING, e2e=False)
+    batch = _batch_finished(
+        artifacts,
+        status="failure",
+        batch_jobs=[_batch_job_result(job, _workflow_job("j1", "failure"), job_dir)],
+    )
+    gatherer.process_message(batch)
+    gatherer.process_message(dataclasses.replace(batch, id="replay"))
+
+    assert [record.value for record in sink.records_named("jobs.failed")] == [1]
+    assert [record.value for record in sink.records_named("batches.failed")] == [1]
+    assert [record.value for record in sink.records_named("operations.count")] == [1]
+
+
 def test_failure_path_records_failed_steps_and_reports(tmp_path: Path):
     artifacts = tmp_path / "artifacts" / "100"
     job_dir = _make_job_tree(artifacts, "j1", junit=JUNIT_FAILING, e2e=False)
@@ -730,13 +831,16 @@ def test_missing_workflow_job_raises(tmp_path: Path):
     # Correlation is the runner's job; a job without a workflow job on a non-timed-out batch is a bug.
     artifacts = tmp_path / "artifacts" / "100"
     job_dir = _make_job_tree(artifacts, "j1")
+    monitoring, sink = recording_runtime()
+    gatherer = _make_gatherer(tmp_path, monitor=monitoring.component("test-gatherer"))
 
-    gatherer = _make_gatherer(tmp_path)
     with pytest.raises(ValueError, match="No workflow job correlated"):
         gatherer.process_message(
             _batch_finished(artifacts, batch_jobs=[_batch_job_result(make_job("j1"), None, job_dir)])
         )
 
+    assert [record.value for record in sink.records_named("operations.failed")] == [1]
+
 
 def test_empty_batch_jobs_has_no_entry_in_the_registry(tmp_path: Path) -> None:
     # Nothing gathered, so the registry stays empty; only the aggregate can say "finished empty".
@@ -1061,8 +1165,9 @@ def test_timed_out_batch_is_recorded_on_the_batch(tmp_path: Path) -> None:
 
 def test_concurrent_batches_produce_one_revision_each(tmp_path: Path) -> None:
     # Concurrent batches land on different threads: one revision each, no loss, no repeat.
+    monitoring, sink = recording_runtime()
     plan = {f"b{index}": [_scenario_batch_job(f"int{index}")] for index in range(1, 6)}
-    gatherer = _make_gatherer(tmp_path, plan)
+    gatherer = _make_gatherer(tmp_path, plan, monitor=monitoring.component("test-gatherer"))
 
     messages = []
     for index in range(1, 6):
@@ -1087,6 +1192,8 @@ def gather(message) -> None:
     final = max(updates, key=lambda update: update.revision)
     assert {batch.state for batch in final.progress.batches} == {ExecutionState.FINISHED}
     assert (final.progress.passed, final.progress.complete, final.progress.total) == (5, 5, 5)
+    # Five concurrent commits, each accepted once: no batch is reported twice.
+    assert [record.value for record in sink.records_named("batches.failed")] == [0] * 5
 
 
 def test_missing_artifact_dir_is_recorded_as_an_attempt_error(tmp_path: Path) -> None:
diff --git a/ddev/tests/cli/ci/tests/test_task_test_runner.py b/ddev/tests/cli/ci/tests/test_task_test_runner.py
index 7c57e76a1278c..e4d278167316b 100644
--- a/ddev/tests/cli/ci/tests/test_task_test_runner.py
+++ b/ddev/tests/cli/ci/tests/test_task_test_runner.py
@@ -6,8 +6,6 @@
 from __future__ import annotations
 
 import asyncio
-import base64
-import gzip
 import json
 import logging
 import secrets
@@ -17,8 +15,8 @@
 import pytest
 from pydantic import ValidationError
 
+from ddev.cli.ci.dispatch_run import ResolvedRun
 from ddev.cli.ci.tests import messages
-from ddev.cli.ci.tests.dispatcher import DispatcherContext
 from ddev.cli.ci.tests.dispatcher_attributes import run_fields
 from ddev.cli.ci.tests.messages import BatchFinished, BatchJob, TestBatch
 from ddev.cli.ci.tests.progress import ExecutionState
@@ -29,8 +27,11 @@
     JobListTooLargeError,
     TaskTestRunner,
     TestRunnerOptions,
+    workflow_duration_seconds,
 )
 from ddev.event_bus.exceptions import FatalProcessingError
+from ddev.monitoring import ComponentMonitor
+from ddev.monitoring.metrics import MetricKind
 from ddev.utils.github_async import GitHubResponse
 from ddev.utils.github_async.models import (
     Artifact,
@@ -41,14 +42,16 @@
     WorkflowJobStatus,
     WorkflowRun,
 )
+from tests.cli.ci.helpers import decode_job_list
 from tests.cli.ci.tests.helpers import (
     RecordingBus,
     drain_queue,
     invalid_response_error,
     make_job,
+    recording_runtime,
 )
 from tests.helpers.github_async import DEFAULT_DISPATCH_HTML_URL, FakeAsyncGitHubClient
-from tests.helpers.monitoring import RecordingJsonHandler, make_monitor
+from tests.helpers.monitoring import RecordingJsonHandler, RecordingSink, make_monitor
 
 # ---------------------------------------------------------------------------
 # Helpers
@@ -59,10 +62,6 @@ def wrap(data: Any) -> GitHubResponse[Any]:
     return GitHubResponse(data=data, headers={})
 
 
-def decode_job_list(encoded: str) -> list[dict[str, Any]]:
-    return json.loads(gzip.decompress(base64.b64decode(encoded)).decode())
-
-
 DEFAULT_URL = object()
 
 
@@ -78,13 +77,20 @@ def make_artifact(idx: int, expired: bool = False, archive_download_url: Any = D
     )
 
 
-def make_workflow_run(status: str = "completed", conclusion: str | None = "success") -> WorkflowRun:
+def make_workflow_run(
+    status: str = "completed",
+    conclusion: str | None = "success",
+    run_started_at: str | None = None,
+    updated_at: str | None = None,
+) -> WorkflowRun:
     return WorkflowRun(
         id=123,
         name="test-batch",
         status=status,
         conclusion=conclusion if status == "completed" else None,
         html_url="https://github.com/o/r/actions/runs/123",
+        run_started_at=run_started_at,
+        updated_at=updated_at,
     )
 
 
@@ -119,28 +125,28 @@ def make_runner(
     origin_run_url: str | None = None,
     pr_number: int | None = None,
     handler: logging.Handler | None = None,
-    context: DispatcherContext | None = None,
+    run: ResolvedRun | None = None,
+    tags: tuple[str, ...] = (),
+    monitor: ComponentMonitor | None = None,
 ) -> TaskTestRunner:
-    context = context or DispatcherContext(
-        owner="DataDog",
-        repo="integrations-core",
-        workflow="test-batch.yaml",
-        workflow_ref="master",
+    run = run or ResolvedRun(
+        repository="DataDog/integrations-core",
         head_sha="head-sha-aaa",
         checkout_sha="merge-sha-bbb",
         head_branch="a-branch",
+        all_targets=False,
         base_branch="master",
         base_sha="base-sha-ccc",
         pr_number=123,
         is_fork=is_fork,
     )
     options = TestRunnerOptions(
-        owner=context.owner,
-        repo=context.repo,
-        workflow_id=context.workflow,
-        ref=context.workflow_ref,
-        run_fields=run_fields(context),
-        concurrency_key=context.concurrency_key,
+        owner=run.owner,
+        repo=run.repo,
+        workflow_id="test-batch.yaml",
+        ref="master",
+        run_fields=run_fields(run, tags=tags),
+        concurrency_key=run.concurrency_key,
         artifacts_base_path=tmp_path,
         poll_interval_seconds=0.0,
         pytest_args=pytest_args,
@@ -152,7 +158,7 @@ def make_runner(
         client=client,  # type: ignore[arg-type]
         options=options,
         artifact_client=artifact_client or client,  # type: ignore[arg-type]
-        monitor=make_monitor('test-runner', handler=handler),
+        monitor=monitor or make_monitor('test-runner', handler=handler),
     )
     runner.bus = RecordingBus()  # type: ignore[assignment]
     return runner
@@ -195,6 +201,126 @@ async def run_happy_path(tmp_path: Path) -> tuple[FakeAsyncGitHubClient, BatchFi
     return fake, finished
 
 
+@pytest.mark.asyncio
+async def test_healthy_attempts_report_zero_operation_failures(tmp_path: Path):
+    fake = FakeAsyncGitHubClient()
+    fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
+    mock_artifacts(fake, [make_artifact(1), make_artifact(2)])
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
+
+    await runner.process_message(
+        TestBatch(
+            id="msg-1",
+            batch_id="batch-1",
+            job_list=[make_job("j1"), make_job("j2")],
+            jobs_count=2,
+            integrations=["ntp", "kafka"],
+        )
+    )
+
+    assert finished_messages(runner)
+    counted = {}
+    for record in sink.records_named('operations.count'):
+        counted[record.tags['dispatcher.operation']] = counted.get(record.tags['dispatcher.operation'], 0) + 1
+    assert counted == {"dispatch_batch": 1, "fetch_workflow": 1, "refresh_jobs": 2, "collect_artifacts": 1}
+    assert failed_by_operation(sink) == {
+        "dispatch_batch": 0,
+        "fetch_workflow": 0,
+        "refresh_jobs": 0,
+        "collect_artifacts": 0,
+    }
+    assert {record.tags['dispatcher.component'] for record in sink.records_named('operations.count')} == {'test-runner'}
+    assert [record.kind.value for record in sink.records_named('artifacts.download.duration')] == ['distribution']
+
+
+@pytest.mark.asyncio
+async def test_an_accepted_dispatch_counts_the_batch_and_its_jobs(tmp_path: Path):
+    fake = FakeAsyncGitHubClient()
+    fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
+    mock_artifacts(fake, [])
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
+    batch = TestBatch(
+        id="msg-1",
+        batch_id="batch-1",
+        job_list=[make_job("j1"), make_job("j2", target="kafka")],
+        jobs_count=2,
+        integrations=["ntp", "kafka"],
+    )
+
+    await runner.process_message(batch)
+
+    assert [record.value for record in sink.records_named('batches.count')] == [1]
+    assert [record.value for record in sink.records_named('batch.jobs.count')] == [2]
+    counted = sink.records_named('jobs.count')
+    assert [record.value for record in counted] == [1, 1]
+    assert [record.tags['dispatcher.batch.job.integration'] for record in counted] == ['ntp', 'kafka']
+    assert {record.tags['dispatcher.component'] for record in counted} == {'test-runner'}
+
+
+@pytest.mark.asyncio
+async def test_a_completed_workflow_reports_its_own_running_duration(tmp_path: Path):
+    """`batch.duration` is the workflow's running time, not the dispatcher's."""
+    fake = FakeAsyncGitHubClient()
+    fake.mock_response(
+        "get_workflow_run",
+        make_workflow_run(
+            "completed", "success", run_started_at="2026-01-01T10:00:00Z", updated_at="2026-01-01T10:01:30Z"
+        ),
+    )
+    mock_artifacts(fake, [])
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
+
+    await runner.process_message(make_batch())
+
+    duration = sink.records_named("batch.duration")
+    assert [record.value for record in duration] == [90.0]
+    assert duration[0].kind is MetricKind.DISTRIBUTION
+
+
+@pytest.mark.asyncio
+async def test_a_completed_workflow_without_timing_reports_no_batch_duration(tmp_path: Path):
+    """A missing start is omitted, not approximated from another clock."""
+    fake = FakeAsyncGitHubClient()
+    fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
+    mock_artifacts(fake, [])
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
+
+    await runner.process_message(make_batch())
+
+    assert sink.records_named("batch.duration") == []
+
+
+@pytest.mark.parametrize(
+    ("run_started_at", "updated_at", "expected"),
+    [
+        pytest.param("2026-01-01T10:00:00Z", "2026-01-01T10:01:30Z", 90.0, id="valid"),
+        pytest.param("2026-01-01T10:00:00Z", "2026-01-01T10:00:00Z", 0.0, id="genuine-zero"),
+        pytest.param("2026-01-01T10:00:00", "2026-01-01T10:01:30", 90.0, id="naive-but-consistent"),
+        pytest.param("2026-01-01T10:00:00Z", "2026-01-01T10:01:30", None, id="mixed-zones"),
+        pytest.param("2026-01-01T10:01:30Z", "2026-01-01T10:00:00Z", None, id="reversed"),
+        pytest.param("not-a-timestamp", "2026-01-01T10:00:00Z", None, id="invalid-start"),
+        pytest.param(None, "2026-01-01T10:00:00Z", None, id="missing-start"),
+        pytest.param("2026-01-01T10:00:00Z", None, None, id="missing-end"),
+    ],
+)
+def test_workflow_duration_comes_only_from_valid_ordered_timestamps(
+    run_started_at: str | None, updated_at: str | None, expected: float | None
+):
+    run = WorkflowRun(
+        id=123,
+        status="completed",
+        html_url="https://github.com/o/r/actions/runs/123",
+        run_started_at=run_started_at,
+        updated_at=updated_at,
+    )
+
+    assert workflow_duration_seconds(run) == expected
+
+
 # ---------------------------------------------------------------------------
 # conclusion_to_status
 # ---------------------------------------------------------------------------
@@ -241,6 +367,28 @@ async def observe_first_poll(*args: Any, **kwargs: Any) -> GitHubResponse[Workfl
     await runner.process_message(make_batch())
 
 
+def failed_by_operation(sink: RecordingSink) -> dict[str, float]:
+    totals: dict[str, float] = {}
+    for record in sink.records_named('operations.failed'):
+        totals[record.tags['dispatcher.operation']] = totals.get(record.tags['dispatcher.operation'], 0) + record.value
+    return totals
+
+
+async def test_a_dispatch_github_refuses_fails_the_dispatch_operation_only(tmp_path: Path):
+    client = FakeAsyncGitHubClient()
+    client.mock_response("create_workflow_dispatch", RuntimeError("dispatch refused"))
+    monitoring, sink = recording_runtime()
+    runner = make_runner(client, tmp_path, monitor=monitoring.component("test-runner"))
+
+    with pytest.raises(RuntimeError, match="dispatch refused"):
+        await runner.process_message(make_batch())
+
+    assert sink.records_named("batches.count") == []
+    assert sink.records_named("jobs.count") == []
+    assert failed_by_operation(sink) == {"dispatch_batch": 1}
+    assert [record.value for record in sink.records_named("operations.count")] == [1]
+
+
 async def test_collection_publishes_outcomes_before_a_cancellable_artifact_request(tmp_path: Path):
     client, artifacts = FakeAsyncGitHubClient(), FakeAsyncGitHubClient()
     batch = make_batch()
@@ -254,7 +402,8 @@ async def blocked_download(*args: Any, **kwargs: Any) -> None:
         await asyncio.Future()
 
     artifacts.download_artifact = blocked_download  # type: ignore[method-assign]
-    runner = make_runner(client, tmp_path, artifact_client=artifacts)
+    monitoring, sink = recording_runtime()
+    runner = make_runner(client, tmp_path, artifact_client=artifacts, monitor=monitoring.component("test-runner"))
     task = asyncio.create_task(runner.process_message(batch))
     try:
         async with asyncio.timeout(5):
@@ -272,6 +421,16 @@ async def blocked_download(*args: Any, **kwargs: Any) -> None:
         with pytest.raises(asyncio.CancelledError):
             await task
 
+    # A cancelled download never settled, so it counts as no operation outcome at all, but the
+    # time it took is still reported.
+    settled = [
+        record
+        for record in sink.records_named("operations.count")
+        if record.tags["dispatcher.operation"] == "collect_artifacts"
+    ]
+    assert settled == []
+    assert [record.kind for record in sink.records_named("artifacts.download.duration")] == [MetricKind.DISTRIBUTION]
+
     await runner.cancel_dispatched_runs()
     assert client.calls_to("cancel_workflow_run") == []
     assert drain_queue(runner.bus.queue) == []
@@ -344,35 +503,34 @@ async def test_dispatches_workflow_with_job_list_payload(tmp_path: Path):
 
 
 @pytest.mark.parametrize(
-    ("context", "expected_context", "has_pr_tags"),
+    ("run", "tags", "expected_context", "concurrency_key", "has_pr_tags"),
     [
-        pytest.param(None, "pr", True, id="pull-request"),
+        pytest.param(None, (), "pr", "pr-123", True, id="pull-request"),
         pytest.param(
-            DispatcherContext(
-                owner="DataDog",
-                repo="integrations-core",
-                workflow="test-batch.yaml",
-                workflow_ref="master",
+            ResolvedRun(
+                repository="DataDog/integrations-core",
                 head_sha="master-sha",
                 checkout_sha="master-sha",
                 head_branch="master",
+                all_targets=False,
             ),
+            (),
             "master",
+            "master-sha",
             False,
             id="master",
         ),
         pytest.param(
-            DispatcherContext(
-                owner="DataDog",
-                repo="integrations-core",
-                workflow="test-batch.yaml",
-                workflow_ref="master",
+            ResolvedRun(
+                repository="DataDog/integrations-core",
                 head_sha="agent-sha",
                 checkout_sha="agent-sha",
                 head_branch="test-agent",
-                tags=("context:test-agent",),
+                all_targets=False,
             ),
+            ("context:test-agent",),
             "test-agent",
+            "agent-sha",
             False,
             id="custom-context",
         ),
@@ -380,38 +538,38 @@ async def test_dispatches_workflow_with_job_list_payload(tmp_path: Path):
 )
 def test_run_identity_and_context_reach_workflow_and_job_tags(
     tmp_path: Path,
-    context: DispatcherContext | None,
+    run: ResolvedRun | None,
+    tags: tuple[str, ...],
     expected_context: str,
+    concurrency_key: str,
     has_pr_tags: bool,
 ):
-    runner = make_runner(FakeAsyncGitHubClient(), tmp_path, context=context)
+    runner = make_runner(FakeAsyncGitHubClient(), tmp_path, run=run, tags=tags)
 
     inputs = runner._build_inputs(make_batch("batch-context"))
     [job] = decode_job_list(inputs["job_list"])
-    tags = job["additional_tags"].split(",")
+    job_tags = job["additional_tags"].split(",")
 
     assert inputs["context"] == expected_context
-    assert inputs["checkout_sha"] == (context.checkout_sha if context else "merge-sha-bbb")
-    assert inputs["head_sha"] == (context.head_sha if context else "head-sha-aaa")
-    assert inputs["head_branch"] == (context.head_branch if context else "a-branch")
-    assert f"dispatcher.context:{expected_context}" in tags
-    assert ("dispatcher.pr.number:123" in tags) is has_pr_tags
-    assert ("dispatcher.base_branch:master" in tags) is has_pr_tags
-    assert not any(tag.startswith(("git.", "dispatcher.head_sha", "dispatcher.head_branch")) for tag in tags)
+    assert inputs["concurrency_key"] == concurrency_key
+    assert inputs["checkout_sha"] == (run.checkout_sha if run else "merge-sha-bbb")
+    assert inputs["head_sha"] == (run.head_sha if run else "head-sha-aaa")
+    assert inputs["head_branch"] == (run.head_branch if run else "a-branch")
+    assert f"dispatcher.context:{expected_context}" in job_tags
+    assert ("dispatcher.pr.number:123" in job_tags) is has_pr_tags
+    assert ("dispatcher.base_branch:master" in job_tags) is has_pr_tags
+    assert not any(tag.startswith(("git.", "dispatcher.head_sha", "dispatcher.head_branch")) for tag in job_tags)
 
 
 def test_job_tags_keep_caller_values_inside_one_transport_field(tmp_path: Path):
-    context = DispatcherContext(
-        owner="DataDog",
-        repo="integrations-core",
-        workflow="test-batch.yaml",
-        workflow_ref="master",
+    run = ResolvedRun(
+        repository="DataDog/integrations-core",
         head_sha="agent-sha",
         checkout_sha="agent-sha",
         head_branch="test-agent",
-        tags=("context:release,candidate\nunsafe\rvalue",),
+        all_targets=False,
     )
-    runner = make_runner(FakeAsyncGitHubClient(), tmp_path, context=context)
+    runner = make_runner(FakeAsyncGitHubClient(), tmp_path, run=run, tags=("context:release,candidate\nunsafe\rvalue",))
 
     inputs = runner._build_inputs(make_batch("batch-context"))
     [job] = decode_job_list(inputs["job_list"])
@@ -683,7 +841,8 @@ async def test_process_message_emits_batch_finished_when_listing_jobs_fails(tmp_
     fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
     mock_artifacts(fake, [])
     fake.mock_response("list_workflow_jobs", RuntimeError("boom-list-jobs"))
-    runner = make_runner(fake, tmp_path)
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
 
     # A failure listing jobs must not abort the batch: BatchFinished is still emitted, each
     # correlated job carrying no workflow job.
@@ -692,6 +851,11 @@ async def test_process_message_emits_batch_finished_when_listing_jobs_fails(tmp_
     finished = finished_messages(runner)[0]
     assert finished.status == "success"
     assert all(result.workflow_job is None for result in finished.batch_jobs)
+    assert [
+        record.value
+        for record in sink.records_named("operations.failed")
+        if record.tags["dispatcher.operation"] == "refresh_jobs"
+    ] == [1, 1]
 
 
 @pytest.mark.parametrize(
@@ -796,28 +960,30 @@ async def test_final_results_preserve_only_completed_observations(tmp_path: Path
     assert finished.batch_jobs[1].workflow_job is None
 
 
+@pytest.mark.parametrize(
+    "unavailable",
+    [
+        pytest.param(make_artifact(2, expired=True), id="expired"),
+        pytest.param(make_artifact(2, archive_download_url=None), id="missing-url"),
+    ],
+)
 @pytest.mark.asyncio
-async def test_process_message_skips_expired_artifacts(tmp_path: Path):
+async def test_unavailable_artifacts_degrade_collection(tmp_path: Path, unavailable: Artifact):
     fake = FakeAsyncGitHubClient()
     fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
-    mock_artifacts(
-        fake,
-        [
-            make_artifact(1),
-            make_artifact(2, expired=True),
-            make_artifact(3, archive_download_url=None),
-        ],
-    )
-    runner = make_runner(fake, tmp_path)
+    mock_artifacts(fake, [make_artifact(1), unavailable])
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
 
     await runner.process_message(
         TestBatch(id="batch-4", batch_id="batch-4", job_list=[make_job()], jobs_count=1, integrations=["ntp"])
     )
 
-    # Only the non-expired artifact with a download URL should be fetched.
     download_calls = fake.calls_to("download_artifact")
     assert len(download_calls) == 1
     assert download_calls[0].kwargs["archive_download_url"] == "https://api.github.com/artifact/1/zip"
+    assert failed_by_operation(sink)["collect_artifacts"] == 1
+    assert [finished.status for finished in finished_messages(runner)] == [Status.SUCCESS]
 
 
 @pytest.mark.asyncio
@@ -825,7 +991,8 @@ async def test_process_message_emits_batch_finished_when_listing_artifacts_fails
     fake = FakeAsyncGitHubClient()
     fake.mock_response("get_workflow_run", make_workflow_run("completed", "success"))
     fake.mock_response("list_workflow_run_artifacts", RuntimeError("boom-list-artifacts"))
-    runner = make_runner(fake, tmp_path)
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
 
     # A failure listing artifacts must not abort the batch: exactly one BatchFinished is still
     # emitted, with the workflow's real conclusion.
@@ -835,6 +1002,13 @@ async def test_process_message_emits_batch_finished_when_listing_artifacts_fails
     assert len(submitted) == 1
     finished = submitted[0]
     assert finished.status == "success"
+    assert failed_by_operation(sink) == {
+        "dispatch_batch": 0,
+        "fetch_workflow": 0,
+        "refresh_jobs": 0,
+        "collect_artifacts": 1,
+    }
+    assert [record.kind for record in sink.records_named("artifacts.download.duration")] == [MetricKind.DISTRIBUTION]
 
 
 @pytest.mark.asyncio
@@ -847,7 +1021,8 @@ async def test_download_failure_for_one_artifact_does_not_abort_others(tmp_path:
         RuntimeError("download failure for artifact 2"),
         archive_download_url="https://api.github.com/artifact/2/zip",
     )
-    runner = make_runner(fake, tmp_path)
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
 
     await runner.process_message(make_batch())
 
@@ -861,6 +1036,8 @@ async def test_download_failure_for_one_artifact_does_not_abort_others(tmp_path:
     submitted = finished_messages(runner)
     assert len(submitted) == 1
     assert submitted[0].status == "success"
+    assert failed_by_operation(sink)["collect_artifacts"] == 1
+    assert [record.kind for record in sink.records_named("artifacts.download.duration")] == [MetricKind.DISTRIBUTION]
 
 
 # ---------------------------------------------------------------------------
@@ -879,16 +1056,57 @@ def running_run() -> WorkflowRun:
 
 
 @pytest.mark.parametrize(
-    ("failure_point", "operation", "workflow_running", "cancelled_runs"),
+    ("failure_point", "operation", "workflow_running", "cancelled_runs", "operation_failures", "durations"),
     [
         # The dispatch response could not be parsed, so a dispatched run's ID is unknown and
         # nothing is tracked to cancel.
-        pytest.param("create_workflow_dispatch", "dispatching the batch", False, [], id="dispatch-response"),
-        pytest.param("get_workflow_run", "polling workflow status", True, [123], id="poll-response"),
+        pytest.param(
+            "create_workflow_dispatch",
+            "dispatching the batch",
+            False,
+            [],
+            {"dispatch_batch": 1},
+            [],
+            id="dispatch-response",
+        ),
+        pytest.param(
+            "get_workflow_run",
+            "polling workflow status",
+            True,
+            [123],
+            {"dispatch_batch": 0, "fetch_workflow": 1},
+            [],
+            id="poll-response",
+        ),
         # Jobs are refreshed on every poll, so the page can fail while the workflow is still going.
-        pytest.param("list_workflow_jobs", "listing workflow jobs", True, [123], id="jobs-page"),
+        pytest.param(
+            "list_workflow_jobs",
+            "listing workflow jobs",
+            True,
+            [123],
+            {"dispatch_batch": 0, "fetch_workflow": 0, "refresh_jobs": 1},
+            [],
+            id="jobs-page",
+        ),
+        pytest.param(
+            "list_workflow_jobs",
+            "listing workflow jobs",
+            False,
+            [],
+            {"dispatch_batch": 0, "fetch_workflow": 0, "refresh_jobs": 1},
+            [90.0],
+            id="completed-jobs-page",
+        ),
         # Artifacts are collected after completion, when the run is already released.
-        pytest.param("list_workflow_run_artifacts", "listing workflow artifacts", False, [], id="artifact-page"),
+        pytest.param(
+            "list_workflow_run_artifacts",
+            "listing workflow artifacts",
+            False,
+            [],
+            {"dispatch_batch": 0, "fetch_workflow": 0, "refresh_jobs": 0, "collect_artifacts": 1},
+            [90.0],
+            id="artifact-page",
+        ),
     ],
 )
 @pytest.mark.asyncio
@@ -898,13 +1116,16 @@ async def test_an_unparsable_response_stops_the_batch_and_keeps_its_run_cancella
     operation: str,
     workflow_running: bool,
     cancelled_runs: list[int],
+    operation_failures: dict[str, float],
+    durations: list[float],
 ):
     """An invalid response stops the batch without losing a known unfinished run."""
     fake = FakeAsyncGitHubClient()
     if workflow_running:
         fake.mock_response("get_workflow_run", running_run())
     fake.mock_response(failure_point, invalid_response_error())
-    runner = make_runner(fake, tmp_path)
+    monitoring, sink = recording_runtime()
+    runner = make_runner(fake, tmp_path, monitor=monitoring.component("test-runner"))
 
     # Bound the test if an invalid response is retried indefinitely.
     with pytest.raises(FatalProcessingError, match=f"Invalid GitHub response while {operation}"):
@@ -912,6 +1133,8 @@ async def test_an_unparsable_response_stops_the_batch_and_keeps_its_run_cancella
             await runner.process_message(make_batch())
 
     assert finished_messages(runner) == []
+    assert failed_by_operation(sink) == operation_failures
+    assert [record.value for record in sink.records_named('batch.duration')] == durations
 
     await runner.cancel_dispatched_runs()
     cancelled = [call.kwargs["run_id"] for call in fake.calls_to("cancel_workflow_run")]
diff --git a/ddev/tests/helpers/github_async.py b/ddev/tests/helpers/github_async.py
index 0bd95d5cee2a0..8b07ce314b6ab 100644
--- a/ddev/tests/helpers/github_async.py
+++ b/ddev/tests/helpers/github_async.py
@@ -177,6 +177,8 @@ def _default_response_factories() -> dict[str, Callable[[], Any]]:
                 status='completed',
                 conclusion='success',
                 html_url='https://github.com/o/r/actions/runs/123',
+                run_started_at='2026-01-01T10:00:00Z',
+                updated_at='2026-01-01T10:01:30Z',
             ),
             headers={},
         ),
diff --git a/ddev/tests/monitoring/test_monitoring.py b/ddev/tests/monitoring/test_monitoring.py
index 246bcf88990ae..de83edcfee0f6 100644
--- a/ddev/tests/monitoring/test_monitoring.py
+++ b/ddev/tests/monitoring/test_monitoring.py
@@ -55,9 +55,9 @@ def test_component_binding_and_shared_scopes_enrich_logs_and_metrics(
     runtime = MonitoringRuntime(
         metrics_sink=sink,
         metrics_tag_projector=projector_for('repo', 'component', 'operation', 'batch_id'),
+        base_fields={'repo': 'DataDog/integrations-core'},
     )
     runtime.add_log_handler(handler)
-    runtime.set_run_fields(repo='DataDog/integrations-core')
     monitor = runtime.component('planner').bind(operation='dispatch')
 
     with runtime.component('dispatcher').scope(batch_id='batch-01'):
@@ -80,7 +80,7 @@ def test_component_binding_and_shared_scopes_enrich_logs_and_metrics(
 
 
 def test_resolving_run_fields_affects_future_events_not_emitted_ones(handler: RecordingJsonHandler):
-    runtime = MonitoringRuntime()
+    runtime = MonitoringRuntime(base_fields={'commit': 'unresolved'})
     runtime.add_log_handler(handler)
     monitor = runtime.component('resolution')
     monitor.logger.info('Resolving run')
@@ -88,7 +88,7 @@ def test_resolving_run_fields_affects_future_events_not_emitted_ones(handler: Re
     runtime.set_run_fields(commit='head-sha')
     monitor.logger.info('Resolved run')
 
-    assert [event.get('commit') for event in handler.events] == [None, 'head-sha']
+    assert [event['commit'] for event in handler.events] == ['unresolved', 'head-sha']
 
 
 @pytest.mark.parametrize('error', [ValueError('boom'), asyncio.CancelledError()])