Bound metric routes and honor configured recorder selection

.github/workflows/ci.yml

@@ -152,7 +152,7 @@ jobs:
         uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7  # clean-trust
         with:
           clients: pip,python
-      - run: pip install -e '.[dev]'
+      - run: pip install -e '.[dev,prometheus]'
       - name: Run unit tests with stalled-thread diagnostics
         id: unit_tests
         shell: bash

CHANGELOG.md

@@ -17,6 +17,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Bound client metric routes to reviewed operation templates, including
+  update-validation tasks, worker sessions, streams and Nexus operations.
+  Unknown paths emit `unknown`; static poll routes retain `/poll`. Variable
+  identifiers and query strings do not create metric series.
+- Honor explicitly configured and inherited metric recorders even when they
+  evaluate to false, rather than silently replacing them with the no-op.
 - Back off workflow, activity and query polling independently after repeated
   HTTP or transport failures, with jitter and a 30-second cap. Successful polls
   reset the delay, and repeated authentication warnings are rate-limited.

docs/reference/metrics.md

@@ -1,3 +1,44 @@
 # Metrics
 
+## Recorder contract
+
+The expanded route normalization described here is available in unreleased
+SDK source. Published packages retain their existing route labels until the
+next package release.
+
+Client request, worker poll and task dispatch observations follow the
+[shared SDK metrics contract](https://github.com/durable-workflow/.github/tree/main/sdk-metrics).
+It defines six recorder names, counter or histogram instruments, count or
+second units, and the exact tag keys. The default recorder drops observations.
+A worker inherits its client's recorder unless given an explicit recorder.
+Selection uses `None` to mean unspecified; an empty or false-valued recorder
+still receives observations. Pass `NoopMetrics()` explicitly to discard them.
+
+Each instrumented operation increments its counter by one and records one
+nonnegative monotonic duration in seconds. Client observations include the
+request's transport retries; they do not count each HTTP attempt. Poll and task
+observations describe local worker activity, not durable execution counts,
+workflow completion, billing units or schedule-to-start latency.
+
+Client route tags use reviewed path templates, without `/api`, query strings,
+or variable IDs. For example, update-validation results use
+`/worker/update-validation-tasks/{task_id}/approve`, and session renewal uses
+`/worker/sessions/{session_id}/heartbeat`. Static poll operations retain their
+`/poll` route. Unknown operations emit the single `unknown` route value.
+Names of workflows, runs, activities, workers, sessions, queues, streams and
+service operations in paths are replaced with placeholders. Request payloads,
+credentials and exception messages are not core metric labels.
+
+The worker's configured `task_queue` label remains application-controlled.
+Use bounded queue names rather than one per run. Exporter resource labels need
+their own cardinality review. Missing observations do not establish zero work
+or a healthy process; abrupt shutdown can prevent the final observation.
+
+`PrometheusMetrics` uses normal Prometheus exporter conventions: counter series
+end in `_total`, and duration histograms expose bucket, sum and count series.
+Those suffixes do not change the recorder names. Install the `prometheus` extra
+to use this adapter. Applications own scrape/export configuration and additional
+resource labels. Custom recorders should remain fast and handle their exporter
+failures without throwing into the client or worker.
+
 ::: durable_workflow.metrics

docs/sdk-reference.md

@@ -1024,6 +1024,12 @@ client = Client("http://server:8080", token="dev-token-123", metrics=metrics)
 
 Custom recorders implement `increment(name, value=1.0, tags=None)` and `record(name, value, tags=None)`.
 
+The [metrics reference](reference/metrics.md#recorder-contract) explains the
+shared names, count/second units, recorder inheritance and exporter conventions.
+In unreleased source, client routes use reviewed templates without variable IDs or query strings;
+unknown operations use `unknown`. Keep configured worker queue names and any
+exporter resource labels bounded.
+
 ## Worker interceptors
 
 Use `Worker(interceptors=[...])` when instrumentation needs the task payload,

src/durable_workflow/_metric_routes.py

@@ -0,0 +1,57 @@
+"""Bound request labels to reviewed SDK operations, never arbitrary URL values."""
+
+from __future__ import annotations
+
+_ROUTES = (
+    "/", "/health", "/cluster/info", "/activities", "/activities/{activity_id}",
+    "/namespaces", "/namespaces/{name}", "/namespaces/{name}/external-storage",
+    "/search-attributes", "/search-attributes/{name}", "/storage/test",
+    "/task-queues", "/task-queues/{task_queue}", "/task-queues/{task_queue}/build-ids",
+    *(f"/task-queues/{{task_queue}}/build-ids/{action}" for action in ("drain", "promote", "resume")),
+    *(f"/system/{operation}{suffix}" for operation in ("activity-timeouts", "repair", "retention")
+      for suffix in ("", "/pass")),
+    "/workers", "/workers/{worker_id}",
+    "/worker/register", "/worker/heartbeat", "/worker/registrations/{worker_id}",
+    "/worker/sessions", "/worker/sessions/{session_id}", "/worker/sessions/{session_id}/heartbeat",
+    *(f"/worker/{kind}-tasks/poll" for kind in ("workflow", "activity", "query", "update-validation")),
+    *(f"/worker/{kind}-tasks/{{task_id}}/{action}" for kind, actions in (
+        ("workflow", ("complete", "fail", "heartbeat", "history", "deliver-cancellation")),
+        ("activity", ("complete", "fail", "heartbeat", "status", "renew-lease", "acknowledge-cancellation")),
+        ("query", ("complete", "fail", "status")),
+        ("update-validation", ("approve", "reject")),
+    ) for action in actions),
+    *(f"/worker/workflow-tasks/{{task_id}}/cancellation-scopes/{action}"
+      for action in ("open", "prepare", "deliver")),
+    *(f"/worker/workflow-tasks/{{task_id}}/local-activities/{action}"
+      for action in ("checkpoint", "checkpoint-group", "prepare", "recover")),
+    *(f"/worker/workflow-tasks/{{task_id}}/local-activities/{{activity_attempt_id}}/{action}"
+      for action in ("control", "heartbeat", "outcome", "acknowledge-cancellation")),
+    "/workflows", "/workflows/{workflow_id}",
+    *(f"/workflows/{{workflow_id}}/{action}"
+      for action in ("archive", "cancel", "repair", "terminate", "request-cancellation", "runs")),
+    *(f"/workflows/{{workflow_id}}/{action}/{{name}}" for action in ("signal", "query", "update")),
+    "/workflows/{workflow_id}/message-streams/{name}/messages",
+    "/workflows/{workflow_id}/runs/{run_id}",
+    *(f"/workflows/{{workflow_id}}/runs/{{run_id}}/{action}"
+      for action in ("history", "history/export", "redrive", "request-cancellation", "streams")),
+    "/workflows/{workflow_id}/runs/{run_id}/streams/{name}",
+    *(f"/workflows/{{workflow_id}}/runs/{{run_id}}/streams/{{name}}/{action}" for action in ("items", "close")),
+    "/schedules", "/schedules/{schedule_id}",
+    *(f"/schedules/{{schedule_id}}/{action}" for action in ("backfill", "pause", "resume", "trigger", "history")),
+    "/bridge-adapters/webhook/{adapter}",
+    "/service-endpoints/{endpoint_name}/services/{service_name}/operations/{operation_name}/execute",
+)
+_TEMPLATES = tuple((route, tuple(route.strip("/").split("/"))) for route in _ROUTES)
+
+
+def route_for_metrics(path: str) -> str:
+    """Return a supported operation template or the single bounded fallback."""
+    clean_path = path.split("?", 1)[0].split("#", 1)[0]
+    parts = tuple(clean_path.strip("/").split("/"))
+    for route, template in _TEMPLATES:
+        if len(parts) == len(template) and all(
+            expected == actual or expected.startswith("{") and bool(actual)
+            for actual, expected in zip(parts, template, strict=True)
+        ):
+            return route
+    return "unknown"

src/durable_workflow/client.py

@@ -48,6 +48,7 @@
     scope_timestamp,
 )
 from ._cooperative_cancellation import CancellationDelivery, CancellationRequest
+from ._metric_routes import route_for_metrics as _route_for_metrics
 from ._query_task_authority import QueryTaskAuthority, current_query_authority
 from .errors import (
     REQUEST_ID_CONFLICT,
@@ -290,38 +291,6 @@ def _command_request_failure(exc: Exception, *, command: str, request_id: str) -
     return exc
 
 
-def _route_for_metrics(path: str) -> str:
-    clean_path = path.split("?", 1)[0]
-    parts = [part for part in clean_path.strip("/").split("/") if part]
-    if not parts:
-        return "/"
-
-    if parts[0] == "workflows" and len(parts) >= 2:
-        parts[1] = "{workflow_id}"
-        if len(parts) >= 4 and parts[2] in {"signal", "query", "update"}:
-            parts[3] = "{name}"
-        if len(parts) >= 4 and parts[2] == "runs":
-            parts[3] = "{run_id}"
-    elif parts[0] == "schedules" and len(parts) >= 2:
-        parts[1] = "{schedule_id}"
-    elif parts[0] in {"namespaces", "search-attributes"} and len(parts) >= 2:
-        parts[1] = "{name}"
-    elif parts[0] == "workers" and len(parts) >= 2:
-        parts[1] = "{worker_id}"
-    elif parts[:2] == ["worker", "registrations"] and len(parts) >= 3:
-        parts[2] = "{worker_id}"
-    elif parts[:2] == ["bridge-adapters", "webhook"] and len(parts) >= 3:
-        parts[2] = "{adapter}"
-    elif (
-        parts[:2] == ["worker", "workflow-tasks"]
-        or parts[:2] == ["worker", "activity-tasks"]
-        or parts[:2] == ["worker", "query-tasks"]
-    ) and len(parts) >= 3:
-        parts[2] = "{task_id}"
-
-    return "/" + "/".join(parts)
-
-
 def _resolve_namespace_name(
     name: str | None,
     namespace_alias: str | None,
@@ -1673,7 +1642,7 @@ def __init__(
         self.timeout = timeout
         self._tls_context = tls_context
         self.retry_policy = retry_policy or TransportRetryPolicy()
-        self.metrics = metrics or NOOP_METRICS
+        self.metrics = metrics if metrics is not None else NOOP_METRICS
         self.payload_size_warning_config = (
             serializer.PayloadSizeWarningConfig(
                 limit_bytes=payload_size_limit_bytes,

src/durable_workflow/worker.py

@@ -1197,7 +1197,7 @@ def __init__(
         self._last_cpu_sample_at: float | None = None
         self._last_cpu_total_seconds: float = 0.0
         configured_metrics = metrics if metrics is not None else getattr(client, "metrics", NOOP_METRICS)
-        self.metrics: MetricsRecorder = configured_metrics or NOOP_METRICS
+        self.metrics: MetricsRecorder = configured_metrics if configured_metrics is not None else NOOP_METRICS
         self.interceptors = tuple(interceptors)
         self.external_storage = (
             external_storage

tests/fixtures/sdk_metrics/contract.json

@@ -0,0 +1,63 @@
+{
+  "schema": "durable-workflow.sdk-metrics/v1",
+  "revision": 1,
+  "reference": {
+    "repository": "durable-workflow/sdk-python",
+    "commit": "2da02a7ac8a68a7cd4267e86ef10e0198a309592",
+    "paths": [
+      "src/durable_workflow/metrics.py",
+      "src/durable_workflow/client.py",
+      "src/durable_workflow/worker.py"
+    ]
+  },
+  "metrics": [
+    {
+      "name": "durable_workflow_client_requests",
+      "type": "counter",
+      "unit": "1",
+      "recorder_method": "increment",
+      "scope": "client_request",
+      "tags": ["method", "route", "plane", "status_code", "outcome"]
+    },
+    {
+      "name": "durable_workflow_client_request_duration_seconds",
+      "type": "histogram",
+      "unit": "s",
+      "recorder_method": "record",
+      "scope": "client_request",
+      "tags": ["method", "route", "plane", "status_code", "outcome"]
+    },
+    {
+      "name": "durable_workflow_worker_polls",
+      "type": "counter",
+      "unit": "1",
+      "recorder_method": "increment",
+      "scope": "worker_poll",
+      "tags": ["task_kind", "task_queue", "outcome"]
+    },
+    {
+      "name": "durable_workflow_worker_poll_duration_seconds",
+      "type": "histogram",
+      "unit": "s",
+      "recorder_method": "record",
+      "scope": "worker_poll",
+      "tags": ["task_kind", "task_queue", "outcome"]
+    },
+    {
+      "name": "durable_workflow_worker_tasks",
+      "type": "counter",
+      "unit": "1",
+      "recorder_method": "increment",
+      "scope": "worker_task",
+      "tags": ["task_kind", "task_queue", "outcome"]
+    },
+    {
+      "name": "durable_workflow_worker_task_duration_seconds",
+      "type": "histogram",
+      "unit": "s",
+      "recorder_method": "record",
+      "scope": "worker_task",
+      "tags": ["task_kind", "task_queue", "outcome"]
+    }
+  ]
+}

tests/fixtures/sdk_metrics/provenance.json

@@ -0,0 +1,4 @@
+{
+  "source": "https://github.com/durable-workflow/.github/blob/73e8fc323e4f46322945a007e54a546395904f79/sdk-metrics/contract.json",
+  "sha256": "c456f4f434c601fa26f0caba07ef0fd29f548ca8f5fc29ba2f6430f01e32c0f0"
+}