diff --git a/.github/workflows/server-parity.yml b/.github/workflows/server-parity.yml
index 7e134b458..8365e6a7c 100644
--- a/.github/workflows/server-parity.yml
+++ b/.github/workflows/server-parity.yml
@@ -37,6 +37,7 @@ on:
       - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs
       - tests/Unit/ServerParityServiceCatalogContractTest.mjs
       - tests/Unit/ServerParitySystemMetricsContractTest.mjs
+      - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs
       - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
@@ -88,6 +89,7 @@ on:
       - tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs
       - tests/Unit/ServerParityServiceCatalogContractTest.mjs
       - tests/Unit/ServerParitySystemMetricsContractTest.mjs
+      - tests/Unit/ServerParityPrometheusMetricsContractTest.mjs
       - tests/Unit/ServerParityNamespaceDeleteContractTest.mjs
       - tests/Unit/ServerParityWaitingHistoryContractTest.mjs
       - tests/Unit/ServerParityTaskRetryContractTest.mjs
@@ -1506,7 +1508,7 @@ jobs:
             -e COMPOSER_HOME=/tmp/composer -v "$PWD:/source" -w /source "$image" \
             install --working-dir=scripts/conformance/server-parity --no-interaction --no-progress
           docker run --rm --user=1000:1000 --entrypoint node -v "$PWD:/source:ro" -w /source \
-            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt
+            "$image" --test tests/Unit/ServerParityRunnerTest.mjs tests/Unit/ServerParityCooperativeContractTest.mjs tests/Unit/ServerParityChildCancellationContractTest.mjs tests/Unit/ServerParityScheduleContractTest.mjs tests/Unit/ServerParityVisibilityContractTest.mjs tests/Unit/ServerParityAdmissionContractTest.mjs tests/Unit/ServerParityRoleTokenContractTest.mjs tests/Unit/ServerParityWorkerDeregistrationContractTest.mjs tests/Unit/ServerParitySearchAttributeContractTest.mjs tests/Unit/ServerParityWorkerListingContractTest.mjs tests/Unit/ServerParityWorkerReplacementContractTest.mjs tests/Unit/ServerParityPatchDeploymentContractTest.mjs tests/Unit/ServerParityWaitingHistoryContractTest.mjs tests/Unit/ServerParityTaskRetryContractTest.mjs tests/Unit/ServerParityPublishedPatchContractTest.mjs tests/Unit/ServerParityPublishedPatchTransportTest.mjs tests/Unit/ServerParityRecordFileTest.mjs tests/Unit/ServerParityLateWorkflowCompletionContractTest.mjs tests/Unit/ServerParityStorageTestContractTest.mjs tests/Unit/ServerParitySchedulePoliciesContractTest.mjs tests/Unit/ServerParitySignalsBeforeWorkerContractTest.mjs tests/Unit/ServerParityScheduleCatalogContractTest.mjs tests/Unit/ServerParityOperatorMetricsContractTest.mjs tests/Unit/ServerParityActivityLeaseRenewalContractTest.mjs tests/Unit/ServerParityServiceCatalogContractTest.mjs tests/Unit/ServerParitySystemMetricsContractTest.mjs tests/Unit/ServerParityPrometheusMetricsContractTest.mjs tests/Unit/ServerParityNamespaceDeleteContractTest.mjs > parity-evidence/comparator-tests.txt
           docker run --rm --user=1000:1000 --cpus=1 --memory=256m --memory-swap=256m \
             --network=none --entrypoint node -v "$PWD:/source:ro" -w /source \
             "$image" --test tests/Unit/ServerParityCodecDeadlineLoadTest.mjs > parity-evidence/codec-load-tests.txt
diff --git a/docs/rust-server-port-checkpoints/2026-10-22-prometheus-metrics.md b/docs/rust-server-port-checkpoints/2026-10-22-prometheus-metrics.md
new file mode 100644
index 000000000..b3e4feda1
--- /dev/null
+++ b/docs/rust-server-port-checkpoints/2026-10-22-prometheus-metrics.md
@@ -0,0 +1,58 @@
+# October 22, Prometheus metrics summary
+
+- **Source.** Server 2.11.1 `SystemController::prometheusMetrics` and
+  `PrometheusMetricsSummary::snapshot`, with `BoundedMetricPolicy` and the
+  declared dimensions in `config/dw-bounded-growth.php`.
+- **Served.** Native answers admin-only `GET /api/system/prometheus-metrics`
+  in the default or a named namespace:
+  - workflow series by task queue and workflow type: started, completed,
+    failed, cancelled and terminated totals, running runs and a latency
+    histogram from run summaries;
+  - activity series by task queue, workflow type and activity type, with the
+    same counters and a latency histogram from activity executions;
+  - task-queue runtime state: ready, due, delayed, leased and expired tasks,
+    tasks added and dispatched in the last minute, and fresh worker pollers,
+    slot capacity and available slots.
+
+  Each family keeps the first 100 series in PHP's order. Native reads at
+  most 101 candidate label sets, then aggregates only the series up to the
+  last candidate. The disclosure reports observed, reported and suppressed
+  series and the suppressed starts, with `lower_bound` precision once the
+  limit is reached. Native refuses the series-limit settings, so the limit is
+  always PHP's default of 100.
+- **Times.** Task and worker comparisons bind whole seconds, as Laravel does
+  and as native's operator metrics already do. A task made available within
+  the current second still counts as delayed. Workers are fresh within
+  `DW_WORKER_STALE_AFTER_SECONDS`, by default `max(3 × heartbeat, 30)`.
+- **Decision: cumulative latency buckets.** Published PHP binds each bucket
+  bound as text. SQLite orders every number before every text value, so PHP
+  on SQLite counts every duration in every bucket
+  ([server#683](https://github.com/durable-workflow/server/issues/683)).
+  Native compares durations with numeric bounds and reports the cumulative
+  histogram on every backend. The fixture accepts PHP's SQLite shape as a
+  recorded reference defect; native's tests pin the cumulative buckets with
+  fixed durations on SQLite, PostgreSQL and MySQL.
+- **Correction: undispatched activity tasks.** PHP's worker completion
+  publishes only the timer tasks it creates. An activity task scheduled by a
+  workflow task therefore keeps null `last_dispatched_at` and
+  `last_dispatch_attempt_at` until a poller claims it, and its queue's
+  dispatched-in-the-last-minute counter stays 0. Native stamped it
+  dispatched. It now leaves those columns null for that path, as PHP does.
+  Other task paths keep their stamps.
+- **Fixture.** `prometheus-metrics` reads a fresh namespace, then:
+  - registers a worker with four workflow and three activity slots on the
+    first queue;
+  - completes a run there after one activity;
+  - terminates a second run there;
+  - starts a run on a second queue that has no worker.
+
+  Counters, task-queue state, worker capacity and the disclosure are exact.
+  Each counted run and activity lasts at least 1.1 seconds, so latency sums
+  and the empty low buckets are checked against that bound.
+- **Evidence.** Native tests pass on SQLite, PostgreSQL and MySQL, and the
+  native suite with `--all-targets` passes on SQLite (649 tests).
+  A local copy of the SQLite parity step passed all 53 fixtures on embedded
+  PHP and native, including `prometheus-metrics`. Published PHP passed
+  `prometheus-metrics` too; its only difference was `storage-test`, an
+  intermittent 503 from PHP's external payload registration on that loaded
+  host, which passed in the previous local run.
diff --git a/docs/rust-server-port.md b/docs/rust-server-port.md
index bfe201c7a..fbb50e0dc 100644
--- a/docs/rust-server-port.md
+++ b/docs/rust-server-port.md
@@ -102,14 +102,17 @@ These published PHP routes on this tree have no native handler yet, by family.
 
 - **Used by the CLI or Waterline.**
   - Standalone activities: list, describe and start.
-  - The Prometheus metrics summary, the repair pass, and external payload
-    cleanup and its pass. (Namespace delete is served with the
-    `namespace-delete` fixture. The JSON `GET|POST /api/system/metrics` is
-    served with the `system-metrics` fixture: namespace failure, durable-state
-    and payload counts are exact; native enforces no admission, durable-state
-    or payload quota, so limits are null and rejections zero, as on an
-    unconfigured PHP server. Projection drift is server-wide, and native's
-    only drift source is a run without its summary.)
+  - The repair pass, and external payload cleanup and its pass. (Namespace
+    delete is served with the `namespace-delete` fixture. The JSON
+    `GET|POST /api/system/metrics` is served with the `system-metrics`
+    fixture: namespace failure, durable-state and payload counts are exact;
+    native enforces no admission, durable-state or payload quota, so limits
+    are null and rejections zero, as on an unconfigured PHP server.
+    Projection drift is server-wide, and native's only drift source is a run
+    without its summary. The Prometheus summary
+    `GET /api/system/prometheus-metrics` is served with the
+    `prometheus-metrics` fixture, with PHP's default limit of 100 series per
+    family.)
   - Task queue priority and fairness.
 - **Worker protocol features.**
   - Worker sessions.
diff --git a/rust/src/runtime/execution.rs b/rust/src/runtime/execution.rs
index 762114d58..3eb9ca2bb 100644
--- a/rust/src/runtime/execution.rs
+++ b/rust/src/runtime/execution.rs
@@ -340,6 +340,7 @@ impl Runtime {
     delegate!(operator_dashboard(config: &super::operator_metrics::MetricsConfig, bounded: bool) -> Result<Value>);
     delegate!(repair_status() -> Result<Value>);
     delegate!(system_metrics() -> Result<Value>);
+    delegate!(prometheus_metrics() -> Result<Value>);
     delegate!(retention_status(limit: i64) -> Result<Value>);
     delegate!(retention_pass(limit: i64, run_ids: Vec<String>) -> Result<(Value, bool)>);
     delegate!(activity_timeout_status(limit: i64) -> Result<Value>);
diff --git a/rust/src/runtime/http.rs b/rust/src/runtime/http.rs
index 87d7dbda7..d9b6deecb 100644
--- a/rust/src/runtime/http.rs
+++ b/rust/src/runtime/http.rs
@@ -204,6 +204,7 @@ pub fn router(runtime: Runtime) -> Router {
         .route("/api/system/operator-metrics", get(operator_metrics))
         .route("/api/system/health", get(system_health))
         .route("/api/system/metrics", get(system_metrics).post(system_metrics))
+        .route("/api/system/prometheus-metrics", get(prometheus_metrics))
         .route("/api/system/operator-dashboard", get(operator_dashboard))
         .route("/api/system/operator-dashboard/bounded", get(bounded_operator_dashboard))
         .route(
@@ -650,6 +651,9 @@ fn named_namespace_operation(method: &str, path: &str) -> bool {
     if path == "/api/system/metrics" {
         return matches!(method, "GET" | "POST");
     }
+    if path == "/api/system/prometheus-metrics" {
+        return method == "GET";
+    }
     if let Some(path) = path.strip_prefix("/api/workflows/") {
         let parts: Vec<_> = path.split('/').collect();
         if matches!(
@@ -1482,6 +1486,10 @@ async fn system_metrics(Extension(runtime): Extension<Runtime>) -> Result<Json<V
     Ok(Json(runtime.system_metrics().await?))
 }
 
+async fn prometheus_metrics(Extension(runtime): Extension<Runtime>) -> Result<Json<Value>> {
+    Ok(Json(runtime.prometheus_metrics().await?))
+}
+
 /// PHP answers 503 while a check reports an error.
 async fn system_health(Extension(runtime): Extension<Runtime>) -> Result<Response> {
     let (unavailable, body) = runtime.system_health(&metrics_config(&runtime)).await?;
diff --git a/rust/src/runtime/mod.rs b/rust/src/runtime/mod.rs
index 43733c7fd..59f3042c0 100644
--- a/rust/src/runtime/mod.rs
+++ b/rust/src/runtime/mod.rs
@@ -73,6 +73,7 @@ mod payload_limits;
 mod poll_readiness;
 pub mod postgres;
 mod previews;
+mod prometheus_metrics;
 mod queries;
 mod query_authority;
 mod repair_candidates;
diff --git a/rust/src/runtime/operator_metrics.rs b/rust/src/runtime/operator_metrics.rs
index c0ec3cac5..2574ab2f5 100644
--- a/rust/src/runtime/operator_metrics.rs
+++ b/rust/src/runtime/operator_metrics.rs
@@ -25,7 +25,7 @@ use sqlx::{ColumnIndex, Decode, Encode, Executor, IntoArguments, Type};
 pub(super) const REDISPATCH_AFTER_SECONDS: i64 = 3;
 
 /// Laravel binds a bare timestamp as `Y-m-d H:i:s`.
-fn whole_second(at: DateTime<Utc>) -> DateTime<Utc> {
+pub(super) fn whole_second(at: DateTime<Utc>) -> DateTime<Utc> {
     at.with_nanosecond(0).unwrap_or(at)
 }
 
diff --git a/rust/src/runtime/prometheus_metrics.rs b/rust/src/runtime/prometheus_metrics.rs
new file mode 100644
index 000000000..2c41497c8
--- /dev/null
+++ b/rust/src/runtime/prometheus_metrics.rs
@@ -0,0 +1,508 @@
+//! `GET /api/system/prometheus-metrics`, as published Server 2.11 serves it
+//! (`PrometheusMetricsSummary::snapshot`): the namespace's workflow, activity
+//! and task-queue series for Prometheus scrapers, each family bounded with its
+//! suppression disclosed.
+//!
+//! Native refuses the series-limit settings, so every family uses PHP's
+//! default limit. Latency buckets compare each duration with a numeric bound
+//! on every backend. Published PHP binds the bounds as text, which SQLite
+//! orders after every number, so PHP on SQLite counts every duration in every
+//! bucket (server#683); native reports the cumulative histogram there too.
+//! Task and worker times compare with whole seconds, as Laravel binds them.
+use super::Result;
+use super::operator_metrics::whole_second;
+use super::store::now;
+use chrono::{DateTime, Duration, Utc};
+use serde_json::{Map, Value, json};
+use std::collections::HashMap;
+
+/// PHP's default series limit for each family.
+const SERIES_LIMIT: usize = 100;
+
+/// `LATENCY_BUCKETS_SECONDS`, keyed as PHP casts each bound to a string.
+const BUCKETS: [(&str, f64); 10] = [
+    ("0.1", 0.1),
+    ("0.5", 0.5),
+    ("1", 1.0),
+    ("2.5", 2.5),
+    ("5", 5.0),
+    ("10", 10.0),
+    ("30", 30.0),
+    ("60", 60.0),
+    ("300", 300.0),
+    ("900", 900.0),
+];
+
+const WORKFLOW_SELECTION: &str = "bounded_task_queue_and_workflow_type_ascending";
+const ACTIVITY_SELECTION: &str = "bounded_task_queue_workflow_type_and_activity_type_ascending";
+const QUEUE_SELECTION: &str = "task_queue_name_ascending";
+
+/// Cumulative bucket counts of a duration expression, one column per bound.
+macro_rules! buckets {
+    ($available:literal, $duration:literal) => {
+        concat!(
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=0.1 THEN 1 END) AS bucket_0,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=0.5 THEN 1 END) AS bucket_1,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=1 THEN 1 END) AS bucket_2,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=2.5 THEN 1 END) AS bucket_3,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=5 THEN 1 END) AS bucket_4,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=10 THEN 1 END) AS bucket_5,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=30 THEN 1 END) AS bucket_6,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=60 THEN 1 END) AS bucket_7,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=300 THEN 1 END) AS bucket_8,",
+            "COUNT(CASE WHEN ", $available, " AND ", $duration, "<=900 THEN 1 END) AS bucket_9"
+        )
+    };
+}
+
+/// The workflow series up to and including the last candidate (queue, type).
+macro_rules! workflow_rows {
+    ($float:literal) => {
+        concat!(
+            "SELECT COALESCE(queue,'default') AS task_queue,workflow_type,COUNT(*) AS started_total,",
+            "COUNT(CASE WHEN status='completed' THEN 1 END) AS completed_total,",
+            "COUNT(CASE WHEN status='failed' THEN 1 END) AS failed_total,",
+            "COUNT(CASE WHEN status='cancelled' THEN 1 END) AS cancelled_total,",
+            "COUNT(CASE WHEN status='terminated' THEN 1 END) AS terminated_total,",
+            "COUNT(CASE WHEN status_bucket='running' THEN 1 END) AS running,",
+            "COUNT(duration_ms) AS latency_count,",
+            "CAST(COALESCE(SUM(duration_ms/1000.0),0) AS ", $float, ") AS latency_sum,",
+            buckets!("duration_ms IS NOT NULL", "duration_ms/1000.0"),
+            " FROM workflow_run_summaries WHERE namespace=$1",
+            " AND (COALESCE(queue,'default')<$2 OR (COALESCE(queue,'default')=$3 AND workflow_type<=$4))",
+            " GROUP BY COALESCE(queue,'default'),workflow_type"
+        )
+    };
+}
+
+/// The activity series up to and including the last candidate key.
+macro_rules! activity_rows {
+    ($float:literal, $duration:literal) => {
+        concat!(
+            "SELECT COALESCE(activity_executions.queue,workflow_runs.queue,'default') AS task_queue,",
+            "workflow_runs.workflow_type AS workflow_type,activity_executions.activity_type AS activity_type,",
+            "COUNT(*) AS started_total,",
+            "COUNT(CASE WHEN activity_executions.status='completed' THEN 1 END) AS completed_total,",
+            "COUNT(CASE WHEN activity_executions.status='failed' THEN 1 END) AS failed_total,",
+            "COUNT(CASE WHEN activity_executions.status='cancelled' THEN 1 END) AS cancelled_total,",
+            "COUNT(CASE WHEN activity_executions.status IN ('pending','running') THEN 1 END) AS running,",
+            "COUNT(CASE WHEN activity_executions.started_at IS NOT NULL AND activity_executions.closed_at IS NOT NULL THEN 1 END) AS latency_count,",
+            "CAST(COALESCE(SUM(CASE WHEN activity_executions.started_at IS NOT NULL AND activity_executions.closed_at IS NOT NULL THEN ",
+            $duration,
+            " END),0) AS ",
+            $float,
+            ") AS latency_sum,",
+            buckets!(
+                "activity_executions.started_at IS NOT NULL AND activity_executions.closed_at IS NOT NULL",
+                $duration
+            ),
+            " FROM activity_executions JOIN workflow_runs ON workflow_runs.id=activity_executions.workflow_run_id",
+            " WHERE workflow_runs.namespace=$1 AND (COALESCE(activity_executions.queue,workflow_runs.queue,'default')<$2",
+            " OR (COALESCE(activity_executions.queue,workflow_runs.queue,'default')=$3 AND (workflow_runs.workflow_type<$4",
+            " OR (workflow_runs.workflow_type=$5 AND activity_executions.activity_type<=$6))))",
+            " GROUP BY COALESCE(activity_executions.queue,workflow_runs.queue,'default'),workflow_runs.workflow_type,activity_executions.activity_type"
+        )
+    };
+}
+
+/// `whereTaskRuntimeVisible`: open work, or work created or dispatched within
+/// the reporting minute.
+macro_rules! visible_tasks {
+    ($namespace:literal, $created:literal, $dispatched:literal) => {
+        concat!(
+            " FROM workflow_tasks WHERE namespace=", $namespace,
+            " AND task_type IN ('workflow','activity') AND (status IN ('ready','leased') OR created_at>=", $created,
+            " OR (last_dispatched_at IS NOT NULL AND last_dispatched_at>=", $dispatched, "))"
+        )
+    };
+}
+
+/// A float as PHP's `json_encode` writes it: whole values without a fraction.
+fn php_float(value: f64) -> Value {
+    if value.fract() == 0.0 && value.abs() < 9.0e15 {
+        json!(value as i64)
+    } else {
+        json!(value)
+    }
+}
+
+/// PHP's `round($value, 6)` for the finite sums reported here.
+fn round6(value: f64) -> f64 {
+    (value * 1_000_000.0).round() / 1_000_000.0
+}
+
+/// `seriesSnapshot`.
+struct Snapshot {
+    series: Vec<Value>,
+    observed: usize,
+    suppressed_started: i64,
+    lower_bound: bool,
+}
+
+impl Snapshot {
+    fn precision(&self) -> &'static str {
+        if self.lower_bound { "lower_bound" } else { "exact" }
+    }
+    fn suppressed(&self) -> usize {
+        self.observed.saturating_sub(self.series.len())
+    }
+    /// `seriesLimitDisclosure`.
+    fn disclosure(&self, dimensions: &[&str], selection: &str) -> Value {
+        json!({
+            "label_dimensions": dimensions, "limit": SERIES_LIMIT,
+            "observed_series_count": self.observed, "observed_series_count_precision": self.precision(),
+            "reported_series_count": self.series.len(), "suppressed_series_count": self.suppressed(),
+            "suppressed_started_total": self.suppressed_started, "suppressed_counts_precision": self.precision(),
+            "truncated": self.observed > self.series.len(), "selection": selection,
+        })
+    }
+}
+
+/// `BoundedMetricPolicy::labelSet` for a family's bounded dimensions.
+fn label_set(dimensions: &[&str], selection: &str) -> Value {
+    let mut set = Map::new();
+    set.insert("namespace".into(), json!("request_scope_not_label"));
+    for dimension in dimensions {
+        set.insert(
+            (*dimension).to_owned(),
+            json!({"cardinality_class": "bounded_series", "limit": SERIES_LIMIT, "selection": selection}),
+        );
+    }
+    Value::Object(set)
+}
+
+fn empty_task_queue(queue: &str) -> Map<String, Value> {
+    let mut entry = Map::new();
+    entry.insert("task_queue".into(), json!(queue));
+    for prefix in ["workflow", "activity"] {
+        for field in [
+            "ready_tasks",
+            "ready_due_tasks",
+            "delayed_tasks",
+            "leased_tasks",
+            "expired_leases",
+            "tasks_added_last_minute",
+            "tasks_dispatched_last_minute",
+        ] {
+            entry.insert(format!("{prefix}_{field}"), json!(0));
+        }
+    }
+    for field in [
+        "active_pollers",
+        "workflow_slot_capacity",
+        "activity_slot_capacity",
+        "available_workflow_slots",
+        "available_activity_slots",
+    ] {
+        entry.insert(field.into(), json!(0));
+    }
+    entry
+}
+
+fn add(entry: &mut Map<String, Value>, field: &str, amount: i64) {
+    let current = entry.get(field).and_then(Value::as_i64).unwrap_or_default();
+    entry.insert(field.to_owned(), json!(current + amount));
+}
+
+store_methods! {
+    pub(crate) async fn prometheus_metrics(&self) -> Result<Value> {
+        let now = now();
+        let workflows = self.workflow_series().await?;
+        let activities = self.activity_series().await?;
+        let queues = self.task_queue_series(now).await?;
+        let families = [&workflows, &activities, &queues];
+        let precision = if families.iter().any(|family| family.lower_bound) { "lower_bound" } else { "exact" };
+        let series_count: usize = families.iter().map(|family| family.series.len()).sum();
+        let observed: usize = families.iter().map(|family| family.observed).sum();
+        let suppressed: usize = families.iter().map(|family| family.suppressed()).sum();
+        let workflow_dimensions = ["task_queue", "workflow_type"];
+        let activity_dimensions = ["task_queue", "workflow_type", "activity_type"];
+        let series_limits = json!({
+            "workflows": workflows.disclosure(&workflow_dimensions, WORKFLOW_SELECTION),
+            "activities": activities.disclosure(&activity_dimensions, ACTIVITY_SELECTION),
+            "task_queues": queues.disclosure(&["task_queue"], QUEUE_SELECTION),
+        });
+        let workflow_labels = label_set(&workflow_dimensions, WORKFLOW_SELECTION);
+        let activity_labels = label_set(&activity_dimensions, ACTIVITY_SELECTION);
+        Ok(json!({
+            "schema": "durable-workflow.prometheus-metrics.v1",
+            "generated_at": now.to_rfc3339_opts(chrono::SecondsFormat::Micros, true),
+            "namespace": self.namespace.as_ref(),
+            "latency_buckets_seconds": BUCKETS.iter().map(|(_, bound)| php_float(*bound)).collect::<Vec<_>>(),
+            "series": {"workflows": workflows.series, "activities": activities.series, "task_queues": queues.series},
+            "series_count": series_count,
+            "observed_series_count": observed,
+            "observed_series_count_precision": precision,
+            "suppressed_series_count": suppressed,
+            "suppressed_series_count_precision": precision,
+            "cardinality": {
+                "series_limits": series_limits,
+                "metric_label_sets": {
+                    "dw_workflow_runs_total": workflow_labels,
+                    "dw_workflow_run_latency_seconds": workflow_labels,
+                    "dw_activity_executions_total": activity_labels,
+                    "dw_activity_execution_latency_seconds": activity_labels,
+                    "dw_task_queue_runtime_state": label_set(&["task_queue"], QUEUE_SELECTION),
+                },
+            },
+        }))
+    }
+
+    /// The counters and latency histogram shared by both runtime families.
+    fn runtime_series(row: &DB::Row, labels: Map<String, Value>, terminated: bool) -> Result<Value> {
+        let mut series = labels;
+        for field in ["started_total", "completed_total", "failed_total", "cancelled_total"] {
+            series.insert(field.into(), json!(DB::number(row, field)?));
+        }
+        if terminated {
+            series.insert("terminated_total".into(), json!(DB::number(row, "terminated_total")?));
+        }
+        series.insert("running".into(), json!(DB::number(row, "running")?));
+        let mut buckets = Map::new();
+        for (index, (bound, _)) in BUCKETS.iter().enumerate() {
+            buckets.insert((*bound).to_owned(), json!(DB::number(row, &format!("bucket_{index}"))?));
+        }
+        let sum = DB::optional_float(row, "latency_sum")?.unwrap_or_default();
+        series.insert("latency_seconds".into(), json!({
+            "count": DB::number(row, "latency_count")?, "sum": php_float(round6(sum)), "buckets": buckets,
+        }));
+        Ok(Value::Object(series))
+    }
+
+    /// `workflowSeries`: the first series by task queue and workflow type.
+    async fn workflow_series(&self) -> Result<Snapshot> {
+        let candidates: Vec<(String, String)> = Self::query(
+            "SELECT DISTINCT COALESCE(queue,'default') AS task_queue,workflow_type FROM workflow_run_summaries WHERE namespace=$1 ORDER BY task_queue,workflow_type LIMIT 101",
+        )
+        .bind(self.namespace.as_ref())
+        .fetch_all(&self.pool)
+        .await?
+        .iter()
+        .map(|row| Ok((DB::string(row, "task_queue")?, DB::optional_string(row, "workflow_type")?.unwrap_or_default())))
+        .collect::<Result<_>>()?;
+        let Some((last_queue, last_type)) = candidates.last().cloned() else {
+            return Ok(Snapshot { series: Vec::new(), observed: 0, suppressed_started: 0, lower_bound: false });
+        };
+        let sql = match DB::NAME {
+            "PostgreSQL" => workflow_rows!("DOUBLE PRECISION"),
+            "MySQL" => workflow_rows!("DOUBLE"),
+            _ => workflow_rows!("REAL"),
+        };
+        let rows = Self::query(sql)
+            .bind(self.namespace.as_ref())
+            .bind(last_queue.clone())
+            .bind(last_queue)
+            .bind(last_type)
+            .fetch_all(&self.pool)
+            .await?;
+        let mut keyed = HashMap::new();
+        for row in &rows {
+            let key = (DB::string(row, "task_queue")?, DB::optional_string(row, "workflow_type")?.unwrap_or_default());
+            keyed.insert(key, row);
+        }
+        let mut series = Vec::new();
+        for key in candidates.iter().take(SERIES_LIMIT) {
+            if let Some(row) = keyed.get(key) {
+                let mut labels = Map::new();
+                labels.insert("task_queue".into(), json!(key.0));
+                labels.insert("workflow_type".into(), json!(key.1));
+                series.push(Self::runtime_series(row, labels, true)?);
+            }
+        }
+        let mut suppressed_started = 0;
+        for key in candidates.iter().skip(SERIES_LIMIT) {
+            if let Some(row) = keyed.get(key) {
+                suppressed_started += DB::number(row, "started_total")?;
+            }
+        }
+        Ok(Snapshot { series, observed: candidates.len(), suppressed_started, lower_bound: candidates.len() > SERIES_LIMIT })
+    }
+
+    /// `activitySeries`: the first series by task queue, workflow type and
+    /// activity type.
+    async fn activity_series(&self) -> Result<Snapshot> {
+        let candidates: Vec<(String, String, String)> = Self::query(
+            "SELECT DISTINCT COALESCE(activity_executions.queue,workflow_runs.queue,'default') AS task_queue,workflow_runs.workflow_type AS workflow_type,activity_executions.activity_type AS activity_type FROM activity_executions JOIN workflow_runs ON workflow_runs.id=activity_executions.workflow_run_id WHERE workflow_runs.namespace=$1 ORDER BY task_queue,workflow_type,activity_type LIMIT 101",
+        )
+        .bind(self.namespace.as_ref())
+        .fetch_all(&self.pool)
+        .await?
+        .iter()
+        .map(|row| {
+            Ok((
+                DB::string(row, "task_queue")?,
+                DB::optional_string(row, "workflow_type")?.unwrap_or_default(),
+                DB::optional_string(row, "activity_type")?.unwrap_or_default(),
+            ))
+        })
+        .collect::<Result<_>>()?;
+        let Some((last_queue, last_type, last_activity)) = candidates.last().cloned() else {
+            return Ok(Snapshot { series: Vec::new(), observed: 0, suppressed_started: 0, lower_bound: false });
+        };
+        let sql = match DB::NAME {
+            "PostgreSQL" => activity_rows!(
+                "DOUBLE PRECISION",
+                "EXTRACT(EPOCH FROM (activity_executions.closed_at-activity_executions.started_at))"
+            ),
+            "MySQL" => activity_rows!(
+                "DOUBLE",
+                "TIMESTAMPDIFF(MICROSECOND,activity_executions.started_at,activity_executions.closed_at)/1000000.0"
+            ),
+            _ => activity_rows!(
+                "REAL",
+                "(julianday(activity_executions.closed_at)-julianday(activity_executions.started_at))*86400.0"
+            ),
+        };
+        let rows = Self::query(sql)
+            .bind(self.namespace.as_ref())
+            .bind(last_queue.clone())
+            .bind(last_queue)
+            .bind(last_type.clone())
+            .bind(last_type)
+            .bind(last_activity)
+            .fetch_all(&self.pool)
+            .await?;
+        let mut keyed = HashMap::new();
+        for row in &rows {
+            let key = (
+                DB::string(row, "task_queue")?,
+                DB::optional_string(row, "workflow_type")?.unwrap_or_default(),
+                DB::optional_string(row, "activity_type")?.unwrap_or_default(),
+            );
+            keyed.insert(key, row);
+        }
+        let mut series = Vec::new();
+        for key in candidates.iter().take(SERIES_LIMIT) {
+            if let Some(row) = keyed.get(key) {
+                let mut labels = Map::new();
+                labels.insert("task_queue".into(), json!(key.0));
+                labels.insert("workflow_type".into(), json!(key.1));
+                labels.insert("activity_type".into(), json!(key.2));
+                series.push(Self::runtime_series(row, labels, false)?);
+            }
+        }
+        let mut suppressed_started = 0;
+        for key in candidates.iter().skip(SERIES_LIMIT) {
+            if let Some(row) = keyed.get(key) {
+                suppressed_started += DB::number(row, "started_total")?;
+            }
+        }
+        Ok(Snapshot { series, observed: candidates.len(), suppressed_started, lower_bound: candidates.len() > SERIES_LIMIT })
+    }
+
+    /// `taskQueueSeries`: open and recent task counts and fresh worker
+    /// capacity for the first task queues by name.
+    async fn task_queue_series(&self, now: DateTime<Utc>) -> Result<Snapshot> {
+        // Laravel binds these times in whole seconds, as operator metrics do.
+        let bound = whole_second(now);
+        let minute_ago = whole_second(now - Duration::minutes(1));
+        let stale_cutoff = whole_second(now - Duration::seconds(self.worker_stale_after_seconds));
+        let mut queues: Vec<String> = Vec::new();
+        for row in Self::query(concat!("SELECT DISTINCT COALESCE(queue,'default') AS task_queue", visible_tasks!("$1", "$2", "$3"), " ORDER BY task_queue LIMIT 101"))
+            .bind(self.namespace.as_ref())
+            .bind(DB::bind_time(minute_ago))
+            .bind(DB::bind_time(minute_ago))
+            .fetch_all(&self.pool)
+            .await?
+        {
+            queues.push(DB::string(&row, "task_queue")?);
+        }
+        for row in Self::query("SELECT DISTINCT COALESCE(task_queue,'default') AS task_queue FROM workflow_worker_registrations WHERE namespace=$1 AND status='active' AND last_heartbeat_at IS NOT NULL AND last_heartbeat_at>=$2 ORDER BY task_queue LIMIT 101")
+            .bind(self.namespace.as_ref())
+            .bind(DB::bind_time(stale_cutoff))
+            .fetch_all(&self.pool)
+            .await?
+        {
+            queues.push(DB::string(&row, "task_queue")?);
+        }
+        queues.sort();
+        queues.dedup();
+        queues.truncate(SERIES_LIMIT + 1);
+        let observed = queues.len();
+        queues.truncate(SERIES_LIMIT);
+        let mut entries: std::collections::BTreeMap<String, Map<String, Value>> = std::collections::BTreeMap::new();
+        let rows = Self::query(concat!(
+            "SELECT COALESCE(queue,'default') AS task_queue,task_type,status,COUNT(*) AS n,",
+            "COUNT(CASE WHEN status='ready' AND available_at>$1 THEN 1 END) AS delayed_count,",
+            "COUNT(CASE WHEN status='leased' AND lease_expires_at IS NOT NULL AND lease_expires_at<=$2 THEN 1 END) AS expired_lease_count,",
+            "COUNT(CASE WHEN created_at>=$3 THEN 1 END) AS added_last_minute,",
+            "COUNT(CASE WHEN last_dispatched_at IS NOT NULL AND last_dispatched_at>=$4 THEN 1 END) AS dispatched_last_minute",
+            visible_tasks!("$5", "$6", "$7"),
+            " GROUP BY COALESCE(queue,'default'),task_type,status"
+        ))
+        .bind(DB::bind_time(bound))
+        .bind(DB::bind_time(bound))
+        .bind(DB::bind_time(minute_ago))
+        .bind(DB::bind_time(minute_ago))
+        .bind(self.namespace.as_ref())
+        .bind(DB::bind_time(minute_ago))
+        .bind(DB::bind_time(minute_ago))
+        .fetch_all(&self.pool)
+        .await?;
+        for row in &rows {
+            let queue = DB::string(row, "task_queue")?;
+            if !queues.contains(&queue) {
+                continue;
+            }
+            let entry = entries.entry(queue.clone()).or_insert_with(|| empty_task_queue(&queue));
+            let prefix = if DB::string(row, "task_type")? == "activity" { "activity" } else { "workflow" };
+            let status = DB::string(row, "status")?;
+            let count = DB::number(row, "n")?;
+            if status == "ready" {
+                let delayed = DB::number(row, "delayed_count")?;
+                add(entry, &format!("{prefix}_ready_tasks"), count);
+                add(entry, &format!("{prefix}_delayed_tasks"), delayed);
+                add(entry, &format!("{prefix}_ready_due_tasks"), (count - delayed).max(0));
+            }
+            if status == "leased" {
+                add(entry, &format!("{prefix}_leased_tasks"), count);
+                add(entry, &format!("{prefix}_expired_leases"), DB::number(row, "expired_lease_count")?);
+            }
+            add(entry, &format!("{prefix}_tasks_added_last_minute"), DB::number(row, "added_last_minute")?);
+            add(entry, &format!("{prefix}_tasks_dispatched_last_minute"), DB::number(row, "dispatched_last_minute")?);
+        }
+        // Capacity is summed here rather than in SQL, where MySQL's SUM is a
+        // decimal.
+        let workers = Self::query("SELECT COALESCE(task_queue,'default') AS task_queue,max_concurrent_workflow_tasks,max_concurrent_activity_tasks,available_workflow_slots,available_activity_slots FROM workflow_worker_registrations WHERE namespace=$1 AND status='active' AND last_heartbeat_at IS NOT NULL AND last_heartbeat_at>=$2")
+            .bind(self.namespace.as_ref())
+            .bind(DB::bind_time(stale_cutoff))
+            .fetch_all(&self.pool)
+            .await?;
+        let mut capacity: HashMap<String, [i64; 5]> = HashMap::new();
+        for row in &workers {
+            let queue = DB::string(row, "task_queue")?;
+            if !queues.contains(&queue) {
+                continue;
+            }
+            let workflow = DB::optional_number(row, "max_concurrent_workflow_tasks")?;
+            let activity = DB::optional_number(row, "max_concurrent_activity_tasks")?;
+            let totals = capacity.entry(queue).or_default();
+            totals[0] += 1;
+            totals[1] += workflow.unwrap_or_default();
+            totals[2] += activity.unwrap_or_default();
+            totals[3] += DB::optional_number(row, "available_workflow_slots")?.or(workflow).unwrap_or_default();
+            totals[4] += DB::optional_number(row, "available_activity_slots")?.or(activity).unwrap_or_default();
+        }
+        for (queue, totals) in capacity {
+            let entry = entries.entry(queue.clone()).or_insert_with(|| empty_task_queue(&queue));
+            for (field, value) in [
+                "active_pollers",
+                "workflow_slot_capacity",
+                "activity_slot_capacity",
+                "available_workflow_slots",
+                "available_activity_slots",
+            ]
+            .iter()
+            .zip(totals)
+            {
+                entry.insert((*field).to_owned(), json!(value));
+            }
+        }
+        Ok(Snapshot {
+            series: entries.into_values().map(Value::Object).collect(),
+            observed,
+            suppressed_started: 0,
+            lower_bound: observed > SERIES_LIMIT,
+        })
+    }
+}
diff --git a/rust/src/runtime/store.rs b/rust/src/runtime/store.rs
index 643c246d1..4c3b50966 100644
--- a/rust/src/runtime/store.rs
+++ b/rust/src/runtime/store.rs
@@ -2363,12 +2363,15 @@ where
                     Some(recorded_at),
                 )
                 .await?;
-                let activity_task = Self::create_task(
+                // PHP publishes only timer tasks a completion creates, so this
+                // activity task waits for a poller undispatched.
+                let activity_task = Self::create_task_dispatched(
                     &mut tx,
                     &run_id,
                     &queue,
                     "activity",
                     json!({"activity_execution_id": activity_id}),
+                    false,
                 )
                 .await?;
                 Self::create_activity_wait(
@@ -3154,6 +3157,19 @@ where
         queue: &str,
         kind: &str,
         payload: Value,
+    ) -> Result<String> {
+        Self::create_task_dispatched(tx, run_id, queue, kind, payload, true).await
+    }
+
+    /// A new ready task. As PHP in poll mode, a task it dispatches is stamped
+    /// dispatched when created; one it leaves to pollers is not.
+    pub(super) async fn create_task_dispatched(
+        tx: &mut Transaction<'_, DB>,
+        run_id: &str,
+        queue: &str,
+        kind: &str,
+        payload: Value,
+        dispatched: bool,
     ) -> Result<String> {
         tx.summary_changed(run_id);
         let task_id = id();
@@ -3163,12 +3179,12 @@ where
             .fetch_one(&mut **tx)
             .await?;
         let namespace = DB::string(&run, "namespace")?;
-        // As PHP in poll mode, a task is dispatched to pollers when it is created.
         let created = now();
+        let dispatched_at = dispatched.then_some(created);
         let fairness_key = DB::optional_string(&run, "fairness_key")?;
         Self::query("INSERT INTO workflow_tasks(id,workflow_run_id,task_type,status,payload,queue,available_at,namespace,created_at,updated_at,last_dispatched_at,last_dispatch_attempt_at,priority,fairness_key,fairness_weight) VALUES ($1,$2,$3,'ready',$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)")
         .bind(&task_id).bind(run_id).bind(kind).bind(DB::document(&payload)).bind(queue).bind(DB::bind_time(created)).bind(&namespace)
-        .bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::bind_time(created))
+        .bind(DB::bind_time(created)).bind(DB::bind_time(created)).bind(DB::optional_time(dispatched_at)).bind(DB::optional_time(dispatched_at))
         .bind(DB::number(&run, "priority")?).bind(fairness_key.as_deref()).bind(DB::number(&run, "fairness_weight")?)
         .execute(&mut **tx).await?;
         Ok(task_id)
diff --git a/rust/tests/execution.rs b/rust/tests/execution.rs
index d23b871c5..631096079 100644
--- a/rust/tests/execution.rs
+++ b/rust/tests/execution.rs
@@ -136,6 +136,9 @@ mod activity_lease_renewal;
 #[path = "execution/system_metrics.rs"]
 mod system_metrics;
 
+#[path = "execution/prometheus_metrics.rs"]
+mod prometheus_metrics;
+
 #[path = "execution/activity_deadlines.rs"]
 mod activity_deadlines;
 
diff --git a/rust/tests/execution/prometheus_metrics.rs b/rust/tests/execution/prometheus_metrics.rs
new file mode 100644
index 000000000..16add676b
--- /dev/null
+++ b/rust/tests/execution/prometheus_metrics.rs
@@ -0,0 +1,170 @@
+use super::*;
+
+const PROMETHEUS: &str = "/api/system/prometheus-metrics";
+
+async fn summary(app: &Router) -> Value {
+    let (status, body) = request(app, "GET", PROMETHEUS, Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    assert_eq!(
+        [&body["schema"], &body["namespace"]],
+        [&json!("durable-workflow.prometheus-metrics.v1"), &json!("default")]
+    );
+    body
+}
+
+async fn start_on(app: &Router, workflow: &str, workflow_type: &str, queue: &str) {
+    let (status, body) = request(app, "POST", "/api/workflows", json!({"workflow_id": workflow,
+        "workflow_type": workflow_type, "task_queue": queue, "input": envelope(Payload::Array(vec![]))})).await;
+    assert_eq!(status, StatusCode::CREATED, "{body}");
+}
+
+fn buckets(values: [i64; 10]) -> Value {
+    let bounds = ["0.1", "0.5", "1", "2.5", "5", "10", "30", "60", "300", "900"];
+    Value::Object(bounds.iter().zip(values).map(|(bound, value)| ((*bound).to_owned(), json!(value))).collect())
+}
+
+#[tokio::test]
+async fn prometheus_summary_counts_runs_activities_and_queues_with_cumulative_latency() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    let empty = summary(&app).await;
+    assert_eq!(empty["series"], json!({"workflows": [], "activities": [], "task_queues": []}));
+    assert_eq!(empty["latency_buckets_seconds"].to_string(), "[0.1,0.5,1,2.5,5,10,30,60,300,900]");
+    assert_eq!(
+        [&empty["series_count"], &empty["observed_series_count_precision"]],
+        [&json!(0), &json!("exact")]
+    );
+    assert_eq!(empty["cardinality"]["series_limits"]["workflows"]["limit"], 100);
+
+    let (status, body) = request(&app, "POST", "/api/worker/register", json!({"worker_id": "prom-worker",
+        "task_queue": "test", "runtime": "php", "supported_workflow_types": ["echo", "other"],
+        "supported_activity_types": ["act"], "max_concurrent_workflow_tasks": 4, "max_concurrent_activity_tasks": 3})).await;
+    assert_eq!(status, StatusCode::CREATED, "{body}");
+    start_on(&app, "prom-completed", "echo", "test").await;
+    let task = poll(&app, "prom-worker", "workflow").await;
+    let (status, body) = finish_task(&app, &task, json!([{"type": "schedule_activity", "activity_type": "act",
+        "arguments": envelope(Payload::Array(vec![])), "retry_policy": {"max_attempts": 1, "backoff_seconds": []},
+        "start_to_close_timeout": 300}])).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let activity = poll(&app, "prom-worker", "activity").await;
+    let (status, body) = request(&app, "POST",
+        &format!("/api/worker/activity-tasks/{}/complete", activity["task_id"].as_str().unwrap()),
+        json!({"activity_attempt_id": activity["activity_attempt_id"], "lease_owner": activity["lease_owner"],
+            "result": envelope(Payload::Array(vec![]))})).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    let task = poll(&app, "prom-worker", "workflow").await;
+    let (status, body) = finish_task(&app, &task,
+        json!([{"type": "complete_workflow", "result": envelope(Payload::Array(vec![]))}])).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    start_on(&app, "prom-terminated", "other", "test").await;
+    let (status, body) = request(&app, "POST", "/api/workflows/prom-terminated/terminate", json!({"reason": "prom"})).await;
+    assert_eq!(status, StatusCode::OK, "{body}");
+    start_on(&app, "prom-waiting", "echo", "second").await;
+    // Fixed durations: the completed run 2 s, the terminated run 50 ms and the
+    // activity 1.5 s, so each falls in known buckets on every backend.
+    database.execute("UPDATE workflow_run_summaries SET duration_ms=2000 WHERE workflow_type='echo' AND status='completed'").await;
+    database.execute("UPDATE workflow_run_summaries SET duration_ms=50 WHERE workflow_type='other'").await;
+    database.execute("UPDATE activity_executions SET started_at='2026-10-22 10:00:00', closed_at='2026-10-22 10:00:01.5'").await;
+    // As in PHP, a task made available within the current second still counts
+    // as delayed; this one became available earlier.
+    database.execute("UPDATE workflow_tasks SET available_at='2026-10-22 09:00:00' WHERE queue='second'").await;
+
+    let read = summary(&app).await;
+    let workflows = read["series"]["workflows"].as_array().unwrap();
+    assert_eq!(workflows.len(), 3, "{read}");
+    assert_eq!(
+        workflows[0],
+        json!({"task_queue": "second", "workflow_type": "echo", "started_total": 1, "completed_total": 0,
+            "failed_total": 0, "cancelled_total": 0, "terminated_total": 0, "running": 1,
+            "latency_seconds": {"count": 0, "sum": 0, "buckets": buckets([0; 10])}})
+    );
+    assert_eq!(
+        workflows[1],
+        json!({"task_queue": "test", "workflow_type": "echo", "started_total": 1, "completed_total": 1,
+            "failed_total": 0, "cancelled_total": 0, "terminated_total": 0, "running": 0,
+            "latency_seconds": {"count": 1, "sum": 2, "buckets": buckets([0, 0, 0, 1, 1, 1, 1, 1, 1, 1])}})
+    );
+    assert_eq!(
+        [&workflows[2]["terminated_total"], &workflows[2]["latency_seconds"]],
+        [&json!(1), &json!({"count": 1, "sum": 0.05, "buckets": buckets([1; 10])})]
+    );
+    // SQLite measures activity durations with PHP's julianday difference, a
+    // float accurate to about a millisecond.
+    let mut activities = read["series"]["activities"].clone();
+    let sum = activities[0]["latency_seconds"]["sum"].take().as_f64().unwrap();
+    assert!((sum - 1.5).abs() < 0.001, "{sum}");
+    assert_eq!(
+        activities,
+        json!([{"task_queue": "test", "workflow_type": "echo", "activity_type": "act", "started_total": 1,
+            "completed_total": 1, "failed_total": 0, "cancelled_total": 0, "running": 0,
+            "latency_seconds": {"count": 1, "sum": null, "buckets": buckets([0, 0, 0, 1, 1, 1, 1, 1, 1, 1])}}])
+    );
+    let queues = read["series"]["task_queues"].as_array().unwrap();
+    assert_eq!(
+        queues.iter().map(|queue| queue["task_queue"].as_str().unwrap()).collect::<Vec<_>>(),
+        ["second", "test"]
+    );
+    assert_eq!(
+        [&queues[0]["workflow_ready_tasks"], &queues[0]["workflow_ready_due_tasks"], &queues[0]["active_pollers"]],
+        [&json!(1), &json!(1), &json!(0)]
+    );
+    assert_eq!(
+        [
+            &queues[1]["workflow_ready_tasks"],
+            &queues[1]["activity_tasks_added_last_minute"],
+            &queues[1]["active_pollers"],
+            &queues[1]["workflow_slot_capacity"],
+            &queues[1]["activity_slot_capacity"],
+            &queues[1]["available_workflow_slots"],
+            &queues[1]["available_activity_slots"],
+        ],
+        [&json!(0), &json!(1), &json!(1), &json!(4), &json!(3), &json!(4), &json!(3)]
+    );
+    assert_eq!(
+        [&read["series_count"], &read["observed_series_count"], &read["suppressed_series_count"]],
+        [&json!(6), &json!(6), &json!(0)]
+    );
+
+    // The summary is namespace-scoped: a named namespace sees only its own.
+    super::namespaces::create(&app, "prom-named").await;
+    let (status, named) = super::namespaces::scoped(&app, "prom-named", "GET", PROMETHEUS, Value::Null).await;
+    assert_eq!(status, StatusCode::OK, "{named}");
+    assert_eq!(
+        [&named["namespace"], &named["series_count"]],
+        [&json!("prom-named"), &json!(0)]
+    );
+}
+
+#[tokio::test]
+async fn prometheus_summary_bounds_workflow_series_and_discloses_the_suppressed_starts() {
+    let database = TestDatabase::new().await;
+    let runtime = database.open().await.unwrap();
+    let app = router(runtime.clone());
+    for index in 0..101 {
+        start_on(&app, &format!("bounded-{index:03}"), &format!("type-{index:03}"), "test").await;
+    }
+    let read = summary(&app).await;
+    let workflows = read["series"]["workflows"].as_array().unwrap();
+    assert_eq!(workflows.len(), 100);
+    assert_eq!(
+        [&workflows[0]["workflow_type"], &workflows[99]["workflow_type"]],
+        [&json!("type-000"), &json!("type-099")]
+    );
+    assert_eq!(
+        read["cardinality"]["series_limits"]["workflows"],
+        json!({"label_dimensions": ["task_queue", "workflow_type"], "limit": 100, "observed_series_count": 101,
+            "observed_series_count_precision": "lower_bound", "reported_series_count": 100,
+            "suppressed_series_count": 1, "suppressed_started_total": 1, "suppressed_counts_precision": "lower_bound",
+            "truncated": true, "selection": "bounded_task_queue_and_workflow_type_ascending"})
+    );
+    assert_eq!(
+        [
+            &read["observed_series_count"],
+            &read["observed_series_count_precision"],
+            &read["suppressed_series_count"],
+            &read["suppressed_series_count_precision"],
+        ],
+        [&json!(102), &json!("lower_bound"), &json!(1), &json!("lower_bound")]
+    );
+}
diff --git a/scripts/conformance/server-parity/contract.mjs b/scripts/conformance/server-parity/contract.mjs
index 936504479..03e8203bc 100644
--- a/scripts/conformance/server-parity/contract.mjs
+++ b/scripts/conformance/server-parity/contract.mjs
@@ -15,6 +15,7 @@ import {checkStorageTest} from './storage-test-contract.mjs';
 import {checkActivityLeaseRenewal} from './activity-lease-renewal-contract.mjs';
 import {checkServiceCatalog} from './service-catalog-contract.mjs';
 import {checkSystemMetrics} from './system-metrics-contract.mjs';
+import {checkPrometheusMetrics} from './prometheus-metrics-contract.mjs';
 import {checkNamespaceDelete} from './namespace-delete-contract.mjs';
 import {checkSchedulePolicies} from './schedule-policies-contract.mjs';
 import {checkScheduleCatalog} from './schedule-catalog-contract.mjs';
@@ -148,6 +149,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) {
   const activityLeaseRenewal = fixture.activity_lease_renewal ? checkActivityLeaseRenewal(fixture, observation, workflowId) : null;
   const serviceCatalog = fixture.service_catalog ? checkServiceCatalog(fixture, observation, workflowId) : null;
   const systemMetrics = fixture.system_metrics ? checkSystemMetrics(fixture, observation, workflowId) : null;
+  const prometheusMetrics = fixture.prometheus_metrics ? checkPrometheusMetrics(fixture, observation, workflowId) : null;
   const namespaceDelete = fixture.namespace_delete ? checkNamespaceDelete(fixture, observation, workflowId) : null;
   const schedulePolicies = fixture.schedule_policies ? checkSchedulePolicies(fixture, observation, workflowId) : null;
   const scheduleCatalog = fixture.schedule_catalog ? checkScheduleCatalog(fixture, observation, workflowId) : null;
@@ -698,6 +700,7 @@ export function checkObservation(fixture, observation, workflowId, artifacts) {
     ...(activityLeaseRenewal ? {activity_lease_renewal: activityLeaseRenewal} : {}),
     ...(serviceCatalog ? {service_catalog: serviceCatalog} : {}),
     ...(systemMetrics ? {system_metrics: systemMetrics} : {}),
+    ...(prometheusMetrics ? {prometheus_metrics: prometheusMetrics} : {}),
     ...(namespaceDelete ? {namespace_delete: namespaceDelete} : {}),
     ...(schedulePolicies ? {schedule_policies: schedulePolicies} : {}),
     ...(scheduleCatalog ? {schedule_catalog: scheduleCatalog} : {}),
diff --git a/scripts/conformance/server-parity/probe.php b/scripts/conformance/server-parity/probe.php
index c05c88677..95cadcd7d 100644
--- a/scripts/conformance/server-parity/probe.php
+++ b/scripts/conformance/server-parity/probe.php
@@ -433,6 +433,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace,
     $activityLeaseRenewal = isset($fixture['activity_lease_renewal']) ? finishHttpActivityLeaseRenewal($client, $fixture, $workflowId, $url) : null;
     $serviceCatalog = isset($fixture['service_catalog']) ? finishHttpServiceCatalog($client, $fixture, $workflowId, $url) : null;
     $systemMetrics = isset($fixture['system_metrics']) ? finishHttpSystemMetrics($client, $fixture, $workflowId, $url) : null;
+    $prometheusMetrics = isset($fixture['prometheus_metrics']) ? finishHttpPrometheusMetrics($client, $fixture, $workflowId, $url) : null;
     $namespaceDelete = isset($fixture['namespace_delete']) ? finishHttpNamespaceDelete($client, $fixture, $workflowId, $queue, $url) : null;
     $schedulePolicies = isset($fixture['schedule_policies']) ? finishHttpSchedulePolicies($client, $fixture, $workflowId, $url) : null;
     $scheduleCatalog = isset($fixture['schedule_catalog'])
@@ -536,6 +537,7 @@ function httpObservation(array $fixture, string $workflowId, string $namespace,
         'activity_lease_renewal' => $activityLeaseRenewal,
         'service_catalog' => $serviceCatalog,
         'system_metrics' => $systemMetrics,
+        'prometheus_metrics' => $prometheusMetrics,
         'namespace_delete' => $namespaceDelete,
         'schedule_policies' => $schedulePolicies,
         'schedule_catalog' => $scheduleCatalog,
@@ -843,6 +845,7 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa
     require __DIR__.'/activity-lease-renewal-probe.php';
     require __DIR__.'/service-catalog-probe.php';
     require __DIR__.'/system-metrics-probe.php';
+    require __DIR__.'/prometheus-metrics-probe.php';
     require __DIR__.'/namespace-delete-probe.php';
     require __DIR__.'/schedule-policies-probe.php';
     require __DIR__.'/signals-before-worker-probe.php';
@@ -1011,6 +1014,9 @@ function embeddedObservation(array $fixture, string $workflowId, string $namespa
     if ($mode === 'embedded' && isset($fixture['system_metrics'])) {
         $observation['system_metrics'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_metrics'];
     }
+    if ($mode === 'embedded' && isset($fixture['prometheus_metrics'])) {
+        $observation['prometheus_metrics'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_metrics'];
+    }
     if ($mode === 'embedded' && isset($fixture['namespace_delete'])) {
         $observation['namespace_delete'] = ['applicable' => false, 'reason' => 'embedded_has_no_http_namespaces'];
     }
diff --git a/scripts/conformance/server-parity/prometheus-metrics-contract.mjs b/scripts/conformance/server-parity/prometheus-metrics-contract.mjs
new file mode 100644
index 000000000..c4e72f1ab
--- /dev/null
+++ b/scripts/conformance/server-parity/prometheus-metrics-contract.mjs
@@ -0,0 +1,142 @@
+import assert from 'node:assert/strict';
+
+const MICROS = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$/;
+const BOUNDS = ['0.1', '0.5', '1', '2.5', '5', '10', '30', '60', '300', '900'];
+const LIMIT = 100;
+const SELECTIONS = {
+  workflows: 'bounded_task_queue_and_workflow_type_ascending',
+  activities: 'bounded_task_queue_workflow_type_and_activity_type_ascending',
+  task_queues: 'task_queue_name_ascending',
+};
+const DIMENSIONS = {
+  workflows: ['task_queue', 'workflow_type'],
+  activities: ['task_queue', 'workflow_type', 'activity_type'],
+  task_queues: ['task_queue'],
+};
+
+const zeros = keys => Object.fromEntries(keys.map(key => [key, 0]));
+
+function labelSet(family) {
+  return {namespace: 'request_scope_not_label', ...Object.fromEntries(DIMENSIONS[family]
+    .map(dimension => [dimension, {cardinality_class: 'bounded_series', limit: LIMIT, selection: SELECTIONS[family]}]))};
+}
+
+function cardinality(counts) {
+  const disclosure = family => ({
+    label_dimensions: DIMENSIONS[family], limit: LIMIT, observed_series_count: counts[family],
+    observed_series_count_precision: 'exact', reported_series_count: counts[family], suppressed_series_count: 0,
+    suppressed_started_total: 0, suppressed_counts_precision: 'exact', truncated: false, selection: SELECTIONS[family],
+  });
+  return {
+    series_limits: {workflows: disclosure('workflows'), activities: disclosure('activities'), task_queues: disclosure('task_queues')},
+    metric_label_sets: {
+      dw_workflow_runs_total: labelSet('workflows'), dw_workflow_run_latency_seconds: labelSet('workflows'),
+      dw_activity_executions_total: labelSet('activities'), dw_activity_execution_latency_seconds: labelSet('activities'),
+      dw_task_queue_runtime_state: labelSet('task_queues'),
+    },
+  };
+}
+
+// A latency histogram for durations that each last at least `minimum`
+// seconds. Its bounds are cumulative. Published PHP on SQLite compares each
+// duration with a bound bound as text, which SQLite orders after every number,
+// so every bucket holds every duration; that recorded reference defect is
+// accepted beside the cumulative histogram (see the fixture's contract).
+function checkLatency(latency, count, minimum, label) {
+  assert.deepStrictEqual(Object.keys(latency).sort(), ['buckets', 'count', 'sum'], `${label} latency fields`);
+  assert.equal(latency.count, count, `${label} latency count`);
+  assert.deepStrictEqual(Object.keys(latency.buckets).sort(), [...BOUNDS].sort(), `${label} bucket bounds`);
+  const buckets = BOUNDS.map(bound => latency.buckets[bound]);
+  assert.ok(buckets.every(Number.isInteger), `${label} integer buckets`);
+  if (count === 0) {
+    assert.deepStrictEqual([latency.sum, buckets], [0, BOUNDS.map(() => 0)], `${label} empty histogram`);
+    return;
+  }
+  assert.ok(typeof latency.sum === 'number' && latency.sum >= count * minimum, `${label} latency sum covers each duration`);
+  assert.ok(buckets.every((value, index) => index === 0 || value >= buckets[index - 1]), `${label} cumulative buckets`);
+  assert.equal(buckets.at(-1), count, `${label} every duration within 900 seconds`);
+  const cumulative = BOUNDS.filter(bound => Number(bound) < minimum).every(bound => latency.buckets[bound] === 0);
+  const textBounds = buckets.every(value => value === count);
+  assert.ok(cumulative || textBounds, `${label} buckets below the minimum duration are empty`);
+}
+
+function queueEntry(queue, values = {}) {
+  const fields = [];
+  for (const prefix of ['workflow', 'activity']) {
+    for (const field of ['ready_tasks', 'ready_due_tasks', 'delayed_tasks', 'leased_tasks', 'expired_leases',
+      'tasks_added_last_minute', 'tasks_dispatched_last_minute']) {
+      fields.push(`${prefix}_${field}`);
+    }
+  }
+  fields.push('active_pollers', 'workflow_slot_capacity', 'activity_slot_capacity', 'available_workflow_slots',
+    'available_activity_slots');
+  return {task_queue: queue, ...zeros(fields), ...values};
+}
+
+function checkSummary(recorded, namespace, expected, label) {
+  assert.equal(recorded.status, 200, `${label} status`);
+  const body = recorded.response;
+  assert.deepStrictEqual(Object.keys(body).sort(), ['cardinality', 'generated_at', 'latency_buckets_seconds', 'namespace',
+    'observed_series_count', 'observed_series_count_precision', 'schema', 'series', 'series_count',
+    'suppressed_series_count', 'suppressed_series_count_precision'], `${label} fields`);
+  assert.match(body.generated_at, MICROS, `${label} generated_at`);
+  const counts = {workflows: expected.workflows.length, activities: expected.activities.length, task_queues: expected.task_queues.length};
+  const total = counts.workflows + counts.activities + counts.task_queues;
+  assert.deepStrictEqual({...body, generated_at: null, series: null},
+    {schema: 'durable-workflow.prometheus-metrics.v1', generated_at: null, namespace,
+      latency_buckets_seconds: BOUNDS.map(Number), series: null, series_count: total, observed_series_count: total,
+      observed_series_count_precision: 'exact', suppressed_series_count: 0, suppressed_series_count_precision: 'exact',
+      cardinality: cardinality(counts)}, `${label} summary`);
+  assert.deepStrictEqual(Object.keys(body.series).sort(), ['activities', 'task_queues', 'workflows'], `${label} families`);
+  for (const family of ['workflows', 'activities']) {
+    assert.equal(body.series[family].length, expected[family].length, `${label} ${family} series`);
+    body.series[family].forEach((series, index) => {
+      const {latency_seconds: latency, ...counters} = series;
+      const {latency_count: count, ...wanted} = expected[family][index];
+      assert.deepStrictEqual(counters, wanted, `${label} ${family}[${index}]`);
+      checkLatency(latency, count, expected.minimum, `${label} ${family}[${index}]`);
+    });
+  }
+  assert.deepStrictEqual(body.series.task_queues, expected.task_queues, `${label} task queues`);
+}
+
+// Reviewed Prometheus summary (SystemController::prometheusMetrics). Counters,
+// task-queue state, worker capacity and the bounded-series disclosure are
+// exact; latency sums and buckets depend on elapsed time, so the contract
+// checks their bounds instead.
+export function checkPrometheusMetrics(fixture, observation, workflowId) {
+  const definition = fixture.prometheus_metrics;
+  const state = observation.prometheus_metrics;
+  const projection = {scope: 'prometheus_metrics', series: 6};
+  assert.ok(state && typeof state === 'object', 'actual Prometheus metrics observation');
+  if (observation.mode === 'embedded') {
+    assert.deepStrictEqual(state, {applicable: false, reason: 'embedded_has_no_http_metrics'}, 'embedded metrics explicitly inapplicable');
+    return projection;
+  }
+  const namespace = `${workflowId}-prometheus`.toLowerCase();
+  const [first, second] = [`${namespace}-q1`, `${namespace}-q2`];
+  assert.deepStrictEqual([state.applicable, state.namespace, state.queues], [true, namespace, [first, second]], 'scenario identity');
+  assert.deepStrictEqual(state.statuses, definition.statuses, 'scenario requests');
+  const minimum = definition.minimum_duration_ms / 1000;
+  checkSummary(state.empty, namespace, {workflows: [], activities: [], task_queues: [], minimum}, 'empty namespace');
+  const [typeA, typeB] = definition.workflow_types;
+  const run = (queue, type, values) => ({task_queue: queue, workflow_type: type, started_total: 1, completed_total: 0,
+    failed_total: 0, cancelled_total: 0, terminated_total: 0, running: 0, latency_count: 0, ...values});
+  checkSummary(state.read, namespace, {
+    minimum,
+    workflows: [
+      run(first, typeA, {completed_total: 1, latency_count: 1}),
+      run(first, typeB, {terminated_total: 1, latency_count: 1}),
+      run(second, typeA, {running: 1}),
+    ],
+    activities: [{task_queue: first, workflow_type: typeA, activity_type: definition.activity_type, started_total: 1,
+      completed_total: 1, failed_total: 0, cancelled_total: 0, running: 0, latency_count: 1}],
+    task_queues: [
+      queueEntry(first, {...definition.first_queue, active_pollers: 1, workflow_slot_capacity: definition.workflow_slots,
+        activity_slot_capacity: definition.activity_slots, available_workflow_slots: definition.workflow_slots,
+        available_activity_slots: definition.activity_slots}),
+      queueEntry(second, definition.second_queue),
+    ],
+  }, 'after the scenario');
+  return projection;
+}
diff --git a/scripts/conformance/server-parity/prometheus-metrics-probe.php b/scripts/conformance/server-parity/prometheus-metrics-probe.php
new file mode 100644
index 000000000..3fbbe285c
--- /dev/null
+++ b/scripts/conformance/server-parity/prometheus-metrics-probe.php
@@ -0,0 +1,81 @@
+<?php
+
+declare(strict_types=1);
+
+use DurableWorkflow\Client;
+use GuzzleHttp\Client as PrometheusMetricsHttp;
+
+// The Prometheus summary of a fresh namespace before and after a known
+// sequence over raw HTTP: a worker with declared slots on the first queue; a
+// run there that completes after one activity; a second run there that is
+// terminated; and a run on a second queue with no worker. Responses are
+// recorded raw.
+function finishHttpPrometheusMetrics(Client $admin, array $fixture, string $workflowId, string $url): array
+{
+    $definition = $fixture['prometheus_metrics'];
+    $namespace = strtolower($workflowId.'-prometheus');
+    $first = $namespace.'-q1';
+    $second = $namespace.'-q2';
+    $worker = $namespace.'-worker';
+    $admin->createNamespace($namespace, 'prometheus metrics λ', 30);
+    $token = getenv('DW_PARITY_TOKEN') ?: null;
+    $http = new PrometheusMetricsHttp(['base_uri' => rtrim($url, '/').'/api/', 'http_errors' => false,
+        'connect_timeout' => 1, 'timeout' => 30]);
+    $raw = static function (string $method, string $path, ?array $body = null, bool $worker = false) use ($http, $token, $namespace): array {
+        $response = $http->request($method, $path, ['headers' => array_filter([
+            'Authorization' => $token === null ? null : 'Bearer '.$token,
+            'Accept' => 'application/json', 'X-Namespace' => $namespace,
+            $worker ? 'X-Durable-Workflow-Protocol-Version' : 'X-Durable-Workflow-Control-Plane-Version' => $worker ? '1.20' : '2',
+        ]), ...($body === null ? [] : ['json' => $body])]);
+
+        return ['status' => $response->getStatusCode(),
+            'response' => json_decode((string) $response->getBody(), true, flags: JSON_THROW_ON_ERROR)];
+    };
+    [$typeA, $typeB] = $definition['workflow_types'];
+    $empty = ['codec' => 'avro', 'blob' => $definition['empty_blob']];
+    $poll = ['worker_id' => $worker, 'task_queue' => $first, 'timeout_seconds' => 0];
+    $state = ['applicable' => true, 'namespace' => $namespace, 'queues' => [$first, $second], 'statuses' => []];
+    $step = static function (string $label, array $recorded) use (&$state): array {
+        $state['statuses'][$label] = $recorded['status'];
+
+        return $recorded['response'];
+    };
+    $state['empty'] = $raw('GET', 'system/prometheus-metrics');
+    $step('register', $raw('POST', 'worker/register', ['worker_id' => $worker, 'task_queue' => $first, 'runtime' => 'php',
+        'sdk_version' => 'raw-http', 'supported_workflow_types' => [$typeA, $typeB],
+        'supported_activity_types' => [$definition['activity_type']], 'capabilities' => [],
+        'capability_manifest' => array_fill_keys(['local_activities', 'worker_sessions', 'sticky_execution'],
+            ['supported' => false, 'minimum_protocol_version' => '1.18', 'reason' => 'raw_http']),
+        'max_concurrent_workflow_tasks' => $definition['workflow_slots'],
+        'max_concurrent_activity_tasks' => $definition['activity_slots']], true));
+    $step('start completed', $raw('POST', 'workflows', ['workflow_id' => $namespace.'-completed',
+        'workflow_type' => $typeA, 'task_queue' => $first, 'input' => $empty]));
+    $task = $step('claim', $raw('POST', 'worker/workflow-tasks/poll', $poll, true))['task'] ?? [];
+    $step('schedule activity', $raw('POST', 'worker/workflow-tasks/'.($task['task_id'] ?? 'none').'/complete', [
+        'lease_owner' => $task['lease_owner'] ?? null, 'workflow_task_attempt' => $task['workflow_task_attempt'] ?? null,
+        'commands' => [['type' => 'schedule_activity', 'activity_type' => $definition['activity_type'], 'arguments' => $empty,
+            'retry_policy' => ['max_attempts' => 1, 'backoff_seconds' => []], 'start_to_close_timeout' => 300]]], true));
+    $activity = $step('claim activity', $raw('POST', 'worker/activity-tasks/poll', $poll, true))['task'] ?? [];
+    // Each run and activity lasts at least this long, so its duration is above
+    // the lowest latency bounds.
+    usleep($definition['minimum_duration_ms'] * 1000);
+    $step('complete activity', $raw('POST', 'worker/activity-tasks/'.($activity['task_id'] ?? 'none').'/complete', [
+        'activity_attempt_id' => $activity['activity_attempt_id'] ?? null, 'lease_owner' => $activity['lease_owner'] ?? null,
+        'result' => $empty], true));
+    $task = $step('claim resume', $raw('POST', 'worker/workflow-tasks/poll', $poll, true))['task'] ?? [];
+    $step('complete run', $raw('POST', 'worker/workflow-tasks/'.($task['task_id'] ?? 'none').'/complete', [
+        'lease_owner' => $task['lease_owner'] ?? null, 'workflow_task_attempt' => $task['workflow_task_attempt'] ?? null,
+        'commands' => [['type' => 'complete_workflow', 'result' => $empty]]], true));
+    $step('start terminated', $raw('POST', 'workflows', ['workflow_id' => $namespace.'-terminated',
+        'workflow_type' => $typeB, 'task_queue' => $first, 'input' => $empty]));
+    usleep($definition['minimum_duration_ms'] * 1000);
+    $step('terminate', $raw('POST', 'workflows/'.$namespace.'-terminated/terminate', ['reason' => 'prometheus fixture']));
+    $step('start waiting', $raw('POST', 'workflows', ['workflow_id' => $namespace.'-waiting',
+        'workflow_type' => $typeA, 'task_queue' => $second, 'input' => $empty]));
+    // Past the second of its creation, the waiting run's task is due on every
+    // backend, including PHP's second-precision comparison on SQLite.
+    usleep($definition['minimum_duration_ms'] * 1000);
+    $state['read'] = $raw('GET', 'system/prometheus-metrics');
+
+    return $state;
+}
diff --git a/tests/Fixtures/ServerParity/prometheus-metrics.json b/tests/Fixtures/ServerParity/prometheus-metrics.json
new file mode 100644
index 000000000..512910178
--- /dev/null
+++ b/tests/Fixtures/ServerParity/prometheus-metrics.json
@@ -0,0 +1,30 @@
+{
+  "schema": "durable-workflow.server-parity-fixture/v1",
+  "id": "prometheus-metrics",
+  "review_state": "reviewed",
+  "workflow_type": "parity.v1.echo",
+  "input": {"message": "prometheus metrics λ", "count": 9007199254740993, "optional": null},
+  "typed_value": {"type": "map", "value": {
+    "message": {"type": "string", "value": "prometheus metrics λ"},
+    "count": {"type": "int64", "value": "9007199254740993"},
+    "optional": {"type": "null", "value": null}
+  }},
+  "activity": false,
+  "expected_events": ["StartAccepted", "WorkflowStarted", "WorkflowCompleted"],
+  "prometheus_metrics": {
+    "workflow_types": ["parity.v1.prom_a", "parity.v1.prom_b"],
+    "activity_type": "parity.v1.prom_act",
+    "empty_blob": "wwHioz3/VYAiNwwA",
+    "workflow_slots": 4,
+    "activity_slots": 3,
+    "minimum_duration_ms": 1100,
+    "statuses": {
+      "register": 201, "start completed": 201, "claim": 200, "schedule activity": 200, "claim activity": 200,
+      "complete activity": 200, "claim resume": 200, "complete run": 200, "start terminated": 201, "terminate": 200,
+      "start waiting": 201
+    },
+    "first_queue": {"workflow_tasks_added_last_minute": 3, "workflow_tasks_dispatched_last_minute": 3, "activity_tasks_added_last_minute": 1},
+    "second_queue": {"workflow_ready_tasks": 1, "workflow_ready_due_tasks": 1, "workflow_tasks_added_last_minute": 1, "workflow_tasks_dispatched_last_minute": 1}
+  },
+  "contract": "GET /api/system/prometheus-metrics (published Server 2.11) summarizes a namespace for Prometheus scrapers: workflow series by task queue and workflow type, activity series by task queue, workflow type and activity type, and task-queue runtime state, each family capped at 100 series with its suppression disclosed and its label policy declared. A fresh namespace reports no series. After a worker with four workflow and three activity slots registers on the first queue, a run there completes after one activity, a second run there is terminated, and a run waits on a second queue without a worker, the summary holds three workflow series, one activity series and two task queues with exact counters, task counts and worker capacity. Each counted run and activity lasts at least 1.1 seconds. Latency sums cover those durations, buckets are cumulative, and every duration falls within the 900-second bound. The buckets below the minimum duration are empty, except on published PHP with SQLite: it binds each bound as text, which SQLite orders after every number, so every bucket holds every duration. That reference defect is recorded, not adopted; native reports the cumulative histogram on every backend. Embedded applications have no HTTP metrics."
+}
diff --git a/tests/Unit/ServerParityPrometheusMetricsContractTest.mjs b/tests/Unit/ServerParityPrometheusMetricsContractTest.mjs
new file mode 100644
index 000000000..42cc12799
--- /dev/null
+++ b/tests/Unit/ServerParityPrometheusMetricsContractTest.mjs
@@ -0,0 +1,116 @@
+import assert from 'node:assert/strict';
+import {readFileSync} from 'node:fs';
+import test from 'node:test';
+import {checkPrometheusMetrics} from '../../scripts/conformance/server-parity/prometheus-metrics-contract.mjs';
+
+const fixture = JSON.parse(readFileSync(new URL('../Fixtures/ServerParity/prometheus-metrics.json', import.meta.url)));
+const definition = fixture.prometheus_metrics;
+const id = 'parity-v1-prometheus-metrics';
+const namespace = `${id}-prometheus`;
+const [first, second] = [`${namespace}-q1`, `${namespace}-q2`];
+const [typeA, typeB] = definition.workflow_types;
+const bounds = [0.1, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 900];
+const selection = {
+  workflows: 'bounded_task_queue_and_workflow_type_ascending',
+  activities: 'bounded_task_queue_workflow_type_and_activity_type_ascending',
+  task_queues: 'task_queue_name_ascending',
+};
+const dimensions = {workflows: ['task_queue', 'workflow_type'], activities: ['task_queue', 'workflow_type', 'activity_type'],
+  task_queues: ['task_queue']};
+
+// A histogram of durations, cumulative or with PHP-on-SQLite's text bounds.
+function latency(durations, textBounds = false) {
+  return {count: durations.length, sum: durations.reduce((sum, value) => sum + value, 0),
+    buckets: Object.fromEntries(bounds.map(bound => [String(bound),
+      textBounds ? durations.length : durations.filter(value => value <= bound).length]))};
+}
+
+function queue(name, values = {}) {
+  const entry = {task_queue: name};
+  for (const prefix of ['workflow', 'activity']) {
+    for (const field of ['ready_tasks', 'ready_due_tasks', 'delayed_tasks', 'leased_tasks', 'expired_leases',
+      'tasks_added_last_minute', 'tasks_dispatched_last_minute']) {
+      entry[`${prefix}_${field}`] = 0;
+    }
+  }
+  for (const field of ['active_pollers', 'workflow_slot_capacity', 'activity_slot_capacity', 'available_workflow_slots',
+    'available_activity_slots']) {
+    entry[field] = 0;
+  }
+  return {...entry, ...values};
+}
+
+function summary(series) {
+  const counts = Object.fromEntries(Object.entries(series).map(([family, list]) => [family, list.length]));
+  const total = counts.workflows + counts.activities + counts.task_queues;
+  const labels = family => ({namespace: 'request_scope_not_label', ...Object.fromEntries(dimensions[family]
+    .map(dimension => [dimension, {cardinality_class: 'bounded_series', limit: 100, selection: selection[family]}]))});
+  const disclosure = family => ({label_dimensions: dimensions[family], limit: 100, observed_series_count: counts[family],
+    observed_series_count_precision: 'exact', reported_series_count: counts[family], suppressed_series_count: 0,
+    suppressed_started_total: 0, suppressed_counts_precision: 'exact', truncated: false, selection: selection[family]});
+  return {status: 200, response: {schema: 'durable-workflow.prometheus-metrics.v1', generated_at: '2026-10-22T22:29:44.981017Z',
+    namespace, latency_buckets_seconds: bounds, series, series_count: total, observed_series_count: total,
+    observed_series_count_precision: 'exact', suppressed_series_count: 0, suppressed_series_count_precision: 'exact',
+    cardinality: {series_limits: {workflows: disclosure('workflows'), activities: disclosure('activities'),
+      task_queues: disclosure('task_queues')}, metric_label_sets: {dw_workflow_runs_total: labels('workflows'),
+      dw_workflow_run_latency_seconds: labels('workflows'), dw_activity_executions_total: labels('activities'),
+      dw_activity_execution_latency_seconds: labels('activities'), dw_task_queue_runtime_state: labels('task_queues')}}}};
+}
+
+function observation(textBounds = false) {
+  const run = (queueName, type, values, durations) => ({task_queue: queueName, workflow_type: type, started_total: 1,
+    completed_total: 0, failed_total: 0, cancelled_total: 0, terminated_total: 0, running: 0, ...values,
+    latency_seconds: latency(durations, textBounds)});
+  const read = summary({
+    workflows: [run(first, typeA, {completed_total: 1}, [2.673]), run(first, typeB, {terminated_total: 1}, [1.161]),
+      run(second, typeA, {running: 1}, [])],
+    activities: [{task_queue: first, workflow_type: typeA, activity_type: definition.activity_type, started_total: 1,
+      completed_total: 1, failed_total: 0, cancelled_total: 0, running: 0, latency_seconds: latency([1.102992], textBounds)}],
+    task_queues: [queue(first, {...definition.first_queue, active_pollers: 1, workflow_slot_capacity: 4, activity_slot_capacity: 3,
+      available_workflow_slots: 4, available_activity_slots: 3}), queue(second, definition.second_queue)],
+  });
+  return structuredClone({mode: 'http', prometheus_metrics: {applicable: true, namespace, queues: [first, second],
+    statuses: definition.statuses, empty: summary({workflows: [], activities: [], task_queues: []}), read}});
+}
+
+test('the Prometheus summary accepts the cumulative histogram', () => {
+  assert.deepStrictEqual(checkPrometheusMetrics(fixture, observation(), id), {scope: 'prometheus_metrics', series: 6});
+});
+
+test("the Prometheus summary records PHP on SQLite's text-bound buckets", () => {
+  checkPrometheusMetrics(fixture, observation(true), id);
+});
+
+test('the embedded Prometheus summary is explicitly inapplicable', () => {
+  checkPrometheusMetrics(fixture, {mode: 'embedded', prometheus_metrics: {applicable: false, reason: 'embedded_has_no_http_metrics'}}, id);
+  assert.throws(() => checkPrometheusMetrics(fixture, {mode: 'embedded', prometheus_metrics: {applicable: true}}, id));
+});
+
+const read = state => state.read.response;
+for (const [label, mutate] of [
+  ['an unimplemented route', state => { state.read = {status: 501, response: {reason: 'rust_capability_not_implemented'}}; }],
+  ['a refused scenario request', state => { state.statuses.terminate = 422; }],
+  ['another namespace', state => { read(state).namespace = 'default'; }],
+  ['a missing series', state => { read(state).series.workflows.pop(); }],
+  ['series out of order', state => { read(state).series.workflows.reverse(); }],
+  ['a terminated run counted as cancelled', state => { Object.assign(read(state).series.workflows[1], {terminated_total: 0, cancelled_total: 1}); }],
+  ['a waiting run not running', state => { read(state).series.workflows[2].running = 0; }],
+  ['a duration for a waiting run', state => { read(state).series.workflows[2].latency_seconds.count = 1; }],
+  ['a duration in a bucket below its length', state => { read(state).series.workflows[0].latency_seconds.buckets['0.5'] = 1; }],
+  ['non-cumulative buckets', state => { read(state).series.workflows[0].latency_seconds.buckets['900'] = 0; }],
+  ['a sum below the minimum duration', state => { read(state).series.activities[0].latency_seconds.sum = 0.2; }],
+  ['a missing bucket', state => { delete read(state).series.activities[0].latency_seconds.buckets['300']; }],
+  ['a delayed due task', state => { Object.assign(read(state).series.task_queues[1], {workflow_ready_due_tasks: 0, workflow_delayed_tasks: 1}); }],
+  ['missing worker capacity', state => { read(state).series.task_queues[0].workflow_slot_capacity = 0; }],
+  ['a worker counted on the second queue', state => { read(state).series.task_queues[1].active_pollers = 1; }],
+  ['an undisclosed truncation', state => { read(state).cardinality.series_limits.workflows.truncated = true; }],
+  ['another series limit', state => { read(state).cardinality.metric_label_sets.dw_workflow_runs_total.task_queue.limit = 50; }],
+  ['a millisecond time', state => { read(state).generated_at = '2026-10-22T22:29:44.981Z'; }],
+  ['an extra field', state => { read(state).control_plane = {}; }],
+]) {
+  test(`the Prometheus summary rejects ${label}`, () => {
+    const value = observation();
+    mutate(value.prometheus_metrics);
+    assert.throws(() => checkPrometheusMetrics(fixture, value, id));
+  });
+}