diff --git a/crates/libsy/README.md b/crates/libsy/README.md
index 5ac6ac835..3e6beaf37 100644
--- a/crates/libsy/README.md
+++ b/crates/libsy/README.md
@@ -114,13 +114,23 @@ the id it calls — they can differ (`"strong"` → `"openai/gpt-4o"`) or coinci
 
 ## Running a request
 
-Hold the algorithm as `Arc<dyn Algorithm>` and choose one of two entry points:
+Hold the algorithm as `Arc<dyn Algorithm>` and pick an entry point. They vary on two
+independent axes: **who serves the model calls**, and **whether the final call is made at
+all**.
+
+|  | libsy serves the calls | you serve the calls |
+|---|---|---|
+| **serve the final call** | `run` | `run_stream` |
+| **hand the final call back** | `decide` | `run_decision_only_stream` |
 
 ```rust
 // run: libsy drives the request to completion, serving each call with the target's
 // client, and returns (trace, response). Errors if a routed target has no client.
 let (trace, response) = algo.clone().run(Context::default(), req).await?;
 
+// decide: same, but stops one step short — see "Deciding without calling" below.
+let (trace, decided) = algo.clone().decide(Context::default(), req).await?;
+
 // run_stream: "ask, don't call" — you drive the stream and make the calls.
 let stream = algo.clone().run_stream(Context::default(), req);
 ```
@@ -192,6 +202,67 @@ while let Some(step) = stream.next().await {
 }
 ```
 
+## Deciding without calling (`decide`)
+
+Sometimes you want the routing answer, not the completion — you have your own transport,
+a cache to check first, or a proxy that will forward the request itself. `decide` runs the
+algorithm normally and stops one step short: instead of a `Response` you get the
+**decision, the request to serve it with, and any response already obtained**.
+
+"One step short" means libsy stops *committing* to the call, not that no model was called.
+Deciding routinely costs model calls of its own and those still happen — so whether the
+selected model has already been called depends on how the algorithm decides:
+
+- **Deciding from the request** — a judge scores the prompt and picks a tier. The judge is
+  called; the selected model is not. `response` is `None`.
+- **Deciding from a response** — the algorithm needs the model's *output* to decide, so it
+  calls one and analyzes the answer (escalate if it looks weak, keep it if it doesn't).
+  That call already happened, and `response` is `Some` — the selected model's answer.
+
+The routed call is the only thing left unmade. The decision still binds whatever state the
+algorithm retains — session affinity latches, and later turns follow that assignment
+whether or not you served this one — exactly as under `run`. Read `decide` as "route this
+turn, I will serve it myself", not "what would you do if I asked".
+
+```rust
+let (trace, (decision, request, response)) = algo.clone().decide(Context::default(), req).await?;
+println!("route to {}", decision.selected_model());
+
+match response {
+    // The selected model has already answered — deciding needed its output. Use this
+    // response as-is, or drop it and call `decision.selected_model()` again; both are
+    // valid, it is your cost/latency tradeoff.
+    Some(response) => { /* use it, or re-call */ }
+    // Not called yet: serve `request` against `decision.selected_model()` yourself.
+    None => { /* your call */ }
+}
+```
+
+Either way `response`, when present, corresponds to `decision` — it is that target's
+answer to `request`, not some intermediate the algorithm discarded.
+
+`run_decision_only_stream` is the "you serve the calls" form: the same `CallLlm` /
+`Decision` steps as `run_stream`, ending in `DecisionOnlyStep::ReturnToAgent` carrying that
+same triple.
+
+An algorithm does not branch on any of this. The mode is fixed by the entry point, recorded
+on the `Driver`, and applied by `Driver::final_decision` — so an algorithm that ends on
+`final_decision` supports all four entry points without mentioning them:
+
+```rust
+// the last thing create_run_task does: conclude on the winning target
+driver.final_decision(ctx, &target, request, decision, &mut already_served).await
+```
+
+`already_served` is an `Option<Response>` the algorithm may have picked up on the way (a
+classifier whose deciding call also answered the turn hands it back through
+`Classifier::score`). It is borrowed, not moved, because `Response` is not `Clone` —
+`final_decision` takes it only on the branch that consumes it.
+
+An algorithm that answers without routing — `Noop`, or one that builds a `Response`
+directly rather than concluding through `final_decision` — has no route to hand back, so
+`decide` on it fails with `LibsyError::AlgorithmError`.
+
 ## Building an algorithm (`Algorithm`)
 
 Implement `Algorithm` to add a strategy. You write `create_run_task` — one call per
@@ -206,11 +277,13 @@ pub trait Algorithm: Send + Sync + 'static {
     fn name(&self) -> &str;
     // `self: Arc<Self>` (not `&mut`): one algorithm serves requests concurrently — use
     // interior mutability for state. Offload calls/decisions on `driver`.
+    // Ends on `driver.final_decision(..)`, which yields `Response` or `Decision` according
+    // to the run's mode — so one implementation serves every entry point.
     async fn create_run_task(self: Arc<Self>, ctx: Context, driver: Driver, request: Request)
-        -> switchyard_libsy::Result<Response>;
+        -> switchyard_libsy::Result<ResponseOrDecision>;
     async fn process_signals(self: Arc<Self>, signals: Signals)
         -> switchyard_libsy::Result<()>;
-    // provided: run(ctx, request) -> (trace, response), run_stream(ctx, request) -> Stream<Step>
+    // provided: run / decide -> (trace, ..), run_stream / run_decision_only_stream -> Stream<..>
 }
 
 pub trait Decision: Send + Sync {
diff --git a/crates/libsy/examples/ensemble.rs b/crates/libsy/examples/ensemble.rs
index 27520403f..3742c6c2f 100644
--- a/crates/libsy/examples/ensemble.rs
+++ b/crates/libsy/examples/ensemble.rs
@@ -24,7 +24,9 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use parking_lot::Mutex;
 
-use switchyard_libsy::{Algorithm, Driver, LibsyError, LlmTarget, LlmTargetSet, Result};
+use switchyard_libsy::{
+    Algorithm, Driver, LibsyError, LlmTarget, LlmTargetSet, ResponseOrDecision, Result,
+};
 use switchyard_llm_client::{Backend, HttpBackendConfig, ModelConfig, TranslatingLlmClient};
 use switchyard_protocol::{
     Context, Decision, Request, Response, RoutedLlmClient, completion_text, prompt_text,
@@ -181,7 +183,7 @@ impl EnsembleOrchAlgo {
         ctx: Context,
         request: Request,
         model: String,
-    ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
+    ) -> Result<(Vec<Arc<dyn Decision>>, ResponseOrDecision)> {
         let target = self.target_set.get_target(&model)?;
         let decision: Arc<dyn Decision> = Arc::new(EnsembleDecision {
             reasoning: format!(
@@ -201,20 +203,21 @@ impl EnsembleOrchAlgo {
             raw_request: request.raw_request,
             metadata: request.metadata,
         };
+        // The committed model is this run's final call, so it concludes the run.
         let response = driver
-            .call_llm_target(ctx, &target, routed, decision.clone())
+            .final_decision(ctx, &target, routed, decision.clone(), &mut None)
             .await?;
         Ok((vec![decision], response))
     }
 
     /// One exploration turn: fan out to every candidate, judge the survivors,
-    /// tally the winner, and return its response.
+    /// tally the winner, and conclude on its response.
     async fn ensemble_turn(
         &self,
         driver: &Driver,
         ctx: Context,
         request: Request,
-    ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
+    ) -> Result<(Vec<Arc<dyn Decision>>, ResponseOrDecision)> {
         let user_prompt = prompt_text(&request.llm_request);
         // The agent's inbound name rides through every sub-call unchanged; the model
         // each call hits is carried by its decision, not stamped onto the request.
@@ -284,7 +287,12 @@ impl EnsembleOrchAlgo {
                 metadata: request.metadata.clone(),
             };
             let judge_response = driver
-                .call_llm_target(ctx, &judge_target, judge_request, judge_decision.clone())
+                .call_llm_target(
+                    ctx.clone(),
+                    &judge_target,
+                    judge_request,
+                    judge_decision.clone(),
+                )
                 .await?;
             // Fail open: an unparseable pick falls back to the first response.
             let choice = parse_choice(
@@ -310,19 +318,38 @@ impl EnsembleOrchAlgo {
             state.turns += 1;
         }
 
+        let winner_target = self.target_set.get_target(&winner_model)?;
         let winner_decision: Arc<dyn Decision> = Arc::new(EnsembleDecision {
             reasoning: format!("judge selected '{winner_model}' as best response"),
             selected_model: winner_model,
             phase: EnsemblePhase::Winner,
         });
 
+        // The winner was already served as one of the candidate calls, so concluding
+        // here hands that response back rather than paying for the turn twice. The
+        // request is the one the winner actually answered, not the raw inbound.
+        let winner_request = Request {
+            llm_request: text_request(inbound, user_prompt),
+            raw_request: request.raw_request,
+            metadata: request.metadata,
+        };
+        let terminal = driver
+            .final_decision(
+                ctx,
+                &winner_target,
+                winner_request,
+                winner_decision.clone(),
+                &mut Some(winner_response),
+            )
+            .await?;
+
         // Trace order: [candidate calls..., judge?, winner].
         let mut trace = candidate_decisions;
         if let Some(judge_decision) = judge_decision {
             trace.push(judge_decision);
         }
         trace.push(winner_decision);
-        Ok((trace, winner_response))
+        Ok((trace, terminal))
     }
 }
 
@@ -382,11 +409,11 @@ impl Algorithm for EnsembleOrchAlgo {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         // Fast path: exploration is over — route straight to the committed model;
         // otherwise run a full ensemble turn. Both return a decision trace plus the
         // final response.
-        let (trace, response) = if let Some(model) = self.resolve_committed()? {
+        let (trace, terminal) = if let Some(model) = self.resolve_committed()? {
             self.route_committed(&driver, ctx.clone(), request, model)
                 .await?
         } else {
@@ -397,7 +424,7 @@ impl Algorithm for EnsembleOrchAlgo {
         for decision in trace {
             driver.info(ctx.clone(), decision).await?;
         }
-        Ok(response)
+        Ok(terminal)
     }
 }
 
@@ -450,7 +477,7 @@ async fn main() -> Result<()> {
         metadata: None,
     };
 
-    let (_, response) = algorithm.run(Context::default(), request).await?;
+    let (_, response) = algorithm.run(Context::default(), request, None).await?;
     println!(
         "{}",
         completion_text(
@@ -604,7 +631,7 @@ mod tests {
         // Judge prefers b/model; it should win and be returned.
         let (algo, calls) = algo(&["a/model", "b/model"], "judge/haiku", "b/model", 100);
         let (trace, response) = orch(algo)
-            .run(Context::default(), request("solve it"))
+            .run(Context::default(), request("solve it"), None)
             .await?;
         assert_eq!(
             response
@@ -639,15 +666,22 @@ mod tests {
         let orch = orch(algo);
 
         // Two exploration turns.
-        orch.clone().run(Context::default(), request("t1")).await?;
-        orch.clone().run(Context::default(), request("t2")).await?;
+        orch.clone()
+            .run(Context::default(), request("t1"), None)
+            .await?;
+        orch.clone()
+            .run(Context::default(), request("t2"), None)
+            .await?;
         let judge_calls_after_exploration =
             calls.lock().iter().filter(|c| *c == "judge/haiku").count();
         assert_eq!(judge_calls_after_exploration, 2);
 
         // Third request: committed fast path — routes straight to b/model with no
         // fan-out to a/model and no judge call.
-        let (trace, response) = orch.clone().run(Context::default(), request("t3")).await?;
+        let (trace, response) = orch
+            .clone()
+            .run(Context::default(), request("t3"), None)
+            .await?;
         assert_eq!(
             response
                 .llm_response
@@ -672,7 +706,9 @@ mod tests {
     #[tokio::test]
     async fn single_candidate_skips_the_judge() -> Result<()> {
         let (algo, calls) = algo(&["only/model"], "judge/haiku", "only/model", 100);
-        let (trace, response) = orch(algo).run(Context::default(), request("hi")).await?;
+        let (trace, response) = orch(algo)
+            .run(Context::default(), request("hi"), None)
+            .await?;
         assert_eq!(
             response
                 .llm_response
@@ -695,7 +731,10 @@ mod tests {
         let (algo, calls) = algo(&["a/model", "b/model"], "judge/haiku", "b/model", 0);
         let orch = orch(algo);
         for _ in 0..3 {
-            let (trace, _) = orch.clone().run(Context::default(), request("x")).await?;
+            let (trace, _) = orch
+                .clone()
+                .run(Context::default(), request("x"), None)
+                .await?;
             // Always a full ensemble turn (never a lone Committed decision).
             assert_eq!(
                 as_ensemble(&trace[trace.len() - 1])?.phase,
@@ -744,7 +783,7 @@ mod tests {
         );
         assert!(
             orch(algo)
-                .run(Context::default(), request("x"))
+                .run(Context::default(), request("x"), None)
                 .await
                 .is_err()
         );
@@ -836,7 +875,7 @@ mod tests {
         let run = |session: Arc<dyn Algorithm>, prompt: &'static str| {
             tokio::spawn(async move {
                 session
-                    .run(Context::default(), request(prompt))
+                    .run(Context::default(), request(prompt), None)
                     .await
                     .map(|(_, response)| {
                         response
@@ -878,15 +917,15 @@ mod tests {
             tokio::spawn(async move {
                 session
                     .clone()
-                    .run(Context::default(), request("t1"))
+                    .run(Context::default(), request("t1"), None)
                     .await?;
                 session
                     .clone()
-                    .run(Context::default(), request("t2"))
+                    .run(Context::default(), request("t2"), None)
                     .await?;
                 let (trace, response) = session
                     .clone()
-                    .run(Context::default(), request("t3"))
+                    .run(Context::default(), request("t3"), None)
                     .await?;
                 let phase = trace
                     .last()
diff --git a/crates/libsy/examples/research_agent.rs b/crates/libsy/examples/research_agent.rs
index 3f2f6d424..187abbf04 100644
--- a/crates/libsy/examples/research_agent.rs
+++ b/crates/libsy/examples/research_agent.rs
@@ -83,7 +83,11 @@ impl ResearchAgent {
                 metadata: None,
             };
 
-            let (_trace, response) = self.algo.clone().run(Context::default(), request).await?;
+            let (_trace, response) = self
+                .algo
+                .clone()
+                .run(Context::default(), request, None)
+                .await?;
             let aggregate = response
                 .llm_response
                 .into_agg()
diff --git a/crates/libsy/src/algorithms/fall_through.rs b/crates/libsy/src/algorithms/fall_through.rs
index 1b41f08e5..6ee5314af 100644
--- a/crates/libsy/src/algorithms/fall_through.rs
+++ b/crates/libsy/src/algorithms/fall_through.rs
@@ -22,7 +22,10 @@ use async_trait::async_trait;
 use parking_lot::Mutex;
 use tokio::sync::Mutex as AsyncMutex;
 
-use crate::core::algorithm::{self, Algorithm, Driver, LlmTarget, LlmTargetSet, SessionEvictions};
+use crate::core::algorithm::{
+    self, Algorithm, Driver, LlmTarget, LlmTargetSet, ResponseOrDecision, SessionEvictions,
+    call_llm_with_overflow_fallback,
+};
 use crate::core::classifier::{Classification, Classifier, Score};
 use crate::core::processor::{Event, Processor};
 use crate::{LibsyError, Result};
@@ -169,7 +172,7 @@ where
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         // The request is threaded mutably through the whole fold: any component may rewrite
         // it, later components see the rewrite, and the final value reaches the model.
         let mut request = request;
@@ -198,19 +201,28 @@ where
         // for twice. There is no outbound call left to overflow, so the fallback is skipped.
         // Nothing reads it on the way out: streamed or buffered, it reaches the caller
         // untouched.
+        // Fixed for the whole run by the entry point that started it.
+        let decision_only = driver.decision_only();
         match served {
-            Some(response) => Ok(response),
-            None => {
-                algorithm::call_llm_with_overflow_fallback(
+            Some(response) if !decision_only => {
+                Ok(ResponseOrDecision::Response(Box::new(response)))
+            }
+            // Everything else concludes through the fallback: it lends the response to each
+            // attempt and takes it only when one concludes, so a retry still has it.
+            served => {
+                call_llm_with_overflow_fallback(
                     ctx,
                     &driver,
                     &self.targets,
                     target,
                     decision,
                     request,
+                    served,
                     session.as_deref(),
                     &self.session_evictions,
                     |from, to| self.fallback_decision(from, to),
+                    // This call is the answer the cascade returns, so it concludes the run.
+                    true,
                 )
                 .await
             }
@@ -277,9 +289,11 @@ where
             });
         };
 
-        // 3. Resolve the target and publish the decision. When an excluded target sends
-        //    the request elsewhere, the tier and reasoning describe where it actually went.
-        let target = self.targets.resolve_target(&score.target, ctx)?;
+        // 3. Resolve the target and publish the decision.
+        let target = match served {
+            Some(_) => self.targets.get_target(&score.target)?,
+            None => self.targets.resolve_target(&score.target, ctx)?,
+        };
         let reasoning = if target.semantic_name == score.target {
             (self.decision_reason)(&self.name, &score)
         } else {
@@ -344,7 +358,7 @@ where
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         self.execute(ctx, driver, request).await
     }
 }
@@ -511,6 +525,7 @@ mod tests {
                     raw_request: None,
                     metadata: None,
                 },
+                None,
             )
             .await?;
         let call = client.0.lock().take();
@@ -557,6 +572,38 @@ mod tests {
         }
     }
 
+    /// A classifier whose decision *is* a model call: it answers the turn while deciding
+    /// it, and hands that response back alongside its score.
+    struct AnswersWhileDeciding {
+        target: String,
+        completion: &'static str,
+    }
+
+    #[async_trait]
+    impl Classifier for AnswersWhileDeciding {
+        async fn score(
+            &self,
+            _state: &mut (),
+            _request: &mut Request,
+            _driver: Option<&Driver>,
+        ) -> Result<(Classification, Option<Response>)> {
+            Ok((
+                Classification::Scores(vec![score(&self.target, 1.0)]),
+                Some(Response {
+                    llm_response: LlmResponse::Agg(text_response(None, self.completion)),
+                    metadata: None,
+                }),
+            ))
+        }
+    }
+
+    fn answers_while_deciding(target: &str, completion: &'static str) -> Arc<dyn Classifier> {
+        Arc::new(AnswersWhileDeciding {
+            target: target.to_string(),
+            completion,
+        })
+    }
+
     fn score(target: &str, confidence: f64) -> Score {
         Score {
             confidence,
@@ -591,7 +638,10 @@ mod tests {
     where
         S: Default + Send + 'static,
     {
-        let (trace, response) = router.clone().run(Context::default(), request).await?;
+        let (trace, response) = router
+            .clone()
+            .run(Context::default(), request, None)
+            .await?;
         let text = response
             .llm_response
             .into_agg()
@@ -912,7 +962,7 @@ mod tests {
         );
         let mut ctx = Context::default();
         ctx.exclude_target("weak");
-        let (trace, response) = router.run(ctx, request()).await?;
+        let (trace, response) = router.run(ctx, request(), None).await?;
         let text = response
             .llm_response
             .into_agg()
@@ -1183,4 +1233,110 @@ mod tests {
         assert_eq!(anonymous2, "weak");
         Ok(())
     }
+
+    // --- a classifier that answers the turn while deciding it ---------------------------
+
+    #[tokio::test]
+    async fn a_classifier_response_answers_the_turn_without_a_second_call() -> Result<()> {
+        // `EchoClient` echoes the target name, so a completion that is not "strong" proves
+        // the routed target was never called and the turn was not paid for twice.
+        let router: Arc<dyn Algorithm> = Arc::new(
+            FallThrough::new(target_set(&["strong"]))
+                .with_classifier(answers_while_deciding("strong", "answered while deciding")),
+        );
+
+        let (trace, response) = router.run(Context::default(), request(), None).await?;
+
+        assert_eq!(
+            response.llm_response.as_agg().map(completion_text),
+            Some("answered while deciding".to_string())
+        );
+        assert_eq!(trace.last().map(|d| d.selected_model()), Some("strong"));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn a_decision_only_run_hands_back_the_classifier_response_with_the_route() -> Result<()> {
+        // The same turn under `decide`: the response the classifier already obtained rides
+        // along with the route rather than being returned on its own.
+        let router: Arc<dyn Algorithm> = Arc::new(
+            FallThrough::new(target_set(&["strong"]))
+                .with_classifier(answers_while_deciding("strong", "answered while deciding")),
+        );
+
+        let (_, (decision, request, response)) =
+            router.decide(Context::default(), request(), None).await?;
+
+        assert_eq!(decision.selected_model(), "strong");
+        assert_eq!(request.requested_model(), Some("auto"));
+        let response = response.ok_or_else(|| test_error("expected the classifier response"))?;
+        assert_eq!(
+            response.llm_response.as_agg().map(completion_text),
+            Some("answered while deciding".to_string())
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn a_classifier_response_keeps_the_target_it_was_answered_by() -> Result<()> {
+        // The scored target is excluded, so ordinary routing would substitute another one.
+        // The classifier has already answered from the target it scored, though, and
+        // substituting here would hand the caller one target's answer under another's name.
+        let router: Arc<dyn Algorithm> = Arc::new(
+            FallThrough::new(target_set(&["weak", "strong"]))
+                .with_classifier(answers_while_deciding("weak", "weak answered")),
+        );
+        let mut ctx = Context::default();
+        ctx.exclude_target("weak");
+
+        let (_, (decision, _, response)) = router.decide(ctx, request(), None).await?;
+
+        assert_eq!(decision.selected_model(), "weak");
+        let response = response.ok_or_else(|| test_error("expected the classifier response"))?;
+        assert_eq!(
+            response.llm_response.as_agg().map(completion_text),
+            Some("weak answered".to_string())
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn a_served_turn_publishes_the_target_that_answered_it() -> Result<()> {
+        // The same substitution under `run`: nothing is returned to the caller but the
+        // answer, so the published decision is all that names the model behind it — cost
+        // and stats key off it.
+        let router = Arc::new(
+            FallThrough::new(target_set(&["weak", "strong"]))
+                .with_classifier(answers_while_deciding("weak", "weak answered")),
+        );
+        let mut ctx = Context::default();
+        ctx.exclude_target("weak");
+
+        let (trace, response) = router.run(ctx, request(), None).await?;
+
+        assert_eq!(trace.last().map(|d| d.selected_model()), Some("weak"));
+        assert_eq!(
+            response.llm_response.as_agg().map(completion_text),
+            Some("weak answered".to_string())
+        );
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn a_decision_only_run_without_a_classifier_response_carries_none() -> Result<()> {
+        // The ordinary cascade: nothing was called before the routing decision, so there is
+        // no response to hand back with it.
+        let router: Arc<dyn Algorithm> = Arc::new(
+            FallThrough::new(target_set(&["weak", "strong"]))
+                .with_classifier(fixed(vec![score("strong", 1.0)])),
+        );
+
+        let (trace, (decision, _, response)) =
+            router.decide(Context::default(), request(), None).await?;
+
+        assert_eq!(decision.selected_model(), "strong");
+        assert!(response.is_none());
+        assert_eq!(trace.last().map(|d| d.selected_model()), Some("strong"));
+        Ok(())
+    }
 }
diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs
index 8e39f0d0f..0ad65a0d9 100644
--- a/crates/libsy/src/algorithms/llm_class.rs
+++ b/crates/libsy/src/algorithms/llm_class.rs
@@ -17,7 +17,7 @@ use super::fall_through::{DefaultTarget, FallThrough};
 use super::util::affinity::AffinityRouter;
 use super::util::escalation::{self, EscalationJudge, EscalationJudgeConfig, EscalationPolicy};
 use super::util::llm_judge::{self, Judge, JudgeClassifier, JudgeConfig, JudgePolicy};
-use crate::core::algorithm::{Algorithm, Driver, LlmTarget, LlmTargetSet};
+use crate::core::algorithm::{Algorithm, Driver, LlmTarget, LlmTargetSet, ResponseOrDecision};
 use crate::core::classifier::{Classification, Classifier, Score};
 use crate::core::state::{State, StateValue};
 use crate::{LibsyError, Result};
@@ -560,7 +560,7 @@ impl Algorithm for LlmTaskClassifier {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         self.route.execute(ctx, driver, request).await
     }
 }
@@ -577,7 +577,8 @@ mod tests {
         LlmClientError, Metadata, completion_text, text_request, text_response,
     };
 
-    use crate::core::algorithm::Algorithm;
+    use crate::{Algorithm, DecisionOnlyStep, RunObservation, RunObserver};
+    use futures::StreamExt;
     use switchyard_protocol::{Context, LlmResponse, Response, RoutedLlmClient};
 
     const TEST_THRESHOLD: f64 = 0.5;
@@ -723,11 +724,161 @@ mod tests {
         request
     }
 
+    // --- decision-only runs -------------------------------------------------------------
+    //
+    // The classifier must consult the judge to decide at all, so these exercise a route
+    // whose decision costs an intermediate model call. What decision-only changes is the
+    // *final* call: it is handed back rather than served.
+
+    #[tokio::test]
+    async fn decide_makes_the_judge_call_but_not_the_routed_one() -> Result<()> {
+        let client = Arc::new(PerRequestClient::default());
+        let router = router(client.clone())?;
+
+        let (trace, (decision, request, response)) = router
+            .decide(Context::default(), classify_request(), None)
+            .await?;
+
+        // The judge is still called — deciding requires it — but the routed target is not:
+        // that call is precisely what the caller is being handed.
+        assert_eq!(client.calls(), vec!["judge"]);
+        assert_eq!(decision.selected_model(), "efficient");
+        assert_eq!(trace.last().map(|d| d.selected_model()), Some("efficient"));
+        // The handed-back request is the one the target should be served with; libsy never
+        // overwrites the agent's inbound model name.
+        assert_eq!(request.requested_model(), Some("auto"));
+        // Nothing served the final call, so there is no response to carry.
+        assert!(response.is_none());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn deciding_reports_the_call_it_made_to_the_observer() -> Result<()> {
+        // The judge call is paid for on a decide run, so it has to reach the observer that
+        // feeds stats. The routed call does not: it never happened here, and reporting it
+        // is the caller's job once they serve it.
+        let client = Arc::new(PerRequestClient::default());
+        let observations = Arc::new(Mutex::new(Vec::new()));
+        let observed = observations.clone();
+        let observer: RunObserver = Arc::new(move |observation| observed.lock().push(observation));
+
+        router(client.clone())?
+            .decide(Context::default(), classify_request(), Some(observer))
+            .await?;
+
+        assert_eq!(client.calls(), vec!["judge"]);
+        let observations = observations.lock();
+        let calls: Vec<_> = observations
+            .iter()
+            .filter_map(|observation| match observation {
+                RunObservation::LlmCall(call) => Some(call),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(calls.len(), 1);
+        assert_eq!(calls[0].selected_model, "judge");
+        // A judge consultation is routing overhead, not the routed call.
+        assert!(!calls[0].is_routed);
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn run_and_decide_agree_on_the_route_but_not_on_serving_it() -> Result<()> {
+        // Same algorithm, same request: `run` serves the routed call, `decide` stops one
+        // step short. The routing decision itself must be identical.
+        let run_client = Arc::new(PerRequestClient::default());
+        let (_, served) = router(run_client.clone())?
+            .run(Context::default(), classify_request(), None)
+            .await?;
+
+        let decide_client = Arc::new(PerRequestClient::default());
+        let (_, (decision, _, _)) = router(decide_client.clone())?
+            .decide(Context::default(), classify_request(), None)
+            .await?;
+
+        assert_eq!(run_client.calls(), vec!["judge", "efficient"]);
+        assert_eq!(decide_client.calls(), vec!["judge"]);
+        assert_eq!(
+            served.llm_response.as_agg().map(completion_text),
+            Some("answer from efficient".to_string())
+        );
+        assert_eq!(decision.selected_model(), "efficient");
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn decision_only_stream_offloads_the_judge_call_then_hands_back_the_route() -> Result<()>
+    {
+        // Driving the stream by hand: the judge call arrives as a `CallLlm` step we serve
+        // ourselves, and the run ends with the route instead of a response.
+        let client = Arc::new(PerRequestClient::default());
+        let stream = router(client.clone())?.run_decision_only_stream(
+            Context::default(),
+            classify_request(),
+            None,
+        );
+        tokio::pin!(stream);
+
+        let mut offloaded = Vec::new();
+        let mut published = Vec::new();
+        let mut decided = None;
+        while let Some(step) = stream.next().await {
+            match step? {
+                DecisionOnlyStep::CallLlm(call) => {
+                    let routed = call.get_routed().clone();
+                    let target = routed.decision.selected_model().to_string();
+                    offloaded.push(target.clone());
+                    let target_client = routed.default_client.clone().ok_or_else(|| {
+                        LibsyError::AlgorithmError {
+                            message: "expected a default client".to_string(),
+                        }
+                    })?;
+                    let result = target_client
+                        .call(routed.ctx, routed.request, routed.decision)
+                        .await
+                        .map_err(|error| LibsyError::client_call(target, error));
+                    call.respond(result)?;
+                }
+                DecisionOnlyStep::Decision(decision) => {
+                    published.push(decision.selected_model().to_string())
+                }
+                DecisionOnlyStep::ReturnToAgent(route) => decided = Some(route),
+            }
+        }
+
+        // Only the judge was offloaded; the routed call never reached the stream.
+        assert_eq!(offloaded, vec!["judge"]);
+        assert_eq!(published, vec!["efficient"]);
+        let (decision, _, response) = decided.ok_or_else(|| LibsyError::AlgorithmError {
+            message: "no ReturnToAgent step".to_string(),
+        })?;
+        assert_eq!(decision.selected_model(), "efficient");
+        assert!(response.is_none());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn a_decision_only_run_survives_an_unreachable_judge() -> Result<()> {
+        // The judge failing is a routing outcome, not a run failure: the cascade falls open
+        // to capable, and decision-only still hands that route back.
+        let router = router(Arc::new(UnreachableJudgeClient))?;
+
+        let (_, (decision, _, response)) = router
+            .decide(Context::default(), classify_request(), None)
+            .await?;
+
+        assert_eq!(decision.selected_model(), "capable");
+        assert!(response.is_none());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn an_unreachable_judge_routes_capable_instead_of_failing_the_request() -> Result<()> {
         let router = router(Arc::new(UnreachableJudgeClient))?;
 
-        let (trace, response) = router.run(Context::default(), classify_request()).await?;
+        let (trace, response) = router
+            .run(Context::default(), classify_request(), None)
+            .await?;
 
         assert_eq!(trace.last().map(|d| d.selected_model()), Some("capable"));
         assert_eq!(
@@ -743,8 +894,14 @@ mod tests {
         let router = router(client.clone())?;
         let request = classify_request;
 
-        router.clone().run(Context::default(), request()).await?;
-        router.clone().run(Context::default(), request()).await?;
+        router
+            .clone()
+            .run(Context::default(), request(), None)
+            .await?;
+        router
+            .clone()
+            .run(Context::default(), request(), None)
+            .await?;
 
         assert_eq!(
             client.calls(),
@@ -772,11 +929,11 @@ mod tests {
 
         router
             .clone()
-            .run(Context::default(), classify_session_request())
+            .run(Context::default(), classify_session_request(), None)
             .await?;
         router
             .clone()
-            .run(Context::default(), classify_session_request())
+            .run(Context::default(), classify_session_request(), None)
             .await?;
 
         assert_eq!(client.calls(), vec!["judge", "efficient", "efficient"]);
@@ -804,11 +961,11 @@ mod tests {
 
         router
             .clone()
-            .run(Context::default(), classify_request())
+            .run(Context::default(), classify_request(), None)
             .await?;
         router
             .clone()
-            .run(Context::default(), classify_follow_up_request())
+            .run(Context::default(), classify_follow_up_request(), None)
             .await?;
 
         assert_eq!(client.calls(), vec!["judge", "efficient", "efficient"]);
@@ -1194,7 +1351,7 @@ mod tests {
         let router = escalation_router(model_client, judge_client)?;
         let request = classify_request();
 
-        let (trace, response) = router.run(Context::default(), request).await?;
+        let (trace, response) = router.run(Context::default(), request, None).await?;
 
         // The efficient model is the serving target, and the response comes from its call.
         assert_eq!(trace.last().map(|d| d.selected_model()), Some("efficient"));
@@ -1214,7 +1371,7 @@ mod tests {
         let router = escalation_router(model_client, judge_client)?;
         let request = classify_request();
 
-        let (trace, response) = router.run(Context::default(), request).await?;
+        let (trace, response) = router.run(Context::default(), request, None).await?;
 
         assert_eq!(trace.last().map(|d| d.selected_model()), Some("capable"));
         assert_eq!(
@@ -1235,11 +1392,11 @@ mod tests {
         let session_request = classify_session_request();
         router
             .clone()
-            .run(Context::default(), session_request.clone())
+            .run(Context::default(), session_request.clone(), None)
             .await?;
         let (trace, _) = router
             .clone()
-            .run(Context::default(), session_request)
+            .run(Context::default(), session_request, None)
             .await?;
 
         assert_eq!(trace.last().map(|d| d.selected_model()), Some("capable"));
diff --git a/crates/libsy/src/algorithms/noop.rs b/crates/libsy/src/algorithms/noop.rs
index 86179a4fc..6c9db886a 100644
--- a/crates/libsy/src/algorithms/noop.rs
+++ b/crates/libsy/src/algorithms/noop.rs
@@ -11,7 +11,7 @@ use switchyard_protocol::{
 };
 
 use crate::Result;
-use crate::core::algorithm::{Algorithm, Driver};
+use crate::core::algorithm::{Algorithm, Driver, ResponseOrDecision};
 use switchyard_protocol::{Context, Decision};
 
 /// A routing algorithm that does not route. It returns a hard-coded response.
@@ -46,7 +46,7 @@ impl Algorithm for Noop {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         let model = request
             .requested_model()
             .unwrap_or("switchyard/noop")
@@ -68,11 +68,12 @@ impl Algorithm for Noop {
             }],
             ..Default::default()
         });
+        // No target and no model call, so there is no final decision to hand back.
         let response = Response {
             llm_response,
             metadata: request.metadata.clone(),
         };
-        Ok(response)
+        Ok(ResponseOrDecision::Response(Box::new(response)))
     }
 }
 
@@ -96,7 +97,7 @@ mod tests {
         };
 
         let a: Arc<dyn Algorithm> = Arc::new(Noop {});
-        let (decisions, response) = a.run(Context::default(), request).await?;
+        let (decisions, response) = a.run(Context::default(), request, None).await?;
         let Some(decision) = decisions.first() else {
             panic!("Expected exactly one Decision");
         };
diff --git a/crates/libsy/src/algorithms/passthrough.rs b/crates/libsy/src/algorithms/passthrough.rs
index 06b73a02d..e76f3c15e 100644
--- a/crates/libsy/src/algorithms/passthrough.rs
+++ b/crates/libsy/src/algorithms/passthrough.rs
@@ -7,10 +7,10 @@
 
 use std::sync::Arc;
 
-use switchyard_protocol::{Request, Response};
+use switchyard_protocol::Request;
 
 use crate::Result;
-use crate::core::algorithm::{Algorithm, Driver, LlmTarget};
+use crate::core::algorithm::{Algorithm, Driver, LlmTarget, ResponseOrDecision};
 use switchyard_protocol::{Context, Decision, RoutedLlmClient};
 
 /// See module comment
@@ -60,13 +60,13 @@ impl Algorithm for Passthrough {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         let decision: Arc<dyn Decision> = Arc::new(PassthroughDecision {
             model_id: self.target.semantic_name.clone(),
         });
         driver.info(ctx.clone(), decision.clone()).await?;
         driver
-            .call_llm_target(ctx, &self.target, request, decision)
+            .final_decision(ctx, &self.target, request, decision, &mut None)
             .await
     }
 }
@@ -113,7 +113,7 @@ mod tests {
             semantic_name: MODEL_ID.to_string(),
             llm_client: Some(Arc::new(EchoClient)),
         }));
-        let (trace, response) = algorithm.run(Context::default(), request).await?;
+        let (trace, response) = algorithm.run(Context::default(), request, None).await?;
 
         assert_eq!(
             response
diff --git a/crates/libsy/src/algorithms/rand.rs b/crates/libsy/src/algorithms/rand.rs
index 2eb93f024..320085258 100644
--- a/crates/libsy/src/algorithms/rand.rs
+++ b/crates/libsy/src/algorithms/rand.rs
@@ -16,7 +16,7 @@ use rand::distr::{Distribution, weighted::WeightedIndex};
 use rand::rngs::StdRng;
 
 use crate::algorithms::fall_through::{FallThrough, FallThroughDecision};
-use crate::core::algorithm::{Algorithm, Driver, LlmTargetSet};
+use crate::core::algorithm::{Algorithm, Driver, LlmTargetSet, ResponseOrDecision};
 use crate::core::classifier::{Classification, Classifier, Score};
 use crate::{LibsyError, Result};
 use switchyard_protocol::{Context, Request, Response, RoutedLlmClient};
@@ -171,7 +171,7 @@ impl Algorithm for Random {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         self.inner.execute(ctx, driver, request).await
     }
 }
@@ -183,10 +183,9 @@ mod tests {
 
     use switchyard_protocol::{Metadata, completion_text, text_request, text_response};
 
-    use crate::DriverError;
     use crate::algorithms::util::affinity::AffinityRouter;
-    use crate::core::algorithm::LlmTarget;
-    use switchyard_protocol::{Decision, LlmResponse, Request, RoutedLlmClient, Signals};
+    use crate::{DriverError, LlmTarget};
+    use switchyard_protocol::{Decision, LlmResponse, Request, Response, RoutedLlmClient, Signals};
 
     /// Echoes the selected target so tests can inspect which target was called.
     struct EchoClient;
@@ -247,7 +246,10 @@ mod tests {
     async fn selected_models(algorithm: Arc<dyn Algorithm>, count: usize) -> Result<Vec<String>> {
         let mut selected = Vec::with_capacity(count);
         for _ in 0..count {
-            let (_, response) = algorithm.clone().run(Context::default(), request()).await?;
+            let (_, response) = algorithm
+                .clone()
+                .run(Context::default(), request(), None)
+                .await?;
             selected.push(
                 response
                     .llm_response
@@ -262,7 +264,7 @@ mod tests {
     #[tokio::test]
     async fn single_target_is_always_selected_and_called() -> Result<()> {
         let algorithm = shared_algorithm(&["only/model"])?;
-        let (trace, response) = algorithm.run(Context::default(), request()).await?;
+        let (trace, response) = algorithm.run(Context::default(), request(), None).await?;
 
         assert_eq!(
             response
@@ -283,7 +285,10 @@ mod tests {
         let algorithm = shared_algorithm(&names)?;
 
         for _ in 0..50 {
-            let (trace, response) = algorithm.clone().run(Context::default(), request()).await?;
+            let (trace, response) = algorithm
+                .clone()
+                .run(Context::default(), request(), None)
+                .await?;
             let selected = response
                 .llm_response
                 .as_agg()
@@ -304,7 +309,10 @@ mod tests {
         let mut seen = HashSet::new();
 
         for _ in 0..100 {
-            let (_, response) = algorithm.clone().run(Context::default(), request()).await?;
+            let (_, response) = algorithm
+                .clone()
+                .run(Context::default(), request(), None)
+                .await?;
             seen.insert(
                 response
                     .llm_response
@@ -370,7 +378,7 @@ mod tests {
 
         let (_, first) = algorithm
             .clone()
-            .run(Context::default(), request_for_session("session-1"))
+            .run(Context::default(), request_for_session("session-1"), None)
             .await?;
         let selected = first
             .llm_response
@@ -391,7 +399,7 @@ mod tests {
         );
 
         let (_, second) = algorithm
-            .run(Context::default(), request_for_session("session-1"))
+            .run(Context::default(), request_for_session("session-1"), None)
             .await?;
         assert_eq!(
             second
@@ -444,7 +452,7 @@ mod tests {
     #[tokio::test]
     async fn decision_is_inspectable_and_downcasts() -> Result<()> {
         let algorithm = shared_algorithm(&["only/model"])?;
-        let (trace, _) = algorithm.run(Context::default(), request()).await?;
+        let (trace, _) = algorithm.run(Context::default(), request(), None).await?;
         let decision = &trace[0];
 
         assert_eq!(decision.selected_model(), "only/model");
diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs
index 8e952a72e..21b39c278 100644
--- a/crates/libsy/src/algorithms/stage.rs
+++ b/crates/libsy/src/algorithms/stage.rs
@@ -27,7 +27,7 @@ use super::util::stage::{
     record_decision_source,
 };
 use super::util::tool_signals::{DEFAULT_RECENT_WINDOW, ToolSignalProcessor};
-use crate::core::algorithm::{Algorithm, Driver, LlmTarget, LlmTargetSet};
+use crate::core::algorithm::{Algorithm, Driver, LlmTarget, LlmTargetSet, ResponseOrDecision};
 use crate::core::classifier::{Classification, Classifier};
 use crate::core::state::State;
 use crate::{LibsyError, Result};
@@ -153,7 +153,7 @@ impl Algorithm for StageRouter {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response> {
+    ) -> Result<ResponseOrDecision> {
         self.route.execute(ctx, driver, request).await
     }
 }
@@ -512,8 +512,11 @@ mod tests {
         let router = recording_router(client.clone(), config_with_notes())?;
         let ctx = Context::default();
 
-        router.clone().run(ctx.clone(), turn_request(false)).await?;
-        router.run(ctx, turn_request(true)).await?;
+        router
+            .clone()
+            .run(ctx.clone(), turn_request(false), None)
+            .await?;
+        router.run(ctx, turn_request(true), None).await?;
 
         let calls = client.routed();
         assert_eq!(calls[0].target, "weak");
@@ -539,7 +542,9 @@ mod tests {
         let client = Arc::new(RecordingClient::default());
         let router = recording_router(client.clone(), config_with_judge(&client, 0.1))?;
 
-        router.run(Context::default(), turn_request(false)).await?;
+        router
+            .run(Context::default(), turn_request(false), None)
+            .await?;
 
         assert!(
             client.calls.lock().iter().any(|c| c.target == JUDGE),
@@ -554,7 +559,9 @@ mod tests {
         let client = Arc::new(RecordingClient::default());
         let router = recording_router(client.clone(), config_with_judge(&client, 0.9))?;
 
-        router.run(Context::default(), turn_request(true)).await?;
+        router
+            .run(Context::default(), turn_request(true), None)
+            .await?;
 
         assert!(
             !client.calls.lock().iter().any(|c| c.target == JUDGE),
@@ -570,9 +577,12 @@ mod tests {
         let router = recording_router(client.clone(), config_with_judge(&client, 0.1))?;
         let ctx = Context::default();
 
-        router.clone().run(ctx.clone(), turn_request(false)).await?;
+        router
+            .clone()
+            .run(ctx.clone(), turn_request(false), None)
+            .await?;
         *client.judge_p_solve.lock() = 0.9;
-        router.run(ctx, turn_request(false)).await?;
+        router.run(ctx, turn_request(false), None).await?;
 
         let routed = client.routed();
         assert_eq!(routed[0].target, "strong");
@@ -595,7 +605,9 @@ mod tests {
         let client = Arc::new(RecordingClient::default());
         let router = recording_router(client.clone(), config_with_judge(&client, 42.0))?;
 
-        router.run(Context::default(), turn_request(false)).await?;
+        router
+            .run(Context::default(), turn_request(false), None)
+            .await?;
 
         assert_eq!(client.routed()[0].target, "weak");
         Ok(())
@@ -606,7 +618,9 @@ mod tests {
         let client = Arc::new(RecordingClient::default());
         let router = recording_router(client.clone(), config_with_judge(&client, 0.9))?;
 
-        router.run(Context::default(), turn_request(false)).await?;
+        router
+            .run(Context::default(), turn_request(false), None)
+            .await?;
 
         let judged = client
             .calls
diff --git a/crates/libsy/src/algorithms/subagent_affinity_tests.rs b/crates/libsy/src/algorithms/subagent_affinity_tests.rs
index 1dd70cbb6..932338bc0 100644
--- a/crates/libsy/src/algorithms/subagent_affinity_tests.rs
+++ b/crates/libsy/src/algorithms/subagent_affinity_tests.rs
@@ -104,7 +104,7 @@ fn router() -> Arc<FallThrough> {
 async fn turn(router: &Arc<FallThrough>, headers: &[(&str, &str)]) -> Result<String> {
     let (_, response) = router
         .clone()
-        .run(Context::default(), request(headers))
+        .run(Context::default(), request(headers), None)
         .await?;
     Ok(response
         .llm_response
diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs
index 4e6d3e75c..78e094c47 100644
--- a/crates/libsy/src/algorithms/util/llm_judge.rs
+++ b/crates/libsy/src/algorithms/util/llm_judge.rs
@@ -346,7 +346,7 @@ mod tests {
     /// Serves the single offloaded judge call with `reply`. The stream is taken first
     /// because the driver refuses to publish a step until a consumer exists.
     async fn score_served_with(reply: Result<Response>) -> Result<String> {
-        let driver = Driver::new();
+        let driver = Driver::new(false);
         let mut steps = Box::pin(driver.stream());
         let classifier = classifier();
         let mut state = State::default();
diff --git a/crates/libsy/src/core/algorithm.rs b/crates/libsy/src/core/algorithm.rs
index 1e124ded3..e9dc18e98 100644
--- a/crates/libsy/src/core/algorithm.rs
+++ b/crates/libsy/src/core/algorithm.rs
@@ -32,6 +32,21 @@ use crate::{DriverError, LibsyError, Result, observability};
 /// [`Algorithm::run_stream`]. Boxed so the trait method that produces it keeps
 /// `Arc<dyn Algorithm>` object-safe.
 pub type StepStream = Pin<Box<dyn Stream<Item = Result<Step>> + Send>>;
+/// A boxed, `Send` stream of [`DecisionOnlyStep`]s — the output of
+/// [`Algorithm::run_decision_only_stream`]. The decision-only counterpart of
+/// [`StepStream`]: it ends on a route to serve rather than on a served answer.
+pub type DecisionOnlyStepStream = Pin<Box<dyn Stream<Item = Result<DecisionOnlyStep>> + Send>>;
+
+/// Either step stream, as produced by [`Algorithm::run_stream_inner`].
+///
+/// The two run modes share one implementation, so the variant is fixed by the
+/// `decision_only` flag the run was started with rather than chosen per step.
+pub enum AnyStepStream {
+    /// A served run: every call is performed and the stream ends with an answer.
+    Step(StepStream),
+    /// A decision-only run: the stream ends with the route for the caller to serve.
+    DecisionOnly(DecisionOnlyStepStream),
+}
 
 /// One completed model call observed at the algorithm offload boundary.
 #[derive(Clone, Debug)]
@@ -146,23 +161,35 @@ pub struct Driver {
     // How long the call that served this run took. We need this to calculate routing overhead.
     routed_call: Arc<Mutex<Option<Duration>>>,
     observer: Option<RunObserver>,
+    // Whether this run concludes with a handoff instead of serving its final call.
+    decision_only: bool,
 }
 
 impl Driver {
     /// Build an empty driver with its step channel ready. Created per call by
-    /// [`run_stream`](Algorithm::run_stream).
-    pub(crate) fn new() -> Self {
-        Self::with_observer(None)
+    /// [`run_stream`](Algorithm::run_stream), which fixes the run's `decision_only` mode.
+    pub(crate) fn new(decision_only: bool) -> Self {
+        Self::with_observer(decision_only, None)
     }
 
-    fn with_observer(observer: Option<RunObserver>) -> Self {
+    fn with_observer(decision_only: bool, observer: Option<RunObserver>) -> Self {
         Self {
             driver: TypeErasedDriver::new(),
             routed_call: Arc::new(Mutex::new(None)),
             observer,
+            decision_only,
         }
     }
 
+    /// Whether this run concludes with a handoff rather than serving its final call.
+    ///
+    /// Fixed for the whole run by the entry point that started it, and applied by
+    /// [`final_decision`](Self::final_decision). An algorithm reads it only when the mode
+    /// changes what it does *before* concluding.
+    pub fn decision_only(&self) -> bool {
+        self.decision_only
+    }
+
     /// How long the call that served this run took, if one has succeeded.
     pub(crate) fn routed_call_duration(&self) -> Option<Duration> {
         *self.routed_call.lock()
@@ -263,6 +290,33 @@ impl Driver {
         .await
     }
 
+    /// Conclude a run on `decision`: either serve the final call here, or hand it back.
+    pub async fn final_decision(
+        &self,
+        ctx: Context,
+        target: &LlmTarget,
+        request: Request,
+        decision: Arc<dyn Decision>,
+        response: &mut Option<Response>,
+    ) -> Result<ResponseOrDecision> {
+        if self.decision_only() {
+            Ok(ResponseOrDecision::Decision((
+                decision,
+                Box::new(request),
+                response.take().map(Box::new),
+            )))
+        } else {
+            let res = match response.take() {
+                Some(res) => res,
+                None => {
+                    self.call_llm_target(ctx.clone(), target, request, decision.clone())
+                        .await?
+                }
+            };
+            Ok(ResponseOrDecision::Response(Box::new(res)))
+        }
+    }
+
     /// Publish a routing [`Decision`] as a [`Step::Decision`] on the stream.
     /// Each successfully published decision is counted and logged with its
     /// reasoning; a decision the stream never accepted is not recorded.
@@ -275,9 +329,13 @@ impl Driver {
     /// Emit the terminal step: [`Step::ReturnToAgent`] on `Ok`, or an `Err` stream
     /// item on failure. Internal: called once by [`run_stream`](Algorithm::run_stream)
     /// when the algorithm finishes.
-    pub(crate) async fn finish(&self, ctx: Context, result: Result<Response>) -> Result<()> {
+    pub(crate) async fn finish(
+        &self,
+        ctx: Context,
+        result: Result<ResponseOrDecision>,
+    ) -> Result<()> {
         match result {
-            Ok(response) => self.driver.done(ctx, response).await,
+            Ok(terminal) => self.driver.done(ctx, terminal).await,
             Err(err) => self.driver.fail(ctx, err).await,
         }
     }
@@ -297,25 +355,95 @@ impl Driver {
                     }
                     .into()
                 }),
-            DriverStep::Done(payload) => payload
-                .downcast::<Response>()
-                .map(Step::ReturnToAgent)
+            // The run task rejects a terminal its mode cannot deliver before publishing it
+            // (see `check_response_type`), so the mismatch arm is a decode-boundary guard.
+            DriverStep::Done(payload) => match payload.downcast::<ResponseOrDecision>() {
+                Ok(terminal) => match *terminal {
+                    ResponseOrDecision::Response(response) => Ok(Step::ReturnToAgent(response)),
+                    ResponseOrDecision::Decision(_) => Err(LibsyError::AlgorithmError {
+                        message: DECISION_IN_SERVED_RUN.to_string(),
+                    }),
+                },
+                Err(_) => Err(DriverError::TypeMismatch {
+                    expected: "ResponseOrDecision",
+                }
+                .into()),
+            },
+        })
+    }
+
+    pub(crate) fn decision_only_stream(
+        &self,
+    ) -> impl Stream<Item = Result<DecisionOnlyStep>> + use<> {
+        self.driver.stream().map(|item| match item? {
+            DriverStep::Request(req) => Ok(DecisionOnlyStep::CallLlm(Box::new(
+                CallLlmRequest::new(req),
+            ))),
+            DriverStep::Info(payload) => payload
+                .downcast::<Arc<dyn Decision>>()
+                .map(|decision| DecisionOnlyStep::Decision(*decision))
                 .map_err(|_| {
                     DriverError::TypeMismatch {
-                        expected: "Response",
+                        expected: "Arc<dyn Decision>",
                     }
                     .into()
                 }),
+            // The run task rejects a terminal its mode cannot deliver before publishing it
+            // (see `check_response_type`), so the mismatch arm is a decode-boundary guard.
+            DriverStep::Done(payload) => match payload.downcast::<ResponseOrDecision>() {
+                Ok(terminal) => match *terminal {
+                    ResponseOrDecision::Response(_) => Err(LibsyError::AlgorithmError {
+                        message: RESPONSE_IN_DECISION_ONLY_RUN.to_string(),
+                    }),
+                    ResponseOrDecision::Decision((decision, request, response)) => Ok(
+                        DecisionOnlyStep::ReturnToAgent((decision, request, response)),
+                    ),
+                },
+                Err(_) => Err(DriverError::TypeMismatch {
+                    expected: "ResponseOrDecision",
+                }
+                .into()),
+            },
         })
     }
 }
 
 impl Default for Driver {
     fn default() -> Self {
-        Self::new()
+        Self::new(false)
     }
 }
 
+/// What an algorithm concludes a run with
+pub enum ResponseOrDecision {
+    /// A served run's answer to the request.
+    Response(Box<Response>),
+    /// A decision-only run's route for the caller to serve — the [`DecidedCall`].
+    Decision((Arc<dyn Decision>, Box<Request>, Option<Box<Response>>)),
+}
+
+/// A terminal a decision-only run cannot hand back: it has no route to the agent.
+const RESPONSE_IN_DECISION_ONLY_RUN: &str =
+    "algorithm returned a response result, which the decision-only step stream cannot serve";
+/// A terminal a served run cannot hand back: its caller expects an answer, not a route.
+const DECISION_IN_SERVED_RUN: &str =
+    "algorithm returned a decision-only result, which the step stream cannot serve";
+
+/// Check the return variant matches the run mode
+fn check_response_type(
+    terminal: ResponseOrDecision,
+    decision_only: bool,
+) -> Result<ResponseOrDecision> {
+    let message = match (&terminal, decision_only) {
+        (ResponseOrDecision::Response(_), true) => RESPONSE_IN_DECISION_ONLY_RUN,
+        (ResponseOrDecision::Decision(_), false) => DECISION_IN_SERVED_RUN,
+        _ => return Ok(terminal),
+    };
+    Err(LibsyError::AlgorithmError {
+        message: message.to_string(),
+    })
+}
+
 /// One item in the stream returned by `Driver::stream` / [`Algorithm::run_stream`].
 pub enum Step {
     /// The algorithm needs this model call performed. The host serves it (optionally
@@ -329,6 +457,176 @@ pub enum Step {
     ReturnToAgent(Box<Response>),
 }
 
+/// One item in the stream returned by [`Algorithm::run_decision_only_stream`].
+pub enum DecisionOnlyStep {
+    /// The algorithm needs this model call performed. The host serves it (optionally
+    /// via [`RoutedRequest::default_client`]) and fulfills it with
+    /// [`CallLlmRequest::respond`]. Boxed: it is by far the largest variant.
+    CallLlm(Box<CallLlmRequest>),
+    /// A routing decision the algorithm made, published via [`Driver::info`] as it
+    /// happens (rather than collected into a trace returned at the end).
+    Decision(Arc<dyn Decision>),
+    /// The algorithm finished with its final decision
+    ReturnToAgent(DecidedCall),
+}
+
+/// The final routing decision of a decision-only run: the decision to act on, the request
+/// to serve it with, and that call's response when the algorithm already made it.
+pub type DecidedCall = (Arc<dyn Decision>, Box<Request>, Option<Box<Response>>);
+
+/// One step of a run as the driving loop sees it.
+///
+/// [`Step`] and [`DecisionOnlyStep`] differ only in what their terminal variant carries, so
+/// both convert into this and one loop can drive either.
+enum StepKind {
+    CallLlm(Box<CallLlmRequest>),
+    Decision(Arc<dyn Decision>),
+    Terminal(ResponseOrDecision),
+}
+
+impl From<Step> for StepKind {
+    fn from(step: Step) -> Self {
+        match step {
+            Step::CallLlm(call) => StepKind::CallLlm(call),
+            Step::Decision(decision) => StepKind::Decision(decision),
+            Step::ReturnToAgent(response) => {
+                StepKind::Terminal(ResponseOrDecision::Response(response))
+            }
+        }
+    }
+}
+
+impl From<DecisionOnlyStep> for StepKind {
+    fn from(step: DecisionOnlyStep) -> Self {
+        match step {
+            DecisionOnlyStep::CallLlm(call) => StepKind::CallLlm(call),
+            DecisionOnlyStep::Decision(decision) => StepKind::Decision(decision),
+            DecisionOnlyStep::ReturnToAgent(decided) => {
+                StepKind::Terminal(ResponseOrDecision::Decision(decided))
+            }
+        }
+    }
+}
+
+/// Drive a step stream to completion: serve every offloaded call with its target's default
+/// client, collect the decisions published along the way, and return the terminal step.
+///
+/// Generic over the step shape so the served-response and decision-only runs share one
+/// loop; the [`StepKind`] conversion normalizes their terminal steps into a
+/// [`ResponseOrDecision`].
+async fn drive<T: Into<StepKind>>(
+    stream: impl Stream<Item = Result<T>>,
+) -> Result<(Vec<Arc<dyn Decision>>, ResponseOrDecision)> {
+    // Serve one offloaded call with its target's default client. A failed *model*
+    // call is forwarded to the algorithm via `respond`; this errors only on an
+    // infrastructure failure (no default client, or the promise was dropped).
+    // `serve` makes the one API call libsy itself performs, so it gets its
+    // own `libsy.client_call` span.
+    #[tracing::instrument(
+        target = "libsy",
+        name = "libsy.client_call",
+        skip_all,
+        fields(
+            algorithm = observability::algorithm_label(&call.get_routed().ctx),
+            switchyard.algorithm = observability::algorithm_label(&call.get_routed().ctx),
+            switchyard.routing.tier = tracing::field::Empty,
+            selected_model = call.get_decision().selected_model(),
+            otel.kind = "client",
+            otel.name = %format_args!("chat {}", call.get_decision().selected_model()),
+            openinference.span.kind = "LLM",
+            gen_ai.operation.name = "chat",
+            gen_ai.request.model = call.get_decision().selected_model(),
+            gen_ai.request.stream = tracing::field::Empty,
+            gen_ai.request.temperature = tracing::field::Empty,
+            gen_ai.request.top_p = tracing::field::Empty,
+            gen_ai.request.top_k = tracing::field::Empty,
+            gen_ai.request.max_tokens = tracing::field::Empty,
+            gen_ai.request.reasoning.level = tracing::field::Empty,
+            gen_ai.output.type = tracing::field::Empty,
+            gen_ai.conversation.id = tracing::field::Empty,
+            server.address = tracing::field::Empty,
+            server.port = tracing::field::Empty,
+            gen_ai.response.id = tracing::field::Empty,
+            gen_ai.response.model = tracing::field::Empty,
+            gen_ai.usage.input_tokens = tracing::field::Empty,
+            gen_ai.usage.output_tokens = tracing::field::Empty,
+            gen_ai.usage.cache_read.input_tokens = tracing::field::Empty,
+            gen_ai.usage.cache_creation.input_tokens = tracing::field::Empty,
+            gen_ai.usage.reasoning.output_tokens = tracing::field::Empty,
+            outcome = tracing::field::Empty,
+            otel.status_code = tracing::field::Empty,
+            error.type = tracing::field::Empty,
+            error = tracing::field::Empty,
+        )
+    )]
+    async fn serve(call: CallLlmRequest) -> Result<()> {
+        let span = tracing::Span::current();
+        observability::record_gen_ai_request(&span, &call.get_routed().request.llm_request);
+        if let Some(tier) = call.get_decision().routing_tier() {
+            span.record("switchyard.routing.tier", tier);
+        }
+        if let Some(session_id) = call
+            .get_routed()
+            .request
+            .metadata
+            .as_ref()
+            .and_then(|metadata| metadata.session_id.as_deref())
+        {
+            span.record("gen_ai.conversation.id", session_id);
+        }
+        let routed = call.get_routed().clone();
+        let target = routed.decision.selected_model().to_string();
+        let client = routed
+            .default_client
+            .clone()
+            .ok_or_else(|| LibsyError::MissingClient {
+                target: target.clone(),
+            })?;
+        let result = client
+            .call(routed.ctx, routed.request, routed.decision)
+            .await
+            .map_err(|source| LibsyError::client_call(target, source));
+        let result = observability::observe_client_call(result);
+        // An algorithm may abandon a call it no longer needs. In that case we just return ok.
+        // TODO Ideally we would signal the caller so that the actual client request can be canceld
+        match call.respond(result) {
+            Err(LibsyError::Driver(DriverError::ResponseDropped)) => Ok(()),
+            result => result,
+        }
+    }
+
+    tokio::pin!(stream);
+
+    let mut trace: Vec<Arc<dyn Decision>> = Vec::new();
+    let mut in_flight = futures::stream::FuturesUnordered::new();
+    let mut terminal: Option<ResponseOrDecision> = None;
+
+    loop {
+        tokio::select! {
+            Some(result) = in_flight.next() => match result {
+                Ok(()) => {}, // CallLlm completed successfully
+                Err(err) => return Err(err), // CallLlm failed, propagate the error
+            },
+            step = stream.next() => {
+                match step {
+                    None => break, // stream has ended, no more steps
+                    Some(item) => match item?.into() {
+                        StepKind::CallLlm(call) => in_flight.push(serve(*call)),
+                        StepKind::Decision(decision) => trace.push(decision),
+                        StepKind::Terminal(step) => {
+                            terminal = Some(step);
+                            break;
+                        }
+                    }
+                }
+            },
+        }
+    }
+    terminal
+        .map(|terminal| (trace, terminal))
+        .ok_or(LibsyError::MissingFinalResponse)
+}
+
 /// Abort guard
 struct AbortOnDrop(tokio::task::AbortHandle);
 
@@ -490,6 +788,15 @@ pub(crate) fn exclude_evicted(
 /// caller's request-side work and retained state still see exactly one turn.
 /// `fallback_decision` builds the [`Decision`] published for a `from -> to` hop, and each
 /// overflow is recorded against `session` so later turns skip that target outright.
+///
+/// Set `final_decision` only when this call's answer is the one returned to the agent: it
+/// concludes the run through [`Driver::final_decision`], which honours decision-only mode.
+/// A side call — a judge or classifier consultation — passes `false` and is always served,
+/// even on a decision-only run, because deciding is what it is for.
+///
+/// `response` carries a response the caller already obtained; on a decision-only run it is
+/// handed back with the decision instead of `target` being called. It is meaningful only
+/// alongside `final_decision`; a side call has no answer to hand back and passes `None`.
 #[allow(clippy::too_many_arguments)]
 pub(crate) async fn call_llm_with_overflow_fallback(
     mut ctx: Context,
@@ -498,14 +805,33 @@ pub(crate) async fn call_llm_with_overflow_fallback(
     mut target: LlmTarget,
     mut decision: Arc<dyn Decision>,
     request: Request,
+    mut response: Option<Response>,
     session: Option<&str>,
     evictions: &SessionEvictions,
     fallback_decision: impl Fn(&LlmTarget, &LlmTarget) -> Arc<dyn Decision>,
-) -> Result<Response> {
+    final_decision: bool,
+) -> Result<ResponseOrDecision> {
     loop {
-        let result = driver
-            .call_llm_target(ctx.clone(), &target, request.clone(), decision.clone())
-            .await;
+        // The response stays owned here and is only lent to each attempt: `Response` is
+        // not `Clone`, so moving it in would leave the next attempt with nothing.
+        // `final_decision` takes it only when it concludes the run.
+        let result = match final_decision {
+            true => {
+                driver
+                    .final_decision(
+                        ctx.clone(),
+                        &target,
+                        request.clone(),
+                        decision.clone(),
+                        &mut response,
+                    )
+                    .await
+            }
+            false => driver
+                .call_llm_target(ctx.clone(), &target, request.clone(), decision.clone())
+                .await
+                .map(|r| ResponseOrDecision::Response(Box::new(r))),
+        };
         let Err(error) = result else { return result };
         let LibsyError::ClientCall {
             target: failed,
@@ -528,6 +854,218 @@ pub(crate) async fn call_llm_with_overflow_fallback(
         driver.info(ctx.clone(), decision.clone()).await?;
     }
 }
+struct AlgoInner<A: Algorithm + ?Sized> {
+    algo: Arc<A>,
+}
+impl<A: Algorithm + ?Sized> AlgoInner<A> {
+    fn new(algo: Arc<A>) -> Self {
+        Self { algo }
+    }
+    /// Process a request to completion, returning a stream of [`Step`]s.
+    /// Each [`Step::CallLlm`] is an offloaded model call the consumer must serve.
+    /// The stream ends with a [`Step::ReturnToAgent`] on success, or an `Err` item on failure.
+    /// Report each model call to `observer`.
+    fn run_stream_inner(
+        &self,
+        ctx: Context,
+        request: Request,
+        decision_only: bool,
+        observer: Option<RunObserver>,
+    ) -> AnyStepStream {
+        // Stamp the algorithm's telemetry label into the request context; the
+        // context rides on every driver call, so its telemetry is attributed.
+        let mut ctx = ctx;
+        ctx.values.insert(
+            observability::ALGORITHM_KEY.to_string(),
+            self.algo.name().to_string(),
+        );
+        let driver = Driver::with_observer(decision_only, observer);
+        let task_driver = driver.clone();
+        let task_ctx = ctx.clone();
+        // Take the consumer stream before the task starts: the driver refuses to publish a
+        // step until a consumer exists. The two shapes read the same driver, so exactly one
+        // of them may be taken. Boxing here erases the two opaque stream types to the one
+        // the variant holds.
+        let steps = if decision_only {
+            AnyStepStream::DecisionOnly(Box::pin(task_driver.decision_only_stream()))
+        } else {
+            AnyStepStream::Step(Box::pin(task_driver.stream()))
+        };
+        // One `libsy.run` span covers the whole algorithm task; the driver's
+        // `libsy.llm_call` spans and decision logs nest inside it via `tracing`'s
+        // contextual parenting.
+        let span = observability::run_span(self.algo.name(), &request);
+        let observed_driver = task_driver.clone();
+        let algo_task = self
+            .algo
+            .clone()
+            .create_run_task(task_ctx.clone(), task_driver, request);
+        let handle = tokio::spawn(
+            async move {
+                observability::observe_run(task_ctx.clone(), observed_driver, async move {
+                    //let terminal = self.algo.clone().create_run_task(task_ctx, task_driver, request).await?;
+                    let terminal = algo_task.await?;
+                    check_response_type(terminal, decision_only)
+                })
+                .await
+            }
+            .instrument(span),
+        );
+        // Dropping the stream aborts the algorithm task, so it doesn't keep running after the
+        let abort_guard = AbortOnDrop(handle.abort_handle());
+
+        let finish_driver = driver.clone();
+        let finish_ctx = ctx;
+
+        // awaits create_run_task append terminal to driver and appends err to step stream
+        fn tail<T: Send + 'static>(
+            handle: tokio::task::JoinHandle<Result<ResponseOrDecision>>,
+            driver: Driver,
+            ctx: Context,
+        ) -> impl Stream<Item = Result<T>> + Send {
+            futures::stream::once(async move {
+                let result = match handle.await {
+                    Ok(response) => response,
+                    Err(source) => Err(LibsyError::AlgorithmTask { source }),
+                };
+                driver.finish(ctx, result).await
+            })
+            .filter_map(|finish_result| async move { finish_result.err().map(Err) })
+        }
+
+        // merge step stream and tail
+        fn merge<T: Send + 'static>(
+            steps: impl Stream<Item = Result<T>> + Send + 'static,
+            tail: impl Stream<Item = Result<T>> + Send + 'static,
+            guard: AbortOnDrop,
+        ) -> Pin<Box<dyn Stream<Item = Result<T>> + Send>> {
+            Box::pin(futures::stream::select(steps, tail).map(move |step| {
+                // link abort guard to stream
+                let _keep_alive = &guard;
+                step
+            }))
+        }
+
+        match steps {
+            AnyStepStream::Step(steps) => AnyStepStream::Step(merge(
+                steps,
+                tail(handle, finish_driver, finish_ctx),
+                abort_guard,
+            )),
+            AnyStepStream::DecisionOnly(steps) => AnyStepStream::DecisionOnly(merge(
+                steps,
+                tail(handle, finish_driver, finish_ctx),
+                abort_guard,
+            )),
+        }
+    }
+
+    /// Process a request to completion, returning a stream of [`Step`]s.
+    fn run_stream(
+        &self,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> StepStream {
+        match self.run_stream_inner(ctx, request, false, observer) {
+            AnyStepStream::Step(stream) => stream,
+            AnyStepStream::DecisionOnly(_) => futures::stream::once(async move {
+                Err(LibsyError::AlgorithmError {
+                    message: "run_stream_inner with decision_only=false should return StepStream"
+                        .to_string(),
+                })
+            })
+            .boxed(),
+        }
+    }
+
+    /// Process a request up to its final routing decision as [`DecisionOnlyStep`]s.
+    fn run_decision_only_stream(
+        &self,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> DecisionOnlyStepStream {
+        match self.run_stream_inner(ctx, request, true, observer) {
+            AnyStepStream::DecisionOnly(stream) => stream,
+            AnyStepStream::Step(_) => futures::stream::once(async move {
+                Err(LibsyError::AlgorithmError {
+                    message: "run_stream_inner with decision_only=true should return \
+                                  DecisionOnlyStepStream"
+                        .to_string(),
+                })
+            })
+            .boxed(),
+        }
+    }
+
+    /// Process a request to completion, serving every offloaded call, and return the
+    /// terminal [`ResponseOrDecision`] plus the trace of [`Decision`]s made along the way.
+    ///
+    /// Both stream shapes drive identically — only their terminal step differs, and the
+    /// shared driving loop normalizes that — so this is one match over the two.
+    async fn run_inner(
+        &self,
+        ctx: Context,
+        request: Request,
+        decision_only: bool,
+        observer: Option<RunObserver>,
+    ) -> Result<(Vec<Arc<dyn Decision>>, ResponseOrDecision)> {
+        match decision_only {
+            true => drive(self.run_decision_only_stream(ctx, request, observer)).await,
+            false => drive(self.run_stream(ctx, request, observer)).await,
+        }
+    }
+
+    /// Process a request to completion, returning the final [`Response`] and the trace of
+    /// [`Decision`]s the algorithm made along the way.
+    async fn run(
+        &self,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
+        self.run_observed(ctx, request, observer).await
+    }
+
+    /// Process a request to completion while reporting each model call to `observer`.
+    async fn run_observed(
+        &self,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
+        let (trace, response) = self.run_inner(ctx, request, false, observer).await?;
+        match response {
+            ResponseOrDecision::Response(response) => Ok((trace, *response)),
+            ResponseOrDecision::Decision(_) => Err(LibsyError::AlgorithmError {
+                message: DECISION_IN_SERVED_RUN.to_string(),
+            }),
+        }
+    }
+
+    /// Process a request up to its final routing decision *without* serving that call:
+    /// returns the decision, the request to serve it with, and any response the algorithm
+    /// already obtained, plus the trace of decisions made along the way.
+    ///
+    /// Only the routed call is left unmade: the decision still binds the algorithm's
+    /// retained state — session affinity latches this session to the target it chose —
+    /// exactly as a served run would. Deciding commits to a route, it does not preview one.
+    async fn decide(
+        &self,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> Result<(Vec<Arc<dyn Decision>>, DecidedCall)> {
+        let (trace, response) = self.run_inner(ctx, request, true, observer).await?;
+        match response {
+            ResponseOrDecision::Decision(decided) => Ok((trace, decided)),
+            ResponseOrDecision::Response(_) => Err(LibsyError::AlgorithmError {
+                message: RESPONSE_IN_DECISION_ONLY_RUN.to_string(),
+            }),
+        }
+    }
+}
 
 /// An optimization strategy. Implement [`create_run_task`](Self::create_run_task);
 /// callers drive it with the provided [`run`](Self::run) (serve calls, get the answer)
@@ -543,7 +1081,8 @@ pub trait Algorithm: Send + Sync + 'static {
     fn name(&self) -> &str;
 
     /// Run one request to completion: make model calls with [`Driver::call_llm_target`],
-    /// publish [`Decision`]s with [`Driver::info`], and return the final [`Response`].
+    /// publish [`Decision`]s with [`Driver::info`], and conclude on the winning target with
+    /// [`Driver::final_decision`].
     /// The method an algorithm implements; [`run`](Self::run) / [`run_stream`](Self::run_stream)
     /// drive it. `ctx` carries the request's cross-cutting values (today: the
     /// algorithm's telemetry label in [`Context::values`]).
@@ -552,7 +1091,7 @@ pub trait Algorithm: Send + Sync + 'static {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> Result<Response>;
+    ) -> Result<ResponseOrDecision>;
 
     /// Feed the algorithm agentic-stack events (tool results, budgets, etc.). The
     /// reference algorithms ignore signals; a stateful algorithm updates its own
@@ -596,189 +1135,50 @@ pub trait Algorithm: Send + Sync + 'static {
     }
 
     /// Process a request to completion, returning a stream of [`Step`]s.
-    /// Each [`Step::CallLlm`] is an offloaded model call the consumer must serve.
-    /// The stream ends with a [`Step::ReturnToAgent`] on success, or an `Err` item on failure.
-    /// Report each model call to `observer`.
     fn run_stream(
         self: Arc<Self>,
         ctx: Context,
         request: Request,
         observer: Option<RunObserver>,
     ) -> StepStream {
-        // Stamp the algorithm's telemetry label into the request context; the
-        // context rides on every driver call, so its telemetry is attributed.
-        let mut ctx = ctx;
-        ctx.values.insert(
-            observability::ALGORITHM_KEY.to_string(),
-            self.name().to_string(),
-        );
-        let driver = Driver::with_observer(observer);
-        let task_driver = driver.clone();
-        let task_ctx = ctx.clone();
-        let stream = task_driver.stream();
-        // One `libsy.run` span covers the whole algorithm task; the driver's
-        // `libsy.llm_call` spans and decision logs nest inside it via `tracing`'s
-        // contextual parenting.
-        let span = observability::run_span(self.name(), &request);
-        let observed_driver = task_driver.clone();
-        let handle = tokio::spawn(
-            async move {
-                observability::observe_run(
-                    task_ctx.clone(),
-                    observed_driver,
-                    self.create_run_task(task_ctx, task_driver, request),
-                )
-                .await
-            }
-            .instrument(span),
-        );
-        // Dropping the stream aborts the algorithm task, so it doesn't keep running after the
-        let abort_guard = AbortOnDrop(handle.abort_handle());
-
-        let finish_driver = driver.clone();
-        let finish_ctx = ctx;
-        let tail: StepStream = Box::pin(
-            futures::stream::once(async move {
-                let result = match handle.await {
-                    Ok(response) => response,
-                    Err(source) => Err(LibsyError::AlgorithmTask { source }),
-                };
-                finish_driver.finish(finish_ctx, result).await
-            })
-            .filter_map(|finish_result| async move { finish_result.err().map(Err) }),
-        );
+        AlgoInner::new(self).run_stream(ctx, request, observer)
+    }
 
-        let stream: StepStream = Box::pin(stream);
-        Box::pin(futures::stream::select(stream, tail).map(move |step| {
-            // link abort guard to stream
-            let _keep_alive = &abort_guard;
-            step
-        }))
+    /// Process a request up to its final routing decision as [`DecisionOnlyStep`]s.
+    fn run_decision_only_stream(
+        self: Arc<Self>,
+        ctx: Context,
+        request: Request,
+        observer: Option<RunObserver>,
+    ) -> DecisionOnlyStepStream {
+        AlgoInner::new(self).run_decision_only_stream(ctx, request, observer)
     }
 
     /// Process a request to completion, returning the final [`Response`] and the trace of
     /// [`Decision`]s the algorithm made along the way.
-    ///
     async fn run(
         self: Arc<Self>,
         ctx: Context,
         request: Request,
+        observer: Option<RunObserver>,
     ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
-        self.run_observed(ctx, request, None).await
+        AlgoInner::new(self).run(ctx, request, observer).await
     }
 
-    /// Process a request to completion while reporting each model call to `observer`.
-    async fn run_observed(
+    /// Process a request up to its final routing decision *without* serving that call:
+    /// returns the decision, the request to serve it with, and any response the algorithm
+    /// already obtained, plus the trace of decisions made along the way.
+    ///
+    /// Only the routed call is left unmade: the decision still binds the algorithm's
+    /// retained state — session affinity latches this session to the target it chose —
+    /// exactly as a served run would. Deciding commits to a route, it does not preview one.
+    async fn decide(
         self: Arc<Self>,
         ctx: Context,
         request: Request,
         observer: Option<RunObserver>,
-    ) -> Result<(Vec<Arc<dyn Decision>>, Response)> {
-        // Serve one offloaded call with its target's default client. A failed *model*
-        // call is forwarded to the algorithm via `respond`; this errors only on an
-        // infrastructure failure (no default client, or the promise was dropped).
-        // `serve` makes the one API call libsy itself performs, so it gets its
-        // own `libsy.client_call` span.
-        #[tracing::instrument(
-            target = "libsy",
-            name = "libsy.client_call",
-            skip_all,
-            fields(
-                algorithm = observability::algorithm_label(&call.get_routed().ctx),
-                switchyard.algorithm = observability::algorithm_label(&call.get_routed().ctx),
-                switchyard.routing.tier = tracing::field::Empty,
-                selected_model = call.get_decision().selected_model(),
-                otel.kind = "client",
-                otel.name = %format_args!("chat {}", call.get_decision().selected_model()),
-                openinference.span.kind = "LLM",
-                gen_ai.operation.name = "chat",
-                gen_ai.request.model = call.get_decision().selected_model(),
-                gen_ai.request.stream = tracing::field::Empty,
-                gen_ai.request.temperature = tracing::field::Empty,
-                gen_ai.request.top_p = tracing::field::Empty,
-                gen_ai.request.top_k = tracing::field::Empty,
-                gen_ai.request.max_tokens = tracing::field::Empty,
-                gen_ai.request.reasoning.level = tracing::field::Empty,
-                gen_ai.output.type = tracing::field::Empty,
-                gen_ai.conversation.id = tracing::field::Empty,
-                server.address = tracing::field::Empty,
-                server.port = tracing::field::Empty,
-                gen_ai.response.id = tracing::field::Empty,
-                gen_ai.response.model = tracing::field::Empty,
-                gen_ai.usage.input_tokens = tracing::field::Empty,
-                gen_ai.usage.output_tokens = tracing::field::Empty,
-                gen_ai.usage.cache_read.input_tokens = tracing::field::Empty,
-                gen_ai.usage.cache_creation.input_tokens = tracing::field::Empty,
-                gen_ai.usage.reasoning.output_tokens = tracing::field::Empty,
-                outcome = tracing::field::Empty,
-                otel.status_code = tracing::field::Empty,
-                error.type = tracing::field::Empty,
-                error = tracing::field::Empty,
-            )
-        )]
-        async fn serve(call: CallLlmRequest) -> Result<()> {
-            let span = tracing::Span::current();
-            observability::record_gen_ai_request(&span, &call.get_routed().request.llm_request);
-            if let Some(tier) = call.get_decision().routing_tier() {
-                span.record("switchyard.routing.tier", tier);
-            }
-            if let Some(session_id) = call
-                .get_routed()
-                .request
-                .metadata
-                .as_ref()
-                .and_then(|metadata| metadata.session_id.as_deref())
-            {
-                span.record("gen_ai.conversation.id", session_id);
-            }
-            let routed = call.get_routed().clone();
-            let target = routed.decision.selected_model().to_string();
-            let client =
-                routed
-                    .default_client
-                    .clone()
-                    .ok_or_else(|| LibsyError::MissingClient {
-                        target: target.clone(),
-                    })?;
-            let result = client
-                .call(routed.ctx, routed.request, routed.decision)
-                .await
-                .map_err(|source| LibsyError::client_call(target, source));
-            let result = observability::observe_client_call(result);
-            call.respond(result)
-        }
-
-        let stream = self.run_stream(ctx, request, observer);
-        tokio::pin!(stream);
-
-        let mut trace: Vec<Arc<dyn Decision>> = Vec::new();
-        let mut in_flight = futures::stream::FuturesUnordered::new();
-        let mut final_response: Option<Response> = None;
-
-        loop {
-            tokio::select! {
-                Some(result) = in_flight.next() => match result {
-                    Ok(()) => {}, // CallLlm completed successfully
-                    Err(err) => return Err(err), // CallLlm failed, propagate the error
-                },
-                step = stream.next() => {
-                    match step {
-                        None => break, // stream has ended, no more steps
-                        Some(item) => match item? {
-                            Step::CallLlm(call) => in_flight.push(serve(*call)),
-                            Step::Decision(decision) => trace.push(decision),
-                            Step::ReturnToAgent(response) => {
-                                final_response = Some(*response);
-                                break;
-                            }
-                        }
-                    }
-                },
-            }
-        }
-        final_response
-            .map(|response| (trace, response))
-            .ok_or(LibsyError::MissingFinalResponse)
+    ) -> Result<(Vec<Arc<dyn Decision>>, DecidedCall)> {
+        AlgoInner::new(self).decide(ctx, request, observer).await
     }
 }
 
@@ -853,7 +1253,7 @@ mod tests {
             ctx: Context,
             driver: Driver,
             request: Request,
-        ) -> Result<Response> {
+        ) -> Result<ResponseOrDecision> {
             let target = self
                 .target_set
                 .targets()
@@ -865,7 +1265,7 @@ mod tests {
             });
             driver.info(ctx.clone(), decision.clone()).await?;
             driver
-                .call_llm_target(ctx, &target, request, decision)
+                .final_decision(ctx, &target, request, decision, &mut None)
                 .await
         }
     }
@@ -901,7 +1301,7 @@ mod tests {
         let observed = observations.clone();
         let observer: RunObserver = Arc::new(move |observation| observed.lock().push(observation));
         let (_, response) = orch(target_set(&[("direct/model", true)]))
-            .run_observed(Context::default(), request(), Some(observer))
+            .run(Context::default(), request(), Some(observer))
             .await?;
         assert_eq!(
             response.llm_response.as_agg().map(completion_text),
@@ -984,7 +1384,7 @@ mod tests {
                 reason: Some("stop".to_string()),
             },
         ]);
-        let (trace, response) = orch.run(Context::default(), request()).await?;
+        let (trace, response) = orch.run(Context::default(), request(), None).await?;
         // `run` handed back the live stream; the caller folds it to a buffered aggregate.
         let agg = response
             .llm_response
@@ -1010,7 +1410,7 @@ mod tests {
                 message: "upstream exploded".to_string(),
             },
         ]);
-        let (_, response) = orch.run(Context::default(), request()).await?;
+        let (_, response) = orch.run(Context::default(), request(), None).await?;
         match response.llm_response.into_agg().await {
             Ok(_) => panic!("expected a mid-stream error, got an aggregate"),
             Err(err) => {
@@ -1124,7 +1524,7 @@ mod tests {
         // Every target has a client, so run serves every call via the
         // default client and returns the trace + final response.
         let (trace, response) = orch(target_set(&[("direct/model", true)]))
-            .run(Context::default(), request())
+            .run(Context::default(), request(), None)
             .await?;
         // TestAlgo calls the first target; EchoClient echoes its name.
         assert_eq!(
@@ -1139,12 +1539,142 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn concluding_through_final_decision_honours_the_run_mode() -> Result<()> {
+        // `TestAlgo` never mentions the mode: it just ends on `Driver::final_decision`.
+        // That alone is enough to support both entry points — `run` gets the served
+        // response, `decide` gets the route with the call left unmade.
+        let algo = orch(target_set(&[("direct/model", true)]));
+
+        let (_, response) = algo
+            .clone()
+            .run(Context::default(), request(), None)
+            .await?;
+        assert_eq!(
+            response
+                .llm_response
+                .as_agg()
+                .map(completion_text)
+                .unwrap_or_default(),
+            "direct/model"
+        );
+
+        let (_, (decision, _, served)) = algo.decide(Context::default(), request(), None).await?;
+        assert_eq!(decision.selected_model(), "direct/model");
+        assert!(served.is_none(), "the final call was handed back, not made");
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn decide_errors_when_the_algorithm_cannot_hand_the_call_back() -> Result<()> {
+        // `Noop` answers without ever routing a call, so it builds its response directly
+        // instead of concluding through `final_decision` — it has no route to hand back.
+        // The mismatch is caught where the terminal payload is decoded, so `decide` never
+        // sees a response to mislabel.
+        let algo: Arc<dyn Algorithm> = Arc::new(crate::Noop {});
+        let error = algo
+            .decide(Context::default(), request(), None)
+            .await
+            .err()
+            .ok_or_else(|| test_error("expected a decision-only mismatch"))?;
+        assert!(matches!(
+            error,
+            LibsyError::AlgorithmError { message } if message.contains("decision-only step stream")
+        ));
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn the_driver_reports_the_mode_the_run_was_started_in() -> Result<()> {
+        // The mode is fixed by the entry point and read back by the algorithm, so a
+        // composition cannot disagree with the stream shape it is being driven as.
+        use std::sync::atomic::{AtomicBool, Ordering};
+
+        struct ModeRecording(Arc<AtomicBool>);
+
+        #[async_trait]
+        impl Algorithm for ModeRecording {
+            fn name(&self) -> &str {
+                "mode_recording"
+            }
+
+            async fn create_run_task(
+                self: Arc<Self>,
+                _ctx: Context,
+                driver: Driver,
+                _request: Request,
+            ) -> Result<ResponseOrDecision> {
+                self.0.store(driver.decision_only(), Ordering::SeqCst);
+                Err(test_error("stop after recording the mode"))
+            }
+        }
+
+        for expected in [false, true] {
+            let seen = Arc::new(AtomicBool::new(!expected));
+            let algo: Arc<dyn Algorithm> = Arc::new(ModeRecording(seen.clone()));
+            // Both entry points fail here by design; only the recorded mode matters.
+            let _ = AlgoInner::new(algo)
+                .run_inner(Context::default(), request(), expected, None)
+                .await;
+            assert_eq!(seen.load(Ordering::SeqCst), expected);
+        }
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn concluding_with_an_answer_already_obtained_does_not_call_again() -> Result<()> {
+        // The target has no client, so any outbound call fails: the run can only succeed by
+        // concluding on the response the algorithm already had.
+        struct ConcludesWithServedAnswer(LlmTargetSet);
+
+        #[async_trait]
+        impl Algorithm for ConcludesWithServedAnswer {
+            fn name(&self) -> &str {
+                "concludes-with-served-answer"
+            }
+
+            async fn create_run_task(
+                self: Arc<Self>,
+                ctx: Context,
+                driver: Driver,
+                request: Request,
+            ) -> Result<ResponseOrDecision> {
+                let target = self.0.get_target("offload/model")?;
+                let decision: Arc<dyn Decision> = Arc::new(TestDecision {
+                    model: target.semantic_name.clone(),
+                });
+                let mut served = Some(Response {
+                    llm_response: LlmResponse::Agg(text_response(None, "already answered")),
+                    metadata: None,
+                });
+                driver
+                    .final_decision(ctx, &target, request, decision, &mut served)
+                    .await
+            }
+        }
+
+        let algo: Arc<dyn Algorithm> = Arc::new(ConcludesWithServedAnswer(target_set(&[(
+            "offload/model",
+            false,
+        )])));
+        let (_trace, response) = algo.run(Context::default(), request(), None).await?;
+        assert_eq!(
+            response
+                .llm_response
+                .as_agg()
+                .map(completion_text)
+                .unwrap_or_default(),
+            "already answered"
+        );
+        Ok(())
+    }
+
     #[tokio::test]
     async fn run_errors_when_a_target_lacks_a_client() -> Result<()> {
         // A client-less target has no default client to serve its offloaded call, so
         // driving it to completion errors.
         let error = orch(target_set(&[("offload/model", false)]))
-            .run(Context::default(), request())
+            .run(Context::default(), request(), None)
             .await
             .err()
             .ok_or_else(|| test_error("expected a missing-client error"))?;
@@ -1204,7 +1734,7 @@ mod tests {
         for _ in 0..N {
             let algo = algo.clone();
             handles.push(tokio::spawn(async move {
-                algo.run(Context::default(), request())
+                algo.run(Context::default(), request(), None)
                     .await
                     .map(|(_, response)| {
                         response
@@ -1294,7 +1824,7 @@ mod tests {
                 _ctx: Context,
                 _driver: Driver,
                 _request: Request,
-            ) -> Result<Response> {
+            ) -> Result<ResponseOrDecision> {
                 let _guard = DropGuard(self.dropped.clone());
                 let _ = self.started.send(());
                 // Await forever without ever touching the driver.
@@ -1342,7 +1872,7 @@ mod tests {
                 _ctx: Context,
                 _driver: Driver,
                 _request: Request,
-            ) -> Result<Response> {
+            ) -> Result<ResponseOrDecision> {
                 panic!("boom");
             }
         }
@@ -1383,13 +1913,13 @@ mod tests {
                 _ctx: Context,
                 _driver: Driver,
                 _request: Request,
-            ) -> Result<Response> {
+            ) -> Result<ResponseOrDecision> {
                 panic!("boom");
             }
         }
 
         let algo: Arc<dyn Algorithm> = Arc::new(Panicky);
-        match algo.run(Context::default(), request()).await {
+        match algo.run(Context::default(), request(), None).await {
             Ok(_) => Err(test_error(
                 "expected run to surface the algorithm panic as an error",
             )),
@@ -1431,7 +1961,7 @@ mod tests {
                 _ctx: Context,
                 _driver: Driver,
                 _request: Request,
-            ) -> Result<Response> {
+            ) -> Result<ResponseOrDecision> {
                 let _guard = DropGuard(self.dropped.clone());
                 let _ = self.started.send(());
                 // Hang forever without ever touching the driver, so only cancellation
@@ -1450,7 +1980,8 @@ mod tests {
 
         // Drive `run` on its own task, wait until the algorithm task is up, then cancel
         // `run` — dropping its future (and the `run_stream` stream it holds).
-        let run_task = tokio::spawn(async move { algo.run(Context::default(), request()).await });
+        let run_task =
+            tokio::spawn(async move { algo.run(Context::default(), request(), None).await });
         started_rx
             .recv()
             .await
@@ -1528,6 +2059,8 @@ mod tests {
     struct Hedge {
         winner: LlmTarget,
         loser: LlmTarget,
+        /// Work the algorithm does after the race, before it concludes.
+        post_select_delay: Option<std::time::Duration>,
     }
 
     #[async_trait]
@@ -1541,7 +2074,7 @@ mod tests {
             ctx: Context,
             driver: Driver,
             request: Request,
-        ) -> Result<Response> {
+        ) -> Result<ResponseOrDecision> {
             let dec_w: Arc<dyn Decision> = Arc::new(TestDecision {
                 model: self.winner.semantic_name.clone(),
             });
@@ -1551,16 +2084,29 @@ mod tests {
             let win = driver.call_llm_target(ctx.clone(), &self.winner, request.clone(), dec_w);
             let lose = driver.call_llm_target(ctx, &self.loser, request, dec_l);
             // First to resolve wins; `select!` drops the losing future (and its promise).
-            tokio::select! {
+            let winner = tokio::select! {
                 res = win => res,
                 res = lose => res,
+            };
+            if let Some(delay) = self.post_select_delay {
+                tokio::time::sleep(delay).await;
             }
+            Ok(ResponseOrDecision::Response(Box::new(winner?)))
         }
     }
 
     /// Builds a hedging algo whose winner is gated behind the loser starting, and whose
     /// loser finishes after `loser_delay` (or never, when `None`).
     fn hedge(loser_delay: Option<std::time::Duration>) -> Arc<dyn Algorithm> {
+        hedge_concluding_after(loser_delay, None)
+    }
+
+    /// A hedge that keeps working for `post_select_delay` after the race, so a late loser
+    /// resolves into its dropped promise while the run is still in progress.
+    fn hedge_concluding_after(
+        loser_delay: Option<std::time::Duration>,
+        post_select_delay: Option<std::time::Duration>,
+    ) -> Arc<dyn Algorithm> {
         let started = Arc::new(tokio::sync::Notify::new());
         let winner = LlmTarget {
             semantic_name: "winner".to_string(),
@@ -1575,7 +2121,11 @@ mod tests {
                 delay: loser_delay,
             })),
         };
-        Arc::new(Hedge { winner, loser })
+        Arc::new(Hedge {
+            winner,
+            loser,
+            post_select_delay,
+        })
     }
 
     #[tokio::test]
@@ -1583,7 +2133,7 @@ mod tests {
         // The loser responds 50ms after the winner has already won. `run` must return the
         // winner, not the loser's `respond`-to-a-dropped-receiver error.
         let (_trace, response) = hedge(Some(std::time::Duration::from_millis(50)))
-            .run(Context::default(), request())
+            .run(Context::default(), request(), None)
             .await?;
         assert_eq!(
             response
@@ -1596,11 +2146,34 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn a_late_loser_resolving_into_a_dropped_promise_does_not_fail_the_run() -> Result<()> {
+        // Abandoning a call is how an algorithm hedges, so the response the loser resolves
+        // into a dropped promise is discarded rather than failing a run the winner already
+        // answered. Unlike the test above, this hedge keeps working after the race, so that
+        // discarded response reaches the consumer well before the terminal step instead of
+        // racing it.
+        let algorithm = hedge_concluding_after(
+            Some(std::time::Duration::from_millis(20)),
+            Some(std::time::Duration::from_millis(200)),
+        );
+        let (_trace, response) = algorithm.run(Context::default(), request(), None).await?;
+        assert_eq!(
+            response
+                .llm_response
+                .as_agg()
+                .map(completion_text)
+                .unwrap_or_default(),
+            "winner"
+        );
+        Ok(())
+    }
+
     #[tokio::test]
     async fn run_returns_the_winner_without_hanging_on_a_pending_loser() -> Result<()> {
         // The loser never resolves. `run` must return the winner promptly, not hang
         // waiting for the in-flight loser.
-        let run = hedge(None).run(Context::default(), request());
+        let run = hedge(None).run(Context::default(), request(), None);
         let (_trace, response) = tokio::time::timeout(std::time::Duration::from_secs(1), run)
             .await
             .map_err(|error| LibsyError::external("waiting for pending loser", error))??;
@@ -1665,7 +2238,7 @@ mod tests {
                 ctx: Context,
                 driver: Driver,
                 request: Request,
-            ) -> Result<Response> {
+            ) -> Result<ResponseOrDecision> {
                 let offloads = futures::future::join_all((0..self.n).map(|i| {
                     let decision: Arc<dyn Decision> = Arc::new(TestDecision {
                         model: format!("m{i}"),
@@ -1698,7 +2271,7 @@ mod tests {
 
         // With the cap gone, `run` keeps polling the stream even with N calls in flight, so
         // the terminal error surfaces promptly instead of hanging.
-        let run = algo.run(Context::default(), request());
+        let run = algo.run(Context::default(), request(), None);
         let result = tokio::time::timeout(std::time::Duration::from_millis(500), run)
             .await
             .map_err(|error| {
diff --git a/crates/libsy/src/core/classifier.rs b/crates/libsy/src/core/classifier.rs
index c417622eb..fd635b9f8 100644
--- a/crates/libsy/src/core/classifier.rs
+++ b/crates/libsy/src/core/classifier.rs
@@ -43,6 +43,17 @@ impl Classification {
             }
         }
     }
+
+    /// The top-scoring [`Score`]s, or an empty set when the classifier abstained (an empty set).
+    pub fn max_classification(&self) -> Result<Classification> {
+        let max = self.argmax(true)?;
+        match self {
+            Classification::Scores(_) => Ok(Classification::Scores(max.into_iter().collect())),
+            Classification::Ambiguous(_) => {
+                Ok(Classification::Ambiguous(max.into_iter().collect()))
+            }
+        }
+    }
 }
 
 /// The highest-confidence score, or `None` when the set is empty (the classifier abstained).
diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs
index bff81c6d7..39521fa09 100644
--- a/crates/libsy/src/lib.rs
+++ b/crates/libsy/src/lib.rs
@@ -15,8 +15,8 @@
 //! - An [`Algorithm`] is the optimization *algorithm*. Its
 //!   [`create_run_task`](Algorithm::create_run_task) runs once per request
 //!   and makes as many model calls as it needs — via [`Driver::call_llm_target`], which look
-//!   like ordinary calls — publishes its [`Decision`](switchyard_protocol::Decision)s with [`Driver::info`], and
-//!   returns the final [`Response`](switchyard_protocol::Response). The provided
+//!   like ordinary calls — publishes its [`Decision`]s with [`Driver::info`], and
+//!   concludes on the winning target with [`Driver::final_decision`]. The provided
 //!   [`run_stream`](Algorithm::run_stream) drives that on its own task and hands
 //!   back a stream of [`Step`]s; [`run`](Algorithm::run) runs
 //!   it to completion with the targets' default clients.
@@ -28,7 +28,13 @@
 //!
 //! ## Running a request
 //!
-//! Hold the algorithm as `Arc<dyn Algorithm>` and call one of two provided methods:
+//! Hold the algorithm as `Arc<dyn Algorithm>` and call one of four provided methods. They
+//! vary on two axes: who serves the model calls, and whether the *final* call is made.
+//!
+//! | | libsy serves the calls | you serve the calls |
+//! |---|---|---|
+//! | serve the final call | [`run`](Algorithm::run) | [`run_stream`](Algorithm::run_stream) |
+//! | hand the final call back | [`decide`](Algorithm::decide) | [`run_decision_only_stream`](Algorithm::run_decision_only_stream) |
 //!
 //! - [`run`](Algorithm::run) — run to completion, serving each
 //!   offloaded call via its [`RoutedRequest::default_client`], and return the decision
@@ -42,6 +48,20 @@
 //!   [`Step::ReturnToAgent`] carrying the final response. The step stream is bounded,
 //!   so pulling it paces the algorithm one step at a time — an "ask, don't call" mode
 //!   that lets a host that owns its transport keep control of every call.
+//! - [`decide`](Algorithm::decide) — stop before *committing* to the final call: return
+//!   the routing [`Decision`], the [`Request`] to serve it with, and the selected model's
+//!   [`Response`] when the algorithm already called it. Model calls the decision itself
+//!   needs are still made, so which of those you get depends on how the algorithm
+//!   decides: judging the request leaves `None` (the call is yours to make), while
+//!   analyzing a response leaves `Some` (reuse it, or call again — your tradeoff).
+//!   [`run_decision_only_stream`](Algorithm::run_decision_only_stream) is the streamed
+//!   form, ending in [`DecisionOnlyStep::ReturnToAgent`]. Only that final call is left
+//!   unmade: the decision still binds retained state — session affinity latches — exactly
+//!   as under [`run`](Algorithm::run).
+//!
+//! An algorithm does not branch on the mode: it is fixed by the entry point, recorded on
+//! the [`Driver`], and applied by [`Driver::final_decision`], so concluding there is
+//! enough to support all four.
 //!
 //! ## Concurrency
 //!
@@ -78,8 +98,9 @@
 
 mod core;
 pub use core::algorithm::{
-    Algorithm, CallLlmRequest, Driver, LlmCallObservation, LlmTarget, LlmTargetSet, RoutedRequest,
-    RunObservation, RunObserver, Step, StepStream,
+    Algorithm, CallLlmRequest, DecisionOnlyStep, DecisionOnlyStepStream, Driver,
+    LlmCallObservation, LlmTarget, LlmTargetSet, ResponseOrDecision, RoutedRequest, RunObservation,
+    RunObserver, Step, StepStream,
 };
 pub use core::classifier::{Classification, Classifier, Score};
 pub use core::processor::{Event, Processor};
diff --git a/crates/libsy/src/observability.rs b/crates/libsy/src/observability.rs
index f972bd0c1..ca7b282fd 100644
--- a/crates/libsy/src/observability.rs
+++ b/crates/libsy/src/observability.rs
@@ -152,11 +152,11 @@ pub(crate) fn run_span(algorithm: &str, request: &Request) -> Span {
 /// histogram, routing overhead, span outcome, and failure log when it resolves.
 /// Executes inside the `libsy.run` span its caller instruments the task with.
 /// `driver` is the run's own, holding the duration of the call that served it.
-pub(crate) async fn observe_run(
+pub(crate) async fn observe_run<T>(
     ctx: Context,
     driver: Driver,
-    run: impl Future<Output = Result<Response>>,
-) -> Result<Response> {
+    run: impl Future<Output = Result<T>>,
+) -> Result<T> {
     let started = Instant::now();
     let result = run.await;
     let duration = started.elapsed();
@@ -447,7 +447,7 @@ fn record_client_error(span: &Span, error_type: &str, error: &dyn std::fmt::Disp
 /// Records the end of one algorithm run: the run counter and duration
 /// histogram, the `outcome`/`error` fields on `span`, and a warn log when the
 /// run failed.
-fn record_run(algorithm: &str, duration: Duration, result: &Result<Response>, span: &Span) {
+fn record_run<T>(algorithm: &str, duration: Duration, result: &Result<T>, span: &Span) {
     let outcome = outcome_value(result);
     span.record("outcome", outcome);
     if let Err(error) = result {
diff --git a/crates/libsy/tests/observability.rs b/crates/libsy/tests/observability.rs
index 64271716b..11dfb9e5b 100644
--- a/crates/libsy/tests/observability.rs
+++ b/crates/libsy/tests/observability.rs
@@ -33,14 +33,12 @@ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt};
 use tracing_subscriber::registry::LookupSpan;
 
 use switchyard_libsy::{
-    Algorithm, Driver, LibsyError, LlmTarget, LlmTargetSet, LlmTaskClassifier, Step,
-    TaskClassifierConfig,
+    Algorithm, Driver, LibsyError, LlmTarget, LlmTargetSet, LlmTaskClassifier, ResponseOrDecision,
+    Step, TaskClassifierConfig,
 };
 use switchyard_protocol::{
-    Context, Decision, LlmResponse, Metadata, Request, Response, RoutedLlmClient, Usage,
-};
-use switchyard_protocol::{
-    LlmClientError, LlmResponseChunk, StopReason, text_request, text_response,
+    Context, Decision, LlmClientError, LlmResponse, LlmResponseChunk, Metadata, Request, Response,
+    RoutedLlmClient, StopReason, Usage, text_request, text_response,
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -408,7 +406,7 @@ impl Algorithm for SingleCallAlgo {
         ctx: Context,
         driver: Driver,
         request: Request,
-    ) -> switchyard_libsy::Result<Response> {
+    ) -> switchyard_libsy::Result<ResponseOrDecision> {
         let target = self
             .target_set
             .targets()
@@ -421,7 +419,7 @@ impl Algorithm for SingleCallAlgo {
         });
         driver.info(ctx.clone(), decision.clone()).await?;
         driver
-            .call_llm_target(ctx, &target, request, decision)
+            .final_decision(ctx, &target, request, decision, &mut None)
             .await
     }
 }
@@ -528,7 +526,7 @@ async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_l
     request.llm_request.output.response_format = Some(json!({"type": "json_schema"}));
     request.llm_request.reasoning.effort = Some("high".to_string());
     let (trace, _response) = algo(ALGO, MODEL, Some(client))
-        .run(Context::default(), request)
+        .run(Context::default(), request, None)
         .await?;
     assert_eq!(trace.len(), 1);
 
@@ -801,7 +799,7 @@ async fn streamed_usage_updates_the_client_call_span() -> switchyard_libsy::Resu
     let mut request = request_with_metadata("obs-stream-session", "obs-stream-corr");
     request.llm_request.stream = true;
     let (_, response) = algo(ALGO, MODEL, Some(client))
-        .run(Context::default(), request)
+        .run(Context::default(), request, None)
         .await?;
     let LlmResponse::Stream(mut stream) = response.llm_response else {
         return Err(test_error("expected a streamed response"));
@@ -848,7 +846,7 @@ async fn dropped_stream_records_cancelled_outcome() -> switchyard_libsy::Result<
     let mut request = request_with_metadata("obs-cancelled-session", "obs-cancelled-corr");
     request.llm_request.stream = true;
     let (_, response) = algo(ALGO, MODEL, Some(client))
-        .run(Context::default(), request)
+        .run(Context::default(), request, None)
         .await?;
     let LlmResponse::Stream(stream) = response.llm_response else {
         return Err(test_error("expected a streamed response"));
@@ -878,6 +876,7 @@ async fn typed_client_failure_records_semantic_error_type() {
     .run(
         Context::default(),
         request_with_metadata("obs-timeout-session", "obs-timeout-corr"),
+        None,
     )
     .await;
     assert!(matches!(
@@ -1061,6 +1060,7 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib
                 raw_request: None,
                 metadata: None,
             },
+            None,
         )
         .await?;
 
@@ -1124,3 +1124,77 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib
     );
     Ok(())
 }
+
+/// Concludes with a served response whatever mode it is run in — the terminal a
+/// decision-only run has no way to deliver.
+struct AlwaysRespondsAlgo {
+    name: String,
+}
+
+#[async_trait]
+impl Algorithm for AlwaysRespondsAlgo {
+    fn name(&self) -> &str {
+        &self.name
+    }
+
+    async fn create_run_task(
+        self: Arc<Self>,
+        _ctx: Context,
+        _driver: Driver,
+        _request: Request,
+    ) -> switchyard_libsy::Result<ResponseOrDecision> {
+        Ok(ResponseOrDecision::Response(Box::new(Response {
+            llm_response: LlmResponse::Agg(text_response(None, "answer")),
+            metadata: None,
+        })))
+    }
+}
+
+#[tokio::test]
+async fn a_run_the_caller_receives_as_an_error_is_metered_as_one() -> switchyard_libsy::Result<()> {
+    let _guard = serialize_test().lock().await;
+    let (store, exporter, provider, _, _) = telemetry();
+    const ALGO: &str = "obs-mode-mismatch-algo";
+    let _before = flushed_metrics(exporter, provider);
+
+    let algorithm: Arc<dyn Algorithm> = Arc::new(AlwaysRespondsAlgo {
+        name: ALGO.to_string(),
+    });
+    // A response terminal under `decide`: the caller receives an error...
+    let result = algorithm
+        .decide(
+            Context::default(),
+            request_with_metadata("obs-session-4", "obs-corr-4"),
+            None,
+        )
+        .await;
+    assert!(
+        result.is_err(),
+        "decide must reject a terminal it cannot deliver"
+    );
+
+    // ...so the run counts as a failure, not a success with a stray latency sample.
+    let snapshots = flushed_metrics(exporter, provider);
+    assert_eq!(
+        u64_counter_value(
+            &snapshots,
+            "switchyard.runs",
+            &[("algorithm", ALGO), ("outcome", "error")]
+        ),
+        Some(1)
+    );
+    assert_eq!(
+        u64_counter_value(
+            &snapshots,
+            "switchyard.runs",
+            &[("algorithm", ALGO), ("outcome", "ok")]
+        ),
+        None
+    );
+    let span = find_span(&store.spans(), "libsy.run", "algorithm", ALGO);
+    assert_eq!(
+        span.fields.get("outcome").map(String::as_str),
+        Some("error")
+    );
+    Ok(())
+}
diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs
index 3131dae01..221520f8e 100644
--- a/crates/switchyard-py/src/libsy_bindings.rs
+++ b/crates/switchyard-py/src/libsy_bindings.rs
@@ -217,7 +217,7 @@ impl PyAlgorithm {
         };
         pyo3_async_runtimes::tokio::future_into_py(py, async move {
             let (decisions, response) = algorithm
-                .run(Context::default(), request)
+                .run(Context::default(), request, None)
                 .await
                 .map_err(py_libsy_error)?;
             let response = response
diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs
index 0c33a10d1..7b8eb980e 100644
--- a/crates/switchyard-server/src/lib.rs
+++ b/crates/switchyard-server/src/lib.rs
@@ -592,7 +592,7 @@ async fn handle_llm_request(
     };
     let observer = stats_observer(state.stats.clone());
     let (trace, response) = match algorithm
-        .run_observed(Context::default(), request, Some(observer))
+        .run(Context::default(), request, Some(observer))
         .await
     {
         Ok(result) => result,