feat: expose aggregate job metrics at GET /metrics

.github/workflows/model-smoke.yml

@@ -18,6 +18,8 @@ on:
       - worker/ai_bridge/jobs.py
       # The worker's HTTP handling, which every model test and the MCP workers' health check go through.
       - worker/ai_bridge/server.py
+      # Imported by both: the counters every model job passes through.
+      - worker/ai_bridge/metrics.py
       - scripts/benchmark.py
       - scripts/ner_measure.py
       - tests/fixtures/ner-sample.json

CHANGELOG.md

@@ -6,6 +6,10 @@ All notable changes to this project are documented in this file. The format foll
 
 ## [Unreleased]
 
+### Added
+
+- `GET /metrics` on the worker, in the Prometheus text format and behind the service token: jobs by task and terminal status, submissions, 429 refusals, queue depth, running jobs, pool size and readiness, the concurrency and capacity settings, and per-task histograms of job and queue durations. Aggregates only, with labels from fixed sets; they live in memory and reset when the worker restarts.
+
 ## [0.2.0] - 2026-10-21
 
 ### Added

README.md

@@ -12,6 +12,7 @@ Includes a FrankenPHP worker-mode example, deterministic fault tests, and an opt
 - Redaction: checksum and pattern rules for card numbers, IBANs, emails, IPv4 and IPv6 addresses and phone numbers on every backend, plus named-entity masking with the int8 cased and uncased `bert-base-NER` models on the ONNX profile. Assistive, not a compliance control.
 - Bounded concurrency, queue/result capacity, retention, request sizes and execution deadlines.
 - Cancellation, progress polling, crash detection and generic errors that omit internal exception text.
+- `GET /metrics` on the worker: aggregate job counts, queue depth, pool readiness and duration histograms in the Prometheus text format, behind the service token and carrying no job content; counters reset on restart ([the protocol](docs/protocol.md#metrics)).
 - A FrankenPHP example that submits work and returns a job ID immediately.
 - A stdio MCP server with one tool per task, a second client of the same protocol beside the PHP client, plus an index builder over the protocol and a `bridge_search` tool that searches what it built, returning offsets and snippets rather than vectors, a compose service that puts the pinned ONNX models behind it, and a ten-question evaluation for LLM hosts with a runner for the Claude Code CLI; see [the MCP guide](docs/mcp.md).
 - CPU inference with `cross-encoder/ms-marco-TinyBERT-L2-v2`, `sentence-transformers/all-MiniLM-L6-v2` and int8 `dslim/bert-base-NER` and `dslim/bert-base-NER-uncased` through ONNX Runtime, using pinned model files and wheel hashes.

docs/protocol.md

@@ -5,6 +5,7 @@ Every route requires `Authorization: Bearer <service token>`. JSON requests requ
 | Route | Result |
 | --- | --- |
 | `GET /healthz` | 200 with protocol version and configured backend; process liveness only, not model readiness |
+| `GET /metrics` | 200 with aggregate counts and timings in the Prometheus text format ([below](#metrics)) |
 | `POST /v1/jobs` | 202 with a queued job, or with the retained job a repeated `idempotency_key` names |
 | `GET /v1/jobs/{id}` | 200 with a snapshot, or 404 |
 | `POST /v1/jobs/{id}/cancel` | Send `{}`; returns the terminal/current snapshot, or 404 |
@@ -40,3 +41,24 @@ With `BRIDGE_STATE_DIR` set, the job table (never a job's input) is written thro
 The PHP client uses a fresh cURL handle per call, refuses redirects, ignores proxy environment variables, caps response bytes, and verifies that GET/cancel responses match the requested ID. Configure its request timeout separately from job execution and local waiting deadlines. There are no transparent submission retries, a 429 included. The MCP server in `mcp_server/` is a second, independent client with the same checks; unlike the PHP client it cancels a job whose local wait expires, because no later request will collect it.
 
 Adding a task requires trusted source changes to `worker/ai_bridge/tasks.py`: an explicit name, input validation, bounded output and execution logic. Add a PHP result type where useful, plus successful, malformed-input and lifecycle tests. Task imports, file paths and command strings must not come from request data.
+
+## Metrics
+
+`GET /metrics` takes the same bearer token as every other route (401 without it; a `POST` is answered as on any unknown route, 404 for `{}`, and another spelling, a query string included, is 404) and answers `Content-Type: text/plain; version=0.0.4; charset=utf-8`, the Prometheus text exposition format, written by the standard library. Every value is an aggregate since the worker process started, held in memory only: a restart resets every counter and histogram to zero, the job table in `BRIDGE_STATE_DIR` does not feed them, and a job read back from that table counts nothing. Each family has one `# HELP` and one `# TYPE` line, and every series exists from the start, at zero. Label values come only from fixed sets: `task` is a task of the allowlist (`rerank`, `embed`, `redact`, plus the test tasks on a worker started with `BRIDGE_TEST_TASKS=1`), `status` one of the four terminal statuses, `reason` is `capacity`, and `backend` the configured backend; no job ID, idempotency key, input, result, error, model path or token appears.
+
+| Family | Type | Labels | What it holds |
+| --- | --- | --- | --- |
+| `bridge_jobs_total` | counter | `task`, `status` | Jobs that reached `succeeded`, `failed`, `timed_out` or `cancelled` |
+| `bridge_jobs_submitted_total` | counter | `task` | Accepted submissions that created a job; a repeated `idempotency_key` creates none and is not counted |
+| `bridge_submissions_rejected_total` | counter | `reason="capacity"` | Submissions answered 429, for every cause the 429 section names |
+| `bridge_queue_depth` | gauge | | Jobs `queued` now |
+| `bridge_jobs_running` | gauge | | Jobs `running` now |
+| `bridge_pool_size` | gauge | | Child processes in the pool, starting or started |
+| `bridge_pool_ready` | gauge | | Of those, the ones that have said they are ready, idle or running a job |
+| `bridge_concurrency` | gauge | | The `BRIDGE_CONCURRENCY` setting |
+| `bridge_capacity` | gauge | | The `BRIDGE_CAPACITY` setting |
+| `bridge_job_duration_seconds` | histogram | `task` | Seconds from submission to terminal status, for every job counted by `bridge_jobs_total` |
+| `bridge_job_queue_seconds` | histogram | `task` | Seconds from submission to `running`, for jobs a child started; one that ended while queued adds nothing |
+| `bridge_info` | gauge | `protocol="1"`, `backend` | Always 1 |
+
+Both histograms have the buckets 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60 and 300 seconds and `+Inf`, cumulative as the format requires, with `_sum` and `_count`. A scrape copies the numbers under the store's lock and formats them without it, so it never waits for a job. An invalid submission (400) creates no job and counts nothing.

docs/security.md

@@ -12,6 +12,7 @@ This experimental service is designed for trusted application-to-service traffic
 - Requests and results may contain sensitive documents. They live in RAM, may be present in process/core dumps, and are not encrypted by this service. With `BRIDGE_STATE_DIR` set, `jobs.sqlite` (and its `-wal` and `-shm` files) in that directory holds each retained job's ID, task, status, error, progress, deadline, idempotency key, submission and finish times, and its result until retention expires; it never holds a job's input, though results such as redacted text, rankings and embedding vectors reveal much of it. SQLite's `secure_delete` overwrites an expired row in the database file, but older copies of its page can stay in the `-wal` file until a checkpoint reuses it, so keep the directory on tmpfs or an encrypted volume readable only by the service user (compose's `worker` uses a 64 MB tmpfs owned by uid 65532 with mode 0700). Minimize retention and use appropriate host/storage policies. Embedding vectors are derived from their input text and can reveal much of it; store and transmit them with the same care as the text.
 - Redaction is assistive, not a compliance control. The rules find only the structured identifiers they recognise, and the optional named-entity models were trained on English news, so their recall on names from other domains or languages is lower and unmeasured here; review redacted output before treating it as free of personal data. The model files are third-party ONNX conversions of the original weights, pinned by revision and hash but not signed by the original author.
 - The MCP server is a client of this service for an AI agent over stdio, where there is no authentication: the host that launches it inherits its trust, and it holds the token from its environment and never returns it. Running it over HTTP or SSE would be a new authorisation surface that the bearer token does not cover; it is out of scope. An agent can be prompt-injected and can then fill a worker's capacity, so point the MCP server at a dedicated worker, never at one serving a production application; compose's `mcp` and `mcp-model` each have their own. It reads files only under `BRIDGE_MCP_ROOT` and writes none; snippets it returns are the agent's own file content and enter the agent's context like any file it reads. Index files built for `bridge_search` are derived data: `index.json` holds every chunk's text verbatim and `vectors.f32` holds vectors that can reveal it, so keep them with the same care as their sources. The builder runs with the service token like the PHP client and is the only component that writes, into the directory the operator names.
+- `GET /metrics` reveals aggregate numbers only: how many jobs of each task were submitted, refused for capacity and ended in each terminal status, how many are queued and running, the pool's size and readiness, the concurrency and capacity settings, the backend name, and histograms of job and queue durations per task. It never reveals a job ID, idempotency key, input, result, error message, model path or the token; its label values come from fixed sets, never from a request. Counts and timings still describe the traffic (how busy the service is, which tasks run, how long they take), so the endpoint is behind the same service token as every route: scrape it from inside the private network with that token, never through a public route. How often it is scraped is the operator's concern; a scrape copies a few counters under the store's lock and formats them outside it.
 - Access logs are disabled to avoid accidentally recording identifiers and inputs. Internal exceptions become generic task errors. This is not a full observability or compliance solution.
 - Cancel/timeout terminates the task process, but cannot roll back previously completed external effects. No task in the prototype modifies an external system.
 - The default health endpoint proves HTTP service availability, not model integrity, ability to spawn processes or successful inference. An ONNX worker checks before it starts that each model directory holds its tokenizer and model files and refuses to start otherwise; that proves presence, not integrity, which the fetcher's hashes establish. Use a controlled model smoke test for readiness validation.

tests/integration.py

@@ -8,6 +8,7 @@
 from urllib.request import Request, urlopen
 
 BASE = os.environ.get("FRANKENPHP_URL", "http://frankenphp:8080")
+WORKER = os.environ.get("WORKER_URL", "http://worker:8090")
 
 
 def request(path, body=None):
@@ -17,6 +18,20 @@ def request(path, body=None):
         return response.status, json.load(response), int(response.headers["X-Demo-Worker-Requests"])
 
 
+def metrics(token=None):
+    """The worker's counters as {(name, labels): value}; the format itself is checked by the unit tests."""
+    headers = {"Authorization": "Bearer " + (token if token is not None else os.environ["BRIDGE_TOKEN"])}
+    with urlopen(Request(WORKER + "/metrics", headers=headers), timeout=5) as response:
+        assert response.headers["Content-Type"] == "text/plain; version=0.0.4; charset=utf-8"
+        text = response.read().decode("utf-8")
+    samples = {}
+    for line in text.splitlines():
+        if not line.startswith("#"):
+            series, value = line.rsplit(" ", 1)
+            samples[series] = float(value)
+    return text, samples
+
+
 def wait_ready():
     deadline = time.monotonic() + 30
     while time.monotonic() < deadline:
@@ -56,6 +71,7 @@ def scenario(index):
 
 def main():
     wait_ready()
+    _, counted = metrics()
     _, _, before = request("/healthz")
     with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
         ids = list(pool.map(scenario, range(12)))
@@ -132,7 +148,21 @@ def main():
             assert error.code == 400, (body, error.code)
         else:
             raise AssertionError("Out-of-contract redact input was accepted")
-    print("PASS: 12 concurrent isolated jobs through a reused FrankenPHP worker; top_k narrowing; embedding; redaction; invalid and missing-job cases")
+    # The worker counted those jobs, and its metrics hold numbers only, none of the documents.
+    try:
+        metrics("not-the-token")
+    except HTTPError as error:
+        assert error.code == 401, error.code
+    else:
+        raise AssertionError("Metrics were served without the token")
+    text, moved = metrics()
+    for task, least in [("rerank", 13), ("embed", 1), ("redact", 1)]:
+        for name in [f'bridge_jobs_total{{task="{task}",status="succeeded"}}', f'bridge_jobs_submitted_total{{task="{task}"}}',
+                     f'bridge_job_duration_seconds_count{{task="{task}"}}']:
+            assert moved[name] - counted[name] >= least, (name, counted[name], moved[name])
+    for document in ["unique0 token0", "needle token", "x@y.io", "4111 1111 1111 1111", "red apple", "café"]:
+        assert document not in text, document
+    print("PASS: 12 concurrent isolated jobs through a reused FrankenPHP worker; top_k narrowing; embedding; redaction; invalid and missing-job cases; worker metrics")
 
 
 if __name__ == "__main__":