diff --git a/console/web/src/components/chat/ChatView.tsx b/console/web/src/components/chat/ChatView.tsx
index 78b6b37f4..1dbcfb7e8 100644
--- a/console/web/src/components/chat/ChatView.tsx
+++ b/console/web/src/components/chat/ChatView.tsx
@@ -179,6 +179,20 @@ export function ChatView({
   const serverWorking = conversation.status === 'working'
   const streamingIndicator = isStreaming || serverWorking
 
+  // Stop requested but not yet finalized server-side. Disables the stop button
+  // until status-changed flips the indicator off. The ref is the dedupe guard:
+  // synchronous (two clicks in one frame can't both pass, unlike closure
+  // state) and readable outside a state updater (React forbids side effects
+  // inside updaters — Strict Mode double-invokes them).
+  const [stopping, setStopping] = useState(false)
+  const stopRequestedRef = useRef(false)
+  useEffect(() => {
+    if (!streamingIndicator) {
+      stopRequestedRef.current = false
+      setStopping(false)
+    }
+  }, [streamingIndicator])
+
   // Messages queued mid-stream (MOT-3837): shown above the composer until the
   // harness drains them into the transcript. Each draft carries the predicted
   // entry id of its eventual transcript row.
@@ -1278,8 +1292,17 @@ export function ChatView({
   )
 
   const handleStop = useCallback(() => {
+    if (stopRequestedRef.current) return
+    stopRequestedRef.current = true
+    setStopping(true)
     abortRef.current?.abort()
-    void backend.abortRun?.(sessionId).catch(() => {})
+    // Re-enable the button if the stop RPC fails while the server still
+    // reports working — otherwise `stopping` never clears (the reset
+    // effect waits on the indicator) and the user can't retry.
+    void backend.abortRun?.(sessionId).catch(() => {
+      stopRequestedRef.current = false
+      setStopping(false)
+    })
   }, [backend, sessionId])
 
   // Rescue a parked stream loop: the session hit a terminal error server-side
@@ -1729,6 +1752,7 @@ export function ChatView({
             onTextChange={handleComposerTextChange}
             onSubmit={handleSubmit}
             onStop={handleStop}
+            stopping={stopping}
             queuedForEdit={backend.editQueued ? queuedForEdit : undefined}
             onEditQueued={backend.editQueued ? handleEditQueued : undefined}
             onBrowseChange={setBrowsedQueuedId}
diff --git a/console/web/src/components/chat/Composer.tsx b/console/web/src/components/chat/Composer.tsx
index 8ad2dced3..194fef0a5 100644
--- a/console/web/src/components/chat/Composer.tsx
+++ b/console/web/src/components/chat/Composer.tsx
@@ -4,7 +4,7 @@ import {
   $getRoot,
   type LexicalEditor,
 } from 'lexical'
-import { ArrowUp, Square } from 'lucide-react'
+import { ArrowUp, Loader2, Square } from 'lucide-react'
 import { useCallback, useEffect, useMemo, useRef, useState } from 'react'
 import { PermissionModePicker } from '@/components/permissions/PermissionModePicker'
 import type { PermissionMode } from '@/lib/backend/approval-settings'
@@ -87,6 +87,8 @@ interface ComposerProps {
   onPermissionModeChange: (next: PermissionMode) => void
   onSubmit: (payload: ComposerSubmitPayload) => void
   onStop?: () => void
+  /** A stop was requested and the server hasn't finalized the turn yet. */
+  stopping?: boolean
   isStreaming?: boolean
   /**
    * When true, the editor stays unlocked while streaming: a submit queues the
@@ -160,6 +162,7 @@ export function Composer({
   onPermissionModeChange,
   onSubmit,
   onStop,
+  stopping,
   isStreaming,
   queueWhileStreaming,
   blocked,
@@ -389,10 +392,15 @@ export function Composer({
             <button
               type="button"
               onClick={onStop}
-              aria-label="stop generating"
+              disabled={stopping}
+              aria-label={stopping ? 'stopping' : 'stop generating'}
               className={actionButtonClass}
             >
-              <Square size={16} aria-hidden className="fill-black/90" />
+              {stopping ? (
+                <Loader2 size={16} aria-hidden className="animate-spin" />
+              ) : (
+                <Square size={16} aria-hidden className="fill-black/90" />
+              )}
             </button>
           ) : (
             <button
diff --git a/console/web/src/lib/backend/translate.test.ts b/console/web/src/lib/backend/translate.test.ts
index dfab0b545..0372691c2 100644
--- a/console/web/src/lib/backend/translate.test.ts
+++ b/console/web/src/lib/backend/translate.test.ts
@@ -177,7 +177,12 @@ describe('translateTurnSource — turn-completed', () => {
         completed({ status: 'cancelled', reason: 'stopped' }),
       ),
     ).toEqual([
-      { kind: 'stop-reason', reason: 'aborted', message: 'stopped' },
+      {
+        kind: 'stop-reason',
+        reason: 'aborted',
+        message: 'stopped',
+        entryId: 'e_t-1_stopped',
+      },
       { kind: 'assistant-end' },
     ])
   })
diff --git a/console/web/src/lib/backend/translate.ts b/console/web/src/lib/backend/translate.ts
index c9208d755..2367406fb 100644
--- a/console/web/src/lib/backend/translate.ts
+++ b/console/web/src/lib/backend/translate.ts
@@ -108,7 +108,14 @@ export function translateTurnSource(event: TurnSourceEvent): StreamEvent[] {
 function translateTurnCompleted(event: TurnCompletedEvent): StreamEvent[] {
   const out: StreamEvent[] = []
   if (event.status === 'cancelled') {
-    out.push({ kind: 'stop-reason', reason: 'aborted', message: event.reason })
+    out.push({
+      kind: 'stop-reason',
+      reason: 'aborted',
+      message: event.reason,
+      // Same id as the harness's durable "stopped" notice entry, so the live
+      // notice and the reconciled transcript row dedupe instead of doubling.
+      entryId: `e_${event.turn_id}_stopped`,
+    })
   } else if (event.status === 'failed' || event.result_error) {
     const message = event.result_error ?? event.reason
     out.push({
diff --git a/harness/src/clients/router.rs b/harness/src/clients/router.rs
index b7728b222..d49bcf67b 100644
--- a/harness/src/clients/router.rs
+++ b/harness/src/clients/router.rs
@@ -19,7 +19,7 @@ use iii_sdk::protocol::TriggerRequest;
 use iii_sdk::IIIClient;
 use serde_json::{json, Value};
 use tokio::sync::mpsc;
-use tokio::sync::Notify;
+use tokio::sync::{watch, Notify};
 
 use crate::error::HarnessError;
 use crate::types::event::{AssistantMessageEvent, StopReason};
@@ -70,10 +70,17 @@ impl RouterClient {
     }
 
     /// Stream one assistant turn, forwarding coalesced partials to `sink`.
+    ///
+    /// `abort_rx` is the harness-local cancel backstop: `router::abort` is the
+    /// happy path (the router closes the stream → EOF), but it's a no-op when
+    /// it races ahead of the router's stream registration or after a router
+    /// restart — then only this local signal cuts the await (which otherwise
+    /// blocks up to `router_timeout_ms`).
     pub async fn chat(
         &self,
         params: ChatParams,
         sink: &dyn StreamSink,
+        mut abort_rx: watch::Receiver<bool>,
     ) -> Result<ChatOutcome, HarnessError> {
         let channel = create_channel(&self.iii, None)
             .await
@@ -113,6 +120,7 @@ impl RouterClient {
             }
         });
 
+        let request_id = params.request_id.clone();
         let mut payload = json!({
             "writer_ref": channel.writer_ref,
             "request_id": params.request_id,
@@ -186,6 +194,15 @@ impl RouterClient {
         // then fall through to the outcome handling below.
         let mut response: Option<Result<Value, iii_sdk::errors::Error>> = None;
         let mut drain_deadline = Instant::now();
+        // Resolves when harness::stop fires the turn's local cancel; a dropped
+        // sender (registry cleared — no stop can arrive) parks forever instead
+        // of busy-looping the select.
+        let abort_fired = async move {
+            if abort_rx.wait_for(|a| *a).await.is_err() {
+                std::future::pending::<()>().await;
+            }
+        };
+        tokio::pin!(abort_fired);
         loop {
             let frame = if response.is_some() {
                 match tokio::time::timeout(Duration::from_millis(750), rx.recv()).await {
@@ -208,6 +225,52 @@ impl RouterClient {
             } else {
                 tokio::select! {
                     f = rx.recv() => f,
+                    // Local cancel backstop: cut the await now, return the
+                    // partial as Aborted. No arm in the post-ack drain branch
+                    // above — that path is bounded (≤10s) already.
+                    _ = &mut abort_fired => {
+                        // notify_one, not notify_waiters: it stores a permit
+                        // when the pump task hasn't polled its notified() yet
+                        // (a pre-fired cancel can get here before the spawned
+                        // pump first runs), so the wakeup can't be lost and
+                        // pump.await can't hang on next_binary().
+                        cancel.notify_one();
+                        let _ = pump.await;
+                        // Kill the held-open router::chat task; the router's
+                        // own teardown was already requested via router::abort.
+                        trigger.abort();
+                        // Re-request the router abort: stop.rs's router::abort
+                        // can race ahead of the router's stream registration
+                        // and no-op — by now the stream is certainly
+                        // registered, so this lands and the provider stops
+                        // generating instead of streaming into the closed
+                        // channel until its own terminal. Detached: an abort
+                        // against a hung router must not block the stop path.
+                        let client = self.clone();
+                        let rid = request_id.clone();
+                        tokio::spawn(async move {
+                            client.abort(&rid).await;
+                        });
+                        // Prefer a terminal frame's complete message over the
+                        // tracker's partial: a stop landing between a Done
+                        // frame and EOF must not replace the full assistant
+                        // entry with the stale pre-Done partial.
+                        let mut message = final_message
+                            .or_else(|| tracker.current())
+                            .unwrap_or_else(|| {
+                                empty_assistant(
+                                    params.provider.as_deref().unwrap_or(""),
+                                    &params.model,
+                                )
+                            });
+                        message.stop_reason = StopReason::Aborted;
+                        return Ok(ChatOutcome {
+                            ok: false,
+                            stop_reason: Some(StopReason::Aborted),
+                            message,
+                            error: None,
+                        });
+                    }
                     r = &mut trigger => {
                         response = Some(r.map_err(|e| {
                             HarnessError::Internal(format!("router::chat task: {e}"))
@@ -273,7 +336,8 @@ impl RouterClient {
         }
 
         // Stream drained; stop the pump and collect the trigger ack.
-        cancel.notify_waiters();
+        // notify_one: latched — safe even if the pump task hasn't polled yet.
+        cancel.notify_one();
         let _ = pump.await;
         let response = match response {
             Some(r) => r,
diff --git a/harness/src/deps.rs b/harness/src/deps.rs
index d62c79edc..5e715c7bf 100644
--- a/harness/src/deps.rs
+++ b/harness/src/deps.rs
@@ -12,7 +12,7 @@ use crate::configuration::ConfigCell;
 use crate::discovery::{FunctionsCell, FunctionsSnapshot};
 use crate::events::TurnEvents;
 use crate::hooks::HookRegistry;
-use crate::locks::SessionLocks;
+use crate::locks::{SessionLocks, TurnCancels};
 use crate::subscriptions::SubscriptionRegistry;
 
 #[derive(Clone)]
@@ -23,6 +23,7 @@ pub struct Deps {
     pub events: TurnEvents,
     pub hooks: HookRegistry,
     pub locks: SessionLocks,
+    pub cancels: TurnCancels,
     pub subscriptions: Arc<SubscriptionRegistry>,
     /// react's per-subscription fire-rate breaker (loop breaker #3).
     pub react_gate: Arc<crate::functions::react::FireGate>,
@@ -43,6 +44,7 @@ impl Deps {
             events,
             hooks,
             locks: SessionLocks::new(),
+            cancels: TurnCancels::new(),
             subscriptions: Arc::new(SubscriptionRegistry::new()),
             react_gate: Arc::new(crate::functions::react::FireGate::default()),
         }
diff --git a/harness/src/functions/stop.rs b/harness/src/functions/stop.rs
index f4f25faa7..bec88e19e 100644
--- a/harness/src/functions/stop.rs
+++ b/harness/src/functions/stop.rs
@@ -44,10 +44,21 @@ pub async fn handle(deps: &Deps, req: StopRequest) -> Result<StopResponse, Harne
     if record.status.is_terminal() {
         return Ok(StopResponse { stopping: false });
     }
+    // Idempotent: a prior stop already fired the cancel signal, cascaded to
+    // children, and requested the router abort — repeat clicks become one read.
+    if record.abort {
+        return Ok(StopResponse { stopping: true });
+    }
     // Pin the turn we observed so the write under the lock can't land on a newer
     // turn that started in between (matters when `turn_id` was omitted).
     let target_turn = record.turn_id.clone();
 
+    // In-process cancel signal, fired lock-free BEFORE anything that awaits:
+    // it cuts the in-flight `router.chat` await (backstop when router::abort
+    // is a no-op) and is observed between tool executions, where the durable
+    // flag write below is blocked on the session lock.
+    deps.cancels.fire(&target_turn);
+
     // Cascade to non-terminal spawned children BEFORE taking this session's
     // lock (harness.md § Cancellation cascade): each child stop acquires its
     // OWN session lock, so holding the parent lock here could deadlock against a
@@ -100,5 +111,15 @@ pub async fn handle(deps: &Deps, req: StopRequest) -> Result<StopResponse, Harne
     record.updated_at = crate::types::message::AgentMessage::now_ms();
     crate::state::put_turn(&deps.iii, &record, cfg.session_timeout_ms).await?;
 
+    // "stopping" ack on the existing phase-reason channel (status stays
+    // "working" — same semantics as "waiting for <model>"). UNDER the lock and
+    // after the terminal re-check: every finalizer emits its own set_status
+    // while holding this lock, so a lock-free ack here could land AFTER a
+    // concurrent finalize's "done" and leave the session stuck on "working".
+    let session = deps.session().await;
+    let _ = session
+        .set_status(&req.session_id, "working", Some("stopping"))
+        .await;
+
     Ok(StopResponse { stopping: true })
 }
diff --git a/harness/src/locks.rs b/harness/src/locks.rs
index 06ad814d0..0490a4a12 100644
--- a/harness/src/locks.rs
+++ b/harness/src/locks.rs
@@ -35,9 +35,55 @@ impl SessionLocks {
     }
 }
 
+/// Per-turn in-process cancel signals. `harness::stop` fires one lock-free so
+/// the running step can observe the stop where durable state can't reach it:
+/// inside the `router::chat` await (via `watch`) and between tool executions
+/// (via `is_fired`) — the tool phase holds the session lock, so the durable
+/// abort write is blocked behind it by construction. Level-triggered `watch`
+/// channels: a fire before subscribe is still observed. Keyed by turn_id so a
+/// stale fire can never cancel a newer turn. Same single-process caveat as
+/// `SessionLocks` above.
+#[derive(Clone, Default)]
+pub struct TurnCancels {
+    map: Arc<Mutex<HashMap<String, tokio::sync::watch::Sender<bool>>>>,
+}
+
+impl TurnCancels {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Signal cancellation for `turn_id` (idempotent).
+    pub fn fire(&self, turn_id: &str) {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.entry(turn_id.to_string())
+            .or_insert_with(|| tokio::sync::watch::channel(false).0)
+            .send_replace(true);
+    }
+
+    pub fn is_fired(&self, turn_id: &str) -> bool {
+        let map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.get(turn_id).is_some_and(|s| *s.borrow())
+    }
+
+    /// Subscribe to `turn_id`'s cancel signal, creating it on first use.
+    pub fn watch(&self, turn_id: &str) -> tokio::sync::watch::Receiver<bool> {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.entry(turn_id.to_string())
+            .or_insert_with(|| tokio::sync::watch::channel(false).0)
+            .subscribe()
+    }
+
+    /// Drop `turn_id`'s signal once the turn is terminal.
+    pub fn clear(&self, turn_id: &str) {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.remove(turn_id);
+    }
+}
+
 #[cfg(test)]
 mod tests {
-    use super::SessionLocks;
+    use super::{SessionLocks, TurnCancels};
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
     use std::sync::Arc;
 
@@ -84,4 +130,18 @@ mod tests {
         let _other = locks.guard("b").await;
         drop(held);
     }
+
+    /// The property the chat-abort backstop relies on: level-triggering. A fire
+    /// BEFORE anyone subscribes is still observed by a later watch/is_fired.
+    #[tokio::test]
+    async fn cancel_fired_before_subscribe_is_observed() {
+        let cancels = TurnCancels::new();
+        cancels.fire("t1");
+        assert!(cancels.is_fired("t1"));
+        let rx = cancels.watch("t1");
+        assert!(*rx.borrow());
+        assert!(!cancels.is_fired("t2"));
+        cancels.clear("t1");
+        assert!(!cancels.is_fired("t1"));
+    }
 }
diff --git a/harness/src/turn_loop.rs b/harness/src/turn_loop.rs
index 2f92050e2..184fd888f 100644
--- a/harness/src/turn_loop.rs
+++ b/harness/src/turn_loop.rs
@@ -520,6 +520,15 @@ pub async fn run_step(
     record.stream_request_id = Some(params.request_id.clone());
     crate::state::put_turn(&deps.iii, &record, cfg.session_timeout_ms).await?;
 
+    // A stop that landed during context assembly (its durable write blocked on
+    // the guard held here) is only visible via the in-process signal. Bail
+    // before dispatching generation at all — otherwise the provider starts a
+    // stream that is dead on arrival.
+    if deps.cancels.is_fired(&record.turn_id) {
+        record.abort = true;
+        return finalize_cancelled(deps, &session, &mut record, "cancelled").await;
+    }
+
     // Release the per-session lock across the generation RPC. The loop writes no
     // turn record between here and the post-generation cancellation check
     // (router::chat streams into the SESSION, not the turn record), so dropping
@@ -536,7 +545,25 @@ pub async fn run_step(
         .await;
 
     let router = deps.router().await;
-    let outcome = router.chat(params, &sink).await?;
+    // Harness-local cancel backstop: cuts this await even when router::abort
+    // was a no-op (registration race / router restart). Level-triggered, so a
+    // stop fired before this subscribe is still observed.
+    let abort_rx = deps.cancels.watch(&record.turn_id);
+    let mut outcome = router.chat(params, &sink, abort_rx).await?;
+
+    // A stop mid-generation truncates tool-call blocks before their arguments
+    // stream. An argument-less call can never execute (the turn is cancelled)
+    // and renders as an empty generic card — drop it from the persisted
+    // partial. Calls with partial arguments are kept: they show what was
+    // forming when the user stopped.
+    if outcome.message.stop_reason == crate::types::event::StopReason::Aborted {
+        outcome.message.content.retain(|b| {
+            !matches!(b, ContentBlock::FunctionCall { arguments, .. }
+                if arguments.is_null()
+                    || arguments.as_str().is_some_and(str::is_empty)
+                    || arguments.as_object().is_some_and(serde_json::Map::is_empty))
+        });
+    }
 
     // Generation consumed: advance the steering watermark (persisted by the
     // advance()/finalize call that ends this step).
@@ -699,6 +726,16 @@ pub async fn run_step(
                 .await?;
         let filesystem_root = record.options.filesystem_root().map(str::to_string);
         for call in trigger_calls.iter().copied() {
+            // Cancel check between tool calls. Only the in-process signal can
+            // be observed here: this phase holds the session lock, so the
+            // durable abort write in harness::stop is blocked behind it by
+            // construction. Executed calls are already checkpointed per-call,
+            // so finalizing mid-loop loses nothing.
+            if deps.cancels.is_fired(&record.turn_id) {
+                record.abort = true;
+                return finalize_cancelled(deps, &session, &mut record, "cancelled").await;
+            }
+
             // Per-call checkpoint: skip done/pending, recover an interrupted
             // trigger.
             match record.calls.get(&call.id).map(|c| c.state) {
@@ -1161,6 +1198,7 @@ async fn finalize_completed(
     record.result_error = None;
     record.updated_at = AgentMessage::now_ms();
     crate::state::put_turn(&deps.iii, record, cfg.session_timeout_ms).await?;
+    deps.cancels.clear(&record.turn_id);
     let _ = session.set_status(&record.session_id, "done", None).await;
     deps.events
         .emit_completed(
@@ -1204,6 +1242,7 @@ async fn finalize_failed(
     record.updated_at = AgentMessage::now_ms();
     record_failure_telemetry(record, reason, failure);
     crate::state::put_turn(&deps.iii, record, cfg.session_timeout_ms).await?;
+    deps.cancels.clear(&record.turn_id);
     let _ = session
         .append_custom(
             &record.session_id,
@@ -1403,6 +1442,19 @@ async fn finalize_cancelled(
     record.status = TurnStatus::Cancelled;
     record.updated_at = AgentMessage::now_ms();
     crate::state::put_turn(&deps.iii, record, cfg.session_timeout_ms).await?;
+    deps.cancels.clear(&record.turn_id);
+    // Durable stop marker: without it the transcript just ends mid-thought
+    // after a reload. Deterministic id so the live `stop-reason` notice
+    // (translate.ts) dedupes against this entry, mirroring `e_{turn}_error`.
+    let _ = session
+        .append_custom(
+            &record.session_id,
+            "notice",
+            json!({ "reason": "stopped", "message": "stopped by user." }),
+            &format!("e_{}_stopped", record.turn_id),
+            Some(&origin(&record.turn_id)),
+        )
+        .await;
     let _ = session.set_status(&record.session_id, "done", None).await;
     deps.events
         .emit_completed(
@@ -1442,6 +1494,11 @@ pub async fn fail_turn(
 ) -> TurnStepResult {
     let cfg = deps.cfg().await;
     let session = deps.session().await;
+    // Serialize with stop/resolve/sweep like every other turn-record writer
+    // (locks.rs). Safe: the only caller runs after run_step returned, so its
+    // guard is gone. Lock-free, this finalize could interleave with
+    // harness::stop's under-lock "stopping" ack and strand the session status.
+    let _guard = deps.locks.guard(session_id).await;
     let record = crate::state::get_turn(&deps.iii, session_id, cfg.session_timeout_ms)
         .await
         .ok()
diff --git a/llm-router/src/chat/chat.rs b/llm-router/src/chat/chat.rs
index 4c3ce8225..e074b591a 100644
--- a/llm-router/src/chat/chat.rs
+++ b/llm-router/src/chat/chat.rs
@@ -270,7 +270,34 @@ impl ChatPipeline {
             }
             let channel = create_router_channel(&self.iii).await?;
             let mut reader = channel.reader;
-            inflight.set_closer(reader.closer());
+            // router::abort closes the relay AND actively cancels the
+            // provider's upstream via `provider::<id>::abort` — without it the
+            // provider only notices the closed channel on its next (ping)
+            // write, billing tokens meanwhile. Detached + best-effort: a
+            // provider without the function (or a stale request id) is a
+            // harmless error/no-op.
+            inflight.set_closer({
+                let close = reader.closer();
+                let iii = self.iii.clone();
+                let provider = provider.to_string();
+                let rid = request_id.to_string();
+                Arc::new(move || {
+                    close();
+                    let iii = iii.clone();
+                    let function_id = format!("provider::{}::abort", provider);
+                    let payload = serde_json::json!({ "request_id": rid });
+                    tokio::spawn(async move {
+                        let _ = iii
+                            .trigger(TriggerRequest {
+                                function_id,
+                                payload,
+                                action: None,
+                                timeout_ms: Some(10_000),
+                            })
+                            .await;
+                    });
+                })
+            });
 
             let stream_input = build_stream_input(
                 call,
diff --git a/llm-router/src/provider_scaffold/aborts.rs b/llm-router/src/provider_scaffold/aborts.rs
new file mode 100644
index 000000000..f3f3c48bc
--- /dev/null
+++ b/llm-router/src/provider_scaffold/aborts.rs
@@ -0,0 +1,160 @@
+//! request_id → in-flight upstream cancel, backing `provider::<id>::abort`.
+//! Provider-side mirror of the router's `chat::inflight`: the stream fn
+//! registers its request id before spawning the upstream, the abort function
+//! signals it, and `pump_abortable` returns on the signal — dropping the
+//! upstream receiver, which aborts the upstream task and its in-flight HTTP
+//! request. Level-triggered watch channels: an abort that lands between
+//! `watch` and the pump's first poll is still observed. `abort` signals
+//! existing entries only (no create), so stray aborts for finished requests
+//! can't leak entries — the ping-write backstop in the pump covers that race.
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use futures::future::BoxFuture;
+use iii_sdk::errors::Error;
+use tokio::sync::watch;
+
+use crate::types::router::{ProviderAbortRequest, ProviderAbortResponse};
+
+#[derive(Clone, Default)]
+pub struct StreamAborts {
+    map: Arc<Mutex<HashMap<String, watch::Sender<bool>>>>,
+}
+
+impl StreamAborts {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Subscribe for `request_id`, creating the entry. Call before spawning
+    /// the upstream; pair with `remove` when the stream ends.
+    pub fn watch(&self, request_id: &str) -> watch::Receiver<bool> {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.entry(request_id.to_string())
+            .or_insert_with(|| watch::channel(false).0)
+            .subscribe()
+    }
+
+    /// Signal cancellation. `false` when the request is unknown — already
+    /// finished, never started, or previously aborted (idempotent).
+    pub fn abort(&self, request_id: &str) -> bool {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        match map.remove(request_id) {
+            Some(tx) => {
+                tx.send_replace(true);
+                true
+            }
+            None => false,
+        }
+    }
+
+    /// Drop `request_id`'s entry once its stream is over.
+    pub fn remove(&self, request_id: &str) {
+        let mut map = self.map.lock().unwrap_or_else(|p| p.into_inner());
+        map.remove(request_id);
+    }
+
+    /// RAII registration: subscribe on create, deregister on Drop. Register
+    /// BEFORE the stream fn's first await so an abort during setup latches,
+    /// and hold the guard across the call so cleanup runs on every exit —
+    /// including the executor cancelling the handler future mid-await, which
+    /// sequential cleanup after the call would silently miss.
+    pub fn register(&self, request_id: &str) -> AbortGuard {
+        AbortGuard {
+            rx: self.watch(request_id),
+            aborts: self.clone(),
+            request_id: request_id.to_string(),
+        }
+    }
+}
+
+pub struct AbortGuard {
+    aborts: StreamAborts,
+    request_id: String,
+    rx: watch::Receiver<bool>,
+}
+
+impl AbortGuard {
+    /// A receiver for `pump_abortable` (level-triggered; clone per pump).
+    pub fn watch(&self) -> watch::Receiver<bool> {
+        self.rx.clone()
+    }
+
+    /// True once the abort fired (also observable via a pre-fired `watch`).
+    pub fn is_fired(&self) -> bool {
+        *self.rx.borrow()
+    }
+}
+
+impl Drop for AbortGuard {
+    fn drop(&mut self) {
+        // A fired abort already consumed the entry — harmless no-op then.
+        self.aborts.remove(&self.request_id);
+    }
+}
+
+/// The `provider::<id>::abort` handler, ready to register: signals the
+/// stream registered under `request_id` (see `StreamAborts`).
+pub fn make_abort(
+    aborts: StreamAborts,
+) -> impl Fn(ProviderAbortRequest) -> BoxFuture<'static, Result<ProviderAbortResponse, Error>>
+       + Send
+       + Sync
+       + 'static {
+    move |req: ProviderAbortRequest| {
+        let aborted = aborts.abort(&req.request_id);
+        Box::pin(async move { Ok(ProviderAbortResponse { aborted }) })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn abort_signals_registered_watchers_once() {
+        let aborts = StreamAborts::new();
+        let rx = aborts.watch("r1");
+        assert!(!*rx.borrow());
+        assert!(aborts.abort("r1"));
+        assert!(*rx.borrow());
+        // entry consumed: a second abort is an idempotent no-op
+        assert!(!aborts.abort("r1"));
+    }
+
+    #[test]
+    fn abort_of_unknown_request_is_a_noop() {
+        let aborts = StreamAborts::new();
+        assert!(!aborts.abort("never-registered"));
+    }
+
+    /// The sender lives in the registry (not the subscriber), so a watcher
+    /// subscribing and dropping doesn't invalidate a later abort signal.
+    #[test]
+    fn resubscribing_sees_a_pre_subscribe_state() {
+        let aborts = StreamAborts::new();
+        drop(aborts.watch("r1"));
+        let rx = aborts.watch("r1");
+        assert!(aborts.abort("r1"));
+        assert!(*rx.borrow());
+    }
+
+    /// The RAII property the providers rely on: dropping the guard (any exit,
+    /// including a cancelled future) deregisters; a pre-drop abort still
+    /// reaches the guard's receiver.
+    #[test]
+    fn guard_deregisters_on_drop_and_latches_aborts() {
+        let aborts = StreamAborts::new();
+        let guard = aborts.register("r1");
+        assert!(!guard.is_fired());
+        assert!(aborts.abort("r1"));
+        assert!(guard.is_fired());
+        assert!(*guard.watch().borrow());
+        drop(guard);
+        assert!(!aborts.abort("r1"), "entry gone after guard drop");
+
+        let guard2 = aborts.register("r2");
+        drop(guard2);
+        assert!(!aborts.abort("r2"), "unfired entry removed by guard drop");
+    }
+}
diff --git a/llm-router/src/provider_scaffold/mod.rs b/llm-router/src/provider_scaffold/mod.rs
index 22c975ac7..8d8780db2 100644
--- a/llm-router/src/provider_scaffold/mod.rs
+++ b/llm-router/src/provider_scaffold/mod.rs
@@ -8,6 +8,7 @@
 //! genuinely theirs: request building, SSE event decoding, and error
 //! classification.
 
+pub mod aborts;
 pub mod cache;
 pub mod names;
 pub mod pump;
diff --git a/llm-router/src/provider_scaffold/pump.rs b/llm-router/src/provider_scaffold/pump.rs
index 93e53cb73..ad1e4eea2 100644
--- a/llm-router/src/provider_scaffold/pump.rs
+++ b/llm-router/src/provider_scaffold/pump.rs
@@ -8,7 +8,14 @@ use std::time::Duration;
 use tokio::sync::mpsc;
 
 /// Heartbeat cadence while the upstream is silent (spec: at least every 30s).
-pub const PING_INTERVAL: Duration = Duration::from_secs(30);
+///
+/// Also the abort-detection bound: a failed ping write is how a provider
+/// notices the router closed the channel (`router::abort`) while the upstream
+/// is silent — e.g. Anthropic server-buffering a large tool_use input,
+/// generating (billed) tokens with no frames flowing. Worst-case waste after
+/// a stop is ~2 intervals (one ping trips the forwarder, the next observes
+/// it), so keep this small: 2s bounds it to ~4s where 30s allowed ~60s.
+pub const PING_INTERVAL: Duration = Duration::from_secs(2);
 
 /// `Err(())` means only "the sink is gone — stop writing"; there is no
 /// error detail to carry, so the unit error is the whole contract.
@@ -26,12 +33,34 @@ pub fn send_event(sink: &dyn FrameSink, ev: &AssistantMessageEvent) -> Result<()
 /// done / no terminal / caller gone) so providers can react to specific
 /// failures — e.g. dropping a cached credential on `AuthExpired`.
 pub async fn pump(
+    rx: mpsc::Receiver<AssistantMessageEvent>,
+    sink: &dyn FrameSink,
+    ping_interval: Duration,
+) -> Option<ErrorKind> {
+    // Held so the never-fired abort arm stays pending instead of erroring.
+    let (_hold, abort_rx) = tokio::sync::watch::channel(false);
+    pump_abortable(rx, sink, ping_interval, abort_rx).await
+}
+
+/// [`pump`] with an active cancel: returns immediately when `abort_rx` fires
+/// (`provider::<id>::abort` via [`super::aborts::StreamAborts`]) — even
+/// mid-silence, without waiting for a ping write to fail. Returning drops
+/// `rx`, which aborts the upstream task and its in-flight HTTP request.
+pub async fn pump_abortable(
     mut rx: mpsc::Receiver<AssistantMessageEvent>,
     sink: &dyn FrameSink,
     ping_interval: Duration,
+    mut abort_rx: tokio::sync::watch::Receiver<bool>,
 ) -> Option<ErrorKind> {
     loop {
-        match tokio::time::timeout(ping_interval, rx.recv()).await {
+        let recv = tokio::select! {
+            r = tokio::time::timeout(ping_interval, rx.recv()) => r,
+            // Level-triggered: a fire before this poll is still observed.
+            // Err (sender gone before firing) can only follow an abort's
+            // send_replace(true), so wait_for resolves Ok on the value first.
+            _ = abort_rx.wait_for(|a| *a) => return None,
+        };
+        match recv {
             Ok(Some(ev)) => {
                 let terminal = ev.is_terminal();
                 let error_kind = match &ev {
diff --git a/llm-router/src/types/router.rs b/llm-router/src/types/router.rs
index f6ba7a413..ee0a72807 100644
--- a/llm-router/src/types/router.rs
+++ b/llm-router/src/types/router.rs
@@ -241,6 +241,23 @@ pub struct RefreshModelsResponse {
     pub count: usize,
 }
 
+/// Input of a provider's `provider::<id>::abort`: actively cancel the
+/// in-flight upstream stream for `request_id` (the router's `request_id`,
+/// delivered to the provider as `resolution_key`) so billed generation stops
+/// immediately instead of waiting for the provider to notice the closed
+/// channel on its next write.
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
+pub struct ProviderAbortRequest {
+    pub request_id: String,
+}
+
+/// Output of `provider::<id>::abort`. `aborted: false` means the request was
+/// unknown — already finished, never started, or aborted before (idempotent).
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, JsonSchema)]
+pub struct ProviderAbortResponse {
+    pub aborted: bool,
+}
+
 /// Event delivered to a provider's `provider::<id>::on_router_ready` (the
 /// `router::ready` trigger payload, currently `{}`). Unknown fields are ignored.
 #[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, JsonSchema)]
diff --git a/provider-anthropic/Cargo.lock b/provider-anthropic/Cargo.lock
index ef45c990b..beddcf57b 100644
--- a/provider-anthropic/Cargo.lock
+++ b/provider-anthropic/Cargo.lock
@@ -776,7 +776,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-anthropic/src/register.rs b/provider-anthropic/src/register.rs
index f1d6a6f81..af29cef3e 100644
--- a/provider-anthropic/src/register.rs
+++ b/provider-anthropic/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -135,15 +136,28 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .build()
         .expect("reqwest client");
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cache.clone()),
+            make_stream(iii.clone(), http.clone(), cache.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC)
         .metadata(json!({ "internal": true })),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone(), http.clone()))
diff --git a/provider-anthropic/src/stream_fn.rs b/provider-anthropic/src/stream_fn.rs
index fee3b1544..3c85e5301 100644
--- a/provider-anthropic/src/stream_fn.rs
+++ b/provider-anthropic/src/stream_fn.rs
@@ -14,8 +14,9 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
 use llm_router::types::events::ErrorKind;
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
 
@@ -23,15 +24,24 @@ pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cache) = (iii.clone(), http.clone(), cache.clone());
+        let (iii, http, cache, aborts) = (iii.clone(), http.clone(), cache.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, &cache, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, &cache, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -43,6 +53,7 @@ async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -136,6 +147,10 @@ async fn run_stream_call(
     );
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -146,9 +161,13 @@ async fn run_stream_call(
             warnings,
         },
     );
+    let kind = match abort_reg {
+        Some(g) => pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    };
     // An upstream auth terminal means the cached credential was rotated
     // out from under us: drop the cache so the next attempt re-resolves.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if kind == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-anthropic/src/surface.rs b/provider-anthropic/src/surface.rs
index fb20bec53..624500953 100644
--- a/provider-anthropic/src/surface.rs
+++ b/provider-anthropic/src/surface.rs
@@ -9,14 +9,18 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::anthropic::stream";
 pub const STREAM_DESC: &str = "Stream an Anthropic chat completion: resolve credentials, call the \
      upstream Messages API, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::anthropic::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::anthropic::refresh_models";
 pub const REFRESH_MODELS_DESC: &str =
     "Refresh the Anthropic catalog slice from the upstream models \
@@ -59,6 +63,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
     ]
diff --git a/provider-anthropic/src/upstream.rs b/provider-anthropic/src/upstream.rs
index c45d416ad..9c397257c 100644
--- a/provider-anthropic/src/upstream.rs
+++ b/provider-anthropic/src/upstream.rs
@@ -29,7 +29,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-anthropic/tests/golden/schemas/provider.anthropic.abort.json b/provider-anthropic/tests/golden/schemas/provider.anthropic.abort.json
new file mode 100644
index 000000000..367d20aef
--- /dev/null
+++ b/provider-anthropic/tests/golden/schemas/provider.anthropic.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::anthropic::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-anthropic/tests/schemas.rs b/provider-anthropic/tests/schemas.rs
index c2360ca9d..b0dcc844d 100644
--- a/provider-anthropic/tests/schemas.rs
+++ b/provider-anthropic/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::anthropic::stream",
+            "provider::anthropic::abort",
             "provider::anthropic::refresh_models",
             "provider::anthropic::on_router_ready",
         ]
diff --git a/provider-kimi/Cargo.lock b/provider-kimi/Cargo.lock
index e8e56df4c..8031c6130 100644
--- a/provider-kimi/Cargo.lock
+++ b/provider-kimi/Cargo.lock
@@ -717,7 +717,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-kimi/src/register.rs b/provider-kimi/src/register.rs
index fbde20040..b7853741c 100644
--- a/provider-kimi/src/register.rs
+++ b/provider-kimi/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
 };
@@ -114,14 +115,29 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .build()
         .expect("reqwest client");
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone()),
+            make_stream(iii.clone(), http.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        // Control-plane callback (router::abort fan-out) — hide from the
+        // agent-facing catalog like the other providers' abort functions.
+        .metadata(serde_json::json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone(), http.clone()))
diff --git a/provider-kimi/src/sse.rs b/provider-kimi/src/sse.rs
index dd933cfe1..d8828d992 100644
--- a/provider-kimi/src/sse.rs
+++ b/provider-kimi/src/sse.rs
@@ -257,7 +257,7 @@ pub fn handle_chunk(
                 }
                 state.thinking.push_str(reasoning);
                 events.push(AssistantMessageEvent::ThinkingDelta {
-                    partial: build_partial(state, model),
+                    partial: None,
                     delta: reasoning.to_string(),
                 });
             }
@@ -273,7 +273,7 @@ pub fn handle_chunk(
                 }
                 state.text.push_str(text);
                 events.push(AssistantMessageEvent::TextDelta {
-                    partial: build_partial(state, model),
+                    partial: None,
                     delta: text.to_string(),
                 });
             }
@@ -306,7 +306,7 @@ pub fn handle_chunk(
                     if !args.is_empty() {
                         state.function_calls[index].args_json.push_str(args);
                         events.push(AssistantMessageEvent::FunctioncallDelta {
-                            partial: build_partial(state, model),
+                            partial: None,
                             delta: args.to_string(),
                             id: state.function_calls[index].id.clone(),
                         });
diff --git a/provider-kimi/src/stream_fn.rs b/provider-kimi/src/stream_fn.rs
index 3af717c53..de8fba013 100644
--- a/provider-kimi/src/stream_fn.rs
+++ b/provider-kimi/src/stream_fn.rs
@@ -12,6 +12,8 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
+use llm_router::provider_scaffold::pump::pump_abortable;
 use llm_router::types::events::{AssistantMessageEvent, ErrorKind};
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
 use std::time::Duration;
@@ -23,15 +25,24 @@ pub const PING_INTERVAL: Duration = Duration::from_secs(30);
 pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http) = (iii.clone(), http.clone());
+        let (iii, http, aborts) = (iii.clone(), http.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -47,6 +58,7 @@ fn send_event(sink: &dyn FrameSink, ev: &AssistantMessageEvent) -> Result<(), ()
 async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -112,6 +124,10 @@ async fn run_stream_call(
     });
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -122,7 +138,13 @@ async fn run_stream_call(
             warnings,
         },
     );
-    pump(rx, sink, PING_INTERVAL).await;
+    // kimi discards the terminal error_kind (no cached credential to drop).
+    match abort_reg {
+        Some(g) => {
+            pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await;
+        }
+        None => pump(rx, sink, PING_INTERVAL).await,
+    }
 }
 
 /// Forward upstream events to the sink; ping through silence; stop on the
diff --git a/provider-kimi/src/surface.rs b/provider-kimi/src/surface.rs
index fedead985..fd96868d2 100644
--- a/provider-kimi/src/surface.rs
+++ b/provider-kimi/src/surface.rs
@@ -9,8 +9,8 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::kimi::stream";
@@ -18,6 +18,10 @@ pub const STREAM_DESC: &str =
     "Stream a Kimi (Moonshot) chat completion: resolve credentials, call \
      the upstream Chat Completions API, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::kimi::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::kimi::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Refresh the Kimi catalog slice from GET /v1/models and \
      reconcile it through the router; returns the model count written.";
@@ -59,6 +63,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
     ]
diff --git a/provider-kimi/src/upstream.rs b/provider-kimi/src/upstream.rs
index 865e2f41b..22f1845f2 100644
--- a/provider-kimi/src/upstream.rs
+++ b/provider-kimi/src/upstream.rs
@@ -23,7 +23,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-kimi/tests/golden/schemas/provider.kimi.abort.json b/provider-kimi/tests/golden/schemas/provider.kimi.abort.json
new file mode 100644
index 000000000..4239f7704
--- /dev/null
+++ b/provider-kimi/tests/golden/schemas/provider.kimi.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::kimi::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-kimi/tests/schemas.rs b/provider-kimi/tests/schemas.rs
index 570b57b3c..00bdc2ea9 100644
--- a/provider-kimi/tests/schemas.rs
+++ b/provider-kimi/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::kimi::stream",
+            "provider::kimi::abort",
             "provider::kimi::refresh_models",
             "provider::kimi::on_router_ready",
         ]
diff --git a/provider-llamacpp/Cargo.lock b/provider-llamacpp/Cargo.lock
index adc2d990f..67a2c3dc9 100644
--- a/provider-llamacpp/Cargo.lock
+++ b/provider-llamacpp/Cargo.lock
@@ -738,7 +738,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-llamacpp/src/register.rs b/provider-llamacpp/src/register.rs
index fe79aa0e0..53135c770 100644
--- a/provider-llamacpp/src/register.rs
+++ b/provider-llamacpp/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -124,14 +125,27 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .build()
         .expect("reqwest client");
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cache.clone()),
+            make_stream(iii.clone(), http.clone(), cache.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone(), http.clone()))
diff --git a/provider-llamacpp/src/stream_fn.rs b/provider-llamacpp/src/stream_fn.rs
index 1404e0dac..b721f34fb 100644
--- a/provider-llamacpp/src/stream_fn.rs
+++ b/provider-llamacpp/src/stream_fn.rs
@@ -12,8 +12,9 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
 use llm_router::types::events::ErrorKind;
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
 
@@ -21,15 +22,24 @@ pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cache) = (iii.clone(), http.clone(), cache.clone());
+        let (iii, http, cache, aborts) = (iii.clone(), http.clone(), cache.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, &cache, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, &cache, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -41,6 +51,7 @@ async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -112,6 +123,10 @@ async fn run_stream_call(
     });
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -122,9 +137,13 @@ async fn run_stream_call(
             warnings,
         },
     );
+    let kind = match abort_reg {
+        Some(g) => pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    };
     // An upstream auth terminal means the cached credential was rotated
     // out from under us: drop the cache so the next attempt re-resolves.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if kind == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-llamacpp/src/surface.rs b/provider-llamacpp/src/surface.rs
index b6b4eb7ae..ed98366c1 100644
--- a/provider-llamacpp/src/surface.rs
+++ b/provider-llamacpp/src/surface.rs
@@ -9,8 +9,8 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::llamacpp::stream";
@@ -18,6 +18,10 @@ pub const STREAM_DESC: &str = "Stream a llama.cpp server chat completion: resolv
      (optional — most local servers run with no --api-key), call the upstream Chat Completions \
      API, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::llamacpp::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::llamacpp::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Discover the resolved llama.cpp server's live model \
      catalog (GET /v1/models + /props) and reconcile it through the router; returns the model \
@@ -66,6 +70,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
         spec::<crate::embed::EmbedRequest, crate::embed::EmbedResponse>(EMBED_ID, EMBED_DESC),
diff --git a/provider-llamacpp/src/upstream.rs b/provider-llamacpp/src/upstream.rs
index 3a46bd7a6..081608211 100644
--- a/provider-llamacpp/src/upstream.rs
+++ b/provider-llamacpp/src/upstream.rs
@@ -27,7 +27,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-llamacpp/tests/golden/schemas/provider.llamacpp.abort.json b/provider-llamacpp/tests/golden/schemas/provider.llamacpp.abort.json
new file mode 100644
index 000000000..ed23a12ac
--- /dev/null
+++ b/provider-llamacpp/tests/golden/schemas/provider.llamacpp.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::llamacpp::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-llamacpp/tests/schemas.rs b/provider-llamacpp/tests/schemas.rs
index 78eb91a6c..214905c7d 100644
--- a/provider-llamacpp/tests/schemas.rs
+++ b/provider-llamacpp/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::llamacpp::stream",
+            "provider::llamacpp::abort",
             "provider::llamacpp::refresh_models",
             "provider::llamacpp::on_router_ready",
             "provider::llamacpp::embed",
diff --git a/provider-openai-codex/Cargo.lock b/provider-openai-codex/Cargo.lock
index 130690f74..ece981b36 100644
--- a/provider-openai-codex/Cargo.lock
+++ b/provider-openai-codex/Cargo.lock
@@ -776,7 +776,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-openai-codex/src/register.rs b/provider-openai-codex/src/register.rs
index b83c7bd7d..6779ecc8c 100644
--- a/provider-openai-codex/src/register.rs
+++ b/provider-openai-codex/src/register.rs
@@ -11,6 +11,7 @@ use crate::{auth, router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -140,15 +141,28 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .expect("reqwest client");
     let refresh_state = Arc::new(CatalogRefreshState::default());
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cache.clone()),
+            make_stream(iii.clone(), http.clone(), cache.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC)
         .metadata(json!({ "internal": true })),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(
diff --git a/provider-openai-codex/src/stream_fn.rs b/provider-openai-codex/src/stream_fn.rs
index b4a2591b2..0a95c2f0b 100644
--- a/provider-openai-codex/src/stream_fn.rs
+++ b/provider-openai-codex/src/stream_fn.rs
@@ -15,8 +15,9 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
 use llm_router::types::events::ErrorKind;
 use llm_router::types::router::{
     CredentialSource, ProviderResolveResponse, ProviderStreamInput, ProviderStreamOutput,
@@ -26,15 +27,24 @@ pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cache) = (iii.clone(), http.clone(), cache.clone());
+        let (iii, http, cache, aborts) = (iii.clone(), http.clone(), cache.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, &cache, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, &cache, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             Ok(ProviderStreamOutput { ok: true })
         })
@@ -55,6 +65,7 @@ async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -149,6 +160,10 @@ async fn run_stream_call(
     });
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -159,10 +174,14 @@ async fn run_stream_call(
             warnings,
         },
     );
+    let kind = match abort_reg {
+        Some(g) => pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    };
     // An upstream auth terminal usually means the VAULT token expired (the
     // existing vault flow handles refresh); invalidating the resolve/token
     // cache is harmless — the next attempt just re-fetches both.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if kind == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-openai-codex/src/surface.rs b/provider-openai-codex/src/surface.rs
index ff366a945..3b3d4e0fc 100644
--- a/provider-openai-codex/src/surface.rs
+++ b/provider-openai-codex/src/surface.rs
@@ -9,8 +9,8 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::openai-codex::stream";
@@ -18,6 +18,10 @@ pub const STREAM_DESC: &str = "Stream an OpenAI Codex completion: resolve a Chat
      from the vault, call the upstream Responses API, and relay AssistantMessageEvent frames to \
      writer_ref.";
 
+pub const ABORT_ID: &str = "provider::openai-codex::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::openai-codex::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Fetch the authenticated Codex model catalog and replace \
      this provider's namespaced router slice; returns the active model count.";
@@ -59,6 +63,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
     ]
diff --git a/provider-openai-codex/src/upstream.rs b/provider-openai-codex/src/upstream.rs
index 925565b91..eb3f87a9f 100644
--- a/provider-openai-codex/src/upstream.rs
+++ b/provider-openai-codex/src/upstream.rs
@@ -26,7 +26,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-openai-codex/tests/golden/schemas/provider.openai-codex.abort.json b/provider-openai-codex/tests/golden/schemas/provider.openai-codex.abort.json
new file mode 100644
index 000000000..07f029e14
--- /dev/null
+++ b/provider-openai-codex/tests/golden/schemas/provider.openai-codex.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::openai-codex::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-openai-codex/tests/schemas.rs b/provider-openai-codex/tests/schemas.rs
index bcde3cd92..9377bb5c2 100644
--- a/provider-openai-codex/tests/schemas.rs
+++ b/provider-openai-codex/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::openai-codex::stream",
+            "provider::openai-codex::abort",
             "provider::openai-codex::refresh_models",
             "provider::openai-codex::on_router_ready",
         ]
diff --git a/provider-openai/Cargo.lock b/provider-openai/Cargo.lock
index f2ba4357f..1fb6a60a1 100644
--- a/provider-openai/Cargo.lock
+++ b/provider-openai/Cargo.lock
@@ -776,7 +776,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-openai/src/register.rs b/provider-openai/src/register.rs
index 0c3501b01..1be350f72 100644
--- a/provider-openai/src/register.rs
+++ b/provider-openai/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -133,15 +134,28 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .build()
         .expect("reqwest client");
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cache.clone()),
+            make_stream(iii.clone(), http.clone(), cache.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC)
         .metadata(json!({ "internal": true })),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone(), http.clone()))
diff --git a/provider-openai/src/stream_fn.rs b/provider-openai/src/stream_fn.rs
index 0484a66ac..720dc685b 100644
--- a/provider-openai/src/stream_fn.rs
+++ b/provider-openai/src/stream_fn.rs
@@ -13,8 +13,9 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
 use llm_router::types::events::ErrorKind;
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
 
@@ -39,15 +40,24 @@ pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cache) = (iii.clone(), http.clone(), cache.clone());
+        let (iii, http, cache, aborts) = (iii.clone(), http.clone(), cache.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, &cache, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, &cache, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -59,6 +69,7 @@ async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -154,6 +165,10 @@ async fn run_stream_call(
     );
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -164,9 +179,13 @@ async fn run_stream_call(
             warnings,
         },
     );
+    let kind = match abort_reg {
+        Some(g) => pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    };
     // An upstream auth terminal means the cached credential was rotated
     // out from under us: drop the cache so the next attempt re-resolves.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if kind == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-openai/src/surface.rs b/provider-openai/src/surface.rs
index a11e1951a..96dbac5d4 100644
--- a/provider-openai/src/surface.rs
+++ b/provider-openai/src/surface.rs
@@ -9,8 +9,8 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::openai::stream";
@@ -18,6 +18,10 @@ pub const STREAM_DESC: &str =
     "Stream an OpenAI response: resolve credentials, call the configured \
      Responses or Chat Completions endpoint, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::openai::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::openai::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Refresh the OpenAI catalog slice from GET /v1/models and \
      reconcile it through the router; returns the model count written.";
@@ -64,6 +68,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
         spec::<crate::embed::EmbedRequest, crate::embed::EmbedResponse>(EMBED_ID, EMBED_DESC),
diff --git a/provider-openai/src/upstream.rs b/provider-openai/src/upstream.rs
index 6752fc942..d06b5268f 100644
--- a/provider-openai/src/upstream.rs
+++ b/provider-openai/src/upstream.rs
@@ -27,7 +27,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-openai/tests/golden/schemas/provider.openai.abort.json b/provider-openai/tests/golden/schemas/provider.openai.abort.json
new file mode 100644
index 000000000..3ff7f67fe
--- /dev/null
+++ b/provider-openai/tests/golden/schemas/provider.openai.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::openai::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-openai/tests/schemas.rs b/provider-openai/tests/schemas.rs
index fee545de2..000f527af 100644
--- a/provider-openai/tests/schemas.rs
+++ b/provider-openai/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::openai::stream",
+            "provider::openai::abort",
             "provider::openai::refresh_models",
             "provider::openai::on_router_ready",
             "provider::openai::embed",
diff --git a/provider-xai/Cargo.lock b/provider-xai/Cargo.lock
index 3bafd9883..f8d1ab3a6 100644
--- a/provider-xai/Cargo.lock
+++ b/provider-xai/Cargo.lock
@@ -776,7 +776,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-xai/src/register.rs b/provider-xai/src/register.rs
index b66482c1d..ef032da80 100644
--- a/provider-xai/src/register.rs
+++ b/provider-xai/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -166,15 +167,34 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         });
     }
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cell.clone(), cache.clone()),
+            make_stream(
+                iii.clone(),
+                http.clone(),
+                cell.clone(),
+                cache.clone(),
+                aborts.clone(),
+            ),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC)
         .metadata(json!({ "internal": true })),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone(), http.clone()))
diff --git a/provider-xai/src/responses.rs b/provider-xai/src/responses.rs
index d7cbe85a4..99d26f040 100644
--- a/provider-xai/src/responses.rs
+++ b/provider-xai/src/responses.rs
@@ -288,7 +288,16 @@ pub fn spawn_responses_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_responses_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_responses_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-xai/src/stream_fn.rs b/provider-xai/src/stream_fn.rs
index 2809bc30d..ce1b2acd6 100644
--- a/provider-xai/src/stream_fn.rs
+++ b/provider-xai/src/stream_fn.rs
@@ -17,26 +17,52 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
-use llm_router::types::events::ErrorKind;
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
+use llm_router::types::events::{AssistantMessageEvent, ErrorKind};
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
+use tokio::sync::{mpsc, watch};
 
 pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cell: ConfigCell,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cell, cache) = (iii.clone(), http.clone(), cell.clone(), cache.clone());
+        let (iii, http, cell, cache, aborts) = (
+            iii.clone(),
+            http.clone(),
+            cell.clone(),
+            cache.clone(),
+            aborts.clone(),
+        );
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
             let wc = { cell.read().await.clone() };
-            run_stream_call(&iii, http, &cache, &wc, input, sink.as_ref()).await;
+            run_stream_call(
+                &iii,
+                http,
+                &cache,
+                abort_reg.as_ref(),
+                &wc,
+                input,
+                sink.as_ref(),
+            )
+            .await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -44,10 +70,26 @@ pub fn make_stream(
     }
 }
 
+/// Pump the upstream, abortable via `provider::xai::abort` while it's live.
+/// Takes a receiver cloned from the call's `AbortGuard`, whose Drop in
+/// `make_stream` deregisters the entry. Returns the pump's terminal
+/// `ErrorKind`.
+async fn pump_stream(
+    rx: mpsc::Receiver<AssistantMessageEvent>,
+    sink: &dyn FrameSink,
+    abort_rx: Option<watch::Receiver<bool>>,
+) -> Option<ErrorKind> {
+    match abort_rx {
+        Some(abort_rx) => pump_abortable(rx, sink, PING_INTERVAL, abort_rx).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    }
+}
+
 async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     wc: &crate::config::WorkerConfig,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
@@ -127,6 +169,10 @@ async fn run_stream_call(
             cfg.max_tokens,
         );
         let headers = build_headers(&cfg);
+        // Aborted while we were setting up — never start the upstream request.
+        if abort_reg.is_some_and(|g| g.is_fired()) {
+            return;
+        }
         let rx = spawn_responses_upstream(
             http,
             ResponsesUpstreamArgs {
@@ -139,7 +185,8 @@ async fn run_stream_call(
         );
         // An upstream auth terminal means the cached credential was rotated
         // out from under us: drop the cache so the next attempt re-resolves.
-        if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+        if pump_stream(rx, sink, abort_reg.map(|g| g.watch())).await == Some(ErrorKind::AuthExpired)
+        {
             cache.invalidate();
         }
         return;
@@ -184,6 +231,10 @@ async fn run_stream_call(
     });
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -196,7 +247,7 @@ async fn run_stream_call(
     );
     // An upstream auth terminal means the cached credential was rotated
     // out from under us: drop the cache so the next attempt re-resolves.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if pump_stream(rx, sink, abort_reg.map(|g| g.watch())).await == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-xai/src/surface.rs b/provider-xai/src/surface.rs
index 5caf0df81..0dd5f8f8e 100644
--- a/provider-xai/src/surface.rs
+++ b/provider-xai/src/surface.rs
@@ -9,14 +9,18 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::xai::stream";
 pub const STREAM_DESC: &str = "Stream an xAI chat completion: resolve credentials, call the \
      upstream Chat Completions API, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::xai::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::xai::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Refresh the xAI catalog slice from GET /v1/models and \
      reconcile it through the router; returns the model count written.";
@@ -58,6 +62,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
     ]
diff --git a/provider-xai/src/upstream.rs b/provider-xai/src/upstream.rs
index 52ec081ce..59ead3b6a 100644
--- a/provider-xai/src/upstream.rs
+++ b/provider-xai/src/upstream.rs
@@ -26,7 +26,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-xai/tests/golden/schemas/provider.xai.abort.json b/provider-xai/tests/golden/schemas/provider.xai.abort.json
new file mode 100644
index 000000000..d4b2e9bbf
--- /dev/null
+++ b/provider-xai/tests/golden/schemas/provider.xai.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::xai::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-xai/tests/schemas.rs b/provider-xai/tests/schemas.rs
index ae1d9736f..731901ad6 100644
--- a/provider-xai/tests/schemas.rs
+++ b/provider-xai/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::xai::stream",
+            "provider::xai::abort",
             "provider::xai::refresh_models",
             "provider::xai::on_router_ready",
         ]
diff --git a/provider-zai/Cargo.lock b/provider-zai/Cargo.lock
index b1786d0f5..5ae1612b8 100644
--- a/provider-zai/Cargo.lock
+++ b/provider-zai/Cargo.lock
@@ -776,7 +776,7 @@ checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"
 
 [[package]]
 name = "llm-router"
-version = "1.1.3"
+version = "1.2.0"
 dependencies = [
  "async-trait",
  "clap",
diff --git a/provider-zai/src/register.rs b/provider-zai/src/register.rs
index dd074a7ba..812f71df0 100644
--- a/provider-zai/src/register.rs
+++ b/provider-zai/src/register.rs
@@ -9,6 +9,7 @@ use crate::{router_client, state, PROVIDER_ID};
 use iii_sdk::errors::Error;
 use iii_sdk::protocol::RegisterTriggerInput;
 use iii_sdk::{IIIClient, RegisterFunction};
+use llm_router::provider_scaffold::aborts::{make_abort, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
 use llm_router::types::router::{
     ProviderDeclaration, ProviderDefaults, ProviderReadyAck, RouterReadyEvent,
@@ -136,15 +137,28 @@ pub async fn register_provider(iii: IIIClient) -> Result<(), Error> {
         .build()
         .expect("reqwest client");
 
+    // request_id → live upstream cancel, shared by stream (registers) and
+    // abort (signals) — see llm_router::provider_scaffold::aborts.
+    let aborts = StreamAborts::new();
+
     iii.register_function(
         surface::STREAM_ID,
         RegisterFunction::new_async_with_bad_request(
-            make_stream(iii.clone(), http.clone(), cache.clone()),
+            make_stream(iii.clone(), http.clone(), cache.clone(), aborts.clone()),
             invalid_request_from_serde,
         )
         .description(surface::STREAM_DESC)
         .metadata(json!({ "internal": true })),
     );
+    iii.register_function(
+        surface::ABORT_ID,
+        RegisterFunction::new_async_with_bad_request(
+            make_abort(aborts),
+            invalid_request_from_serde,
+        )
+        .description(surface::ABORT_DESC)
+        .metadata(json!({ "internal": true })),
+    );
     iii.register_function(
         surface::REFRESH_MODELS_ID,
         RegisterFunction::new_async(make_refresh_models(iii.clone()))
diff --git a/provider-zai/src/stream_fn.rs b/provider-zai/src/stream_fn.rs
index e6ae888f2..d30940cfd 100644
--- a/provider-zai/src/stream_fn.rs
+++ b/provider-zai/src/stream_fn.rs
@@ -13,8 +13,9 @@ use iii_sdk::errors::Error;
 use iii_sdk::IIIClient;
 use llm_router::channels::open_sink;
 use llm_router::chat::relay::FrameSink;
+use llm_router::provider_scaffold::aborts::{AbortGuard, StreamAborts};
 use llm_router::provider_scaffold::cache::ScaffoldCache;
-use llm_router::provider_scaffold::pump::{pump, send_event, PING_INTERVAL};
+use llm_router::provider_scaffold::pump::{pump, pump_abortable, send_event, PING_INTERVAL};
 use llm_router::types::events::ErrorKind;
 use llm_router::types::router::{ProviderStreamInput, ProviderStreamOutput};
 
@@ -22,15 +23,24 @@ pub fn make_stream(
     iii: IIIClient,
     http: reqwest::Client,
     cache: ScaffoldCache,
+    aborts: StreamAborts,
 ) -> impl Fn(ProviderStreamInput) -> BoxFuture<'static, Result<ProviderStreamOutput, Error>>
        + Send
        + Sync
        + 'static {
     move |input: ProviderStreamInput| {
-        let (iii, http, cache) = (iii.clone(), http.clone(), cache.clone());
+        let (iii, http, cache, aborts) = (iii.clone(), http.clone(), cache.clone(), aborts.clone());
         Box::pin(async move {
+            // Register BEFORE the first await: an abort landing while the sink
+            // opens must latch, not hit an unknown id. The RAII guard
+            // deregisters on every exit — early returns and an executor
+            // cancelling this future mid-await alike.
+            let abort_reg = input
+                .resolution_key
+                .as_ref()
+                .map(|rid| aborts.register(rid));
             let sink = open_sink(&iii, &input.writer_ref).await?;
-            run_stream_call(&iii, http, &cache, input, sink.as_ref()).await;
+            run_stream_call(&iii, http, &cache, abort_reg.as_ref(), input, sink.as_ref()).await;
             sink.close();
             // ProviderStreamOutput (spec § stream contract)
             Ok(ProviderStreamOutput { ok: true })
@@ -42,6 +52,7 @@ async fn run_stream_call(
     iii: &IIIClient,
     http: reqwest::Client,
     cache: &ScaffoldCache,
+    abort_reg: Option<&AbortGuard>,
     input: ProviderStreamInput,
     sink: &dyn FrameSink,
 ) {
@@ -133,6 +144,10 @@ async fn run_stream_call(
     });
     let headers = build_headers(&cfg);
 
+    // Aborted while we were setting up — never start the upstream request.
+    if abort_reg.is_some_and(|g| g.is_fired()) {
+        return;
+    }
     let rx = spawn_upstream(
         http,
         UpstreamArgs {
@@ -143,9 +158,13 @@ async fn run_stream_call(
             warnings,
         },
     );
+    let kind = match abort_reg {
+        Some(g) => pump_abortable(rx, sink, PING_INTERVAL, g.watch()).await,
+        None => pump(rx, sink, PING_INTERVAL).await,
+    };
     // An upstream auth terminal means the cached credential was rotated
     // out from under us: drop the cache so the next attempt re-resolves.
-    if pump(rx, sink, PING_INTERVAL).await == Some(ErrorKind::AuthExpired) {
+    if kind == Some(ErrorKind::AuthExpired) {
         cache.invalidate();
     }
 }
diff --git a/provider-zai/src/surface.rs b/provider-zai/src/surface.rs
index b98d51970..aa41c03f0 100644
--- a/provider-zai/src/surface.rs
+++ b/provider-zai/src/surface.rs
@@ -9,14 +9,18 @@
 //! registration emits.
 
 use llm_router::types::router::{
-    ProviderReadyAck, ProviderStreamInput, ProviderStreamOutput, RefreshModelsRequest,
-    RefreshModelsResponse, RouterReadyEvent,
+    ProviderAbortRequest, ProviderAbortResponse, ProviderReadyAck, ProviderStreamInput,
+    ProviderStreamOutput, RefreshModelsRequest, RefreshModelsResponse, RouterReadyEvent,
 };
 
 pub const STREAM_ID: &str = "provider::zai::stream";
 pub const STREAM_DESC: &str = "Stream a Z.AI chat completion: resolve credentials, call the \
      upstream Chat Completions API, and relay AssistantMessageEvent frames to writer_ref.";
 
+pub const ABORT_ID: &str = "provider::zai::abort";
+pub const ABORT_DESC: &str = "Cancel the in-flight upstream stream for a request_id \
+     (router::abort fan-out), stopping billed generation immediately.";
+
 pub const REFRESH_MODELS_ID: &str = "provider::zai::refresh_models";
 pub const REFRESH_MODELS_DESC: &str = "Reconcile the curated Z.AI catalog slice through the \
      router (Z.AI has no models-listing endpoint); returns the model count written.";
@@ -58,6 +62,7 @@ where
 pub fn catalog() -> Vec<FunctionSpec> {
     vec![
         spec::<ProviderStreamInput, ProviderStreamOutput>(STREAM_ID, STREAM_DESC),
+        spec::<ProviderAbortRequest, ProviderAbortResponse>(ABORT_ID, ABORT_DESC),
         spec::<RefreshModelsRequest, RefreshModelsResponse>(REFRESH_MODELS_ID, REFRESH_MODELS_DESC),
         spec::<RouterReadyEvent, ProviderReadyAck>(ON_ROUTER_READY_ID, ON_ROUTER_READY_DESC),
     ]
diff --git a/provider-zai/src/upstream.rs b/provider-zai/src/upstream.rs
index 46dccbb83..845baf663 100644
--- a/provider-zai/src/upstream.rs
+++ b/provider-zai/src/upstream.rs
@@ -27,7 +27,16 @@ pub fn spawn_upstream(
 ) -> mpsc::Receiver<AssistantMessageEvent> {
     let (tx, rx) = mpsc::channel(64);
     tokio::spawn(async move {
-        run_upstream(client, args, tx).await;
+        // Race the call against receiver-side closure: send errors alone only
+        // observe a dropped receiver at the next send, so a silent upstream
+        // (parked in a chunk read, nothing to send) would otherwise keep the
+        // HTTP stream — and billed generation — alive until the next frame or
+        // the read timeout. This makes the module contract above immediate.
+        let closed = tx.clone();
+        tokio::select! {
+            _ = run_upstream(client, args, tx) => {}
+            _ = closed.closed() => {}
+        }
     });
     rx
 }
diff --git a/provider-zai/tests/golden/schemas/provider.zai.abort.json b/provider-zai/tests/golden/schemas/provider.zai.abort.json
new file mode 100644
index 000000000..de047884b
--- /dev/null
+++ b/provider-zai/tests/golden/schemas/provider.zai.abort.json
@@ -0,0 +1,32 @@
+{
+  "description": "Cancel the in-flight upstream stream for a request_id (router::abort fan-out), stopping billed generation immediately.",
+  "function_id": "provider::zai::abort",
+  "request_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Input of a provider's `provider::<id>::abort`: actively cancel the in-flight upstream stream for `request_id` (the router's `request_id`, delivered to the provider as `resolution_key`) so billed generation stops immediately instead of waiting for the provider to notice the closed channel on its next write.",
+    "properties": {
+      "request_id": {
+        "type": "string"
+      }
+    },
+    "required": [
+      "request_id"
+    ],
+    "title": "ProviderAbortRequest",
+    "type": "object"
+  },
+  "response_schema": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "description": "Output of `provider::<id>::abort`. `aborted: false` means the request was unknown — already finished, never started, or aborted before (idempotent).",
+    "properties": {
+      "aborted": {
+        "type": "boolean"
+      }
+    },
+    "required": [
+      "aborted"
+    ],
+    "title": "ProviderAbortResponse",
+    "type": "object"
+  }
+}
diff --git a/provider-zai/tests/schemas.rs b/provider-zai/tests/schemas.rs
index b96726b3d..e38efd4d8 100644
--- a/provider-zai/tests/schemas.rs
+++ b/provider-zai/tests/schemas.rs
@@ -40,6 +40,7 @@ fn catalog_lists_all_functions_in_registration_order() {
         ids,
         vec![
             "provider::zai::stream",
+            "provider::zai::abort",
             "provider::zai::refresh_models",
             "provider::zai::on_router_ready",
         ]