diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 370fb2b826..61a243d2ac 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -30,6 +30,8 @@ jobs:
       scripts-meta: ${{ steps.meta-scripts.outputs.json }}
       exec-repo: ${{ steps.ghcr-repo.outputs.exec }}
       exec-meta: ${{ steps.meta-exec.outputs.json }}
+      celery-repo: ${{ steps.ghcr-repo.outputs.celery }}
+      celery-meta: ${{ steps.meta-celery.outputs.json }}
 
     # https://docs.docker.com/build/ci/github-actions/local-registry/
     services:
@@ -74,6 +76,7 @@ jobs:
           webapp="ghcr.io/$repo/circ-webapp"
           scripts="ghcr.io/$repo/circ-scripts"
           exec="ghcr.io/$repo/circ-exec"
+          celery="ghcr.io/$repo/circ-celery"
           baseimage="ghcr.io/$repo/circ-baseimage"
           echo "webapp=$webapp"
           echo "webapp=$webapp" >> "$GITHUB_OUTPUT"
@@ -81,6 +84,8 @@ jobs:
           echo "scripts=$scripts" >> "$GITHUB_OUTPUT"
           echo "exec=$exec"
           echo "exec=$exec" >> "$GITHUB_OUTPUT"
+          echo "celery=$celery"
+          echo "celery=$celery" >> "$GITHUB_OUTPUT"
           echo "baseimage=$baseimage"
           echo "baseimage=$baseimage" >> "$GITHUB_OUTPUT"
 
@@ -222,6 +227,29 @@ jobs:
           labels: ${{ steps.meta-exec.outputs.labels }}
           outputs: type=image,"name=${{ steps.ghcr-repo.outputs.exec }}",push-by-digest=true,name-canonical=true,push=true
 
+      - name: Generate tags for circ-celery
+        id: meta-celery
+        uses: docker/metadata-action@v6
+        with:
+          images: ${{ steps.ghcr-repo.outputs.celery }}
+          tags: |
+            type=semver,pattern={{major}}.{{minor}},priority=10
+            type=semver,pattern={{version}},priority=20
+            type=ref,event=branch,priority=30
+            type=sha,priority=40
+
+      - name: Build circ-celery image
+        id: build-celery
+        uses: docker/build-push-action@v7
+        with:
+          context: .
+          file: ./docker/Dockerfile
+          target: celery
+          build-args: |
+            BASE_IMAGE=${{ steps.baseimage.outputs.tag }}
+          labels: ${{ steps.meta-celery.outputs.labels }}
+          outputs: type=image,"name=${{ steps.ghcr-repo.outputs.celery }}",push-by-digest=true,name-canonical=true,push=true
+
       - name: Export digests
         run: |
           mkdir -p ${{ runner.temp }}/digests/webapp
@@ -236,6 +264,10 @@ jobs:
           exec_digest="${{ steps.build-exec.outputs.digest }}"
           touch "${{ runner.temp }}/digests/exec/${exec_digest#sha256:}"
           echo "EXEC_DIGEST=$exec_digest"
+          mkdir -p ${{ runner.temp }}/digests/celery
+          celery_digest="${{ steps.build-celery.outputs.digest }}"
+          touch "${{ runner.temp }}/digests/celery/${celery_digest#sha256:}"
+          echo "CELERY_DIGEST=$celery_digest"
 
       - name: Upload digests
         uses: actions/upload-artifact@v7
@@ -292,6 +324,15 @@ jobs:
           echo "$IMAGE"
           echo "WEBAPP_IMAGE=$IMAGE" >> $GITHUB_ENV
 
+      # This sets the environment variable referenced in the docker-compose file
+      # for the celery containers to the digest of the image built in the build job.
+      - name: Set celery image
+        working-directory: ${{ runner.temp }}/digests/celery
+        run: |
+          IMAGE="${{needs.build.outputs.celery-repo}}$(printf '@sha256:%s' *)"
+          echo "$IMAGE"
+          echo "CELERY_IMAGE=$IMAGE" >> $GITHUB_ENV
+
       # See comment here: https://github.com/actions/runner-images/issues/1187#issuecomment-686735760
       - name: Disable network offload
         run: sudo ethtool -K eth0 tx off rx off
@@ -308,6 +349,9 @@ jobs:
       - name: Run scripts image tests
         run: ./docker/ci/test_scripts.sh scripts
 
+      - name: Run celery image tests
+        run: ./docker/ci/test_celery.sh
+
       - name: Output logs
         if: failure()
         run: docker compose logs
@@ -516,3 +560,10 @@ jobs:
           docker buildx imagetools create
           $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< '${{ needs.build.outputs.exec-meta }}')
           $(printf '${{ needs.build.outputs.exec-repo }}@sha256:%s ' *)
+
+      - name: Create manifest & push circ-celery
+        working-directory: ${{ runner.temp }}/digests/celery
+        run: >
+          docker buildx imagetools create
+          $(jq -cr '.tags | map("-t " + .) | join(" ")' <<< '${{ needs.build.outputs.celery-meta }}')
+          $(printf '${{ needs.build.outputs.celery-repo }}@sha256:%s ' *)
diff --git a/README.md b/README.md
index fbb0499286..a89b9ecfa4 100644
--- a/README.md
+++ b/README.md
@@ -33,6 +33,7 @@ Docker images created from this code are available at:
 - [circ-webapp](https://github.com/ThePalaceProject/circulation/pkgs/container/circ-webapp)
 - [circ-scripts](https://github.com/ThePalaceProject/circulation/pkgs/container/circ-scripts)
 - [circ-exec](https://github.com/ThePalaceProject/circulation/pkgs/container/circ-exec)
+- [circ-celery](https://github.com/ThePalaceProject/circulation/pkgs/container/circ-celery)
 
 Docker images are the preferred way to deploy this code in a production environment.
 
@@ -247,6 +248,28 @@ We support overriding a number of other Celery settings via environment variable
 the defaults should be sufficient. The full list of settings can be found in
 [`service/celery/configuration.py`](src/palace/manager/service/celery/configuration.py).
 
+##### `circ-celery` image
+
+The `circ-celery` image runs a single Celery process per container, so it can be deployed as the beat
+scheduler or one or more autoscaled worker pools — all from the same image. The role is chosen by the
+container command (`beat` or `worker`), and a few launch-time variables (read by the container's
+entrypoint, not by the application) configure the `worker` role:
+
+- `PALACE_CELERY_QUEUES`: Comma-separated list of queues a `worker` container should consume, e.g.
+    `high,default` (**required for the `worker` role**). Each queue's tasks are defined by
+    [`QueueNames`](src/palace/manager/service/celery/celery.py).
+- `PALACE_CELERY_CONCURRENCY`: The number of child processes for a `worker` container. This is a single
+    pool shared across all of the queues that container consumes; to give a queue its own concurrency,
+    run a separate `worker` deployment with its own `PALACE_CELERY_QUEUES`/`PALACE_CELERY_CONCURRENCY`.
+    The default is `1` (optional).
+- `PALACE_CELERY_WORKER_HOSTNAME`: The Celery `--hostname` for a `worker` container. The default is
+    `worker@%h` (optional).
+
+The `beat` role must run as a single instance (a second beat would double-fire scheduled tasks); only
+the `worker` role should be scaled. The queue-depth metrics that drive worker autoscaling
+(`QueueWaiting` / `QueueOldestAge`, in the `Celery` CloudWatch namespace) are published every minute by
+the `publish_queue_stats` task on the beat schedule, so no always-on metrics process is required.
+
 #### Redis
 
 We use Redis as the caching layer for the application. Although you can use the same Redis database for both
diff --git a/docker-compose.yml b/docker-compose.yml
index 66ccda3e47..fb88fb0062 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -1,34 +1,39 @@
 # Common CM setup
 # see: https://github.com/compose-spec/compose-spec/blob/master/spec.md#extension
+
+# The shared environment is its own anchor so a service (e.g. the celery worker)
+# can merge it and add a few of its own variables without repeating the block.
+x-cm-environment: &cm-environment
+  SIMPLIFIED_PRODUCTION_DATABASE: "postgresql://palace:test@pg:5432/circ"
+  PALACE_SEARCH_URL: "http://os:9200"
+  PALACE_STORAGE_ACCESS_KEY: "palace"
+  PALACE_STORAGE_SECRET_KEY: "test123456789"
+  PALACE_STORAGE_ENDPOINT_URL: "http://minio:9000"
+  PALACE_STORAGE_PUBLIC_ACCESS_BUCKET: "public"
+  PALACE_STORAGE_ANALYTICS_BUCKET: "analytics"
+  PALACE_STORAGE_URL_TEMPLATE: "http://localhost:9000/{bucket}/{key}"
+  PALACE_REPORTING_NAME: "TEST CM"
+  PALACE_SECRET_KEY: "SECRET_KEY_USED_FOR_ADMIN_UI_COOKIES"
+  PALACE_PATRON_WEB_HOSTNAMES: "*"
+  PALACE_BASE_URL: "http://localhost:6500"
+  PALACE_CELERY_BROKER_URL: "redis://valkey:6379/0"
+  PALACE_CELERY_RESULT_BACKEND: "redis://valkey:6379/2"
+  PALACE_CELERY_BROKER_TRANSPORT_OPTIONS_GLOBAL_KEYPREFIX: "test"
+  PALACE_CELERY_CLOUDWATCH_STATISTICS_DRYRUN: "true"
+  PALACE_REDIS_URL: "redis://valkey:6379/1"
+  PALACE_GOOGLE_DRIVE_SERVICE_ACCOUNT_INFO_JSON: "${PALACE_GOOGLE_DRIVE_SERVICE_ACCOUNT_INFO_JSON-}"
+
+  # Set up the environment variables used for testing as well
+  PALACE_TEST_DATABASE_URL: "postgresql://palace:test@pg:5432/circ"
+  PALACE_TEST_SEARCH_URL: "http://os:9200"
+  PALACE_TEST_MINIO_URL: "http://minio:9000"
+  PALACE_TEST_MINIO_USER: "palace"
+  PALACE_TEST_MINIO_PASSWORD: "test123456789"
+  PALACE_TEST_REDIS_URL: "redis://valkey:6379/3"
+
 x-cm-variables: &cm
   platform: "${BUILD_PLATFORM-}"
-  environment:
-    SIMPLIFIED_PRODUCTION_DATABASE: "postgresql://palace:test@pg:5432/circ"
-    PALACE_SEARCH_URL: "http://os:9200"
-    PALACE_STORAGE_ACCESS_KEY: "palace"
-    PALACE_STORAGE_SECRET_KEY: "test123456789"
-    PALACE_STORAGE_ENDPOINT_URL: "http://minio:9000"
-    PALACE_STORAGE_PUBLIC_ACCESS_BUCKET: "public"
-    PALACE_STORAGE_ANALYTICS_BUCKET: "analytics"
-    PALACE_STORAGE_URL_TEMPLATE: "http://localhost:9000/{bucket}/{key}"
-    PALACE_REPORTING_NAME: "TEST CM"
-    PALACE_SECRET_KEY: "SECRET_KEY_USED_FOR_ADMIN_UI_COOKIES"
-    PALACE_PATRON_WEB_HOSTNAMES: "*"
-    PALACE_BASE_URL: "http://localhost:6500"
-    PALACE_CELERY_BROKER_URL: "redis://valkey:6379/0"
-    PALACE_CELERY_RESULT_BACKEND: "redis://valkey:6379/2"
-    PALACE_CELERY_BROKER_TRANSPORT_OPTIONS_GLOBAL_KEYPREFIX: "test"
-    PALACE_CELERY_CLOUDWATCH_STATISTICS_DRYRUN: "true"
-    PALACE_REDIS_URL: "redis://valkey:6379/1"
-    PALACE_GOOGLE_DRIVE_SERVICE_ACCOUNT_INFO_JSON: "${PALACE_GOOGLE_DRIVE_SERVICE_ACCOUNT_INFO_JSON-}"
-
-    # Set up the environment variables used for testing as well
-    PALACE_TEST_DATABASE_URL: "postgresql://palace:test@pg:5432/circ"
-    PALACE_TEST_SEARCH_URL: "http://os:9200"
-    PALACE_TEST_MINIO_URL: "http://minio:9000"
-    PALACE_TEST_MINIO_USER: "palace"
-    PALACE_TEST_MINIO_PASSWORD: "test123456789"
-    PALACE_TEST_REDIS_URL: "redis://valkey:6379/3"
+  environment: *cm-environment
 
   depends_on:
     pg:
@@ -67,6 +72,34 @@ services:
       target: scripts
     image: "${SCRIPTS_IMAGE-}"
 
+  # The celery-* services all run from the single circ-celery image; the role is
+  # the container command. In production these become the beat deployment (one
+  # replica) and the autoscaled worker deployment(s). Queue-depth metrics are
+  # published by the beat-scheduled publish_queue_stats task, so there is no
+  # separate metrics container.
+  celery-beat:
+    <<: *cm
+    build:
+      <<: *cm-build
+      target: celery
+    image: "${CELERY_IMAGE-}"
+    command: ["beat"]
+
+  celery-worker:
+    <<: *cm
+    build:
+      <<: *cm-build
+      target: celery
+    image: "${CELERY_IMAGE-}"
+    command: ["worker"]
+    # A single dev/test worker consumes every queue. In production each queue
+    # (or group) is its own deployment with its own queues/concurrency so it can
+    # be autoscaled independently.
+    environment:
+      <<: *cm-environment
+      PALACE_CELERY_QUEUES: "high,default,apply"
+      PALACE_CELERY_CONCURRENCY: "2"
+
   pg:
     image: "postgres:16"
     environment:
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 6b18e69dec..603262a321 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -63,6 +63,52 @@ WORKDIR /home/palace/circulation/bin
 
 CMD ["/sbin/my_init"]
 
+###############################################################################
+## Circ-celery Image
+##
+## A single image that runs one Celery process per container. The role is
+## selected by the container command (beat|worker), so one image backs both the
+## beat scheduler and every autoscaled worker pool. Unlike the scripts image, it
+## does NOT use runit to supervise several processes -- one process per container
+## is what lets the worker pools be scaled horizontally, and the beat scheduler
+## be pinned to a single instance.
+##
+## Like the webapp/scripts images it boots via my_init, so the shared
+## /etc/my_init.d startup scripts run first -- notably the advisory-locked
+## `initialize_instance` (alembic upgrade head) -- before the Celery role starts.
+## This keeps the migration path identical to the other images: every container
+## brings the schema to head on start, so no separate migration step is needed.
+## `--skip-runit` is used because this image has no runit services; the single
+## Celery process is my_init's main command instead.
+###############################################################################
+
+FROM common AS celery
+
+# Set the local timezone so log timestamps line up with the other images. Beat's
+# own schedule is driven by the Celery `timezone` setting, not this.
+ENV TZ=US/Eastern
+RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime
+
+# The Celery process runs as the `palace` user via `celery --uid`, which does not
+# reset HOME. Without this it would inherit root's HOME=/root, and libpq
+# (psycopg2) -- probing $HOME/.postgresql/postgresql.crt on connect -- would hit
+# EACCES on /root (mode 700), abort the SSL handshake, and fall back to a
+# plaintext connection that an SSL-required Postgres rejects. my_init does not
+# override an inherited HOME, so setting it here covers the --uid drop.
+ENV HOME=/home/palace
+
+COPY --chmod=755 docker/celery-entrypoint.sh /celery-entrypoint.sh
+
+VOLUME /var/log
+WORKDIR /var/www/circulation
+
+# The role is provided as the container command, e.g. `docker run circ-celery
+# worker`, with PALACE_CELERY_QUEUES / PALACE_CELERY_CONCURRENCY set for worker
+# pools. my_init runs the startup scripts (incl. the DB migration), then execs
+# the entrypoint with the role appended. See docker/celery-entrypoint.sh for the
+# supported roles and env vars.
+ENTRYPOINT ["/sbin/my_init", "--skip-runit", "--quiet", "--", "/celery-entrypoint.sh"]
+
 ###############################################################################
 ## Circ-webapp Image
 ###############################################################################
diff --git a/docker/README.md b/docker/README.md
index 39a40be577..871e506d71 100644
--- a/docker/README.md
+++ b/docker/README.md
@@ -77,6 +77,42 @@ $ docker run --name search_index_refresh -it \
     ghcr.io/thepalaceproject/circ-exec:main
 ```
 
+### circ-celery
+
+This image runs a single Celery process per container. Unlike `circ-scripts`, which supervises the
+beat scheduler and every worker in one container with runit, `circ-celery` runs exactly one process,
+so each role can be deployed and (for workers) autoscaled independently. The role is chosen by the
+container command: `beat` or `worker`. Queue-depth metrics that drive autoscaling are published by the
+`publish_queue_stats` task on the beat schedule, so there is no separate always-on metrics container.
+
+Like the other images, `circ-celery` boots via `my_init` and so initializes or migrates the database
+on startup (advisory-locked `alembic upgrade head`) before the Celery process starts. No separate
+migration step is required.
+
+```sh
+# The beat scheduler (run exactly one of these).
+$ docker run --name celery-beat -d \
+    -e PALACE_CELERY_BROKER_URL='redis://[host]:6379/0' \
+    -e PALACE_CELERY_RESULT_BACKEND='redis://[host]:6379/2' \
+    -e SIMPLIFIED_PRODUCTION_DATABASE='postgresql://[username]:[password]@[host]:[port]/[database_name]' \
+    ghcr.io/thepalaceproject/circ-celery:main beat
+
+# A worker pool (scale the replica count on queue depth). PALACE_CELERY_QUEUES is required.
+$ docker run --name celery-worker -d \
+    -e PALACE_CELERY_BROKER_URL='redis://[host]:6379/0' \
+    -e PALACE_CELERY_RESULT_BACKEND='redis://[host]:6379/2' \
+    -e SIMPLIFIED_PRODUCTION_DATABASE='postgresql://[username]:[password]@[host]:[port]/[database_name]' \
+    -e PALACE_CELERY_QUEUES='high,default' \
+    -e PALACE_CELERY_CONCURRENCY='8' \
+    ghcr.io/thepalaceproject/circ-celery:main worker
+```
+
+The `worker` launch variables (`PALACE_CELERY_QUEUES`, `PALACE_CELERY_CONCURRENCY`,
+`PALACE_CELERY_WORKER_HOSTNAME`) and the `cloudwatch` flush interval
+(`PALACE_CELERY_CLOUDWATCH_FLUSH_INTERVAL`) are documented in the top-level
+[`README.md`](../README.md). Logs are written to stdout/stderr rather than to files, so nothing is
+lost when an autoscaled worker is scaled away.
+
 ## Environment Variables
 
 Environment variables can be set with the `-e VARIABLE_KEY='variable_value'` option on the `docker run` command.
diff --git a/docker/celery-entrypoint.sh b/docker/celery-entrypoint.sh
new file mode 100755
index 0000000000..a0bc182d94
--- /dev/null
+++ b/docker/celery-entrypoint.sh
@@ -0,0 +1,77 @@
+#!/bin/bash
+#
+# Entrypoint for the circ-celery image.
+#
+# my_init runs this script as its main command (after the shared
+# /etc/my_init.d startup scripts, which apply the DB migration), so by the time
+# we exec Celery the schema is already at head -- the same path the webapp and
+# scripts images use.
+#
+# A single image backs every Celery process we run; the role is chosen by the
+# first argument, so the same image can be deployed as the beat scheduler or any
+# number of autoscaled worker pools:
+#
+#   celery-entrypoint.sh beat
+#   celery-entrypoint.sh worker        (PALACE_CELERY_QUEUES required)
+#
+# Exactly one Celery process runs per container (no runit supervision); that is
+# what lets the worker pools be scaled horizontally, one replica per unit of
+# queue depth. Logs are written to stdout/stderr (as JSON, via the application's
+# logging configuration) rather than to files, so nothing is lost when an
+# autoscaled worker is scaled away.
+#
+# Queue-depth metrics (which drive worker autoscaling) are published by the
+# periodic publish_queue_stats task on the beat schedule, so there is no separate
+# always-on metrics process/role here.
+
+set -euo pipefail
+
+APP="palace.manager.celery.app"
+CELERY="/var/www/circulation/env/bin/celery"
+
+cd /var/www/circulation
+
+role="${1:-}"
+if [[ -z "$role" ]]; then
+  echo "Usage: $(basename "$0") <beat|worker> [extra celery args]" >&2
+  exit 64
+fi
+shift
+
+case "$role" in
+  beat)
+    # Beat MUST run as a singleton -- a second replica would double-fire every
+    # scheduled task. Never autoscale this role; pin it to a single instance.
+    schedule_dir="/var/run/celery"
+    mkdir -p "$schedule_dir"
+    chown palace:palace "$schedule_dir"
+    exec "$CELERY" -A "$APP" beat \
+      --uid palace --gid palace \
+      --schedule "$schedule_dir/beat-schedule" \
+      "$@"
+    ;;
+  worker)
+    # The queue set and concurrency are supplied per-deployment so one image can
+    # back every worker pool. Concurrency is a single pool of child processes
+    # shared across ALL queues this worker consumes -- to give a queue its own
+    # concurrency, run a separate deployment with its own PALACE_CELERY_QUEUES
+    # and PALACE_CELERY_CONCURRENCY.
+    queues="${PALACE_CELERY_QUEUES:-}"
+    if [[ -z "$queues" ]]; then
+      echo "PALACE_CELERY_QUEUES is required for the worker role (comma-separated queue names)." >&2
+      exit 64
+    fi
+    concurrency="${PALACE_CELERY_CONCURRENCY:-1}"
+    hostname="${PALACE_CELERY_WORKER_HOSTNAME:-worker@%h}"
+    exec "$CELERY" -A "$APP" worker \
+      --uid palace --gid palace \
+      --queues "$queues" \
+      --concurrency "$concurrency" \
+      --hostname "$hostname" \
+      "$@"
+    ;;
+  *)
+    echo "Unknown role '$role' (expected: beat or worker)." >&2
+    exit 64
+    ;;
+esac
diff --git a/docker/ci/test_celery.sh b/docker/ci/test_celery.sh
new file mode 100755
index 0000000000..d791f8c0bc
--- /dev/null
+++ b/docker/ci/test_celery.sh
@@ -0,0 +1,50 @@
+#!/bin/bash
+
+# Smoke test for the circ-celery image. Unlike the scripts image, circ-celery
+# runs a single Celery process per container (no runit), so we can't use the
+# `sv check` helpers. Instead we confirm the worker answers a ping over the
+# broker and that the beat container comes up and stays running.
+
+set -ex
+
+CELERY="/var/www/circulation/env/bin/celery"
+APP="palace.manager.celery.app"
+
+# Assert a compose service's container is running (has not exited).
+function assert_running() {
+  service="$1"
+  cid=$(docker compose ps -q "$service")
+  if [[ -z "$cid" ]]; then
+    echo "  FAIL: $service has no container"
+    exit 1
+  fi
+  running=$(docker inspect -f '{{.State.Running}}' "$cid")
+  if [[ "$running" != "true" ]]; then
+    echo "  FAIL: $service is not running (State.Running=$running)"
+    docker compose logs "$service"
+    exit 1
+  fi
+  echo "  OK: $service is running"
+}
+
+# Output the version file for debugging.
+docker compose exec -T celery-worker cat /var/www/circulation/src/palace/manager/_version.py
+
+# Wait for the worker to come up and answer a ping over the broker. `inspect
+# ping` only returns successfully once a worker has connected and is ready, so
+# this exercises the whole path: image -> entrypoint -> celery -> broker.
+timeout 120s bash -c "
+  until docker compose exec -T celery-worker $CELERY -A $APP inspect ping --timeout 5; do
+    sleep 5
+  done
+"
+
+# Beat has no ping; confirm it started and is still running (i.e. the entrypoint
+# launched the right process and it did not immediately exit).
+assert_running celery-beat
+assert_running celery-worker
+
+# Confirm beat has actually started its scheduler loop.
+timeout 60s grep -q -e 'beat: Starting' -e 'Scheduler:' <(docker compose logs celery-beat -f 2>&1)
+
+exit 0
diff --git a/src/palace/manager/celery/monitoring.py b/src/palace/manager/celery/monitoring.py
index d1d751a6de..89b082bc9e 100644
--- a/src/palace/manager/celery/monitoring.py
+++ b/src/palace/manager/celery/monitoring.py
@@ -10,6 +10,7 @@
 import boto3
 from boto3.exceptions import Boto3Error
 from botocore.exceptions import BotoCoreError
+from celery import Celery
 from celery.events.snapshot import Polaroid
 from celery.events.state import State
 from kombu.transport.redis import PrefixedStrictRedis
@@ -96,37 +97,36 @@ class _PrefixedRedis(PrefixedStrictRedis):
     ]
 
 
-class Cloudwatch(Polaroid):
+class QueueStatsReporter:
     """
-    Implements a Celery custom camera that sends queue statistics to Cloudwatch.
-
-    See Celery documentation for more information on custom cameras:
-    https://docs.celeryq.dev/en/stable/userguide/monitoring.html#custom-camera
+    Collects per-queue depth and oldest-message age from the broker Redis and
+    publishes them to Cloudwatch as the ``QueueWaiting`` / ``QueueOldestAge``
+    metrics.
+
+    This is independent of Celery's event machinery, so it can be driven either by
+    the periodic ``publish_queue_stats`` task (see
+    :mod:`palace.manager.celery.tasks.monitoring`) or by the :class:`Cloudwatch`
+    events camera below. Both paths read the same config off the Celery app, so
+    they produce identical metrics (same namespace, ``Manager``/``QueueName``
+    dimensions).
     """
 
-    clear_after = True  # clear after flush (incl, state.event_count).
-
-    def __init__(
-        self,
-        *args: Any,
-        **kwargs: Any,
-    ):
-        super().__init__(*args, **kwargs)
+    def __init__(self, app: Celery):
         # We use logger_for_cls instead of just inheriting from LoggerMixin
-        # because the base class Polaroid already defines a logger attribute,
-        # which conflicts with the logger() method in LoggerMixin.
+        # because the Cloudwatch subclass's Polaroid base already defines a logger
+        # attribute, which conflicts with the logger() method in LoggerMixin.
         self.logger = logger_for_cls(self.__class__)
-        broker_url = self.app.conf.get("broker_url")
+        broker_url = app.conf.get("broker_url")
         broker_type = urlparse(broker_url).scheme if broker_url else None
         if broker_type != "redis":
             raise PalaceValueError(f"Broker type '{broker_type}' is not supported.")
 
-        region = self.app.conf.get("cloudwatch_statistics_region")
-        dryrun = self.app.conf.get("cloudwatch_statistics_dryrun")
+        region = app.conf.get("cloudwatch_statistics_region")
+        dryrun = app.conf.get("cloudwatch_statistics_dryrun")
         self.cloudwatch_client = (
             boto3.client("cloudwatch", region_name=region) if not dryrun else None
         )
-        broker_transport_options = self.app.conf.get("broker_transport_options", {})
+        broker_transport_options = app.conf.get("broker_transport_options", {})
         self.manager_name = broker_transport_options.get("global_keyprefix")
         self.redis_client = self.get_redis_client(
             broker_url,
@@ -135,9 +135,9 @@ def __init__(
             # value keeps health checks on rather than silently disabling them.
             broker_transport_options.get("health_check_interval", 30),
         )
-        self.namespace = self.app.conf.get("cloudwatch_statistics_namespace")
-        self.upload_size = self.app.conf.get("cloudwatch_statistics_upload_size")
-        self.queues = {queue.name for queue in self.app.conf.get("task_queues")}
+        self.namespace = app.conf.get("cloudwatch_statistics_namespace")
+        self.upload_size = app.conf.get("cloudwatch_statistics_upload_size")
+        self.queues = {queue.name for queue in app.conf.get("task_queues")}
 
     @classmethod
     def get_redis_client(
@@ -193,9 +193,9 @@ def get_queue_stats(self) -> dict[str, QueueStats]:
             for queue in self.queues
         }
 
-    def on_shutter(self, state: State) -> None:
-        timestamp = utc_now()
-        self.publish(self.get_queue_stats(), timestamp)
+    def run(self) -> None:
+        """Snapshot every queue and publish its metrics to Cloudwatch once."""
+        self.publish(self.get_queue_stats(), utc_now())
 
     def publish(
         self,
@@ -224,3 +224,26 @@ def publish(
                 self.logger.info("Dry run enabled. Not sending metrics to Cloudwatch.")
                 for data in chunk:
                     self.logger.info(f"Data: {data}")
+
+
+class Cloudwatch(QueueStatsReporter, Polaroid):
+    """
+    A Celery custom camera that publishes queue statistics to Cloudwatch on the
+    events-snapshot interval.
+
+    This is retained for the ``celery events`` deployment used by the circ-scripts
+    container. New deployments publish the same metrics from the periodic
+    ``publish_queue_stats`` task instead, which does not require an always-on
+    events process. See Celery's custom-camera documentation:
+    https://docs.celeryq.dev/en/stable/userguide/monitoring.html#custom-camera
+    """
+
+    clear_after = True  # clear after flush (incl, state.event_count).
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        # Polaroid.__init__ sets self.app; QueueStatsReporter reads its config.
+        Polaroid.__init__(self, *args, **kwargs)
+        QueueStatsReporter.__init__(self, self.app)
+
+    def on_shutter(self, state: State) -> None:
+        self.run()
diff --git a/src/palace/manager/celery/tasks/monitoring.py b/src/palace/manager/celery/tasks/monitoring.py
new file mode 100644
index 0000000000..719339f673
--- /dev/null
+++ b/src/palace/manager/celery/tasks/monitoring.py
@@ -0,0 +1,33 @@
+from celery import shared_task
+
+from palace.manager.celery.monitoring import QueueStatsReporter
+from palace.manager.celery.task import Task
+from palace.manager.service.celery.celery import QueueNames
+
+# The reporter opens a Redis connection pool and a boto3 Cloudwatch client, so we
+# build it once per worker process and reuse it across invocations (the task runs
+# every minute). This mirrors how Task caches its session maker.
+_reporter: QueueStatsReporter | None = None
+
+
+def _get_reporter(task: Task) -> QueueStatsReporter:
+    global _reporter
+    if _reporter is None:
+        _reporter = QueueStatsReporter(task.app)
+    return _reporter
+
+
+@shared_task(queue=QueueNames.high, bind=True)
+def publish_queue_stats(task: Task) -> None:
+    """Publish per-queue depth and oldest-message age to Cloudwatch.
+
+    Scheduled by beat every minute onto the ``high`` queue. ``high`` is the
+    always-staffed, time-sensitive queue, so the metric keeps flowing even while
+    the ``default`` / ``apply`` pools are saturated and being scaled -- which are
+    exactly the situations the autoscaler needs the metric for.
+
+    This replaces the always-on ``celery events`` Cloudwatch camera: the metric is
+    just a periodic Redis read plus a ``put_metric_data`` call, so it does not need
+    a dedicated long-running process.
+    """
+    _get_reporter(task).run()
diff --git a/src/palace/manager/service/celery/celery.py b/src/palace/manager/service/celery/celery.py
index 297d0efe22..50d56d1c1e 100644
--- a/src/palace/manager/service/celery/celery.py
+++ b/src/palace/manager/service/celery/celery.py
@@ -61,6 +61,7 @@ def beat_schedule() -> dict[str, Any]:
         equivalents,
         license_expiration,
         marc,
+        monitoring,
         notifications,
         novelist,
         nyt,
@@ -93,6 +94,10 @@ def beat_schedule() -> dict[str, Any]:
             "task": search.search_indexing.name,
             "schedule": crontab(minute="*"),  # Run every minute
         },
+        "publish_queue_stats": {
+            "task": monitoring.publish_queue_stats.name,
+            "schedule": crontab(minute="*"),  # Run every minute
+        },
         "marc_export": {
             "task": marc.marc_export.name,
             "schedule": crontab(
diff --git a/tests/manager/celery/tasks/test_monitoring.py b/tests/manager/celery/tasks/test_monitoring.py
new file mode 100644
index 0000000000..c1a1ac5efd
--- /dev/null
+++ b/tests/manager/celery/tasks/test_monitoring.py
@@ -0,0 +1,24 @@
+from unittest.mock import patch
+
+import pytest
+
+from palace.manager.celery.tasks import monitoring
+from palace.manager.celery.tasks.monitoring import publish_queue_stats
+from tests.fixtures.celery import CeleryFixture
+
+
+class TestPublishQueueStats:
+    def test_publishes_via_reporter(
+        self,
+        celery_fixture: CeleryFixture,
+        monkeypatch: pytest.MonkeyPatch,
+    ):
+        # The task builds a QueueStatsReporter from its app and runs it. We mock
+        # the reporter because the test app uses a memory:// broker, which the
+        # reporter (Redis-only) would reject.
+        monkeypatch.setattr(monitoring, "_reporter", None)
+        with patch.object(monitoring, "QueueStatsReporter") as mock_reporter_cls:
+            publish_queue_stats.delay().wait()
+
+        mock_reporter_cls.assert_called_once()
+        mock_reporter_cls.return_value.run.assert_called_once_with()
diff --git a/tests/manager/celery/test_monitoring.py b/tests/manager/celery/test_monitoring.py
index 5c2f02ad8f..1859fb58f5 100644
--- a/tests/manager/celery/test_monitoring.py
+++ b/tests/manager/celery/test_monitoring.py
@@ -9,7 +9,7 @@
 from palace.util.log import LogLevel
 
 from palace.manager.celery.celery import Celery
-from palace.manager.celery.monitoring import Cloudwatch, QueueStats
+from palace.manager.celery.monitoring import Cloudwatch, QueueStats, QueueStatsReporter
 
 
 class CloudwatchCameraFixture:
@@ -24,6 +24,11 @@ def create_cloudwatch(self):
             self._mock_get_redis = mock_get_redis
             return Cloudwatch(state=MagicMock(), app=self.app)
 
+    def create_reporter(self):
+        with patch.object(QueueStatsReporter, "get_redis_client") as mock_get_redis:
+            self._mock_get_redis = mock_get_redis
+            return QueueStatsReporter(self.app)
+
     @property
     def mock_get_redis(self):
         if self._mock_get_redis is None:
@@ -339,3 +344,35 @@ def test_publish(
         cloudwatch.publish(queues, timestamp)
         mock_put_metric_data.assert_not_called()
         assert "Dry run enabled. Not sending metrics to Cloudwatch." in caplog.text
+
+
+class TestQueueStatsReporter:
+    def test_logger_name(self, cloudwatch_camera: CloudwatchCameraFixture):
+        reporter = cloudwatch_camera.create_reporter()
+        assert (
+            reporter.logger.name
+            == "palace.manager.celery.monitoring.QueueStatsReporter"
+        )
+
+    def test_run(self, cloudwatch_camera: CloudwatchCameraFixture):
+        # The reporter can be driven directly (no events camera): run() snapshots
+        # every queue and publishes a QueueWaiting metric for each.
+        reporter = cloudwatch_camera.create_reporter()
+        cloudwatch_camera.mock_get_redis.return_value.llen.return_value = 3
+        cloudwatch_camera.mock_get_redis.return_value.lindex.return_value = None
+        put_metric_data = cloudwatch_camera.client.return_value.put_metric_data
+
+        with freeze_time("2021-01-01"):
+            reporter.run()
+
+        put_metric_data.assert_called_once()
+        metric_data = put_metric_data.call_args.kwargs["MetricData"]
+        assert {metric["MetricName"] for metric in metric_data} == {"QueueWaiting"}
+        assert all(metric["Value"] == 3 for metric in metric_data)
+        queue_dimensions = {
+            dimension["Value"]
+            for metric in metric_data
+            for dimension in metric["Dimensions"]
+            if dimension["Name"] == "QueueName"
+        }
+        assert queue_dimensions == {"queue1", "queue2"}