feat(metrics): expose async-operation queue + consolidation backlog as gauges (#1987)

* feat(metrics): expose async-operation queue + consolidation backlog as gauges

The bank-stats endpoint already computes operations_by_status,
pending_consolidation and failed_consolidation, but only as a point-in-time
HTTP response per bank. There's no way to trend or alert on "is the worker
keeping up?" / "is the knowledge base caught up?" from Prometheus.

This adds three observable gauges, fed by a 30s background-refresh cache (the
same pattern as the existing db-pool gauges, so the /metrics scrape path stays
synchronous):

- hindsight_async_operations{operation_type,status} -- worker queue depth for
  non-terminal states. pending = queued backlog (e.g. retain / consolidation),
  processing = in-flight, failed = stranded. Terminal states (completed,
  cancelled) are deliberately excluded: a gauge of finished work grows without
  bound and says nothing about current load. The processing series is the only
  signal that surfaces a hung operation holding a worker slot.
- hindsight_consolidation_backlog -- source memories (experience/world) not yet
  consolidated into observations (pending_consolidation).
- hindsight_consolidation_failed -- source memories whose consolidation
  permanently failed, recoverable via the consolidation recovery endpoint
  (failed_consolidation).

The SQL is lifted from the bank-stats endpoint and is index-backed
(idx_async_operations_status, idx_memory_units_unconsolidated). Per-bank labels
are gated behind the existing metrics_include_bank_id flag (off by default);
when off, counts aggregate per tenant/schema, bounding cardinality to a handful
of series. All queries are PostgreSQL-specific (FILTER, information_schema),
consistent with this collector already being bound to an asyncpg pool.

* review: address feedback on backlog metrics

- Split the consolidation backlog into two separate COUNT(*) queries, each with
  a WHERE matching a partial-index predicate exactly (idx_memory_units_
  unconsolidated / idx_memory_units_consolidation_failed), instead of one
  aggregate with two FILTERs that seq-scans the whole memory_units table on
  every 30s refresh across every schema. GROUP BY bank_id still composes
  (bank_id is each index's lead column).
- Type the gauge cache keys as NamedTuples (_AsyncOpKey, _BacklogKey) instead of
  raw tuples.
- Hoist `import asyncio` to module scope (was imported inside two methods).
- Document that _backlog_task is process-lifetime and intentionally not
  cancelled (no teardown hook to hang it on).
- Add tests for the per_bank=True path (bank_id in the cache key + GROUP BY
  bank_id in the SQL + bank_id gauge attribute) and assert the backlog queries
  are index-matched, not FILTER scans.

* fix(metrics): force index scan for the consolidation backlog count

Splitting the consolidation count into two index-predicate-matched COUNT(*)
queries fixed the failed count (index-only scan) but NOT the backlog count.
Verified on a 114k-row memory_units via EXPLAIN ANALYZE: the backlog query still
seq-scans (~92 ms) because `consolidated_at IS NULL` is true for ~40% of the
table (every observation has a null consolidated_at), so the planner misjudges
selectivity and won't use idx_memory_units_unconsolidated even though the
predicate matches it exactly. ANALYZE doesn't change the plan (structural, not
stale stats); `enable_seqscan=off` confirms the index is usable (~0.1 ms).

Run the backlog count in a scoped transaction with SET LOCAL enable_seqscan=off
to force the partial-index scan (verified ~0.07 ms, transaction-scoped, no
leak). The failed count needs no nudge — consolidation_failed_at IS NOT NULL is
rare, so its index is chosen on cost.

* feat(metrics): gate consolidation backlog gauges behind config flag (off by default)

Add HINDSIGHT_API_METRICS_BACKLOG_ENABLED (default false). The
async-operation queue + consolidation backlog gauges run periodic
per-schema COUNT queries on a background task, so they are now opt-in
rather than always-on when a db pool is set.

* chore: sync embed env template + prettify paperclip README after main merge

---------

Co-authored-by: Nicolò Boschi <boschi1997@gmail.com>

Author: MattJackson

.env.example

@@ -162,6 +162,10 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Custom service name and environment (optional, defaults: hindsight-api, development)
 # HINDSIGHT_API_OTEL_SERVICE_NAME=hindsight-production
 # HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT=production
+#
+# Expose async-operation queue + consolidation-backlog gauges on /metrics.
+# Runs periodic per-schema COUNT queries on a background task (disabled by default).
+# HINDSIGHT_API_METRICS_BACKLOG_ENABLED=true
 
 # -----------------------------------------------------------------------------
 # Control Plane (Optional)

hindsight-api-slim/hindsight_api/config.py

@@ -375,6 +375,7 @@ def normalize_config_dict(config: dict[str, Any]) -> dict[str, Any]:
 ENV_OTEL_SERVICE_NAME = "HINDSIGHT_API_OTEL_SERVICE_NAME"
 ENV_OTEL_DEPLOYMENT_ENVIRONMENT = "HINDSIGHT_API_OTEL_DEPLOYMENT_ENVIRONMENT"
 ENV_METRICS_INCLUDE_BANK_ID = "HINDSIGHT_API_METRICS_INCLUDE_BANK_ID"
+ENV_METRICS_BACKLOG_ENABLED = "HINDSIGHT_API_METRICS_BACKLOG_ENABLED"
 
 # Vertex AI configuration
 ENV_LLM_VERTEXAI_PROJECT_ID = "HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID"
@@ -974,6 +975,7 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_OTEL_SERVICE_NAME = "hindsight-api"
 DEFAULT_OTEL_DEPLOYMENT_ENVIRONMENT = "development"
 DEFAULT_METRICS_INCLUDE_BANK_ID = False  # Disabled by default to avoid high-cardinality OTel metric growth
+DEFAULT_METRICS_BACKLOG_ENABLED = False  # Disabled by default: runs periodic per-schema COUNT queries
 
 # Audit log defaults
 DEFAULT_AUDIT_LOG_ENABLED = False  # Disabled by default
@@ -1651,6 +1653,7 @@ class HindsightConfig:
     otel_service_name: str
     otel_deployment_environment: str
     metrics_include_bank_id: bool
+    metrics_backlog_enabled: bool
 
     # Audit log configuration (static - server-level only)
     audit_log_enabled: bool  # Master switch for audit logging
@@ -2642,6 +2645,8 @@ def from_env(cls) -> "HindsightConfig":
             otel_deployment_environment=os.getenv(ENV_OTEL_DEPLOYMENT_ENVIRONMENT, DEFAULT_OTEL_DEPLOYMENT_ENVIRONMENT),
             metrics_include_bank_id=os.getenv(ENV_METRICS_INCLUDE_BANK_ID, str(DEFAULT_METRICS_INCLUDE_BANK_ID)).lower()
             in ("true", "1", "yes"),
+            metrics_backlog_enabled=os.getenv(ENV_METRICS_BACKLOG_ENABLED, str(DEFAULT_METRICS_BACKLOG_ENABLED)).lower()
+            in ("true", "1", "yes"),
             # Audit log configuration (static, server-level only)
             audit_log_enabled=os.getenv(ENV_AUDIT_LOG_ENABLED, str(DEFAULT_AUDIT_LOG_ENABLED)).lower() == "true",
             audit_log_actions=[

hindsight-api-slim/hindsight_api/metrics.py

@@ -11,6 +11,7 @@
 - Database connection pool metrics
 """
 
+import asyncio
 import importlib
 import logging
 import os
@@ -20,7 +21,7 @@
 import threading
 import time
 from contextlib import contextmanager
-from typing import TYPE_CHECKING, Callable
+from typing import TYPE_CHECKING, Callable, NamedTuple
 
 from opentelemetry import metrics
 from opentelemetry.exporter.prometheus import PrometheusMetricReader
@@ -76,6 +77,28 @@ def _is_client_cancellation(exc: BaseException) -> bool:
 # HTTP request duration buckets (millisecond-level for fast endpoints)
 HTTP_DURATION_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0)
 
+# How often the backlog / queue-depth gauge caches are refreshed (seconds).
+# The counts are aggregate COUNT queries, so a background task refreshes a
+# cache and the observable gauges read from it — keeping the /metrics scrape
+# path synchronous (the same reason the db-pool gauges read cached state).
+BACKLOG_METRICS_REFRESH_SECONDS = 30
+
+
+class _AsyncOpKey(NamedTuple):
+    """Cache / label key for the async-operation queue gauge."""
+
+    tenant: str
+    operation_type: str
+    status: str
+    bank_id: str | None
+
+
+class _BacklogKey(NamedTuple):
+    """Cache / label key for the consolidation backlog and failed gauges."""
+
+    tenant: str
+    bank_id: str | None
+
 
 def get_token_bucket(token_count: int) -> str:
     """
@@ -383,6 +406,13 @@ def __init__(self):
         # DB pool metrics holder (set via set_db_pool)
         self._db_pool: "asyncpg.Pool | None" = None
 
+        # Backlog / queue-depth gauge caches, refreshed by a background task
+        # (see _setup_backlog_metrics) so the scrape path stays synchronous.
+        self._async_ops_counts: dict[_AsyncOpKey, int] = {}
+        self._consolidation_backlog: dict[_BacklogKey, int] = {}
+        self._consolidation_failed: dict[_BacklogKey, int] = {}
+        self._backlog_task: "asyncio.Task | None" = None
+
     @contextmanager
     def record_operation(
         self,
@@ -650,6 +680,10 @@ def set_db_pool(self, pool: "asyncpg.Pool"):
         """
         self._db_pool = pool
         self._setup_db_pool_metrics()
+        from .config import get_config
+
+        if get_config().metrics_backlog_enabled:
+            self._setup_backlog_metrics()
 
     def _setup_db_pool_metrics(self):
         """Set up observable gauges for database pool metrics."""
@@ -715,6 +749,192 @@ def get_pool_max_size(_options):
             unit="{connections}",
         )
 
+    def _setup_backlog_metrics(self):
+        """Observable gauges for the async-operation queue and the
+        consolidation backlog.
+
+        These mirror fields the bank-stats endpoint already computes
+        (``operations_by_status``, ``pending_consolidation``,
+        ``failed_consolidation``) but expose them as scrapable gauges, so
+        queue depth and backlog can be trended and alerted on instead of only
+        polled per-bank over HTTP. The two motivating questions both come for
+        free here: "is the worker keeping up?" (async-op queue) and "is the
+        knowledge base caught up?" (consolidation backlog) — including the
+        ``processing`` state, which is the only signal that surfaces a hung
+        operation stuck holding a worker slot.
+
+        Counts are aggregate ``COUNT`` queries, so a background task refreshes
+        a cache every ``BACKLOG_METRICS_REFRESH_SECONDS`` and these callbacks
+        read it — keeping the scrape path synchronous, the same approach as
+        the db-pool gauges above.
+        """
+        if self._backlog_task is not None:
+            return  # already started for this collector
+
+        def get_async_operations(_options):
+            for key, value in list(self._async_ops_counts.items()):
+                attrs = {"tenant": key.tenant, "operation_type": key.operation_type, "status": key.status}
+                if key.bank_id is not None:
+                    attrs["bank_id"] = key.bank_id
+                yield metrics.Observation(value, attrs)
+
+        def get_consolidation_backlog(_options):
+            for key, value in list(self._consolidation_backlog.items()):
+                attrs = {"tenant": key.tenant}
+                if key.bank_id is not None:
+                    attrs["bank_id"] = key.bank_id
+                yield metrics.Observation(value, attrs)
+
+        def get_consolidation_failed(_options):
+            for key, value in list(self._consolidation_failed.items()):
+                attrs = {"tenant": key.tenant}
+                if key.bank_id is not None:
+                    attrs["bank_id"] = key.bank_id
+                yield metrics.Observation(value, attrs)
+
+        self.meter.create_observable_gauge(
+            name="hindsight.async_operations",
+            callbacks=[get_async_operations],
+            description="Async operations in a non-terminal state, by operation_type and status "
+            "(pending=queued backlog, processing=in-flight, failed=stranded)",
+            unit="{operations}",
+        )
+        self.meter.create_observable_gauge(
+            name="hindsight.consolidation.backlog",
+            callbacks=[get_consolidation_backlog],
+            description="Source memories (experience/world) not yet consolidated into observations",
+            unit="{memories}",
+        )
+        self.meter.create_observable_gauge(
+            name="hindsight.consolidation.failed",
+            callbacks=[get_consolidation_failed],
+            description="Source memories whose consolidation permanently failed "
+            "(recoverable via the consolidation recovery endpoint)",
+            unit="{memories}",
+        )
+
+        # Drive the caches from a background task on the running loop.
+        # set_db_pool runs during async startup, so a loop is normally present;
+        # if not, the gauges simply stay empty rather than crashing collection.
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            logger.warning("No running event loop; backlog metrics disabled")
+            return
+        # Process-lifetime task: there is no collector teardown hook to cancel it
+        # on, so it's torn down with the event loop at process shutdown. If a
+        # shutdown path is ever added, cancel self._backlog_task there.
+        self._backlog_task = loop.create_task(self._backlog_refresh_loop())
+
+    async def _backlog_refresh_loop(self):
+        """Periodically refresh the backlog / queue-depth caches."""
+        while True:
+            try:
+                await self._refresh_backlog()
+            except Exception:
+                logger.debug("Backlog metrics refresh failed", exc_info=True)
+            await asyncio.sleep(BACKLOG_METRICS_REFRESH_SECONDS)
+
+    async def _refresh_backlog(self):
+        """Recount the async-operation queue and consolidation backlog across
+        every provisioned Hindsight schema.
+
+        Per-bank labels are gated behind ``metrics_include_bank_id`` (off by
+        default) to keep cardinality bounded; when off, counts are aggregated
+        per tenant/schema. All SQL here is PostgreSQL-specific (``FILTER``,
+        ``information_schema``), which is consistent with this collector
+        already being bound to an asyncpg pool.
+        """
+        if self._db_pool is None:
+            return
+
+        async_ops: dict[_AsyncOpKey, int] = {}
+        backlog: dict[_BacklogKey, int] = {}
+        failed: dict[_BacklogKey, int] = {}
+        per_bank = self._include_bank_id
+        bank_sel = "bank_id, " if per_bank else ""
+        bank_grp = " GROUP BY bank_id" if per_bank else ""
+
+        async with self._db_pool.acquire() as conn:
+            # memory_units is the central per-tenant table; its presence marks a
+            # provisioned Hindsight schema.
+            schema_rows = await conn.fetch(
+                "SELECT table_schema FROM information_schema.tables WHERE table_name = 'memory_units'"
+            )
+            for schema_row in schema_rows:
+                schema = schema_row["table_schema"]
+
+                # Worker queue depth — mirrors operations_by_status, split by
+                # operation_type. Terminal states (completed/cancelled) are
+                # excluded on purpose: a gauge of finished work grows without
+                # bound and says nothing about current load.
+                # Index: idx_async_operations_status.
+                ops_grp = "operation_type, status" + (", bank_id" if per_bank else "")
+                try:
+                    rows = await conn.fetch(
+                        f"SELECT operation_type, status, {bank_sel}COUNT(*) AS count "
+                        f'FROM "{schema}".async_operations '
+                        "WHERE status IN ('pending', 'processing', 'failed') "
+                        f"GROUP BY {ops_grp}"
+                    )
+                    for row in rows:
+                        bank = row["bank_id"] if per_bank else None
+                        key = _AsyncOpKey(schema, row["operation_type"] or "unknown", row["status"], bank)
+                        async_ops[key] = async_ops.get(key, 0) + int(row["count"])
+                except Exception:
+                    logger.debug("Async-ops queue query failed for schema %s", schema, exc_info=True)
+
+                # Consolidation backlog + stranded counts. Two separate COUNT(*)
+                # queries rather than one with two FILTERs — each WHERE matches a
+                # partial-index predicate exactly:
+                #   idx_memory_units_unconsolidated        WHERE consolidated_at IS NULL ...
+                #   idx_memory_units_consolidation_failed  WHERE consolidation_failed_at IS NOT NULL ...
+                # GROUP BY bank_id still composes — bank_id is each index's lead column.
+                #
+                # The backlog count runs with seqscan disabled in a scoped
+                # transaction. The partial index matches its predicate, but
+                # `consolidated_at IS NULL` is true for a large fraction of the
+                # table (every observation has a null consolidated_at), so the
+                # planner misjudges selectivity and otherwise seq-scans the whole
+                # (largest) table on every refresh — verified on a 114k-row table
+                # via EXPLAIN: seq scan ~92 ms vs index scan ~0.1 ms. SET LOCAL
+                # forces the index path and resets at transaction end. The failed
+                # count below needs no such nudge: `consolidation_failed_at IS NOT
+                # NULL` is rare, so its index is chosen on cost.
+                try:
+                    async with conn.transaction():
+                        await conn.execute("SET LOCAL enable_seqscan = off")
+                        rows = await conn.fetch(
+                            f"SELECT {bank_sel}COUNT(*) AS count "
+                            f'FROM "{schema}".memory_units '
+                            "WHERE consolidated_at IS NULL AND fact_type IN ('experience', 'world')"
+                            f"{bank_grp}"
+                        )
+                    for row in rows:
+                        bank = row["bank_id"] if per_bank else None
+                        key = _BacklogKey(schema, bank)
+                        backlog[key] = backlog.get(key, 0) + int(row["count"])
+                except Exception:
+                    logger.debug("Consolidation backlog query failed for schema %s", schema, exc_info=True)
+
+                try:
+                    rows = await conn.fetch(
+                        f"SELECT {bank_sel}COUNT(*) AS count "
+                        f'FROM "{schema}".memory_units '
+                        "WHERE consolidation_failed_at IS NOT NULL AND fact_type IN ('experience', 'world')"
+                        f"{bank_grp}"
+                    )
+                    for row in rows:
+                        bank = row["bank_id"] if per_bank else None
+                        key = _BacklogKey(schema, bank)
+                        failed[key] = failed.get(key, 0) + int(row["count"])
+                except Exception:
+                    logger.debug("Consolidation failed query failed for schema %s", schema, exc_info=True)
+
+        self._async_ops_counts = async_ops
+        self._consolidation_backlog = backlog
+        self._consolidation_failed = failed
+
 
 # Global metrics collector instance (defaults to no-op)
 _metrics_collector: MetricsCollectorBase = NoOpMetricsCollector()