Expose protected operability metrics

docs/operability.md

@@ -14,6 +14,15 @@ Use `structured_log_context(correlation_id=..., tenant=..., actor=...)` as the
 base log envelope. Adapters should add transport details around that envelope
 without dropping correlation id or tenant.
 
+## Metrics
+
+`GET /metrics` exposes Prometheus text containing the binary
+`user_engine_ready` dependency gauge and bounded aggregate
+`user_engine_records{kind=...}` counters. It does not expose identity values,
+emails, tenant names, correlation identifiers, credential material, or raw
+outbox payloads. Keep this route cluster-internal and authorize ingress only
+from the selected monitoring workload.
+
 ## Outbox Drain
 
 `outbox_diagnostics()` reports pending event count, event type counts, and the

src/user_engine/web.py

@@ -91,6 +91,11 @@ class PortalApplication:
         if path == "/readyz":
             report = self.service.readiness()
             return self._json(start_response, "200 OK" if report.ready else "503 Service Unavailable", _jsonable(report), correlation_id)
+        if path == "/metrics":
+            supplied = str(environ.get("HTTP_X_USER_ENGINE_PROXY_SECRET", ""))
+            if not secrets.compare_digest(supplied, self.trusted_proxy_secret):
+                raise AuthorizationDenied("metrics require the trusted workload marker")
+            return self._metrics(start_response, correlation_id)
         if path in {"/login", "/oidc/start"}:
             location = self.oidc_client.begin() if self.oidc_client else self.login_url
             start_response("303 See Other", [("Location", location), *self._security_headers(correlation_id)])
@@ -536,6 +541,35 @@ a:focus-visible,input:focus-visible,select:focus-visible,button:focus-visible{{o
         start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(data))), *self._security_headers(correlation_id)])
         return [data]
 
+    def _metrics(
+        self, start_response: StartResponse, correlation_id: str
+    ) -> list[bytes]:
+        report = self.service.readiness()
+        counts = self.service.operability_snapshot().metrics
+        lines = [
+            "# HELP user_engine_ready Whether runtime dependencies are ready.",
+            "# TYPE user_engine_ready gauge",
+            f"user_engine_ready {1 if report.ready else 0}",
+            "# HELP user_engine_records Durable logical record counts by kind.",
+            "# TYPE user_engine_records gauge",
+        ]
+        for kind, count in sorted(counts.items()):
+            safe_kind = "".join(
+                character for character in str(kind)
+                if character.isalnum() or character in "_-"
+            )
+            lines.append(f'user_engine_records{{kind="{safe_kind}"}} {int(count)}')
+        data = ("\n".join(lines) + "\n").encode()
+        start_response(
+            "200 OK",
+            [
+                ("Content-Type", "text/plain; version=0.0.4; charset=utf-8"),
+                ("Content-Length", str(len(data))),
+                *self._security_headers(correlation_id),
+            ],
+        )
+        return [data]
+
     def _error(self, start_response: StartResponse, status: str, code: str, message: str, correlation_id: str) -> list[bytes]:
         return self._json(start_response, status, {"error": {"code": code, "message": message, "correlation_id": correlation_id}}, correlation_id)
 

tests/test_web.py

@@ -73,6 +73,19 @@ class PortalApplicationTests(unittest.TestCase):
         self.assertEqual("200 OK", home["status"])
         self.assertIn(b"Sign in with KeyCape", html)
 
+    def test_metrics_expose_only_bounded_aggregate_state(self):
+        denied, _ = invoke(self.app, "/metrics", claims={}, marker="")
+        self.assertEqual("403 Forbidden", denied["status"])
+        result, payload = invoke(self.app, "/metrics", claims={})
+        self.assertEqual("200 OK", result["status"])
+        self.assertEqual(
+            "text/plain; version=0.0.4; charset=utf-8",
+            result["headers"]["Content-Type"],
+        )
+        self.assertIn(b"user_engine_ready 1", payload)
+        self.assertIn(b'user_engine_records{kind="users"} 0', payload)
+        self.assertNotIn(SECRET.encode(), payload)
+
     def test_protected_route_rejects_untrusted_claim_header(self):
         result, payload = invoke(
             self.app, "/api/v1/me", claims=self.claims, marker="attacker"