diff --git a/sdks/python/apache_beam/metrics/cells.py b/sdks/python/apache_beam/metrics/cells.py
index 8300d7f063e9..ca91410cb98c 100644
--- a/sdks/python/apache_beam/metrics/cells.py
+++ b/sdks/python/apache_beam/metrics/cells.py
@@ -181,6 +181,18 @@ def reset(self):
     # type: () -> None
     self.data = DistributionData.identity_element()
 
+  def reset_to(self, value):
+    """Atomically replaces the distribution with a single sample of value.
+
+    Like ``reset()`` followed by ``update(value)``, except that a concurrent
+    reader never sees the empty distribution in between. Useful for publishing
+    the latest value of a gauge-like quantity as a distribution whose
+    cumulative mean (sum / count) is that value.
+    """
+    ivalue = int(value)
+    with self._lock:
+      self.data = DistributionData(ivalue, 1, ivalue, ivalue)
+
   def combine(self, other):
     # type: (DistributionCell) -> DistributionCell
     result = DistributionCell()
diff --git a/sdks/python/apache_beam/metrics/cells_test.py b/sdks/python/apache_beam/metrics/cells_test.py
index 11ea20ed6f6d..0ec509d52b91 100644
--- a/sdks/python/apache_beam/metrics/cells_test.py
+++ b/sdks/python/apache_beam/metrics/cells_test.py
@@ -131,6 +131,37 @@ def test_basic_operations(self):
     d.update(900)
     self.assertEqual(d.get_cumulative(), DistributionData(912, 3, 2, 900))
 
+  def test_reset_to(self):
+    d = DistributionCell()
+    d.update(10)
+    d.update(2)
+    d.reset_to(7)
+    self.assertEqual(d.get_cumulative(), DistributionData(7, 1, 7, 7))
+    d.reset_to(3.6)
+    self.assertEqual(d.get_cumulative(), DistributionData(3, 1, 3, 3))
+    d.update(5)
+    self.assertEqual(d.get_cumulative(), DistributionData(8, 2, 3, 5))
+
+  def test_reset_to_never_exposes_an_empty_distribution(self):
+    d = DistributionCell()
+    d.reset_to(1)
+    stop = threading.Event()
+    counts = set()
+
+    def read():
+      while not stop.is_set():
+        counts.add(d.get_cumulative().count)
+
+    reader = threading.Thread(target=read)
+    reader.start()
+    try:
+      for i in range(20000):
+        d.reset_to(i % 7 + 1)
+    finally:
+      stop.set()
+      reader.join()
+    self.assertEqual(counts, {1})
+
   def test_integer_only(self):
     d = DistributionCell()
     d.update(3.1)
diff --git a/sdks/python/apache_beam/ml/inference/autoscaling_metrics.py b/sdks/python/apache_beam/ml/inference/autoscaling_metrics.py
new file mode 100644
index 000000000000..17945d427c57
--- /dev/null
+++ b/sdks/python/apache_beam/ml/inference/autoscaling_metrics.py
@@ -0,0 +1,591 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+"""Capacity-based autoscaling signal for model handlers that run a model server.
+
+Dataflow's GPU autoscaling reads the process-wide distribution
+``num_loaded_models_<tag>`` in the ``BeamML_ModelManager`` namespace as how
+many requests each worker can handle, and sizes the pool roughly as
+``ceil(backlog / value)``. ``ModelManager`` publishes the number of loaded
+model copies there. A handler that serves inference through an out-of-process
+model server (vLLM, Triton, ...) always has exactly one copy per worker, so it
+publishes an estimate of the worker's capacity under that metric instead.
+
+:class:`PrometheusNumLoadedModelsPublisher` reads the server's metrics over
+HTTP, so the server must:
+
+* run on the same worker as the handler and serve the Prometheus text format
+  at ``http://localhost:<port>/metrics``;
+* export a counter of completed requests (monotonically increasing, reset only
+  when the server restarts), e.g. vLLM's ``vllm:request_success_total``; and
+* export a gauge of the requests currently in flight, e.g. vLLM's
+  ``vllm:num_requests_running``.
+
+A handler for such a server (vLLM, for example) supplies only the two metric
+names, a way to find the port, and the RunInference model identifier::
+
+  publisher = PrometheusNumLoadedModelsPublisher(
+      metrics_port_fn=lambda: server.port if server.is_up else None,
+      completed_metric='vllm:request_success_total',
+      running_metric='vllm:num_requests_running',
+      model_identifier=model_identifier,
+      model_name=model_name,
+      config=autoscaling_config)  # An AutoscalingMetricConfig, or None.
+  publisher.start()  # Once the server is reachable; publisher.stop() at exit.
+
+If the server exports different metrics,
+``AutoscalingMetricConfig.load_signal_fn`` can replace the scrape and the
+estimate: it is given the server's port and returns the capacity directly.
+
+The public API is :class:`PrometheusNumLoadedModelsPublisher` and
+:class:`AutoscalingMetricConfig`; everything else is an implementation detail.
+"""
+
+# pytype: skip-file
+
+import logging
+import re
+import threading
+import time
+import urllib.request
+from collections.abc import Callable
+from collections.abc import Mapping
+from dataclasses import dataclass
+from typing import NamedTuple
+from typing import Optional
+
+from apache_beam.metrics.execution import MetricsEnvironment
+from apache_beam.metrics.metricbase import MetricName
+
+__all__ = [
+    'AutoscalingMetricConfig',
+    'PrometheusNumLoadedModelsPublisher',
+]
+
+# The metric the runner reads (shared with model_manager.py).
+MODEL_MANAGER_METRICS_NAMESPACE = 'BeamML_ModelManager'
+NUM_LOADED_MODELS_METRIC_PREFIX = 'num_loaded_models_'
+
+DEFAULT_POLL_INTERVAL_SECS = 5.0
+
+# Estimator defaults.
+# A larger multiplier lets each worker hold more backlog (fewer workers); a
+# smaller one scales out more aggressively.
+DEFAULT_MULTIPLIER = 1.0
+DEFAULT_SIGNAL_WINDOW_SECS = 1800.0  # 30 minutes.
+DEFAULT_RUNNING_PERCENTILE = 90.0
+DEFAULT_MIN_REQUESTS = 3
+DEFAULT_WARMUP_TIMEOUT_SECS = 10.0
+
+# How _CompletionRateCapacityEstimator turns its samples into an estimate.
+CONCURRENCY = 'concurrency'
+WINDOW_MEAN = 'window_mean'
+WINDOW_MAX = 'window_max'
+LIFETIME_MAX = 'lifetime_max'
+_AGGREGATIONS = (CONCURRENCY, WINDOW_MEAN, WINDOW_MAX, LIFETIME_MAX)
+
+# One sample line of the Prometheus text format:
+#   metric_name{label="value",...} value [timestamp]
+_PROMETHEUS_SAMPLE_RE = re.compile(
+    r'(?P<name>[a-zA-Z_:][a-zA-Z0-9_:]*)'
+    r'(?:\{(?P<labels>(?:[^"}]|"(?:[^"\\]|\\.)*")*)\})?'
+    r'\s+(?P<value>\S+)')
+# One label pair; values are quoted and may contain \\, \" and \n escapes.
+_PROMETHEUS_LABEL_RE = re.compile(
+    r'(?P<name>[a-zA-Z_][a-zA-Z0-9_]*)\s*=\s*"(?P<value>(?:[^"\\]|\\.)*)"')
+_PROMETHEUS_ESCAPE_RE = re.compile(r'\\(.)')
+
+
+def _sanitize_metric_tag(name: str) -> str:
+  """Turns a model name into a metric-tag-safe token.
+
+  Only used for the fallback tag of handlers running outside RunInference; the
+  RunInference model identifier is always used verbatim.
+
+  Args:
+    name: The model name (may contain ``/``, ``:`` etc.).
+
+  Returns:
+    ``name`` with characters outside ``[A-Za-z0-9_]`` replaced by ``_``.
+  """
+  return ''.join(c if (c.isalnum() or c == '_') else '_' for c in name)
+
+
+def _scrape_prometheus_metrics(
+    port: int,
+    timeout_secs: float = 5.0,
+    path: str = '/metrics') -> Optional[str]:
+  """Fetches a local Prometheus endpoint.
+
+  Args:
+    port: Port of the server on ``localhost``.
+    timeout_secs: Request timeout.
+    path: Endpoint path.
+
+  Returns:
+    The response body, or ``None`` on any failure (e.g. the server is still
+    starting or restarting), so callers can simply skip that sample.
+  """
+  url = f'http://localhost:{port}{path}'
+  try:
+    with urllib.request.urlopen(url, timeout=timeout_secs) as response:
+      return response.read().decode('utf-8', errors='replace')
+  except Exception:  # pylint: disable=broad-except
+    return None
+
+
+def _parse_prometheus_labels(label_str: str) -> dict[str, str]:
+  """Parses the inside of ``{k="v",...}``, undoing backslash escapes."""
+  def unescape(match: re.Match) -> str:
+    return '\n' if match.group(1) == 'n' else match.group(1)
+
+  return {
+      match['name']: _PROMETHEUS_ESCAPE_RE.sub(unescape, match['value'])
+      for match in _PROMETHEUS_LABEL_RE.finditer(label_str)
+  }
+
+
+def _parse_prometheus_metric(
+    metrics_text: str,
+    metric_name: str,
+    label_filter: Optional[Mapping[str, str]] = None,
+) -> Optional[float]:
+  """Sums the values of one metric from Prometheus text-format exposition.
+
+  Args:
+    metrics_text: The full body returned by the server's ``/metrics`` endpoint.
+    metric_name: The exact sample name to match (e.g.
+      ``vllm:num_requests_running``).
+    label_filter: If provided, only samples whose labels contain all of these
+      key/value pairs are summed.
+
+  Returns:
+    The sum of the matching sample values, or ``None`` if there are none.
+  """
+  total: Optional[float] = None
+  for line in metrics_text.splitlines():
+    match = _PROMETHEUS_SAMPLE_RE.match(line.strip())
+    # Comment, blank and other-metric lines don't match.
+    if match is None or match['name'] != metric_name:
+      continue
+    if label_filter:
+      labels = _parse_prometheus_labels(match['labels'] or '')
+      if any(labels.get(k) != v for k, v in label_filter.items()):
+        continue
+    try:
+      value = float(match['value'])
+    except ValueError:
+      continue
+    total = value if total is None else total + value
+  return total
+
+
+def _nearest_rank_percentile(values: list[float], q: float) -> float:
+  """Nearest-rank percentile (``q`` in [0, 100]) of a non-empty list."""
+  ordered = sorted(values)
+  q = min(max(q, 0.0), 100.0)
+  # Nearest rank ceil(q / 100 * n), in arithmetic that avoids float rounding.
+  rank = int((q * len(ordered) + 99) // 100)
+  return ordered[min(max(rank - 1, 0), len(ordered) - 1)]
+
+
+class _Interval(NamedTuple):
+  """One sampling interval in the estimator's trailing window."""
+  end_ts: float
+  completed: float  # Requests completed during the interval.
+  running: float  # Requests in flight at the end of the interval.
+  secs: float  # Length of the interval.
+
+
+class _CompletionRateCapacityEstimator:
+  """Estimates how many requests a worker can handle.
+
+  Feed it periodic samples of a model server's completed-requests counter and
+  in-flight gauge through :meth:`update`. Each sample closes an interval, and
+  ``aggregation`` picks how the intervals become an estimate. Every mode
+  returns ``max(min_value, value * multiplier)``. This class does no I/O.
+
+  ``'concurrency'`` (the default) computes, over a trailing ``window_secs``
+  window::
+
+      mu          = sum(completed) / sum(running * secs)
+      running_ref = percentile(running, running_percentile)
+      value       = mu * running_ref
+
+  ``mu`` is the throughput of one in-flight slot (roughly ``1 / latency``),
+  computed as a ratio of windowed sums so discrete completions don't make it
+  jumpy. ``running_ref`` is the concurrency the worker typically sustains; the
+  p90 rather than the max avoids latching onto rare backlog spikes. Their
+  product is the worker's sustained throughput.
+
+  Normalizing by concurrency keeps the estimate steady when load changes: right
+  after a scaling event, when the runner rebalances work and a worker is
+  briefly starved, its raw completion rate collapses, but ``mu`` and
+  ``running_ref`` hold, so the estimate does not crater and trigger a downscale
+  spiral. This mode suits servers whose requests take long enough to be seen
+  in flight by periodic sampling, such as LLM servers with latencies of a
+  second or more; for millisecond latencies the gauge almost always reads 0.
+
+  The other modes use the raw completion rate and ignore concurrency, so
+  they also work for millisecond-latency servers:
+
+  * ``'window_mean'``: completions per second over the trailing window. Drops
+    when a worker is starved.
+  * ``'window_max'``: the highest single-interval completion rate in the
+    trailing window.
+  * ``'lifetime_max'``: the highest single-interval completion rate seen since
+    the estimator was created. Never decays.
+
+  Single-interval rates are noisy when requests take seconds, since a few
+  completions can land in one interval; the max modes latch onto such bursts.
+
+  In every mode, intervals with no completions and nothing in flight are
+  skipped, so idle periods don't drag the estimate down. While requests are in
+  flight but none has completed within the window, there is no estimate either
+  (except in ``'lifetime_max'`` mode once a peak has been seen).
+
+  Units: every mode is in requests per second. The runner reads the published
+  value as how many requests one worker can handle, so each worker is sized
+  for about one second of its sustained throughput; ``multiplier`` scales
+  that, e.g. 2.0 sizes each worker for two seconds' worth of requests.
+
+  Args:
+    multiplier: Dimensionless factor on the estimate; larger means fewer
+      workers.
+    min_value: Floor of the estimate.
+    window_secs: Trailing window; 0 keeps every interval. ``'lifetime_max'``
+      uses it only to count completions for ``min_requests``.
+    running_percentile: Percentile of in-flight concurrency used as
+      ``running_ref``; ``'concurrency'`` mode only.
+    min_requests: Completions a newly started worker needs before it reports,
+      unless ``warmup_timeout_secs`` passes first.
+    warmup_timeout_secs: Longest time after a worker first sees work that it
+      waits for ``min_requests``.
+    aggregation: ``'concurrency'``, ``'window_mean'``, ``'window_max'`` or
+      ``'lifetime_max'``.
+
+  Raises:
+    ValueError: if ``aggregation`` is not one of the above.
+  """
+  def __init__(
+      self,
+      multiplier: float = DEFAULT_MULTIPLIER,
+      min_value: int = 1,
+      window_secs: float = DEFAULT_SIGNAL_WINDOW_SECS,
+      running_percentile: float = DEFAULT_RUNNING_PERCENTILE,
+      min_requests: int = DEFAULT_MIN_REQUESTS,
+      warmup_timeout_secs: float = DEFAULT_WARMUP_TIMEOUT_SECS,
+      aggregation: str = CONCURRENCY,
+  ):
+    if aggregation not in _AGGREGATIONS:
+      raise ValueError(
+          f'Unknown aggregation {aggregation!r}; expected one of '
+          f'{", ".join(_AGGREGATIONS)}.')
+    self._aggregation = aggregation
+    self._multiplier = max(float(multiplier), 0.0)
+    self._min_value = int(min_value)
+    self._window_secs = max(float(window_secs), 0.0)
+    self._running_percentile = min(max(float(running_percentile), 0.0), 100.0)
+    self._min_requests = max(int(min_requests), 0)
+    self._warmup_timeout_secs = max(float(warmup_timeout_secs), 0.0)
+    self._last_count: Optional[float] = None
+    self._last_ts: Optional[float] = None
+    self._first_active_ts: Optional[float] = None
+    self._samples: list[_Interval] = []
+    self._peak_rate = 0.0  # Highest single-interval rate, for lifetime_max.
+
+  def update(
+      self,
+      completed_total: float,
+      running: float,
+      now: Optional[float] = None) -> Optional[float]:
+    """Records one sample and returns the current estimate.
+
+    Args:
+      completed_total: Current value of the completed-requests counter.
+      running: Current number of in-flight requests.
+      now: Sample timestamp in seconds; defaults to ``time.monotonic()``.
+
+    Returns:
+      The capacity estimate, or ``None`` when there is not enough data (first
+      sample, counter reset, idle window or warmup); the caller should then
+      keep its previous value.
+    """
+    if now is None:
+      now = time.monotonic()
+    prev_count, prev_ts = self._last_count, self._last_ts
+    self._last_count, self._last_ts = completed_total, now
+    if prev_count is None or prev_ts is None:
+      return None  # The first sample only sets the baseline.
+    secs = now - prev_ts
+    completed = completed_total - prev_count
+    if secs <= 0 or completed < 0 or running < 0:
+      return None  # Clock skew, counter reset (server restart) or bad gauge.
+
+    if self._window_secs > 0:
+      cutoff = now - self._window_secs
+      self._samples = [s for s in self._samples if s.end_ts >= cutoff]
+    if completed == 0 and running == 0:
+      # Skip fully idle intervals so idle periods don't drag the estimate down.
+      return None
+    self._samples.append(
+        _Interval(now, float(completed), float(running), float(secs)))
+    self._peak_rate = max(self._peak_rate, completed / secs)
+
+    total_completed = sum(s.completed for s in self._samples)
+    slot_secs = sum(s.running * s.secs for s in self._samples)
+    if self._aggregation == CONCURRENCY and slot_secs <= 0:
+      return None  # Nothing was in flight during the window.
+    if self._first_active_ts is None:
+      self._first_active_ts = prev_ts
+    if total_completed <= 0 and not (self._aggregation == LIFETIME_MAX and
+                                     self._peak_rate > 0):
+      # Requests are in flight but none has completed in the window yet, e.g.
+      # slow first requests or the first poll after an idle period. That says
+      # nothing about capacity, so keep the previous value instead of
+      # reporting the floor.
+      return None
+    if (total_completed < self._min_requests and
+        now - self._first_active_ts < self._warmup_timeout_secs):
+      # Wait until a new worker is calibrated, so it doesn't briefly report
+      # the floor value for the whole job.
+      return None
+
+    if self._aggregation == WINDOW_MEAN:
+      value = total_completed / sum(s.secs for s in self._samples)
+    elif self._aggregation == WINDOW_MAX:
+      value = max(s.completed / s.secs for s in self._samples)
+    elif self._aggregation == LIFETIME_MAX:
+      value = self._peak_rate
+    else:
+      mu = total_completed / slot_secs
+      running_ref = _nearest_rank_percentile(
+          [s.running for s in self._samples if s.running > 0],
+          self._running_percentile)
+      value = mu * running_ref
+    return max(float(self._min_value), value * self._multiplier)
+
+
+@dataclass(frozen=True)
+class AutoscalingMetricConfig:
+  """User-facing knobs for the capacity-based autoscaling signal.
+
+  Model handlers that publish the signal take an instance, typically as an
+  ``autoscaling_config`` argument; leave it unset for the defaults.
+
+  Attributes:
+    poll_interval_secs: How often to read the server and publish.
+    multiplier: Factor on the capacity estimate; larger means fewer workers,
+      smaller scales out more aggressively.
+    signal_window_secs: Trailing window of the capacity estimate.
+    min_requests: Completions a newly started worker needs before it
+      publishes, unless ``warmup_timeout_secs`` passes first.
+    warmup_timeout_secs: Longest time a new worker waits for ``min_requests``.
+    aggregation: How the estimate, in requests per second, is computed:
+
+      * ``'concurrency'`` (default): per-slot throughput over the window times
+        the p90 in-flight concurrency. Holds steady when a worker is briefly
+        starved; suits servers with request latencies of a second or more.
+      * ``'window_mean'``: completions per second over the window.
+      * ``'window_max'``: the highest single-interval completion rate in the
+        window.
+      * ``'lifetime_max'``: the highest single-interval completion rate seen
+        so far; never decays.
+    load_signal_fn: Optional replacement for the built-in estimate: given the
+      server's metrics port, returns the capacity directly.
+    tag: Optional metric tag override. Leave unset so the RunInference model
+      identifier is used, which is what the runner looks up and which stays
+      stable across model updates. Only set it if you pass the same value as
+      ``RunInference(model_identifier=...)``.
+  """
+  poll_interval_secs: float = DEFAULT_POLL_INTERVAL_SECS
+  multiplier: float = DEFAULT_MULTIPLIER
+  signal_window_secs: float = DEFAULT_SIGNAL_WINDOW_SECS
+  min_requests: int = DEFAULT_MIN_REQUESTS
+  warmup_timeout_secs: float = DEFAULT_WARMUP_TIMEOUT_SECS
+  aggregation: str = CONCURRENCY
+  load_signal_fn: Optional[Callable[[int], Optional[float]]] = None
+  tag: Optional[str] = None
+
+
+class PrometheusNumLoadedModelsPublisher:
+  """Publishes a Prometheus model server's capacity for autoscaling.
+
+  The capacity is published as ``num_loaded_models_<tag>``.
+
+  The server must meet the requirements in the module docstring: a Prometheus
+  text-format ``/metrics`` endpoint on ``localhost`` exporting a
+  completed-requests counter and an in-flight gauge.
+
+  Every ``poll_interval_secs`` a daemon thread reads that endpoint, feeds the
+  two metrics to a
+  capacity estimator, and writes the estimate, rounded
+  to an int, to the process-wide distribution the runner reads. If
+  ``config.load_signal_fn`` is set, it replaces the scrape and the estimate.
+
+  A poll is skipped, leaving the last published value in place, while the
+  server is down, when the scrape fails, or while the estimator has no
+  estimate yet. The publisher never injects a 0 of its own, on start, on stop
+  or on a skipped poll: a 0 tells the runner the worker has no capacity and can
+  trigger a spurious downscale. Only one publisher per metric name is active in
+  a process; starting a new one (e.g. after a model reload) stops the previous
+  one.
+
+  The metric tag is ``config.tag`` if set; else ``model_identifier``, used
+  verbatim because it must equal the identifier RunInference annotates the
+  step with (see ``ModelHandler.set_model_identifier``); else ``model_name``
+  with characters outside ``[A-Za-z0-9_]`` replaced by ``_``, for handlers
+  used outside RunInference.
+
+  Args:
+    metrics_port_fn: Returns the port serving ``/metrics`` while the server is
+      up, or ``None`` while it is not.
+    completed_metric: Name of the server's counter of completed requests, e.g.
+      ``vllm:request_success_total``.
+    running_metric: Name of the server's gauge of in-flight requests, e.g.
+      ``vllm:num_requests_running``.
+    model_identifier: The identifier RunInference reported through
+      ``ModelHandler.set_model_identifier``, or ``None`` outside RunInference.
+    model_name: Fallback for the metric tag outside RunInference.
+    config: Knobs; ``None`` uses the defaults.
+    label_filter: Optional labels a sample must have to be counted, e.g. to
+      select one model on a multi-model server.
+
+  Raises:
+    ValueError: if no metric tag can be derived.
+  """
+  # The publisher owning each metric name in this process, so a reloaded
+  # model's publisher can supersede the stale one.
+  _active: dict[str, 'PrometheusNumLoadedModelsPublisher'] = {}
+  _active_lock = threading.Lock()
+
+  def __init__(
+      self,
+      *,
+      metrics_port_fn: Callable[[], Optional[int]],
+      completed_metric: str,
+      running_metric: str,
+      model_identifier: Optional[str],
+      model_name: str = '',
+      config: Optional[AutoscalingMetricConfig] = None,
+      label_filter: Optional[Mapping[str, str]] = None,
+  ):
+    config = config or AutoscalingMetricConfig()
+    tag = config.tag or model_identifier or _sanitize_metric_tag(model_name)
+    if not tag:
+      raise ValueError(
+          'Cannot derive a num_loaded_models tag: pass a model_identifier, a '
+          'model_name, or set AutoscalingMetricConfig.tag.')
+    self._metric_name = NUM_LOADED_MODELS_METRIC_PREFIX + tag
+    self._metrics_port_fn = metrics_port_fn
+    self._completed_metric = completed_metric
+    self._running_metric = running_metric
+    self._label_filter = dict(label_filter) if label_filter else None
+    self._load_signal_fn = config.load_signal_fn
+    self._poll_interval_secs = config.poll_interval_secs
+    self._estimator = _CompletionRateCapacityEstimator(
+        multiplier=config.multiplier,
+        window_secs=config.signal_window_secs,
+        min_requests=config.min_requests,
+        warmup_timeout_secs=config.warmup_timeout_secs,
+        aggregation=config.aggregation)
+    self._stop_event = threading.Event()
+    self._thread: Optional[threading.Thread] = None
+
+  def start(self) -> None:
+    """Starts the background polling thread.
+
+    The metric stays absent until the first sample is published. Does nothing
+    if already started or once stopped, so a superseded publisher can't take
+    its metric back from the newer one; create a new publisher instead.
+    """
+    if self._thread is not None or self._stop_event.is_set():
+      return
+    with PrometheusNumLoadedModelsPublisher._active_lock:
+      previous = PrometheusNumLoadedModelsPublisher._active.get(
+          self._metric_name)
+      PrometheusNumLoadedModelsPublisher._active[self._metric_name] = self
+    if previous is not None and previous is not self:
+      logging.info(
+          'Replacing the active publisher for %s (model reload or server '
+          'restart).',
+          self._metric_name)
+      previous.stop()
+    self._thread = threading.Thread(
+        target=self._run, name='num-loaded-models-publisher', daemon=True)
+    self._thread.start()
+
+  def stop(self) -> None:
+    """Stops polling.
+
+    Publishes no final 0: when a worker is drained during a downscale, a 0
+    would shrink the pool's total capacity and could drive a feedback loop of
+    further downscales. The drained worker's value ages out instead.
+    """
+    self._stop_event.set()
+    with PrometheusNumLoadedModelsPublisher._active_lock:
+      if PrometheusNumLoadedModelsPublisher._active.get(
+          self._metric_name) is self:
+        del PrometheusNumLoadedModelsPublisher._active[self._metric_name]
+    thread = self._thread
+    if thread is not None:
+      thread.join(timeout=self._poll_interval_secs + 5)
+      self._thread = None
+
+  def _run(self) -> None:
+    while not self._stop_event.is_set():
+      try:
+        capacity = self._read_capacity()
+        if capacity is not None:
+          self._publish(capacity)
+      except Exception:  # pylint: disable=broad-except
+        logging.exception(
+            'Failed to publish autoscaling metric %s; will retry.',
+            self._metric_name)
+      self._stop_event.wait(self._poll_interval_secs)
+
+  def _read_capacity(self) -> Optional[float]:
+    """Reads the server once; returns its capacity, or None to skip."""
+    port = self._metrics_port_fn()
+    if port is None:
+      return None  # The server is not up.
+    if self._load_signal_fn is not None:
+      return self._load_signal_fn(port)
+    text = _scrape_prometheus_metrics(port)
+    if text is None:
+      return None
+    completed = _parse_prometheus_metric(
+        text, self._completed_metric, self._label_filter)
+    running = _parse_prometheus_metric(
+        text, self._running_metric, self._label_filter)
+    if completed is None or running is None:
+      return None
+    return self._estimator.update(completed, running)
+
+  def _publish(self, value: float) -> None:
+    published = int(round(value))
+    # The SDK harness reports process-wide distributions without resetting
+    # them, and Dataflow reads their cumulative mean (sum / count). Keeping the
+    # cell at (sum=published, count=1) makes the runner see the latest value
+    # rather than an average since worker start. The harness reads the cell
+    # from another thread, so replace the data in one step with reset_to():
+    # reset() then update() would briefly expose an empty cell (count=0).
+    cell = MetricsEnvironment.process_wide_container().get_distribution(
+        MetricName(MODEL_MANAGER_METRICS_NAMESPACE, self._metric_name))
+    current = cell.get_cumulative()
+    if current.count != 1 or current.sum != published:
+      cell.reset_to(published)
diff --git a/sdks/python/apache_beam/ml/inference/autoscaling_metrics_test.py b/sdks/python/apache_beam/ml/inference/autoscaling_metrics_test.py
new file mode 100644
index 000000000000..aec5dae9afda
--- /dev/null
+++ b/sdks/python/apache_beam/ml/inference/autoscaling_metrics_test.py
@@ -0,0 +1,502 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+import io
+import threading
+import unittest
+from unittest import mock
+
+from apache_beam.metrics.cells import DistributionData
+from apache_beam.ml.inference import autoscaling_metrics
+
+
+def _publisher(model_identifier='step-id', port=None, **kwargs):
+  """A publisher for a fake server exposing ``done_total`` and ``running``."""
+  return autoscaling_metrics.PrometheusNumLoadedModelsPublisher(
+      metrics_port_fn=lambda: port,
+      completed_metric='done_total',
+      running_metric='running',
+      model_identifier=model_identifier,
+      **kwargs)
+
+
+def _cell(model_identifier):
+  """The process-wide ``num_loaded_models_<id>`` cell."""
+  return autoscaling_metrics.MetricsEnvironment.process_wide_container(
+  ).get_distribution(
+      autoscaling_metrics.MetricName(
+          autoscaling_metrics.MODEL_MANAGER_METRICS_NAMESPACE,
+          f'num_loaded_models_{model_identifier}'))
+
+
+def _published(model_identifier):
+  return _cell(model_identifier).get_cumulative()
+
+
+class AutoscalingMetricsTest(unittest.TestCase):
+  def test_sanitize_metric_tag(self):
+    self.assertEqual(
+        autoscaling_metrics._sanitize_metric_tag('Qwen/Qwen2.5-7B:v1'),
+        'Qwen_Qwen2_5_7B_v1')
+    self.assertEqual(
+        autoscaling_metrics._sanitize_metric_tag('normal_tag_1'),
+        'normal_tag_1')
+
+  def test_parse_prometheus_metric(self):
+    prom_text = (
+        '# HELP vllm:num_requests_running Running requests\n'
+        '# TYPE vllm:num_requests_running gauge\n'
+        'vllm:num_requests_running{model_name="qwen"} 4.0\n'
+        'vllm:num_requests_running{model_name="other"} 6.0\n')
+    self.assertEqual(
+        autoscaling_metrics._parse_prometheus_metric(
+            prom_text,
+            'vllm:num_requests_running',
+            label_filter={'model_name': 'qwen'}),
+        4.0)
+    self.assertEqual(
+        autoscaling_metrics._parse_prometheus_metric(
+            prom_text, 'vllm:num_requests_running'),
+        10.0)
+    self.assertIsNone(
+        autoscaling_metrics._parse_prometheus_metric(prom_text, 'vllm:missing'))
+
+  def test_parse_prometheus_metric_exposition_details(self):
+    prom_text = (
+        # Series of one counter that differ only by label are summed.
+        'vllm:request_success_total{finished_reason="stop"} 10.0\n'
+        'vllm:request_success_total{finished_reason="length"} 5.0\n'
+        # A metric whose name merely starts with the requested one is ignored.
+        'vllm:request_success_total_created 1.7e9\n'
+        # Values that aren't numbers are skipped.
+        'vllm:request_success_total{finished_reason="abort"} n/a\n'
+        # Optional trailing timestamps are ignored.
+        'nv_inference_request_success{model="a"} 3 1700000000000\n'
+        # Quoted label values may contain commas, braces and escapes.
+        'labeled{path="a,b}c",quote="say \\"hi\\"",n="x\\ny"} 7\n'
+        '  \n'
+        'not a sample line\n')
+    parse = autoscaling_metrics._parse_prometheus_metric
+    self.assertEqual(parse(prom_text, 'vllm:request_success_total'), 15.0)
+    self.assertEqual(parse(prom_text, 'nv_inference_request_success'), 3.0)
+    self.assertEqual(
+        parse(prom_text, 'labeled', label_filter={'path': 'a,b}c'}), 7.0)
+    self.assertEqual(
+        parse(prom_text, 'labeled', label_filter={'quote': 'say "hi"'}), 7.0)
+    self.assertEqual(
+        parse(prom_text, 'labeled', label_filter={'n': 'x\ny'}), 7.0)
+    self.assertIsNone(
+        parse(prom_text, 'labeled', label_filter={'path': 'other'}))
+
+  def test_scrape_prometheus_metrics(self):
+    def fake_urlopen(url, timeout):
+      self.assertEqual(url, 'http://localhost:8000/metrics')
+      self.assertEqual(timeout, 5.0)
+      return io.BytesIO(b'running 4\n')
+
+    with mock.patch('urllib.request.urlopen', side_effect=fake_urlopen):
+      self.assertEqual(
+          autoscaling_metrics._scrape_prometheus_metrics(8000), 'running 4\n')
+    # Any failure (e.g. the server is restarting) -> None.
+    with mock.patch('urllib.request.urlopen', side_effect=OSError('refused')):
+      self.assertIsNone(autoscaling_metrics._scrape_prometheus_metrics(8000))
+
+  def test_capacity_estimator(self):
+    estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+        multiplier=2.0, min_value=1, window_secs=60.0, running_percentile=90.0)
+    # First sample establishes baseline -> None.
+    self.assertIsNone(estimator.update(10.0, 4.0, now=100.0))
+    # delta=20 over dt=5s with running=4 -> mu = 20 / (4 * 5) = 1.0,
+    # running_ref = 4.0, multiplier = 2.0 -> 8.0.
+    self.assertAlmostEqual(estimator.update(30.0, 4.0, now=105.0), 8.0)
+
+  def test_capacity_estimator_skips_counter_reset_and_clock_skew(self):
+    estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+        window_secs=60.0, min_requests=0)
+    self.assertIsNone(estimator.update(50.0, 2.0, now=100.0))
+    # Counter went down (server restart) -> skipped, new baseline.
+    self.assertIsNone(estimator.update(5.0, 2.0, now=105.0))
+    # Non-increasing timestamp -> skipped.
+    self.assertIsNone(estimator.update(9.0, 2.0, now=105.0))
+    # Negative gauge reading -> skipped.
+    self.assertIsNone(estimator.update(9.0, -1.0, now=107.0))
+    # delta=4 over 3s with running=2 -> mu=0.67, p90=2 -> 1.33.
+    self.assertAlmostEqual(estimator.update(13.0, 2.0, now=110.0), 4 / 3)
+
+  def test_capacity_estimator_running_percentile(self):
+    def run(running_percentile):
+      estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+          min_value=0, min_requests=0, running_percentile=running_percentile)
+      estimator.update(0.0, 1.0, now=0.0)
+      # Ten 1s intervals with 1..10 in flight, each completing as many
+      # requests: sum(completed) == sum(running * secs), so mu = 1.
+      for i in range(1, 11):
+        value = estimator.update(i * (i + 1) / 2, float(i), now=float(i))
+      return value
+
+    self.assertEqual(run(90.0), 9.0)  # Nearest rank: the 9th of 10.
+    self.assertEqual(run(50.0), 5.0)
+    self.assertEqual(run(100.0), 10.0)
+
+  def test_capacity_estimator_warmup_min_requests_and_timeout(self):
+    # Requires >= 3 completed requests before the 10s warmup timeout.
+    estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_requests=3, warmup_timeout_secs=10.0)
+    self.assertIsNone(estimator.update(0.0, 2.0, now=100.0))  # Baseline.
+    # 1 completion (< 3) at elapsed=5s (< 10s) -> None.
+    self.assertIsNone(estimator.update(1.0, 2.0, now=105.0))
+    # 4 completions (>= 3) at elapsed=8s -> publishes
+    # (4 / (2*5 + 2*3)) * 2 = 0.5 -> clamped to min_value=1.0.
+    self.assertAlmostEqual(estimator.update(4.0, 2.0, now=108.0), 1.0)
+
+    slow = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_requests=3, warmup_timeout_secs=10.0)
+    self.assertIsNone(slow.update(0.0, 1.0, now=100.0))  # Baseline.
+    # 1 completion (< 3) at elapsed=5s -> None.
+    self.assertIsNone(slow.update(1.0, 1.0, now=105.0))
+    # 2 completions (< 3) but elapsed=10s (>= timeout) -> publishes.
+    self.assertAlmostEqual(slow.update(2.0, 1.0, now=110.0), 1.0)
+
+  def test_capacity_estimator_skips_idle_ticks_without_diluting_p90(self):
+    estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+        multiplier=1.0,
+        min_value=1,
+        window_secs=60.0,
+        running_percentile=90.0,
+        min_requests=1)
+    self.assertIsNone(estimator.update(0.0, 5.0, now=100.0))  # Baseline.
+    # 25 completions over 5s with running=5 -> mu=1.0, p90=5 -> 5.0.
+    self.assertAlmostEqual(estimator.update(25.0, 5.0, now=105.0), 5.0)
+    # Three completely idle ticks (delta=0, running=0) -> None, without adding
+    # r=0 samples that would dilute p90 concurrency.
+    for t in (110.0, 115.0, 120.0):
+      self.assertIsNone(estimator.update(25.0, 0.0, now=t))
+    # 5 completions that already finished before the scrape (delta=5,
+    # running=0): total_completed=30, total_slot_secs=25 -> mu=1.2, p90=5 -> 6.
+    self.assertAlmostEqual(estimator.update(30.0, 0.0, now=125.0), 6.0)
+
+  def test_capacity_estimator_no_estimate_until_a_request_completes(self):
+    """In-flight requests with no completions never report the floor."""
+    # Slow first requests: still nothing completed after the warmup timeout.
+    slow = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_requests=3, warmup_timeout_secs=10.0)
+    self.assertIsNone(slow.update(0.0, 4.0, now=100.0))  # Baseline.
+    for t in (105.0, 110.0, 115.0):
+      self.assertIsNone(slow.update(0.0, 4.0, now=t))
+    # The first completions produce an estimate.
+    self.assertIsNotNone(slow.update(8.0, 4.0, now=120.0))
+
+    # First poll after an idle period longer than the window: the earlier
+    # samples have left the window and the new requests haven't finished yet.
+    for aggregation in ('concurrency', 'window_mean', 'window_max'):
+      resumed = autoscaling_metrics._CompletionRateCapacityEstimator(
+          min_value=1,
+          window_secs=10.0,
+          min_requests=0,
+          aggregation=aggregation)
+      resumed.update(0.0, 2.0, now=0.0)
+      self.assertGreater(resumed.update(20.0, 2.0, now=5.0), 1.0)
+      for t in (10.0, 15.0, 20.0):
+        self.assertIsNone(resumed.update(20.0, 0.0, now=t))  # Idle.
+      self.assertIsNone(resumed.update(20.0, 3.0, now=25.0), aggregation)
+
+    # lifetime_max keeps reporting its peak.
+    peak = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_value=1,
+        window_secs=10.0,
+        min_requests=0,
+        aggregation='lifetime_max')
+    peak.update(0.0, 2.0, now=0.0)
+    self.assertEqual(peak.update(20.0, 2.0, now=5.0), 4.0)
+    self.assertIsNone(peak.update(20.0, 0.0, now=20.0))  # Idle.
+    self.assertEqual(peak.update(20.0, 3.0, now=25.0), 4.0)
+
+  def test_capacity_estimator_defaults_to_monotonic_clock(self):
+    estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_requests=0)
+    with mock.patch.object(autoscaling_metrics.time,
+                           'monotonic',
+                           side_effect=[100.0, 105.0]):
+      self.assertIsNone(estimator.update(0.0, 1.0))
+      self.assertAlmostEqual(estimator.update(10.0, 1.0), 2.0)
+
+  def test_capacity_estimator_completion_rate_aggregations(self):
+    """The completion-rate modes ignore concurrency."""
+    def run(aggregation, **kwargs):
+      estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+          min_value=0,
+          min_requests=0,
+          window_secs=60.0,
+          aggregation=aggregation,
+          **kwargs)
+      # (completed_total, running, now): rates of 2, 6, idle, 1 req/s.
+      samples = [(0, 3, 0.0), (10, 3, 5.0), (40, 8, 10.0), (40, 0, 15.0),
+                 (45, 1, 20.0)]
+      return [estimator.update(c, r, now=t) for c, r, t in samples]
+
+    # 10 + 30 + 5 completions over the three non-idle 5s intervals.
+    self.assertEqual(run('window_mean'), [None, 2.0, 4.0, None, 3.0])
+    self.assertEqual(run('window_max'), [None, 2.0, 6.0, None, 6.0])
+    self.assertEqual(run('lifetime_max'), [None, 2.0, 6.0, None, 6.0])
+    self.assertEqual(
+        run('lifetime_max', multiplier=0.5), [None, 1.0, 3.0, None, 3.0])
+
+  def test_capacity_estimator_max_modes_differ_once_peak_leaves_window(self):
+    def run(aggregation, window_secs=10.0):
+      estimator = autoscaling_metrics._CompletionRateCapacityEstimator(
+          min_value=0,
+          min_requests=0,
+          window_secs=window_secs,
+          aggregation=aggregation)
+      # Rates: 6 req/s, then 1 req/s for three intervals.
+      samples = [(0, 1, 0.0), (30, 1, 5.0), (35, 1, 10.0), (40, 1, 15.0),
+                 (45, 1, 20.0)]
+      return [estimator.update(c, r, now=t) for c, r, t in samples][-1]
+
+    # The 6 req/s interval ended at t=5, outside the 10s window at t=20.
+    self.assertEqual(run('window_max'), 1.0)
+    self.assertEqual(run('lifetime_max'), 6.0)
+    # window_secs=0 keeps every interval.
+    self.assertEqual(run('window_max', window_secs=0), 6.0)
+
+  def test_capacity_estimator_completion_rate_works_without_in_flight(self):
+    """A server whose gauge always reads 0 still gets a completion rate."""
+    concurrency = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_requests=0)
+    rate = autoscaling_metrics._CompletionRateCapacityEstimator(
+        min_value=0, min_requests=0, aggregation='window_mean')
+    for estimator in (concurrency, rate):
+      estimator.update(0, 0, now=0.0)
+    self.assertIsNone(concurrency.update(500, 0, now=5.0))
+    self.assertEqual(rate.update(500, 0, now=5.0), 100.0)
+
+  def test_capacity_estimator_rejects_unknown_aggregation(self):
+    with self.assertRaises(ValueError):
+      autoscaling_metrics._CompletionRateCapacityEstimator(aggregation='p90')
+
+  def test_publisher_publishes_signal_values_as_is(self):
+    """Each capacity is published rounded; None or an error skips the tick."""
+    _cell('sink_test').reset()
+    self.addCleanup(lambda: _cell('sink_test').reset())
+    published = []
+    signal_values = [3.4, None, RuntimeError('scrape broke'), 7.6]
+
+    def load_signal_fn(port):
+      del port
+      if published:
+        # Record what the previous tick left in the cell.
+        published[-1] = _published('sink_test').sum
+      if not signal_values:
+        pub._stop_event.set()  # pylint: disable=protected-access
+        return None
+      published.append(None)
+      value = signal_values.pop(0)
+      if isinstance(value, Exception):
+        raise value
+      return value
+
+    pub = _publisher(
+        'sink_test',
+        port=8000,
+        config=autoscaling_metrics.AutoscalingMetricConfig(
+            poll_interval_secs=0, load_signal_fn=load_signal_fn))
+    # Run the polling loop on this thread so the test is deterministic.
+    with self.assertLogs(level='ERROR') as logs:
+      pub._run()  # pylint: disable=protected-access
+    self.assertIn('num_loaded_models_sink_test', logs.output[0])
+    # 3.4 -> 3; None and the error leave 3 in place; 7.6 -> 8.
+    self.assertEqual(published, [3, 3, 3, 8])
+    self.assertEqual(_published('sink_test').count, 1)
+
+  def test_publish_resets_process_wide_distribution_cell(self):
+    pub = _publisher('reset_test')
+    _cell('reset_test').reset()
+    self.addCleanup(lambda: _cell('reset_test').reset())
+    # A stale value, e.g. from before a model reload.
+    _cell('reset_test').update(7)
+    for _ in range(10):
+      pub._publish(2.0)
+    self.assertEqual(_published('reset_test'), DistributionData(2, 1, 2, 2))
+    pub._publish(5.0)
+    self.assertEqual(_published('reset_test'), DistributionData(5, 1, 5, 5))
+
+  def test_publish_never_exposes_an_empty_cell(self):
+    """A concurrent reader must never see count=0 between updates."""
+    pub = _publisher('atomic_test')
+    self.addCleanup(lambda: _cell('atomic_test').reset())
+    pub._publish(1.0)
+    stop = threading.Event()
+    counts = set()
+
+    def read():
+      while not stop.is_set():
+        counts.add(_published('atomic_test').count)
+
+    reader = threading.Thread(target=read)
+    reader.start()
+    try:
+      for i in range(20000):
+        pub._publish(float(i % 7 + 1))
+    finally:
+      stop.set()
+      reader.join()
+    self.assertEqual(counts, {1})
+
+  def test_publisher_for_same_tag_supersedes_previous(self):
+    """A reloaded model's publisher takes over its tag from the stale one."""
+    active = autoscaling_metrics.PrometheusNumLoadedModelsPublisher._active
+    fast = autoscaling_metrics.AutoscalingMetricConfig(poll_interval_secs=0.01)
+    first = _publisher('reload', config=fast)
+    second = _publisher('reload', config=fast)
+    other = _publisher('other', config=fast)
+    try:
+      first.start()
+      other.start()
+      self.assertTrue(first._thread.is_alive())
+      second.start()
+      # first is stopped and unregistered; second owns the tag; other is
+      # untouched.
+      self.assertTrue(first._stop_event.is_set())
+      self.assertIsNone(first._thread)
+      self.assertTrue(second._thread.is_alive())
+      self.assertTrue(other._thread.is_alive())
+      self.assertIs(active['num_loaded_models_reload'], second)
+      # Starting again is a no-op.
+      thread = second._thread
+      second.start()
+      self.assertIs(second._thread, thread)
+      # The stale one can neither take the tag back nor unregister the owner.
+      first.start()
+      self.assertIsNone(first._thread)
+      first.stop()
+      self.assertIs(active['num_loaded_models_reload'], second)
+      self.assertTrue(second._thread.is_alive())
+    finally:
+      first.stop()
+      second.stop()
+      other.stop()
+    self.assertNotIn('num_loaded_models_reload', active)
+
+  def test_publisher_metric_tag(self):
+    def tag(model_identifier, model_name='', **config):
+      publisher = _publisher(
+          model_identifier,
+          model_name=model_name,
+          config=autoscaling_metrics.AutoscalingMetricConfig(**config))
+      return publisher._metric_name
+
+    # The RunInference identifier is used verbatim (it has to match the step
+    # annotation), even if it contains characters sanitize would change.
+    self.assertEqual(
+        tag('my-step/id', 'org/model:v1'), 'num_loaded_models_my-step/id')
+    # Outside RunInference: the sanitized model name.
+    self.assertEqual(
+        tag(None, 'org/model:v1'), 'num_loaded_models_org_model_v1')
+    # An explicit tag always wins.
+    self.assertEqual(
+        tag('step-id', 'model', tag='custom'), 'num_loaded_models_custom')
+    with self.assertRaises(ValueError):
+      tag(None, '')
+
+  def test_publisher_config_and_defaults(self):
+    publisher = _publisher(
+        config=autoscaling_metrics.AutoscalingMetricConfig(
+            poll_interval_secs=7.0))
+    self.assertEqual(publisher._poll_interval_secs, 7.0)
+    self.assertEqual(
+        _publisher(
+            config=autoscaling_metrics.AutoscalingMetricConfig(
+                aggregation='lifetime_max'))._estimator._aggregation,
+        'lifetime_max')
+    # No config -> defaults.
+    self.assertEqual(
+        _publisher()._poll_interval_secs,
+        autoscaling_metrics.DEFAULT_POLL_INTERVAL_SECS)
+
+  def test_publisher_reads_server_capacity(self):
+    """The handler supplies metric names and a port; the rest is built in."""
+    port = [None]
+    publisher = autoscaling_metrics.PrometheusNumLoadedModelsPublisher(
+        metrics_port_fn=lambda: port[0],
+        completed_metric='done_total',
+        running_metric='running',
+        model_identifier='step-id',
+        config=autoscaling_metrics.AutoscalingMetricConfig(multiplier=2.0))
+    signal = publisher._read_capacity  # What the publisher's thread polls.
+
+    # Server not up yet -> skip the tick without scraping.
+    with mock.patch('urllib.request.urlopen') as urlopen:
+      self.assertIsNone(signal())
+      urlopen.assert_not_called()
+
+    port[0] = 8000
+    responses = [
+        'done_total 10\nrunning 4\n',
+        'done_total 30\nrunning 4\n',
+    ]
+
+    def fake_urlopen(url, timeout=5.0):
+      self.assertEqual(url, 'http://localhost:8000/metrics')
+      del timeout
+      return io.BytesIO(responses.pop(0).encode('utf-8'))
+
+    with mock.patch('urllib.request.urlopen', side_effect=fake_urlopen):
+      with mock.patch.object(autoscaling_metrics.time,
+                             'monotonic',
+                             side_effect=[100.0, 105.0]):
+        self.assertIsNone(signal())  # Baseline sample.
+        # delta=20 over 5s with running=4 -> mu=1, p90=4, multiplier 2 -> 8.
+        self.assertAlmostEqual(signal(), 8.0)
+
+  def test_publisher_skips_unreadable_scrapes(self):
+    publisher = _publisher(
+        port=8000,
+        config=autoscaling_metrics.AutoscalingMetricConfig(min_requests=0),
+        label_filter={'model': 'a'})
+    signal = publisher._read_capacity
+    scrapes = [
+        None,  # Endpoint unreachable.
+        'done_total{model="a"} 1\n',  # Gauge missing.
+        # Only model "a" counts.
+        'done_total{model="a"} 10\ndone_total{model="b"} 99\n'
+        'running{model="a"} 2\nrunning{model="b"} 50\n',
+        'done_total{model="a"} 20\ndone_total{model="b"} 999\n'
+        'running{model="a"} 2\nrunning{model="b"} 50\n',
+    ]
+    with mock.patch.object(autoscaling_metrics,
+                           '_scrape_prometheus_metrics',
+                           side_effect=lambda port: scrapes.pop(0)):
+      with mock.patch.object(autoscaling_metrics.time,
+                             'monotonic',
+                             side_effect=[100.0, 110.0]):
+        self.assertIsNone(signal())
+        self.assertIsNone(signal())
+        self.assertIsNone(signal())  # Baseline sample.
+        # delta=10 over 10s with running=2 -> mu=0.5, p90=2 -> 1.0.
+        self.assertAlmostEqual(signal(), 1.0)
+
+  def test_publisher_load_signal_fn_override(self):
+    publisher = _publisher(
+        port=8000,
+        config=autoscaling_metrics.AutoscalingMetricConfig(
+            load_signal_fn=lambda port: port / 1000.0))
+    with mock.patch('urllib.request.urlopen') as urlopen:
+      self.assertEqual(publisher._read_capacity(), 8.0)
+      urlopen.assert_not_called()
+
+
+if __name__ == '__main__':
+  unittest.main()