From dd256597bb95c1c44bf06aad4622f500f689d717 Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 17:55:02 -0400
Subject: [PATCH 1/6] feat(server): instrument HTTP and store with bounded
 metrics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Wire the observability handle into the hot path without holding the
global lock during export. Replace the default TraceLayer that leaked
raw URIs with a safe middleware that records method, matched route
template (never concrete ID or query, 1000 IDs remain one series),
finite surface (native, runner, broker, results, webhook, git, oidc,
live_logs, public, test, unknown) and status class. The live_logs
WebSocket is excluded from the duration histogram and will be tracked
via livelog connections instead. Active requests are an updown gauge;
durations use the 0.005..10s buckets.

Add a shared MetricsRegistry in preloop-observability (HttpMetrics +
StoreMetrics) with Prometheus text rendering. The registry is
cloneable via the Observability handle (already in AppState) and is
fed only from the cached snapshot or via the InstrumentedStore wrapper
— no await while holding InnerState.

Wrap the private Store trait once in store.rs with InstrumentedStore:
all seven methods (load_into, store_inner, store_meta_only,
store_run_event, store_workflow_run_counter, store_log_chunk,
append_event) record preloop.store.operation.duration with backend,
operation, outcome and the consecutive-failures gauge, preserving
every return/error and the best-effort persistence rule. Bootstrap
wraps the store returned by open_store with the same observability
handle (backend sqlite vs postgres detected from PRELOOP_STORE_URL).

Extend GET /metrics to append the registry's http and store exposition
after the snapshot-based pool/queue gauges, still behind native bearer.

cargo check, fmt, sg-scan-strict and preloop-observability tests
(12, including route normalization and bounded-series) pass; run/job
and broker lifecycle counters remain for the next change.

Entire-Checkpoint: 01M0GJQ8PAGXWHJTR7WHR0SGVS
---
 crates/preloop-observability/src/lib.rs       |   8 +
 crates/preloop-observability/src/metrics.rs   | 405 ++++++++++++++++++
 crates/preloop-runner-server/src/bootstrap.rs |  19 +
 .../preloop-runner-server/src/http_metrics.rs | 103 +++++
 crates/preloop-runner-server/src/lib.rs       |   2 +-
 crates/preloop-runner-server/src/routes.rs    |   5 +-
 crates/preloop-runner-server/src/runs.rs      |   2 +
 crates/preloop-runner-server/src/store.rs     | 143 +++++++
 8 files changed, 685 insertions(+), 2 deletions(-)
 create mode 100644 crates/preloop-observability/src/metrics.rs
 create mode 100644 crates/preloop-runner-server/src/http_metrics.rs

diff --git a/crates/preloop-observability/src/lib.rs b/crates/preloop-observability/src/lib.rs
index 7ebd6531..3da05d42 100644
--- a/crates/preloop-observability/src/lib.rs
+++ b/crates/preloop-observability/src/lib.rs
@@ -11,6 +11,7 @@
 //! - Always retain `stderr`/`journald` even when OTLP is configured.
 //! - `Debug` on config never reveals headers or credential-bearing endpoint parts.
 
+pub mod metrics;
 pub mod status;
 
 use std::collections::HashMap;
@@ -381,6 +382,7 @@ struct Inner {
     config: Arc<ObservabilityConfig>,
     heartbeat: TaskHeartbeat,
     limits: LimitRegistry,
+    metrics: Arc<metrics::MetricsRegistry>,
     is_noop: bool,
 }
 
@@ -407,6 +409,7 @@ impl Observability {
                 config: Arc::new(config),
                 heartbeat: TaskHeartbeat::default(),
                 limits: LimitRegistry::default(),
+                metrics: Arc::new(metrics::MetricsRegistry::default()),
                 is_noop: true,
             }),
         }
@@ -420,6 +423,7 @@ impl Observability {
                 config: Arc::new(config),
                 heartbeat: TaskHeartbeat::default(),
                 limits: LimitRegistry::default(),
+                metrics: Arc::new(metrics::MetricsRegistry::default()),
                 is_noop,
             }),
         };
@@ -451,6 +455,10 @@ impl Observability {
         &self.inner.limits
     }
 
+    pub fn metrics(&self) -> &metrics::MetricsRegistry {
+        &self.inner.metrics
+    }
+
     pub fn config(&self) -> &ObservabilityConfig {
         &self.inner.config
     }
diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs
new file mode 100644
index 00000000..1d504461
--- /dev/null
+++ b/crates/preloop-observability/src/metrics.rs
@@ -0,0 +1,405 @@
+//! Metrics registry for HTTP and store — Step 4.
+//!
+//! In-memory, bounded, no network I/O. Instruments and attribute arrays are
+//! prebuilt where static; gauges are updated from the cached snapshot.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use parking_lot::RwLock;
+
+// ---------------------------------------------------------------------------
+// Histogram buckets
+// ---------------------------------------------------------------------------
+
+pub const HTTP_BUCKETS: &[f64] = &[
+    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
+];
+pub const STORE_BUCKETS: &[f64] = &[
+    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
+];
+pub const QUEUE_BUCKETS: &[f64] = &[
+    0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 900.0,
+];
+
+#[derive(Debug, Default, Clone)]
+struct Histogram {
+    buckets: Vec<(f64, u64)>, // (le, count)
+    count: u64,
+    sum: f64,
+}
+
+impl Histogram {
+    fn new(buckets: &[f64]) -> Self {
+        Self {
+            buckets: buckets.iter().map(|&le| (le, 0)).collect(),
+            count: 0,
+            sum: 0.0,
+        }
+    }
+
+    fn observe(&mut self, value: f64) {
+        self.count += 1;
+        self.sum += value;
+        for (le, cnt) in &mut self.buckets {
+            if value <= *le {
+                *cnt += 1;
+            }
+        }
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Http metrics
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct HttpLabels {
+    pub method: String,
+    pub route: String,
+    pub surface: String,
+    pub status_class: String,
+}
+
+#[derive(Debug, Default)]
+pub struct HttpMetrics {
+    active: RwLock<HashMap<HttpLabels, i64>>,
+    durations: RwLock<HashMap<HttpLabels, Histogram>>,
+}
+
+impl HttpMetrics {
+    pub fn inc_active(&self, labels: &HttpLabels) {
+        *self.active.write().entry(labels.clone()).or_insert(0) += 1;
+    }
+
+    pub fn dec_active(&self, labels: &HttpLabels) {
+        let mut g = self.active.write();
+        if let Some(v) = g.get_mut(labels) {
+            *v -= 1;
+            if *v <= 0 {
+                g.remove(labels);
+            }
+        }
+    }
+
+    pub fn observe_duration(&self, labels: HttpLabels, duration: Duration) {
+        let secs = duration.as_secs_f64();
+        let mut g = self.durations.write();
+        let hist = g
+            .entry(labels)
+            .or_insert_with(|| Histogram::new(HTTP_BUCKETS));
+        hist.observe(secs);
+    }
+
+    pub fn render(&self, out: &mut String) {
+        out.push_str("# HELP http_server_request_duration_seconds API latency — matched route, never raw URI\n");
+        out.push_str("# TYPE http_server_request_duration_seconds histogram\n");
+        let g = self.durations.read();
+        for (labels, hist) in g.iter() {
+            for (le, cnt) in &hist.buckets {
+                out.push_str(&format!(
+                    "http_server_request_duration_seconds_bucket{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\",le=\"{}\"}} {}\n",
+                    labels.method, labels.route, labels.surface, labels.status_class, le, cnt
+                ));
+            }
+            out.push_str(&format!(
+                "http_server_request_duration_seconds_bucket{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\",le=\"+Inf\"}} {}\n",
+                labels.method, labels.route, labels.surface, labels.status_class, hist.count
+            ));
+            out.push_str(&format!(
+                "http_server_request_duration_seconds_sum{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\"}} {}\n",
+                labels.method, labels.route, labels.surface, labels.status_class, hist.sum
+            ));
+            out.push_str(&format!(
+                "http_server_request_duration_seconds_count{{method=\"{}\",route=\"{}\",surface=\"{}\",status_class=\"{}\"}} {}\n",
+                labels.method, labels.route, labels.surface, labels.status_class, hist.count
+            ));
+        }
+        out.push_str("# HELP http_server_active_requests Current HTTP concurrency\n");
+        out.push_str("# TYPE http_server_active_requests gauge\n");
+        let g2 = self.active.read();
+        for (labels, v) in g2.iter() {
+            out.push_str(&format!(
+                "http_server_active_requests{{method=\"{}\",route=\"{}\",surface=\"{}\"}} {}\n",
+                labels.method, labels.route, labels.surface, v
+            ));
+        }
+    }
+
+    #[cfg(test)]
+    pub fn clear(&self) {
+        self.active.write().clear();
+        self.durations.write().clear();
+    }
+
+    #[cfg(test)]
+    pub fn series_count(&self) -> usize {
+        self.durations.read().len()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Store metrics
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct StoreLabels {
+    pub backend: String,
+    pub operation: String,
+    pub outcome: String,
+}
+
+#[derive(Debug, Default)]
+pub struct StoreMetrics {
+    durations: RwLock<HashMap<StoreLabels, Histogram>>,
+    consecutive_failures: RwLock<HashMap<String, u64>>, // backend -> count
+}
+
+impl StoreMetrics {
+    pub fn observe(&self, backend: &str, operation: &str, outcome: &str, duration: Duration) {
+        let labels = StoreLabels {
+            backend: backend.to_string(),
+            operation: operation.to_string(),
+            outcome: outcome.to_string(),
+        };
+        let mut g = self.durations.write();
+        let hist = g
+            .entry(labels)
+            .or_insert_with(|| Histogram::new(STORE_BUCKETS));
+        hist.observe(duration.as_secs_f64());
+
+        // Update consecutive failures
+        let mut cf = self.consecutive_failures.write();
+        if outcome == "error" {
+            *cf.entry(backend.to_string()).or_insert(0) += 1;
+        } else {
+            cf.insert(backend.to_string(), 0);
+        }
+    }
+
+    pub fn render(&self, out: &mut String) {
+        out.push_str("# HELP preloop_store_operation_duration_seconds Store operation latency\n");
+        out.push_str("# TYPE preloop_store_operation_duration_seconds histogram\n");
+        let g = self.durations.read();
+        for (labels, hist) in g.iter() {
+            for (le, cnt) in &hist.buckets {
+                out.push_str(&format!(
+                    "preloop_store_operation_duration_seconds_bucket{{backend=\"{}\",operation=\"{}\",outcome=\"{}\",le=\"{}\"}} {}\n",
+                    labels.backend, labels.operation, labels.outcome, le, cnt
+                ));
+            }
+            out.push_str(&format!(
+                "preloop_store_operation_duration_seconds_bucket{{backend=\"{}\",operation=\"{}\",outcome=\"{}\",le=\"+Inf\"}} {}\n",
+                labels.backend, labels.operation, labels.outcome, hist.count
+            ));
+            out.push_str(&format!(
+                "preloop_store_operation_duration_seconds_sum{{backend=\"{}\",operation=\"{}\",outcome=\"{}\"}} {}\n",
+                labels.backend, labels.operation, labels.outcome, hist.sum
+            ));
+            out.push_str(&format!(
+                "preloop_store_operation_duration_seconds_count{{backend=\"{}\",operation=\"{}\",outcome=\"{}\"}} {}\n",
+                labels.backend, labels.operation, labels.outcome, hist.count
+            ));
+        }
+        out.push_str("# HELP preloop_store_consecutive_failures Restart-durability risk\n");
+        out.push_str("# TYPE preloop_store_consecutive_failures gauge\n");
+        let g2 = self.consecutive_failures.read();
+        for (backend, v) in g2.iter() {
+            out.push_str(&format!(
+                "preloop_store_consecutive_failures{{backend=\"{}\"}} {}\n",
+                backend, v
+            ));
+        }
+    }
+
+    #[cfg(test)]
+    pub fn clear(&self) {
+        self.durations.write().clear();
+        self.consecutive_failures.write().clear();
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Registry
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Default)]
+pub struct MetricsRegistry {
+    pub http: HttpMetrics,
+    pub store: StoreMetrics,
+}
+
+impl MetricsRegistry {
+    pub fn render(&self) -> String {
+        let mut out = String::new();
+        self.http.render(&mut out);
+        self.store.render(&mut out);
+        out
+    }
+
+    #[cfg(test)]
+    pub fn clear(&self) {
+        self.http.clear();
+        self.store.clear();
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Helpers — surface classification and route normalization
+// ---------------------------------------------------------------------------
+
+/// Classify a normalized route template into a finite surface.
+pub fn classify_surface(route: &str) -> &'static str {
+    if route == "/healthz" || route == "/readyz" || route == "/metrics" {
+        return "public";
+    }
+    if route.starts_with("/api/v1") {
+        return "native";
+    }
+    if route.starts_with("/_apis") || route.starts_with("/runner") {
+        return "runner";
+    }
+    if route.starts_with("/broker") {
+        return "broker";
+    }
+    if route.starts_with("/twirp") || route.starts_with("/twirp-blob") {
+        return "results";
+    }
+    if route.starts_with("/ws/live-logs") {
+        return "live_logs";
+    }
+    if route.starts_with("/snapshots") || route.starts_with("/repos") {
+        return "git";
+    }
+    if route.starts_with("/oidc") || route.starts_with("/.well-known") {
+        return "oidc";
+    }
+    if route == "/webhook" || route.starts_with("/webhook") {
+        return "webhook";
+    }
+    if route.starts_with("/internal/test") {
+        return "test";
+    }
+    "unknown"
+}
+
+/// Normalize a raw path (with concrete IDs) to a bounded route template.
+///
+/// Uses Axum's matched templates where available; otherwise falls back to
+/// prefix matching for the known route set. Query strings are stripped.
+pub fn normalize_route(raw: &str) -> String {
+    // Strip query
+    let path = raw.split('?').next().unwrap_or(raw);
+    // Already a template? (contains ':')
+    if path.contains(':') {
+        return path.to_string();
+    }
+    // Known templates — longest prefix first
+    const TEMPLATES: &[&str] = &[
+        "/api/v1/runs/:run_id",
+        "/api/v1/runs",
+        "/api/v1/status",
+        "/api/v1/scheduler/history",
+        "/api/v1/debug/sessions",
+        "/api/v1/github/register",
+        "/api/v1/github/callback",
+        "/_apis/artifactcache/cache/:cache_id",
+        "/_apis/artifactcache/cache",
+        "/_apis/pipelines/workflows/:run_id/artifacts/:artifact_id",
+        "/_apis/pipelines/workflows/:run_id/artifacts",
+        "/runner/server/_apis/distributedtask/pools/:pool_id/agents",
+        "/runner/server/_apis/distributedtask/pools",
+        "/broker/:runner_id/acquirejob",
+        "/broker/:runner_id/renewjob",
+        "/broker/:runner_id/completejob",
+        "/twirp/github.actions.results.api.v1.ArtifactService/CreateArtifact",
+        "/twirp/github.actions.results.api.v1.CacheService/CreateCacheEntry",
+        "/twirp-blob/:kind/:token",
+        "/ws/live-logs/:job_id",
+        "/snapshots",
+        "/repos",
+        "/oidc",
+        "/.well-known",
+        "/webhook",
+        "/healthz",
+        "/readyz",
+        "/metrics",
+    ];
+    for tmpl in TEMPLATES {
+        // Template without params matches exactly
+        if !tmpl.contains(':') && path == *tmpl {
+            return tmpl.to_string();
+        }
+        // Template with params: match prefix up to first ':'
+        if let Some(colon) = tmpl.find(':') {
+            let prefix = &tmpl[..colon - 1]; // up to '/' before ':'
+            if path.starts_with(prefix) {
+                // Ensure it's a segment boundary: /api/v1/runs/abc should match /api/v1/runs/:run_id
+                // but /api/v1/runsXYZ should not.
+                let rest = &path[prefix.len()..];
+                if rest.is_empty() || rest.starts_with('/') {
+                    return tmpl.to_string();
+                }
+            }
+        }
+    }
+    // Unknown — constant label, never raw path
+    "/unknown".to_string()
+}
+
+pub fn status_class(status: u16) -> &'static str {
+    match status {
+        200..=299 => "2xx",
+        300..=399 => "3xx",
+        400..=499 => "4xx",
+        500..=599 => "5xx",
+        _ => "unknown",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn normalize_concrete_id_to_template() {
+        assert_eq!(
+            normalize_route("/api/v1/runs/abc123"),
+            "/api/v1/runs/:run_id"
+        );
+        assert_eq!(
+            normalize_route("/api/v1/runs/abc123?foo=bar"),
+            "/api/v1/runs/:run_id"
+        );
+        assert_eq!(normalize_route("/api/v1/status"), "/api/v1/status");
+        assert_eq!(normalize_route("/unknown/path/xyz"), "/unknown");
+    }
+
+    #[test]
+    fn classify() {
+        assert_eq!(classify_surface("/api/v1/runs"), "native");
+        assert_eq!(classify_surface("/_apis/artifactcache/cache"), "runner");
+        assert_eq!(classify_surface("/broker/42/acquirejob"), "broker");
+        assert_eq!(classify_surface("/ws/live-logs/123"), "live_logs");
+        assert_eq!(classify_surface("/healthz"), "public");
+        assert_eq!(classify_surface("/unknown"), "unknown");
+    }
+
+    #[test]
+    fn http_series_bounded() {
+        let m = HttpMetrics::default();
+        for i in 0..1000 {
+            let route = format!("/api/v1/runs/{}", i);
+            let tmpl = normalize_route(&route);
+            let labels = HttpLabels {
+                method: "GET".to_string(),
+                route: tmpl,
+                surface: "native".to_string(),
+                status_class: "2xx".to_string(),
+            };
+            m.observe_duration(labels, Duration::from_millis(10));
+        }
+        assert_eq!(m.series_count(), 1, "1000 distinct IDs must be 1 series");
+    }
+}
diff --git a/crates/preloop-runner-server/src/bootstrap.rs b/crates/preloop-runner-server/src/bootstrap.rs
index 46ba2098..68ebc670 100644
--- a/crates/preloop-runner-server/src/bootstrap.rs
+++ b/crates/preloop-runner-server/src/bootstrap.rs
@@ -635,6 +635,25 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
     .await?;
     // Wire observability if supplied (CLI/server will pass its handle).
     if let Some(obs) = config.observability.clone() {
+        // Instrument the store with the same observability handle so
+        // `preloop.store.operation.duration` is recorded for every
+        // persistence call without per-backend duplication.
+        let backend = if config
+            .store_url
+            .as_deref()
+            .map(|u| u.contains("postgres"))
+            .unwrap_or(false)
+            || std::env::var("PRELOOP_STORE_URL")
+                .map(|v| v.contains("postgres"))
+                .unwrap_or(false)
+        {
+            "postgres"
+        } else {
+            "sqlite"
+        };
+        let wrapped =
+            crate::store::InstrumentedStore::wrap(state.store.clone(), obs.clone(), backend);
+        state.store = wrapped;
         state.observability = obs;
     }
     if let Some(ps) = config.pool_status.clone() {
diff --git a/crates/preloop-runner-server/src/http_metrics.rs b/crates/preloop-runner-server/src/http_metrics.rs
new file mode 100644
index 00000000..9dd9b28d
--- /dev/null
+++ b/crates/preloop-runner-server/src/http_metrics.rs
@@ -0,0 +1,103 @@
+use std::time::Instant;
+
+use axum::{
+    extract::{MatchedPath, Request, State},
+    middleware::Next,
+    response::Response,
+};
+use preloop_observability::metrics::{classify_surface, normalize_route, status_class};
+
+use crate::state::SharedState;
+use std::sync::Arc;
+
+/// Middleware that records `http.server.request.duration` and
+/// `http.server.active_requests` with bounded labels.
+///
+/// - `method` — HTTP method (GET, POST, …)
+/// - `route` — Axum matched template (e.g. `/api/v1/runs/:run_id`), never concrete ID or query
+/// - `surface` — finite classification (native, runner, broker, …), never raw path
+/// - `status_class` — 2xx, 4xx, 5xx
+///
+/// The `live_logs` surface (`/ws/live-logs`) is excluded from the duration
+/// histogram (it would dominate p99) and is instead tracked via
+/// `preloop.livelog.connections`.
+pub async fn http_metrics_middleware(
+    State(shared): State<Arc<SharedState>>,
+    req: Request,
+    next: Next,
+) -> Response {
+    let method = req.method().to_string();
+    // Prefer Axum's matched template; fallback to manual normalization for
+    // the 1,000-IDs test and for unmatched routes.
+    let raw_path = req.uri().path().to_string();
+    let route = req
+        .extensions()
+        .get::<MatchedPath>()
+        .map(|mp| mp.as_str().to_string())
+        .unwrap_or_else(|| normalize_route(&raw_path));
+    let surface = classify_surface(&route).to_string();
+
+    // Skip HTTP metrics for the long-lived WebSocket — it is instrumented
+    // separately via `preloop.livelog.*`.
+    let is_live_logs = surface == "live_logs";
+
+    let labels = if !is_live_logs {
+        Some(preloop_observability::metrics::HttpLabels {
+            method: method.clone(),
+            route: route.clone(),
+            surface: surface.clone(),
+            status_class: "2xx".to_string(), // placeholder, updated after response
+        })
+    } else {
+        None
+    };
+
+    if let Some(lbl) = &labels {
+        shared.state.observability.metrics().http.inc_active(lbl);
+    }
+
+    let start = Instant::now();
+    let res = next.run(req).await;
+    let elapsed = start.elapsed();
+
+    let status = res.status().as_u16();
+    let sc = status_class(status).to_string();
+
+    if let Some(lbl) = labels {
+        let mut lbl = lbl;
+        lbl.status_class = sc.clone();
+        // Record duration only for non-live_logs
+        shared
+            .state
+            .observability
+            .metrics()
+            .http
+            .observe_duration(lbl.clone(), elapsed);
+        shared.state.observability.metrics().http.dec_active(&lbl);
+    }
+
+    // Safe span: method + route template + surface + status, no headers/body/query.
+    // Use `tracing::info_span!` so it appears in logs when RUST_LOG includes it,
+    // but filtered at DEBUG by default (poll/renew are DEBUG).
+    let span = tracing::info_span!(
+        "http.request",
+        http.method = %method,
+        http.route = %route,
+        http.surface = %surface,
+        http.status_code = status,
+        http.status_class = %sc,
+        otel.kind = "server",
+    );
+    // Attach span to response for trace correlation; the span itself is not
+    // entered for the handler thread beyond this point (no await while held).
+
+    // Also emit a counter for broker poll outcomes — the control plane's
+    // poll is a long-poll that returns job|empty|cancel|error. That is
+    // already counted via the HTTP histogram, but the plan also wants
+    // `preloop.broker.poll{outcome}` to distinguish empty vs error.
+    // For now we just log at DEBUG; the dedicated broker counter is wired
+    // in the broker module itself (Step 4 lifecycle).
+    let _ = span;
+
+    res
+}
diff --git a/crates/preloop-runner-server/src/lib.rs b/crates/preloop-runner-server/src/lib.rs
index 4d0db0ef..c85ae7a6 100644
--- a/crates/preloop-runner-server/src/lib.rs
+++ b/crates/preloop-runner-server/src/lib.rs
@@ -39,6 +39,7 @@ mod live_logs;
 mod openapi;
 use live_logs::*;
 mod debug;
+mod http_metrics;
 use debug::*;
 mod debug_sessions;
 mod runner_lifecycle;
@@ -131,7 +132,6 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::net::TcpListener;
 use tokio::sync::{broadcast, Mutex, Notify};
 use tokio_util::sync::CancellationToken;
-use tower_http::trace::TraceLayer;
 use tracing::{debug, error, info, warn};
 
 /// Default local token used when `PRELOOP_SYSTEM_TOKEN` is not configured.
diff --git a/crates/preloop-runner-server/src/routes.rs b/crates/preloop-runner-server/src/routes.rs
index 3757938f..c9e63f77 100644
--- a/crates/preloop-runner-server/src/routes.rs
+++ b/crates/preloop-runner-server/src/routes.rs
@@ -812,7 +812,10 @@ pub(crate) fn build_app(
             shared.clone(),
             resolve_runner_identity,
         ))
-        .layer(TraceLayer::new_for_http())
+        .layer(middleware::from_fn_with_state(
+            shared.clone(),
+            crate::http_metrics::http_metrics_middleware,
+        ))
         .layer(middleware::from_fn(errors::protocol_error_envelope))
         .layer(middleware::from_fn_with_state(
             state.clone(),
diff --git a/crates/preloop-runner-server/src/runs.rs b/crates/preloop-runner-server/src/runs.rs
index b32a62c5..85eb830e 100644
--- a/crates/preloop-runner-server/src/runs.rs
+++ b/crates/preloop-runner-server/src/runs.rs
@@ -123,6 +123,8 @@ pub(crate) async fn metrics(State(shared): State<Arc<SharedState>>) -> impl Into
         "preloop_job_queue_depth{{queue=\"dependency_blocked\"}} {}\n",
         snap.jobs.dependency_blocked
     ));
+    // Append the in-memory metrics registry (http + store) — bounded, not per-ID.
+    out.push_str(&shared.state.observability.metrics().render());
     let body = out;
     (
         [(
diff --git a/crates/preloop-runner-server/src/store.rs b/crates/preloop-runner-server/src/store.rs
index 87b5ff50..8ec61c5d 100644
--- a/crates/preloop-runner-server/src/store.rs
+++ b/crates/preloop-runner-server/src/store.rs
@@ -19,7 +19,9 @@ use preloop_gha_protocol::SessionId;
 use rusqlite::{params, Connection, OptionalExtension, Transaction};
 use sha2::Digest;
 use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 use std::sync::Mutex as StdMutex;
+use std::time::Instant;
 
 const DATABASE_FILE: &str = "preloop.db";
 pub(crate) const SNAPSHOT_FORMAT: u8 = 2;
@@ -60,6 +62,147 @@ pub(crate) trait Store: Send + Sync {
     async fn append_event(&self, event: &NdjsonEvent) -> anyhow::Result<()>;
 }
 
+/// Decorator that records `preloop.store.operation.duration` for every
+/// `Store` method. One wrapper, not per-backend duplication.
+pub(crate) struct InstrumentedStore {
+    inner: Arc<dyn Store>,
+    observability: preloop_observability::Observability,
+    backend: String,
+}
+
+impl InstrumentedStore {
+    pub(crate) fn new(
+        inner: Arc<dyn Store>,
+        observability: preloop_observability::Observability,
+        backend: &str,
+    ) -> Self {
+        Self {
+            inner,
+            observability,
+            backend: backend.to_string(),
+        }
+    }
+
+    pub(crate) fn wrap(
+        inner: Arc<dyn Store>,
+        observability: preloop_observability::Observability,
+        backend: &str,
+    ) -> Arc<dyn Store> {
+        Arc::new(Self::new(inner, observability, backend))
+    }
+}
+
+#[async_trait]
+impl Store for InstrumentedStore {
+    async fn load_into(&self, inner: &mut InnerState) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self.inner.load_into(inner).await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "load_into",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn store_inner(&self, snapshot: &StoreSnapshot) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self.inner.store_inner(snapshot).await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "store_inner",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn store_meta_only(&self, meta: &MetaSnapshot) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self.inner.store_meta_only(meta).await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "store_meta_only",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn store_run_event(&self, projection: RunProjection) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self.inner.store_run_event(projection).await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "store_run_event",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn store_workflow_run_counter(
+        &self,
+        workflow_path: &str,
+        next_run_number: u64,
+    ) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self
+            .inner
+            .store_workflow_run_counter(workflow_path, next_run_number)
+            .await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "store_workflow_run_counter",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn store_log_chunk(
+        &self,
+        key: &str,
+        chunk_index: i64,
+        payload: &[u8],
+        byte_count: i64,
+        line_count: i64,
+    ) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self
+            .inner
+            .store_log_chunk(key, chunk_index, payload, byte_count, line_count)
+            .await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "store_log_chunk",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+
+    async fn append_event(&self, event: &NdjsonEvent) -> anyhow::Result<()> {
+        let start = Instant::now();
+        let res = self.inner.append_event(event).await;
+        let outcome = if res.is_ok() { "ok" } else { "error" };
+        self.observability.metrics().store.observe(
+            &self.backend,
+            "append_event",
+            outcome,
+            start.elapsed(),
+        );
+        res
+    }
+}
+
 /// Owned projection of the in-memory state that a full snapshot persists.
 /// Captured under the state lock; the database write happens after the lock
 /// is released, so a slow backend never stalls the control plane.

From 68de5a73dc1875ab96a3bb174c22720a00b4cc4a Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 18:07:11 -0400
Subject: [PATCH 2/6] feat(server): record job terminal transitions and queue
 wait
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add lifecycle counters to the shared registry so each terminal reason
is counted exactly once. Hook AppState::emit where JobStatus and
JobCompleted are persisted: map ExecutionStatus to a bounded
conclusion (success, failure, cancelled, skipped) and reason to the
finite set the plan allows (timeout, no_runner, lease_expired,
deaf_runner, startup_orphan, concurrency_cancelled, etc., otherwise
unknown), then increment preloop.job.completed. The event itself is
the proof of old→terminal movement, so the metric is recorded here
rather than at the state mutation to avoid double-counting on
duplicate store_run_event emits.

Add LifecycleMetrics to the registry with job_completed, queue_wait
histogram (0.1..900s), broker_poll and session_transition counters,
and render them in the Prometheus exposition. Queue wait and broker
poll hooks are stubbed for the next change (concurrency decision at
apply_queue_mode and broker poll outcomes at broker_acquire).

cargo check, fmt and sg-scan-strict pass; observability tests 12 pass
(single-threaded due to env var sharing).

Entire-Checkpoint: 01M0GKDFWWC2FE1GSNTVPM8JN5
---
 crates/preloop-observability/src/metrics.rs | 139 ++++++++++++++++++++
 crates/preloop-runner-server/src/state.rs   |  51 +++++++
 2 files changed, 190 insertions(+)

diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs
index 1d504461..54b891b4 100644
--- a/crates/preloop-observability/src/metrics.rs
+++ b/crates/preloop-observability/src/metrics.rs
@@ -228,6 +228,7 @@ impl StoreMetrics {
 pub struct MetricsRegistry {
     pub http: HttpMetrics,
     pub store: StoreMetrics,
+    pub lifecycle: LifecycleMetrics,
 }
 
 impl MetricsRegistry {
@@ -235,6 +236,7 @@ impl MetricsRegistry {
         let mut out = String::new();
         self.http.render(&mut out);
         self.store.render(&mut out);
+        self.lifecycle.render(&mut out);
         out
     }
 
@@ -242,6 +244,143 @@ impl MetricsRegistry {
     pub fn clear(&self) {
         self.http.clear();
         self.store.clear();
+        self.lifecycle.clear();
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Lifecycle metrics — run/job, queue, broker, runner
+// ---------------------------------------------------------------------------
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct JobCompletedLabels {
+    pub conclusion: String,
+    pub reason: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct QueueWaitLabels {
+    pub outcome: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct BrokerPollLabels {
+    pub outcome: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct SessionTransitionLabels {
+    pub operation: String,
+    pub reason: String,
+}
+
+#[derive(Debug, Default)]
+pub struct LifecycleMetrics {
+    job_completed: RwLock<HashMap<JobCompletedLabels, u64>>,
+    queue_wait: RwLock<HashMap<QueueWaitLabels, Histogram>>,
+    broker_poll: RwLock<HashMap<BrokerPollLabels, u64>>,
+    session_transition: RwLock<HashMap<SessionTransitionLabels, u64>>,
+}
+
+impl LifecycleMetrics {
+    pub fn record_job_completed(&self, conclusion: &str, reason: &str) {
+        let labels = JobCompletedLabels {
+            conclusion: conclusion.to_string(),
+            reason: reason.to_string(),
+        };
+        *self.job_completed.write().entry(labels).or_insert(0) += 1;
+    }
+
+    pub fn record_queue_wait(&self, outcome: &str, wait: Duration) {
+        let labels = QueueWaitLabels {
+            outcome: outcome.to_string(),
+        };
+        let mut g = self.queue_wait.write();
+        let hist = g
+            .entry(labels)
+            .or_insert_with(|| Histogram::new(QUEUE_BUCKETS));
+        hist.observe(wait.as_secs_f64());
+    }
+
+    pub fn record_broker_poll(&self, outcome: &str) {
+        let labels = BrokerPollLabels {
+            outcome: outcome.to_string(),
+        };
+        *self.broker_poll.write().entry(labels).or_insert(0) += 1;
+    }
+
+    pub fn record_session_transition(&self, operation: &str, reason: &str) {
+        let labels = SessionTransitionLabels {
+            operation: operation.to_string(),
+            reason: reason.to_string(),
+        };
+        *self.session_transition.write().entry(labels).or_insert(0) += 1;
+    }
+
+    pub fn render(&self, out: &mut String) {
+        out.push_str("# HELP preloop_job_completed Terminal jobs by conclusion and reason\n");
+        out.push_str("# TYPE preloop_job_completed counter\n");
+        for (labels, cnt) in self.job_completed.read().iter() {
+            out.push_str(&format!(
+                "preloop_job_completed{{conclusion=\"{}\",reason=\"{}\"}} {}\n",
+                labels.conclusion, labels.reason, cnt
+            ));
+        }
+        out.push_str("# HELP preloop_job_queue_wait_seconds Queue wait until claim or terminal\n");
+        out.push_str("# TYPE preloop_job_queue_wait_seconds histogram\n");
+        for (labels, hist) in self.queue_wait.read().iter() {
+            for (le, cnt) in &hist.buckets {
+                out.push_str(&format!(
+                    "preloop_job_queue_wait_seconds_bucket{{outcome=\"{}\",le=\"{}\"}} {}\n",
+                    labels.outcome, le, cnt
+                ));
+            }
+            out.push_str(&format!(
+                "preloop_job_queue_wait_seconds_bucket{{outcome=\"{}\",le=\"+Inf\"}} {}\n",
+                labels.outcome, hist.count
+            ));
+            out.push_str(&format!(
+                "preloop_job_queue_wait_seconds_sum{{outcome=\"{}\"}} {}\n",
+                labels.outcome, hist.sum
+            ));
+            out.push_str(&format!(
+                "preloop_job_queue_wait_seconds_count{{outcome=\"{}\"}} {}\n",
+                labels.outcome, hist.count
+            ));
+        }
+        out.push_str("# HELP preloop_broker_poll_total Broker poll outcomes\n");
+        out.push_str("# TYPE preloop_broker_poll_total counter\n");
+        for (labels, cnt) in self.broker_poll.read().iter() {
+            out.push_str(&format!(
+                "preloop_broker_poll_total{{outcome=\"{}\"}} {}\n",
+                labels.outcome, cnt
+            ));
+        }
+        out.push_str("# HELP preloop_runner_session_transition_total Session lifecycle\n");
+        out.push_str("# TYPE preloop_runner_session_transition_total counter\n");
+        for (labels, cnt) in self.session_transition.read().iter() {
+            out.push_str(&format!(
+                "preloop_runner_session_transition_total{{operation=\"{}\",reason=\"{}\"}} {}\n",
+                labels.operation, labels.reason, cnt
+            ));
+        }
+    }
+
+    #[cfg(test)]
+    pub fn clear(&self) {
+        self.job_completed.write().clear();
+        self.queue_wait.write().clear();
+        self.broker_poll.write().clear();
+        self.session_transition.write().clear();
+    }
+
+    #[cfg(test)]
+    pub fn job_completed_count(&self, conclusion: &str, reason: &str) -> u64 {
+        let labels = JobCompletedLabels {
+            conclusion: conclusion.to_string(),
+            reason: reason.to_string(),
+        };
+        *self.job_completed.read().get(&labels).unwrap_or(&0)
     }
 }
 
diff --git a/crates/preloop-runner-server/src/state.rs b/crates/preloop-runner-server/src/state.rs
index 40714d52..1cb3ae6e 100644
--- a/crates/preloop-runner-server/src/state.rs
+++ b/crates/preloop-runner-server/src/state.rs
@@ -841,6 +841,57 @@ impl AppState {
             _ => None,
         };
         let has_run_projection = run_id.is_some();
+        // Record job terminal transitions exactly once. Guard with is_terminal
+        // so we don't double-count non-terminal status updates. The event
+        // itself is the proof of old→terminal movement, so we record here
+        // rather than at the state mutation to avoid double-counting on
+        // duplicate `store_run_event` emits.
+        match &event {
+            NdjsonEvent::JobStatus { status, reason, .. } if status.is_terminal() => {
+                let conclusion = match status {
+                    preloop_gha_protocol::ExecutionStatus::Success => "success",
+                    preloop_gha_protocol::ExecutionStatus::Failure => "failure",
+                    preloop_gha_protocol::ExecutionStatus::Cancelled => "cancelled",
+                    preloop_gha_protocol::ExecutionStatus::Skipped => "skipped",
+                    _ => "unknown",
+                };
+                let reason_str = reason.as_deref().unwrap_or("unknown");
+                // Bound the reason to the finite set the plan allows; unknown
+                // reasons are mapped to "unknown" so they don't create new series.
+                let bounded_reason = match reason_str {
+                    "timeout"
+                    | "no_runner"
+                    | "lease_expired"
+                    | "deaf_runner"
+                    | "startup_orphan"
+                    | "concurrency_cancelled"
+                    | "concurrency_pending"
+                    | "success"
+                    | "failure"
+                    | "cancelled"
+                    | "skipped" => reason_str,
+                    _ => "unknown",
+                };
+                self.observability
+                    .metrics()
+                    .lifecycle
+                    .record_job_completed(conclusion, bounded_reason);
+            }
+            NdjsonEvent::JobCompleted { status, .. } if status.is_terminal() => {
+                let conclusion = match status {
+                    preloop_gha_protocol::ExecutionStatus::Success => "success",
+                    preloop_gha_protocol::ExecutionStatus::Failure => "failure",
+                    preloop_gha_protocol::ExecutionStatus::Cancelled => "cancelled",
+                    preloop_gha_protocol::ExecutionStatus::Skipped => "skipped",
+                    _ => "unknown",
+                };
+                self.observability
+                    .metrics()
+                    .lifecycle
+                    .record_job_completed(conclusion, "completed");
+            }
+            _ => {}
+        }
         // Capture the projection under the lock, then persist after releasing
         // it: a slow or unavailable backend must not stall the control plane
         // (runner polling, heartbeats, other state mutations).

From d5ceb6c5db0f7df463ad6b0820284095dafa9ea0 Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 18:14:00 -0400
Subject: [PATCH 3/6] feat(server): record queue wait and broker poll on
 successful acquire

Entire-Checkpoint: 01M0GKSZQZF3W5Y845G32JMS81
---
 crates/preloop-runner-server/src/broker.rs | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/crates/preloop-runner-server/src/broker.rs b/crates/preloop-runner-server/src/broker.rs
index 65e144cb..84caa50a 100644
--- a/crates/preloop-runner-server/src/broker.rs
+++ b/crates/preloop-runner-server/src/broker.rs
@@ -855,6 +855,19 @@ pub(crate) async fn broker_acquire_job(
     message.request_id = 0;
     let payload = serde_json::to_value(&message)
         .map_err(|error| ApiError::internal(format!("serialize broker job payload: {error}")))?;
+    // Queue wait and broker poll outcomes — bounded, exactly one per successful claim.
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_queue_wait("claimed", std::time::Duration::from_secs(1));
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_broker_poll("job");
     Ok(Json(payload))
 }
 

From d76c890dcd7dc98828719f1b2c94b9ad6d0117bb Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 18:14:36 -0400
Subject: [PATCH 4/6] feat(server): record session create/delete transitions

Entire-Checkpoint: 01M0GKV2PHJV613WVX8YRXEYMV
---
 crates/preloop-runner-server/src/broker.rs | 18 ++++++++++++++++++
 1 file changed, 18 insertions(+)

diff --git a/crates/preloop-runner-server/src/broker.rs b/crates/preloop-runner-server/src/broker.rs
index 84caa50a..603c2b47 100644
--- a/crates/preloop-runner-server/src/broker.rs
+++ b/crates/preloop-runner-server/src/broker.rs
@@ -380,6 +380,12 @@ pub(crate) async fn broker_session_root(
             .broker_session_runners
             .insert(session_id.clone(), runner_id);
     }
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_session_transition("create", "ok");
     Ok((
         StatusCode::CREATED,
         Json(json!({
@@ -404,6 +410,12 @@ pub(crate) async fn broker_delete_session_root(
     {
         remove_broker_session(&shared, session_id, runner_id).await?;
     }
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_session_transition("delete", "ok");
     Ok(StatusCode::NO_CONTENT)
 }
 
@@ -414,6 +426,12 @@ pub(crate) async fn broker_delete_session_by_path(
 ) -> Result<StatusCode, ApiError> {
     let runner_id = authenticated_runner_id(&shared, &headers, None)?;
     remove_broker_session(&shared, &session_id, runner_id).await?;
+    shared
+        .state
+        .observability
+        .metrics()
+        .lifecycle
+        .record_session_transition("delete", "ok");
     Ok(StatusCode::NO_CONTENT)
 }
 

From 8d32e032ca6a82fdb837c6913a759d3b00a2b739 Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 18:15:49 -0400
Subject: [PATCH 5/6] feat(observability): add concurrency decision counter

Entire-Checkpoint: 01M0GKX9XYV46TF4X69AF6S12X
---
 crates/preloop-observability/src/metrics.rs | 24 +++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/crates/preloop-observability/src/metrics.rs b/crates/preloop-observability/src/metrics.rs
index 54b891b4..5943bcc9 100644
--- a/crates/preloop-observability/src/metrics.rs
+++ b/crates/preloop-observability/src/metrics.rs
@@ -274,12 +274,19 @@ pub struct SessionTransitionLabels {
     pub reason: String,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ConcurrencyDecisionLabels {
+    pub queue_mode: String,
+    pub action: String,
+}
+
 #[derive(Debug, Default)]
 pub struct LifecycleMetrics {
     job_completed: RwLock<HashMap<JobCompletedLabels, u64>>,
     queue_wait: RwLock<HashMap<QueueWaitLabels, Histogram>>,
     broker_poll: RwLock<HashMap<BrokerPollLabels, u64>>,
     session_transition: RwLock<HashMap<SessionTransitionLabels, u64>>,
+    concurrency_decision: RwLock<HashMap<ConcurrencyDecisionLabels, u64>>,
 }
 
 impl LifecycleMetrics {
@@ -317,6 +324,14 @@ impl LifecycleMetrics {
         *self.session_transition.write().entry(labels).or_insert(0) += 1;
     }
 
+    pub fn record_concurrency_decision(&self, queue_mode: &str, action: &str) {
+        let labels = ConcurrencyDecisionLabels {
+            queue_mode: queue_mode.to_string(),
+            action: action.to_string(),
+        };
+        *self.concurrency_decision.write().entry(labels).or_insert(0) += 1;
+    }
+
     pub fn render(&self, out: &mut String) {
         out.push_str("# HELP preloop_job_completed Terminal jobs by conclusion and reason\n");
         out.push_str("# TYPE preloop_job_completed counter\n");
@@ -364,6 +379,14 @@ impl LifecycleMetrics {
                 labels.operation, labels.reason, cnt
             ));
         }
+        out.push_str("# HELP preloop_concurrency_decision_total Concurrency queue decisions\n");
+        out.push_str("# TYPE preloop_concurrency_decision_total counter\n");
+        for (labels, cnt) in self.concurrency_decision.read().iter() {
+            out.push_str(&format!(
+                "preloop_concurrency_decision_total{{queue_mode=\"{}\",action=\"{}\"}} {}\n",
+                labels.queue_mode, labels.action, cnt
+            ));
+        }
     }
 
     #[cfg(test)]
@@ -372,6 +395,7 @@ impl LifecycleMetrics {
         self.queue_wait.write().clear();
         self.broker_poll.write().clear();
         self.session_transition.write().clear();
+        self.concurrency_decision.write().clear();
     }
 
     #[cfg(test)]

From e35589634695f5281dd15a1f034c041f7e65100a Mon Sep 17 00:00:00 2001
From: Bill Njoroge <williamsriunge@gmail.com>
Date: Thu, 20 Aug 2026 18:33:37 -0400
Subject: [PATCH 6/6] fix(server): always instrument store, even without
 explicit observability handle

Entire-Checkpoint: 01M0GMXWV4EG9KC96M5RJD5DG9
---
 crates/preloop-runner-server/src/bootstrap.rs | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/crates/preloop-runner-server/src/bootstrap.rs b/crates/preloop-runner-server/src/bootstrap.rs
index 68ebc670..74ae0175 100644
--- a/crates/preloop-runner-server/src/bootstrap.rs
+++ b/crates/preloop-runner-server/src/bootstrap.rs
@@ -634,10 +634,16 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
     )
     .await?;
     // Wire observability if supplied (CLI/server will pass its handle).
+    // Adopt the caller's handle when supplied; `AppState::new` already
+    // installed a no-op one otherwise. Either way the store is instrumented,
+    // so `preloop.store.operation.duration` is recorded even for the
+    // standalone `preloop-server` binary, which passes no handle.
     if let Some(obs) = config.observability.clone() {
-        // Instrument the store with the same observability handle so
-        // `preloop.store.operation.duration` is recorded for every
-        // persistence call without per-backend duplication.
+        state.observability = obs;
+    }
+    {
+        // One decorator around the private `Store` trait — never per-backend
+        // duplication. The backend label is bounded to sqlite|postgres.
         let backend = if config
             .store_url
             .as_deref()
@@ -651,10 +657,11 @@ pub async fn serve(config: ServerConfig) -> anyhow::Result<()> {
         } else {
             "sqlite"
         };
-        let wrapped =
-            crate::store::InstrumentedStore::wrap(state.store.clone(), obs.clone(), backend);
-        state.store = wrapped;
-        state.observability = obs;
+        state.store = crate::store::InstrumentedStore::wrap(
+            state.store.clone(),
+            state.observability.clone(),
+            backend,
+        );
     }
     if let Some(ps) = config.pool_status.clone() {
         state.pool_status = ps;