[Rust Frontend] Add external→internal request-id map for abort() (#45137)

Signed-off-by: Sahil Singh <sahiilsiingh37@gmail.com>

rust/Cargo.lock

@@ -5804,6 +5804,7 @@ dependencies = [
  "enum-as-inner",
  "expect-test",
  "futures",
+ "parking_lot",
  "rmp-serde",
  "serde",
  "serde_json",

rust/src/chat/src/lib.rs

@@ -235,6 +235,12 @@ impl ChatLlm {
         Ok(token_ids)
     }
 
+    /// Abort in-flight requests by their external (user-supplied) request ids.
+    pub async fn abort(&self, external_ids: &[String]) -> Result<()> {
+        self.text.abort(external_ids).await?;
+        Ok(())
+    }
+
     /// Shut down the underlying LLM client and its background tasks.
     pub async fn shutdown(self) -> Result<()> {
         self.text.shutdown().await?;

rust/src/engine-core-client/src/client.rs

@@ -490,6 +490,10 @@ impl EngineCoreClient {
             return Ok(());
         }
 
+        // Finalize the consumer streams first, before the engine round-trip.
+        let all_request_ids: Vec<String> = abortable.values().flatten().cloned().collect();
+        self.inner.abort_requests_locally(&all_request_ids);
+
         for (engine_id, request_ids) in abortable {
             self.inner.do_abort_requests(&engine_id, &request_ids).await?;
         }

rust/src/engine-core-client/src/client/imp.rs

@@ -1,5 +1,6 @@
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
 
 use arc_swap::ArcSwapOption;
 use parking_lot::Mutex;
@@ -126,6 +127,20 @@ impl ClientInner {
         self.request_reg.lock().finish_many(request_ids)
     }
 
+    /// Finalize client-initiated aborts by pushing a terminal `Abort` output
+    /// down each request's stream and removing it from the registry. Returns
+    /// the request ids that were still active. See [`RequestRegistry::abort_many`].
+    pub fn abort_requests_locally<'a>(
+        &self,
+        request_ids: impl IntoIterator<Item = &'a String>,
+    ) -> Vec<String> {
+        let timestamp = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_secs_f64())
+            .unwrap_or(0.0);
+        self.request_reg.lock().abort_many(request_ids, timestamp)
+    }
+
     /// Apply one scheduler stats update for the given engine to the local
     /// routing state. Returns `false` if the engine is unknown to the
     /// client.

rust/src/engine-core-client/src/client/state.rs

@@ -9,7 +9,7 @@ use crate::client::stream::EngineCoreStreamOutput;
 use crate::error::{Error, Result};
 use crate::protocol::stats::SchedulerStats;
 use crate::protocol::utility::UtilityOutput;
-use crate::protocol::{EngineCoreEventType, EngineCoreOutput};
+use crate::protocol::{EngineCoreEventType, EngineCoreFinishReason, EngineCoreOutput};
 use crate::transport::ConnectedEngine;
 
 pub type OutputSender = mpsc::UnboundedSender<Result<EngineCoreStreamOutput>>;
@@ -289,6 +289,34 @@ impl RequestRegistry {
             .collect()
     }
 
+    /// Finalize client-initiated aborts: remove each request and push a
+    /// terminal output with `finish_reason = Abort` down its stream before the
+    /// sender drops. Returns the request ids that were still active.
+    pub fn abort_many<'a>(
+        &mut self,
+        request_ids: impl IntoIterator<Item = &'a String>,
+        timestamp: f64,
+    ) -> Vec<String> {
+        let mut aborted = Vec::new();
+        for request_id in request_ids {
+            let Some((sender, engine_id)) = self.remove(request_id) else {
+                continue;
+            };
+            let output = EngineCoreStreamOutput {
+                engine_index: engine_id.engine_index().unwrap_or(0),
+                timestamp,
+                output: EngineCoreOutput {
+                    request_id: request_id.clone(),
+                    finish_reason: Some(EngineCoreFinishReason::Abort),
+                    ..EngineCoreOutput::default()
+                },
+            };
+            let _ = sender.send(Ok(output));
+            aborted.push(request_id.clone());
+        }
+        aborted
+    }
+
     /// Remove one request from the local registry. Returns the tracked entry if
     /// it exists.
     #[must_use]

rust/src/engine-core-client/src/tests/client.rs

@@ -1939,7 +1939,7 @@ async fn multi_engine_abort_is_grouped_and_utility_fans_out_to_all_engines() {
 
     let (shutdown_tx_0, engine_task_0) = spawn_mock_engine_task(
         handshake_address.clone(),
-        b"engine-0".to_vec(),
+        EngineId::from_engine_index(0).into_frame().to_vec(),
         |dealer, push| {
             Box::pin(async move {
                 let utility = recv_engine_message(dealer).await;
@@ -1993,7 +1993,7 @@ async fn multi_engine_abort_is_grouped_and_utility_fans_out_to_all_engines() {
     tokio::time::sleep(Duration::from_millis(50)).await;
     let (shutdown_tx_1, engine_task_1) = spawn_mock_engine_task(
         handshake_address.clone(),
-        b"engine-1".to_vec(),
+        EngineId::from_engine_index(1).into_frame().to_vec(),
         |dealer, push| {
             Box::pin(async move {
                 let utility = recv_engine_message(dealer).await;

rust/src/llm/Cargo.toml

@@ -11,6 +11,7 @@ test-util = []
 easy-ext.workspace = true
 enum-as-inner.workspace = true
 futures.workspace = true
+parking_lot.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

rust/src/llm/src/inflight.rs

@@ -0,0 +1,179 @@
+//! Tracking of the external→internal request-id mapping for in-flight requests.
+//!
+//! When request-id randomization is enabled (the default), [`crate::Llm`]
+//! rewrites the external (user-supplied) request id into a unique internal
+//! engine id before reaching engine-core. Engine-core only ever knows the
+//! internal id, so aborting a request by its external id requires resolving it
+//! back to the internal id(s) first.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Weak};
+
+use parking_lot::Mutex;
+
+/// external id → internal id → number of live guards holding that edge.
+type InflightMap = HashMap<String, HashMap<String, usize>>;
+
+/// Maps external (user-supplied) request ids to the set of live internal engine
+/// request ids they currently expand into.
+///
+/// One external id may map to multiple internal ids: duplicate external ids
+/// submitted concurrently each get their own randomized internal id, and an
+/// abort by the shared external id must reach all of them. Edges are
+/// refcounted: with randomization disabled the same (external, internal) pair
+/// can be tracked by several guards in sequence (e.g. a finished request whose
+/// stream is still held alongside a fresh submission reusing the id), and the
+/// edge must survive until the last guard drops.
+#[derive(Default)]
+pub(crate) struct InflightRequests {
+    map: Arc<Mutex<InflightMap>>,
+}
+
+impl InflightRequests {
+    pub(crate) fn new() -> Self {
+        Self::default()
+    }
+
+    /// Record that `internal` is now an in-flight engine request for the
+    /// `external` request id, returning a guard that removes the edge when the
+    /// request's output stream is dropped (on clean finish or cancellation).
+    pub(crate) fn track(&self, external: String, internal: String) -> RequestGuard {
+        *self
+            .map
+            .lock()
+            .entry(external.clone())
+            .or_default()
+            .entry(internal.clone())
+            .or_insert(0) += 1;
+        RequestGuard {
+            map: Arc::downgrade(&self.map),
+            external,
+            internal,
+        }
+    }
+
+    /// Resolve external request ids to the internal engine ids currently
+    /// in-flight for them. Unknown or already-finished ids contribute nothing.
+    pub(crate) fn resolve(&self, external_ids: &[String]) -> Vec<String> {
+        let map = self.map.lock();
+        external_ids
+            .iter()
+            .filter_map(|external| map.get(external))
+            .flat_map(|internal_ids| internal_ids.keys())
+            .cloned()
+            .collect()
+    }
+
+    #[cfg(test)]
+    fn is_empty(&self) -> bool {
+        self.map.lock().is_empty()
+    }
+}
+
+/// RAII guard that releases one refcount on a single external→internal edge
+/// when dropped, removing the edge once no live guard holds it.
+///
+/// Held by the per-request output stream, so cleanup runs whether the stream
+/// terminates cleanly or is cancelled. A [`Weak`] handle is used so a stream
+/// outliving its owning [`InflightRequests`] does not keep the map alive.
+pub(crate) struct RequestGuard {
+    map: Weak<Mutex<InflightMap>>,
+    external: String,
+    internal: String,
+}
+
+impl Drop for RequestGuard {
+    fn drop(&mut self) {
+        let Some(map) = self.map.upgrade() else {
+            return;
+        };
+        let mut map = map.lock();
+        if let Some(internal_ids) = map.get_mut(&self.external) {
+            if let Some(count) = internal_ids.get_mut(&self.internal) {
+                *count -= 1;
+                if *count == 0 {
+                    internal_ids.remove(&self.internal);
+                }
+            }
+            if internal_ids.is_empty() {
+                map.remove(&self.external);
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn resolves_external_to_internal() {
+        let inflight = InflightRequests::new();
+        let _guard = inflight.track("ext".to_string(), "ext-abc".to_string());
+
+        assert_eq!(
+            inflight.resolve(&["ext".to_string()]),
+            vec!["ext-abc".to_string()]
+        );
+        assert!(inflight.resolve(&["unknown".to_string()]).is_empty());
+    }
+
+    #[test]
+    fn one_external_maps_to_many_internal() {
+        let inflight = InflightRequests::new();
+        let _g1 = inflight.track("dup".to_string(), "dup-1".to_string());
+        let _g2 = inflight.track("dup".to_string(), "dup-2".to_string());
+
+        let mut resolved = inflight.resolve(&["dup".to_string()]);
+        resolved.sort();
+        assert_eq!(resolved, vec!["dup-1".to_string(), "dup-2".to_string()]);
+    }
+
+    #[test]
+    fn dropping_guard_removes_only_its_own_edge_then_cleans_empty_key() {
+        let inflight = InflightRequests::new();
+        let g1 = inflight.track("dup".to_string(), "dup-1".to_string());
+        let g2 = inflight.track("dup".to_string(), "dup-2".to_string());
+
+        drop(g1);
+        assert_eq!(
+            inflight.resolve(&["dup".to_string()]),
+            vec!["dup-2".to_string()]
+        );
+
+        drop(g2);
+        assert!(inflight.resolve(&["dup".to_string()]).is_empty());
+        assert!(
+            inflight.is_empty(),
+            "empty external key must be removed, not left dangling"
+        );
+    }
+
+    #[test]
+    fn identical_edges_are_refcounted_across_guards() {
+        // With request-id randomization disabled, internal == external, so two
+        // tracked requests can share the exact same edge. Dropping one guard
+        // (e.g. a stale stream, or the error path of a rejected duplicate
+        // submission) must not untrack the other still-live request.
+        let inflight = InflightRequests::new();
+        let g1 = inflight.track("x".to_string(), "x".to_string());
+        let g2 = inflight.track("x".to_string(), "x".to_string());
+
+        drop(g1);
+        assert_eq!(inflight.resolve(&["x".to_string()]), vec!["x".to_string()]);
+
+        drop(g2);
+        assert!(inflight.resolve(&["x".to_string()]).is_empty());
+        assert!(inflight.is_empty());
+    }
+
+    #[test]
+    fn guard_drop_is_a_noop_after_inflight_is_gone() {
+        let guard = {
+            let inflight = InflightRequests::new();
+            inflight.track("ext".to_string(), "ext-abc".to_string())
+        };
+        // Dropping the guard after the owning map is gone must not panic.
+        drop(guard);
+    }
+}

rust/src/llm/src/lib.rs

@@ -2,6 +2,7 @@ use tracing::Span;
 use vllm_engine_core_client::EngineCoreClient;
 
 mod error;
+mod inflight;
 mod log_stats;
 mod output;
 mod request;
@@ -15,18 +16,22 @@ pub use output::{
 pub use request::GenerateRequest;
 pub use vllm_engine_core_client::protocol::logprobs::{Logprobs, PositionLogprobs, TokenLogprob};
 
+use crate::inflight::InflightRequests;
 use crate::log_stats::StatsLogger;
 use crate::request_metrics::RequestMetricsTracker;
 
-/// Thin generate-only facade over [`EngineCoreClient`].
+/// Thin generate-and-abort facade over [`EngineCoreClient`].
 ///
 /// This mirrors the narrow public shape of Python `AsyncLLM.generate()` and
 /// `abort()`, but keeps the boundary close to raw engine-core requests and
-/// outputs.
+/// outputs. It tracks an in-flight external→internal request-id index (see
+/// [`InflightRequests`]) so that aborts issued against external (user-supplied)
+/// ids can be resolved to the internal engine ids that engine-core understands.
 pub struct Llm {
     client: EngineCoreClient,
     randomize_request_id: bool,
     stats_logger: Option<StatsLogger>,
+    inflight: InflightRequests,
 }
 
 impl Llm {
@@ -37,6 +42,7 @@ impl Llm {
             client,
             randomize_request_id: true,
             stats_logger: None,
+            inflight: InflightRequests::new(),
         }
     }
 
@@ -72,9 +78,15 @@ impl Llm {
     pub async fn generate(&self, req: GenerateRequest) -> Result<GenerateOutputStream> {
         let prepared = req.prepare(self.randomize_request_id)?;
         let prompt_token_ids = prepared.prompt_token_ids().into();
+        let external_request_id = prepared
+            .engine_request
+            .external_req_id
+            .clone()
+            .expect("prepare always sets external_req_id");
+        let internal_request_id = prepared.engine_request.request_id.clone();
 
         // Record internal engine-core request ID in the current tracing span.
-        Span::current().record("engine_request_id", &prepared.engine_request.request_id);
+        Span::current().record("engine_request_id", &internal_request_id);
 
         let request_metrics = RequestMetricsTracker::new(
             self.client.model_name().to_string(),
@@ -84,14 +96,32 @@ impl Llm {
             1,
         );
         let stream = self.client.call(prepared.engine_request).await?;
+        let guard = self.inflight.track(external_request_id, internal_request_id);
 
         Ok(GenerateOutputStream::new(
             prompt_token_ids,
             stream,
             request_metrics,
+            guard,
         ))
     }
 
+    /// Abort in-flight requests by their external (user-supplied) request ids.
+    ///
+    /// External ids are resolved to the internal engine ids actually known to
+    /// engine-core (one external id may map to several internal ids). Unknown
+    /// or already-finished ids resolve to nothing and are a safe no-op. The
+    /// tracking entries themselves are removed when the corresponding output
+    /// streams are dropped, not here.
+    pub async fn abort(&self, external_ids: &[String]) -> Result<()> {
+        let internal_ids = self.inflight.resolve(external_ids);
+        if internal_ids.is_empty() {
+            return Ok(());
+        }
+        self.client.abort(&internal_ids).await?;
+        Ok(())
+    }
+
     /// Shut down the underlying engine-core client and its background tasks.
     pub async fn shutdown(self) -> Result<()> {
         self.client.shutdown().await?;

rust/src/llm/src/output.rs

@@ -12,6 +12,7 @@ use vllm_engine_core_client::protocol::{EngineCoreFinishReason, StopReason};
 use vllm_engine_core_client::{AbortCause, EngineCoreOutputStream};
 
 use crate::error::Result;
+use crate::inflight::RequestGuard;
 use crate::request_metrics::{RequestMetricsTracker, current_unix_timestamp_secs};
 
 /// Token usage metadata for one request.
@@ -195,12 +196,17 @@ impl GenerateOutput {
 
 /// Stream of per-request generate outputs for one request.
 ///
-/// - A normal termination of the stream represents a clean completion of the request.
-/// - For errors, unexpected closes, or explicit aborts, the stream terminates with an error.
+/// - A normal termination of the stream represents a clean completion of the
+///   request, including a client-initiated abort, which yields a final output
+///   with `finish_reason = Abort` before the stream ends.
+/// - For errors or unexpected engine-side closes, the stream terminates with an error.
 pub struct GenerateOutputStream {
     pending_prompt_info: Option<GeneratePromptInfo>,
     raw_stream: EngineCoreOutputStream,
     request_metrics: RequestMetricsTracker,
+    /// Removes this request's external→internal tracking edge on drop. Held for
+    /// its `Drop` side effect only; never read directly.
+    _request_guard: RequestGuard,
 }
 
 impl GenerateOutputStream {
@@ -210,6 +216,7 @@ impl GenerateOutputStream {
         prompt_token_ids: Arc<[u32]>,
         raw_stream: EngineCoreOutputStream,
         request_metrics: RequestMetricsTracker,
+        request_guard: RequestGuard,
     ) -> Self {
         Self {
             pending_prompt_info: Some(GeneratePromptInfo {
@@ -218,6 +225,7 @@ impl GenerateOutputStream {
             }),
             raw_stream,
             request_metrics,
+            _request_guard: request_guard,
         }
     }
 

rust/src/llm/src/request_metrics.rs

@@ -98,27 +98,33 @@ impl RequestMetricsTracker {
             self.observe_events(engine_index, events);
         }
 
-        if self.is_prefilling {
-            if let Some(prefill_stats) = &output.prefill_stats {
-                record_prompt_tokens(&self.model_name, engine_index, prefill_stats);
+        // Only outputs that actually carry tokens drive token-timing metrics.
+        // A terminal output with no new tokens (e.g. the synthesized abort
+        // output) must not log a stray time-to-first-token or inter-token
+        // sample.
+        if !output.new_token_ids.is_empty() {
+            if self.is_prefilling {
+                if let Some(prefill_stats) = &output.prefill_stats {
+                    record_prompt_tokens(&self.model_name, engine_index, prefill_stats);
+                }
+                self.first_token_latency = received_at - self.arrival_time;
+                observe_time_to_first_token_seconds(
+                    &self.model_name,
+                    engine_index,
+                    self.first_token_latency,
+                );
+                self.first_token_ts = batch_timestamp;
+                self.is_prefilling = false;
+            } else if self.last_token_ts > 0.0 {
+                observe_inter_token_latency_seconds(
+                    &self.model_name,
+                    engine_index,
+                    batch_timestamp - self.last_token_ts,
+                );
             }
-            self.first_token_latency = received_at - self.arrival_time;
-            observe_time_to_first_token_seconds(
-                &self.model_name,
-                engine_index,
-                self.first_token_latency,
-            );
-            self.first_token_ts = batch_timestamp;
-            self.is_prefilling = false;
-        } else if self.last_token_ts > 0.0 {
-            observe_inter_token_latency_seconds(
-                &self.model_name,
-                engine_index,
-                batch_timestamp - self.last_token_ts,
-            );
-        }
 
-        self.last_token_ts = batch_timestamp;
+            self.last_token_ts = batch_timestamp;
+        }
     }
 
     /// Emit the terminal request metrics once a finished output has been

rust/src/llm/tests/generate.rs

@@ -554,6 +554,142 @@ async fn duplicate_external_request_ids_are_randomized_before_reaching_engine_co
     llm.shutdown().await.unwrap();
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn abort_resolves_external_request_id_to_internal_before_reaching_engine() {
+    let ipc = IpcNamespace::new().unwrap();
+    let handshake_address = ipc.handshake_endpoint();
+    let engine_id = b"engine-abort".to_vec();
+
+    let (shutdown_tx, engine_task) = spawn_mock_engine_task(
+        handshake_address.clone(),
+        engine_id.clone(),
+        |dealer, push| {
+            Box::pin(async move {
+                let add = recv_engine_message(dealer).await;
+                assert_eq!(add[0].as_ref(), &[0x00]);
+                let request: EngineCoreRequest = rmp_serde::from_slice(&add[1]).unwrap();
+                assert_eq!(request.external_req_id.as_deref(), Some("req-abort"));
+                assert!(request.request_id.starts_with("req-abort-"));
+                assert_ne!(request.request_id, "req-abort");
+
+                send_outputs(
+                    push,
+                    EngineCoreOutputs {
+                        outputs: vec![request_output(&request.request_id, vec![7], None)],
+                        ..Default::default()
+                    },
+                )
+                .await;
+
+                // The abort frame must carry the internal engine id, not the
+                // external "req-abort" id the caller aborted by.
+                let abort =
+                    timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap();
+                assert_eq!(abort[0].as_ref(), &[0x01]);
+                let aborted_ids: Vec<String> = rmp_serde::from_slice(&abort[1]).unwrap();
+                assert_eq!(aborted_ids, vec![request.request_id]);
+            })
+        },
+    );
+
+    let llm = connect_async_llm_with_ipc(handshake_address, 0, "test-model", &ipc).await;
+    let mut stream = llm.generate(sample_generate_request("req-abort", 4)).await.unwrap();
+    let internal_id = stream.request_id().to_string();
+    assert_ne!(internal_id, "req-abort");
+
+    assert_eq!(stream.next().await.unwrap().unwrap().token_ids, vec![7]);
+
+    // Abort by the external id; engine-core only knows the internal id.
+    llm.abort(&["req-abort".to_string()]).await.unwrap();
+
+    // The consumer stream is finalized locally with a clean abort terminal
+    // rather than hanging or surfacing as RequestStreamClosed. The engine sends
+    // no final output for a client abort, so this output is synthesized.
+    let terminal = stream.next().await.unwrap().unwrap();
+    assert_eq!(terminal.finish_reason, Some(FinishReason::Abort));
+    assert!(terminal.token_ids.is_empty());
+    assert!(stream.next().await.is_none());
+
+    let _ = shutdown_tx.send(());
+    engine_task.await.unwrap();
+    drop(stream);
+    llm.shutdown().await.unwrap();
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn abort_by_external_id_aborts_all_internal_requests() {
+    let ipc = IpcNamespace::new().unwrap();
+    let handshake_address = ipc.handshake_endpoint();
+    let engine_id = b"engine-abort-many".to_vec();
+
+    let (shutdown_tx, engine_task) = spawn_mock_engine_task(
+        handshake_address.clone(),
+        engine_id.clone(),
+        |dealer, push| {
+            Box::pin(async move {
+                let add_1 = recv_engine_message(dealer).await;
+                assert_eq!(add_1[0].as_ref(), &[0x00]);
+                let request_1: EngineCoreRequest = rmp_serde::from_slice(&add_1[1]).unwrap();
+
+                let add_2 = recv_engine_message(dealer).await;
+                assert_eq!(add_2[0].as_ref(), &[0x00]);
+                let request_2: EngineCoreRequest = rmp_serde::from_slice(&add_2[1]).unwrap();
+
+                assert_eq!(request_1.external_req_id.as_deref(), Some("req-dup-abort"));
+                assert_eq!(request_2.external_req_id.as_deref(), Some("req-dup-abort"));
+                assert_ne!(request_1.request_id, request_2.request_id);
+
+                send_outputs(
+                    push,
+                    EngineCoreOutputs {
+                        outputs: vec![
+                            request_output(&request_1.request_id, vec![7], None),
+                            request_output(&request_2.request_id, vec![8], None),
+                        ],
+                        ..Default::default()
+                    },
+                )
+                .await;
+
+                // A single abort by the shared external id must abort both
+                // internal engine ids it expanded into.
+                let abort =
+                    timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap();
+                assert_eq!(abort[0].as_ref(), &[0x01]);
+                let mut aborted_ids: Vec<String> = rmp_serde::from_slice(&abort[1]).unwrap();
+                aborted_ids.sort();
+                let mut expected = vec![request_1.request_id, request_2.request_id];
+                expected.sort();
+                assert_eq!(aborted_ids, expected);
+            })
+        },
+    );
+
+    let llm = connect_async_llm_with_ipc(handshake_address, 0, "test-model", &ipc).await;
+    let mut stream_1 = llm.generate(sample_generate_request("req-dup-abort", 4)).await.unwrap();
+    let mut stream_2 = llm.generate(sample_generate_request("req-dup-abort", 4)).await.unwrap();
+    assert_ne!(stream_1.request_id(), stream_2.request_id());
+
+    assert_eq!(stream_1.next().await.unwrap().unwrap().token_ids, vec![7]);
+    assert_eq!(stream_2.next().await.unwrap().unwrap().token_ids, vec![8]);
+
+    llm.abort(&["req-dup-abort".to_string()]).await.unwrap();
+
+    // Both internal requests the external id expanded into are finalized with a
+    // clean abort terminal.
+    for stream in [&mut stream_1, &mut stream_2] {
+        let terminal = stream.next().await.unwrap().unwrap();
+        assert_eq!(terminal.finish_reason, Some(FinishReason::Abort));
+        assert!(stream.next().await.is_none());
+    }
+
+    let _ = shutdown_tx.send(());
+    engine_task.await.unwrap();
+    drop(stream_1);
+    drop(stream_2);
+    llm.shutdown().await.unwrap();
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn generate_records_request_metrics_in_prometheus_output() {
     let ipc = IpcNamespace::new().unwrap();

rust/src/text/src/lib.rs

@@ -151,6 +151,12 @@ impl TextLlm {
         Ok((text_request, raw_stream))
     }
 
+    /// Abort in-flight requests by their external (user-supplied) request ids.
+    pub async fn abort(&self, external_ids: &[String]) -> Result<()> {
+        self.llm.abort(external_ids).await?;
+        Ok(())
+    }
+
     /// Shut down the underlying LLM client and its background tasks.
     pub async fn shutdown(self) -> Result<()> {
         self.llm.shutdown().await?;