fix(libsy): in-flight algo count metric

crates/libsy-llm-client/tests/observability.rs

@@ -311,6 +311,34 @@ fn f64_histogram_sum_ms(
     })
 }
 
+/// Most recent value of an `i64` up-down counter for the given attribute set.
+///
+/// Unlike the counter and histogram helpers, this takes the value from the last
+/// snapshot that carries the metric rather than the max across snapshots: an
+/// up-down counter falls as well as rises, so a max would report a past peak and
+/// never observe the return to zero.
+fn i64_up_down_counter_value(
+    snapshots: &[ResourceMetrics],
+    name: &str,
+    wanted: &[(&str, &str)],
+) -> Option<i64> {
+    snapshots
+        .iter()
+        .flat_map(|snapshot| snapshot.scope_metrics())
+        .filter(|scope| scope.scope().name() == "switchyard")
+        .flat_map(|scope| scope.metrics())
+        .filter(|metric| metric.name() == name)
+        .filter_map(|metric| match metric.data() {
+            AggregatedMetrics::I64(MetricData::Sum(sum)) => sum
+                .data_points()
+                .filter(|point| attributes_match(point.attributes(), wanted))
+                .map(|point| point.value())
+                .last(),
+            _ => None,
+        })
+        .last()
+}
+
 /// Latest value of a `u64` observable gauge.
 fn u64_gauge_value(snapshots: &[ResourceMetrics], name: &str) -> Option<u64> {
     latest_metric_value(snapshots, name, |data| match data {
@@ -1436,3 +1464,101 @@ async fn classifier_fail_open_records_each_failure_stage() -> switchyard_libsy::
     }
     Ok(())
 }
+
+#[tokio::test]
+async fn in_flight_gauge_reads_a_run_parked_on_an_unanswered_routing_call()
+-> switchyard_libsy::Result<()> {
+    let _guard = serialize_test().lock().await;
+    let (_, exporter, provider, _, _) = telemetry();
+    const ALGO: &str = "obs-in-flight-algo";
+    const MODEL: &str = "obs-in-flight-model";
+    let algorithm = Arc::new(RoutingCallAlgo {
+        name: ALGO.to_string(),
+        target: MODEL.into(),
+    });
+    let stream = algorithm.run_stream(request_with_metadata("obs-session-if", "obs-corr-if"));
+    tokio::pin!(stream);
+    let attributes = [("algorithm", ALGO)];
+
+    // Take the offloaded call and hold it without responding. This is the shape of a
+    // stalled classifier: the run has started and cannot proceed until the call returns.
+    let Some(Ok(Step::CallModel(call))) = stream.next().await else {
+        return Err(test_error("expected an offloaded routing call"));
+    };
+
+    let snapshots = flushed_metrics(exporter, provider);
+    assert_eq!(
+        i64_up_down_counter_value(&snapshots, "switchyard.algorithms_in_flight", &attributes),
+        Some(1),
+        "a run waiting on an unanswered routing call must read as in flight"
+    );
+    // The gap this gauge fills: every other run metric is recorded on resolution, so
+    // while the call is outstanding they say nothing at all.
+    assert_eq!(
+        u64_counter_value(&snapshots, "switchyard.runs", &attributes),
+        None,
+        "the run counter must not report a run that has not resolved"
+    );
+
+    call.respond(Ok(Response {
+        llm_response: LlmResponse::Agg(text_response(Some(MODEL.to_string()), "answer")),
+        metadata: None,
+    }))?;
+    while stream.next().await.is_some() {}
+
+    let snapshots = flushed_metrics(exporter, provider);
+    assert_eq!(
+        i64_up_down_counter_value(&snapshots, "switchyard.algorithms_in_flight", &attributes),
+        Some(0),
+        "the gauge must fall back to zero once the run resolves"
+    );
+    Ok(())
+}
+
+#[tokio::test]
+async fn in_flight_gauge_clears_when_a_run_is_abandoned() -> switchyard_libsy::Result<()> {
+    let _guard = serialize_test().lock().await;
+    let (_, exporter, provider, _, _) = telemetry();
+    const ALGO: &str = "obs-abandoned-algo";
+    const MODEL: &str = "obs-abandoned-model";
+    let algorithm = Arc::new(RoutingCallAlgo {
+        name: ALGO.to_string(),
+        target: MODEL.into(),
+    });
+    let attributes = [("algorithm", ALGO)];
+
+    // Drop the step stream while the routing call is still outstanding, the way a
+    // disconnected client abandons a run. The run task is aborted mid-await and never
+    // reaches the code that follows it, so only a drop can return the count.
+    {
+        let stream = algorithm.run_stream(request_with_metadata("obs-session-ab", "obs-corr-ab"));
+        tokio::pin!(stream);
+        let Some(Ok(Step::CallModel(_call))) = stream.next().await else {
+            return Err(test_error("expected an offloaded routing call"));
+        };
+        assert_eq!(
+            i64_up_down_counter_value(
+                &flushed_metrics(exporter, provider),
+                "switchyard.algorithms_in_flight",
+                &attributes,
+            ),
+            Some(1),
+        );
+    }
+
+    // Let the runtime finish aborting the task it was told to drop.
+    for _ in 0..10 {
+        tokio::task::yield_now().await;
+    }
+
+    assert_eq!(
+        i64_up_down_counter_value(
+            &flushed_metrics(exporter, provider),
+            "switchyard.algorithms_in_flight",
+            &attributes,
+        ),
+        Some(0),
+        "an abandoned run must not strand the gauge above zero"
+    );
+    Ok(())
+}

crates/libsy/src/observability.rs

@@ -110,13 +110,45 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span {
     span
 }
 
+/// Holds `switchyard.algorithms_in_flight` up by one for as long as it lives.
+struct InFlightRun {
+    algorithm: String,
+}
+
+impl InFlightRun {
+    fn enter(algorithm: &str) -> Self {
+        record_algorithms_in_flight(algorithm, 1);
+        Self {
+            algorithm: algorithm.to_string(),
+        }
+    }
+}
+
+impl Drop for InFlightRun {
+    fn drop(&mut self) {
+        record_algorithms_in_flight(&self.algorithm, -1);
+    }
+}
+
+/// Adds `delta` to the count of algorithm runs that have started and not yet
+/// finished.
+fn record_algorithms_in_flight(algorithm: &str, delta: i64) {
+    meter()
+        .i64_up_down_counter("switchyard.algorithms_in_flight")
+        .build()
+        .add(delta, &[KeyValue::new("algorithm", algorithm.to_string())]);
+}
+
 /// Runs one algorithm task to completion, recording the run counter, duration
-/// histogram, span outcome, and failure log when it resolves.
+/// histogram, span outcome, and failure log when it resolves. Counts the run as
+/// in flight for its whole duration.
 /// Executes inside the `libsy.run` span its caller instruments the task with.
 pub(crate) async fn observe_run<T>(
     algorithm: &str,
     run: impl Future<Output = Result<T>>,
 ) -> Result<T> {
+    // Binding, not `let _ =`: the guard must live until the run resolves.
+    let _in_flight = InFlightRun::enter(algorithm);
     let started = Instant::now();
     let result = run.await;
     let duration = started.elapsed();

docs/internal/metrics_reference.md

@@ -59,6 +59,12 @@ Each histogram emits `_bucket`, `_sum`, and `_count` series. Use
 |---|---|---|
 | `switchyard_routing_overhead_ms{algorithm}` | histogram | Total run time minus the time spent in successful routed model calls, with overlapping hedged calls counted once. Includes classifier calls, failed routed attempts, target resolution, and decision publication; runs with no successful routed call are not recorded. Measured across the whole run, so it does not reconcile with `switchyard_run_duration_ms`, which times only the algorithm task. |
 
+## In-flight runs
+
+| Metric | Type | Meaning |
+|---|---|---|
+| `switchyard_algorithms_in_flight{algorithm}` | gauge | Algorithm runs that have started and not yet finished. |
+
 ## Classifier fail-open counter
 
 | Metric | Type | Meaning |
@@ -149,7 +155,7 @@ into label space.
 | `outcome` | Exactly 3: `success`, `retryable_error`, `other_error`. | Outcome counters |
 | `code` | Bounded: the known-code allowlist (`200`, `400`, `401`, `403`, `404`, `408`, `409`, `422`, `429`, `500`, `502`, `503`, `504`), plus `none` and the per-class buckets `1xx`/`2xx`/`3xx`/`4xx`/`5xx`/`other`. About 20 values max. | `switchyard_upstream_attempts_total` |
 | `le` | The configured histogram bucket boundaries. | Histogram buckets |
-| `algorithm` | One stable value per configured algorithm. | Routing-overhead histogram |
+| `algorithm` | One stable value per configured algorithm. | Routing-overhead histogram, in-flight gauge |
 | `tier` | Small enumerated set, optional. | Per-endpoint counters and histograms on algorithms that supply it |
 | `judge_model` | One per configured judge target. | Classifier fail-open counter |
 | `reason` | Exactly 8 fixed error categories. | Classifier fail-open counter |
@@ -161,5 +167,6 @@ into label space.
 | `model="<unknown>"` rows appear | A routed-call observation did not include a selected model. |
 | All counters at 0 after warm-up | Server just started with no traffic, or the scraper is hitting the wrong port. |
 | `switchyard_routing_overhead_ms_count` stuck at `0` | No successful algorithm run has recorded a successful routed model call. |
+| `switchyard_algorithms_in_flight` stuck above zero with no traffic | Runs are parked on an internal routing call that never returns. Check the classifier or judge target's upstream. |
 | `switchyard_classifier_fail_open_total` rising | The judge target is failing or returning a response the classifier cannot parse. Check `judge_model` and `reason`. |
 | `switchyard_client_responses_total{outcome="retryable_error"}` rising | Either the upstream is genuinely flaky, or retries are exhausting; compare client responses with retryable upstream attempts. |