Skip to main content

fynd_core/worker_pool_router/
mod.rs

1//! Orchestrates multiple solver pools to find the best quote per request.
2//!
3//! The WorkerPoolRouter sits between the API layer and multiple solver pools.
4//! It fans out each order to all configured solvers, manages timeouts,
5//! selects the best quote based on `amount_out_net_gas`, and optionally
6//! encodes the winning solution into an on-chain transaction.
7
8//! # Responsibilities
9//!
10//! 1. **Fan-out**: Distribute each order to solver pools. Its distribution algorithm can be
11//!    customized, but initially it's set to relay to all solvers.
12//! 2. **Timeout**: Cancel if solver response takes too long
13//! 3. **Collection**: Wait for N responses OR timeout per order
14//! 4. **Gas refinement**: Before ranking across worker pools, replace each candidate's naive
15//!    `route.total_gas()` estimate (used internally by algorithms for ranking within a worker pool)
16//!    with the more accurate `estimate_gas_usage` from tycho-execution, which accounts for token
17//!    transfer costs and router overhead. The `amount_out_net_gas` values are rescaled
18//!    proportionally so the final ranking reflects realistic execution cost.
19//! 5. **Selection**: Choose best quote (max refined `amount_out_net_gas`)
20//! 6. **Encoding**: If [`EncodingOptions`](crate::EncodingOptions) are provided in the request,
21//!    encode winning solutions into executable on-chain transactions via the
22//!    [`encoding::encoder::Encoder`](crate::encoding::encoder::Encoder)
23
24mod allocation;
25mod comparison_log;
26pub mod config;
27
28use std::{
29    sync::LazyLock,
30    time::{Duration, Instant},
31};
32
33pub use allocation::ExclusiveAccess;
34use allocation::{allocate, Allocation, OrderClass};
35use comparison_log::{log_quote_comparison, solver_error_label};
36use config::WorkerPoolRouterConfig;
37use futures::stream::{FuturesUnordered, StreamExt};
38use metrics::{counter, gauge, histogram};
39use num_bigint::BigUint;
40use num_traits::ToPrimitive;
41use rustc_hash::{FxHashMap, FxHashSet};
42use serde::{Deserialize, Serialize};
43use tracing::{debug, warn};
44use tycho_execution::encoding::{
45    evm::gas_estimator::estimate_gas_usage,
46    models::{Solution, Strategy},
47};
48use tycho_simulation::tycho_common::{models::Chain, Bytes};
49
50use crate::{
51    encoding::encoder::Encoder, feed::exclusivity::is_exclusive, price_guard::guard::PriceGuard,
52    worker_pool::task_queue::TaskQueueHandle, BlockInfo, EncodingOptions, Order, OrderQuote,
53    OrderSide, Quote, QuoteOptions, QuoteRequest, QuoteStatus, SolveError, SolveParams,
54    SurplusInfo, Swap,
55};
56
57/// Environment variable overriding [`DEFAULT_USER_IMPROVEMENT_SHARE_BPS`]. Read once, on the
58/// first quote that overlays an exclusive-access candidate.
59const ENV_USER_IMPROVEMENT_SHARE_BPS: &str = "EXCLUSIVE_ROUTE_USER_SHARE_BPS";
60
61/// Share of an exclusive route's improvement over the public market that is handed to the user,
62/// in basis points of that improvement: `1_000` gives the user a tenth of it.
63const DEFAULT_USER_IMPROVEMENT_SHARE_BPS: u32 = 1_000;
64
65/// Basis-point denominator: `10_000` bps is the whole improvement.
66const BPS_DENOMINATOR: u32 = 10_000;
67
68/// The configured user share, parsed from [`ENV_USER_IMPROVEMENT_SHARE_BPS`].
69static USER_IMPROVEMENT_SHARE_BPS: LazyLock<u32> = LazyLock::new(user_improvement_share_bps_env);
70
71/// Reads the user share from the environment, falling back to
72/// [`DEFAULT_USER_IMPROVEMENT_SHARE_BPS`] when the variable is unset or unusable.
73fn user_improvement_share_bps_env() -> u32 {
74    let Ok(raw) = std::env::var(ENV_USER_IMPROVEMENT_SHARE_BPS) else {
75        return DEFAULT_USER_IMPROVEMENT_SHARE_BPS;
76    };
77    match parse_user_improvement_share_bps(&raw) {
78        Some(bps) => bps,
79        None => {
80            warn!(
81                value = %raw,
82                default_bps = DEFAULT_USER_IMPROVEMENT_SHARE_BPS,
83                "{ENV_USER_IMPROVEMENT_SHARE_BPS} must be an integer from 0 to \
84                 {BPS_DENOMINATOR} basis points; using the default",
85            );
86            DEFAULT_USER_IMPROVEMENT_SHARE_BPS
87        }
88    }
89}
90
91/// Parses a user share in basis points, rejecting anything above the whole improvement — a share
92/// over `10_000` bps would commit more output than the route produces.
93fn parse_user_improvement_share_bps(raw: &str) -> Option<u32> {
94    raw.trim()
95        .parse::<u32>()
96        .ok()
97        .filter(|bps| *bps <= BPS_DENOMINATOR)
98}
99
100/// Returns the part of `improvement` handed to the user, `user_share_bps` of it.
101///
102/// Rounded up, so a share that would otherwise round away still moves the price in the user's
103/// favour. Never exceeds `improvement`, so the commitment it feeds stays within what the route
104/// produces.
105fn user_margin(improvement: &BigUint, user_share_bps: u32) -> BigUint {
106    let denominator = BigUint::from(BPS_DENOMINATOR);
107    (improvement * BigUint::from(user_share_bps) + (&denominator - 1u32)) / denominator
108}
109
110/// The fee in basis points that the protocol takes from an exclusive leg. It applies when the
111/// public worker pools find no route in the solve timeout.
112///
113/// Without a public quote there is no reference price. The protocol takes this fee and the user
114/// gets the rest. The value is the fee of Ekubo's most used ETH/USDC pool. This keeps the quote
115/// competitive with the other solvers.
116const NO_PUBLIC_ROUTE_FEE_BPS: u32 = 5;
117
118/// Which liquidity a solver pool (a group of workers) routes through, and therefore what its
119/// candidates mean in a quote.
120///
121/// A `PublicOnly` worker pool routes only through public liquidity and provides the committed
122/// (quoted) reference output; its workers filter exclusive components out of their graphs. An
123/// `IncludeExclusive` worker pool applies no filtering — its workers ingest whatever the
124/// deployment's stream delivers. In a deployment opted into exclusive components at the stream
125/// filter, an `IncludeExclusive` worker pool may beat the public reference — in which case the
126/// protocol captures the surplus. Without that opt-in, no exclusive components ever arrive and
127/// the two scopes behave identically.
128///
129/// Serialized in snake_case (`"public_only"` / `"include_exclusive"`) in `worker_pools.toml` via
130/// [`PoolConfig`].
131///
132/// [`PoolConfig`]: crate::PoolConfig
133#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
134#[serde(rename_all = "snake_case")]
135pub enum LiquidityScope {
136    /// Routes through public liquidity only, establishing the committed reference output.
137    #[default]
138    PublicOnly,
139    /// No filtering: routes through whatever the stream delivers, exclusive components
140    /// included if the deployment opted into them. Candidates from this scope may capture
141    /// surplus above the public reference.
142    IncludeExclusive,
143}
144
145/// Handle to a solver pool for dispatching orders.
146#[derive(Clone)]
147pub struct SolverPoolHandle {
148    /// Human-readable name for this worker pool (used in logging & metrics).
149    name: String,
150    /// Queue handle for this worker pool.
151    queue: TaskQueueHandle,
152    /// Which liquidity this worker pool routes through. Decides whether an order is dispatched
153    /// to it ([`SolverPoolHandle::serves`]).
154    liquidity_scope: LiquidityScope,
155}
156
157impl SolverPoolHandle {
158    /// Creates a new solver pool handle with the default [`LiquidityScope`].
159    pub fn new(name: impl Into<String>, queue: TaskQueueHandle) -> Self {
160        Self { name: name.into(), queue, liquidity_scope: LiquidityScope::default() }
161    }
162
163    /// Sets the worker pool's liquidity scope.
164    pub fn with_liquidity_scope(mut self, scope: LiquidityScope) -> Self {
165        self.liquidity_scope = scope;
166        self
167    }
168
169    /// Returns the worker pool name.
170    pub fn name(&self) -> &str {
171        &self.name
172    }
173
174    /// Returns the task queue handle.
175    pub fn queue(&self) -> &TaskQueueHandle {
176        &self.queue
177    }
178
179    /// Returns the worker pool's liquidity scope.
180    pub fn liquidity_scope(&self) -> LiquidityScope {
181        self.liquidity_scope
182    }
183}
184
185/// One worker pool's answer for an order, with the time that pool took to produce it.
186///
187/// "Worker pool" throughout is the solver thread group named in `worker_pools.toml`, never a
188/// liquidity pool.
189#[derive(Debug, Clone)]
190pub(crate) struct WorkerPoolQuote {
191    /// Name of the worker pool that produced this quote.
192    worker_pool: String,
193    /// What that worker pool solved: route, amounts and gas.
194    quote: OrderQuote,
195    /// Wall time the worker spent on this order, in milliseconds.
196    solve_time_ms: u64,
197}
198
199/// Collected responses for a single order from multiple solvers.
200#[derive(Debug)]
201pub(crate) struct OrderResponses {
202    /// ID of the order these responses correspond to.
203    order_id: String,
204    /// Quotes received from each worker pool.
205    quotes: Vec<WorkerPoolQuote>,
206    /// Worker pools that failed with their respective errors (worker_pool_name, error).
207    /// This captures all error types: timeouts, no routes, algorithm errors, etc.
208    failed_solvers: Vec<(String, SolveError)>,
209}
210
211impl OrderResponses {
212    /// Returns a copy keeping only candidates from public-scoped worker pools.
213    ///
214    /// These form the committed reference and the ranked fallback chain (ranked by `rank_quotes`,
215    /// consumed by the price guard); exclusive-access candidates are overlaid separately by
216    /// `combine_with_surplus`. `failed_solvers` is retained so placeholder construction is
217    /// unchanged.
218    fn public_only(&self, pool_scopes: &FxHashMap<String, LiquidityScope>) -> OrderResponses {
219        let quotes = self
220            .quotes
221            .iter()
222            .filter(|wq| {
223                pool_scopes.get(&wq.worker_pool) != Some(&LiquidityScope::IncludeExclusive)
224            })
225            .cloned()
226            .collect();
227        OrderResponses {
228            order_id: self.order_id.clone(),
229            quotes,
230            failed_solvers: self.failed_solvers.clone(),
231        }
232    }
233}
234
235/// Orchestrates multiple solver pools to find the best quote.
236pub struct WorkerPoolRouter {
237    /// All registered solver pools.
238    solver_pools: Vec<SolverPoolHandle>,
239    /// Configuration for the worker router.
240    config: WorkerPoolRouterConfig,
241    /// Encoder for encoding solutions into on-chain transactions.
242    encoder: Encoder,
243    /// Validates solution outputs against external price sources.
244    /// Present when the server has price guard enabled; `None` when disabled.
245    price_guard: Option<PriceGuard>,
246}
247
248impl WorkerPoolRouter {
249    /// Creates a new WorkerPoolRouter with the given solver pools, config, and encoder.
250    pub fn new(
251        solver_pools: Vec<SolverPoolHandle>,
252        config: WorkerPoolRouterConfig,
253        encoder: Encoder,
254    ) -> Self {
255        Self { solver_pools, config, encoder, price_guard: None }
256    }
257
258    /// Makes price guard validation available for this router.
259    ///
260    /// Providers are started and caches stay warm. Validation only runs for
261    /// requests where the client sets `enabled: true` in `PriceGuardConfig`.
262    pub fn with_price_guard(mut self, price_guard: PriceGuard) -> Self {
263        self.price_guard = Some(price_guard);
264        self
265    }
266
267    /// Returns the number of registered solver pools.
268    pub fn num_pools(&self) -> usize {
269        self.solver_pools.len()
270    }
271
272    /// Returns a quote by fanning out to the worker pools that serve the request.
273    ///
274    /// For each order in the request:
275    /// 1. Allocates the worker pools that serve it (see the `allocation` module)
276    /// 2. Sends the order to those worker pools in parallel
277    /// 3. Waits for responses with timeout
278    /// 4. Selects the best quote based on `amount_out_net_gas`
279    /// 5. If `encoding_options` are set on the request, encodes winning solutions into on-chain
280    ///    transactions
281    ///
282    /// `access` is the caller's access to exclusive liquidity, resolved at the trust
283    /// boundary. With [`ExclusiveAccess::Denied`] no exclusive-access worker pool is allocated, so
284    /// such worker pools do no work for the request and the quote is built from public liquidity
285    /// alone.
286    pub async fn quote(
287        &self,
288        request: QuoteRequest,
289        access: ExclusiveAccess,
290    ) -> Result<Quote, SolveError> {
291        let start = Instant::now();
292        let deadline = start + self.effective_timeout(request.options());
293        let min_responses = request
294            .options()
295            .min_responses()
296            .unwrap_or(self.config.min_responses());
297
298        if self.solver_pools.is_empty() {
299            return Err(SolveError::Internal("no solver pools configured".to_string()));
300        }
301
302        let params = match request.options().state_label().cloned() {
303            Some(label) => SolveParams::default().with_state_label(label),
304            None => SolveParams::default(),
305        };
306
307        counter!(
308            "worker_router_exclusive_access_total",
309            "access" => match access {
310                ExclusiveAccess::Granted => "granted",
311                ExclusiveAccess::Denied => "denied",
312            }
313        )
314        .increment(1);
315
316        // One allocation per order: which worker pools serve an order is a property of the order,
317        // not of the request. Today every order in a request classifies identically; once
318        // trade size joins `OrderClass` they will differ.
319        let class = OrderClass::new(access);
320        let allocations: Vec<Allocation<'_>> = request
321            .orders()
322            .iter()
323            .map(|_| allocate(&self.solver_pools, class))
324            .collect();
325
326        if allocations
327            .iter()
328            .any(Allocation::is_empty)
329        {
330            return Err(SolveError::Internal(format!(
331                "no solver pool serves this request: {class:?}"
332            )));
333        }
334
335        // Process each order independently in parallel
336        let order_futures: Vec<_> = request
337            .orders()
338            .iter()
339            .zip(&allocations)
340            .map(|(order, allocation)| {
341                self.solve_order(order.clone(), params.clone(), deadline, min_responses, allocation)
342            })
343            .collect();
344
345        let mut order_responses = futures::future::join_all(order_futures).await;
346
347        // Refine gas estimates for all candidates using estimate_gas_usage before ranking,
348        // so ranking uses accurate gas costs rather than naive route.total_gas().
349        if let Some(encoding_options) = request.options().encoding_options() {
350            refine_gas_estimates(&mut order_responses, encoding_options)?;
351        }
352
353        // Rank quotes for each order (sorted by refined amount_out_net_gas descending).
354        // `rank_quotes` produces the public ranking — the committed reference AND the price-guard
355        // fallback chain. When the allocation holds an exclusive-scope worker pool, the winning
356        // exclusive-access candidate is overlaid onto that ranked list (prepended) by
357        // `combine_with_surplus`, so the fallbacks are preserved. If the public worker pools find
358        // nothing, that ranking is the `NoRouteFound` placeholder. The exclusive candidate then
359        // uses a default fee.
360        let ranked_quotes: Vec<Vec<OrderQuote>> = order_responses
361            .iter()
362            .zip(&allocations)
363            .map(|(responses, allocation)| {
364                if allocation.exclusive_routing_active() {
365                    let public_ranked = self.rank_quotes(
366                        &responses.public_only(allocation.scopes()),
367                        request.options(),
368                    );
369                    combine_with_surplus(
370                        responses,
371                        allocation.scopes(),
372                        request.options(),
373                        public_ranked,
374                        *USER_IMPROVEMENT_SHARE_BPS,
375                        self.encoder.chain(),
376                    )
377                } else {
378                    self.rank_quotes(responses, request.options())
379                }
380            })
381            .collect();
382
383        // `join_all` preserves input order, so orders and responses line up one to one
384        for (order, responses) in request
385            .orders()
386            .iter()
387            .zip(&order_responses)
388        {
389            log_quote_comparison(order, responses, request.options());
390        }
391
392        // Validate against external prices when the client explicitly enables it.
393        let price_guard_config = request
394            .options()
395            .encoding_options()
396            .map(|e| e.price_guard())
397            .filter(|c| c.enabled());
398
399        let mut order_quotes: Vec<OrderQuote> = match (&self.price_guard, price_guard_config) {
400            (Some(guard), Some(config)) => guard
401                .validate(ranked_quotes, config)
402                .map_err(|e| {
403                    warn!(error = %e, "price guard validation error");
404                    SolveError::Internal(e.to_string())
405                })?,
406            (None, Some(_)) => {
407                return Err(SolveError::Internal(
408                    "price guard config provided but price guard is not enabled on this server"
409                        .to_string(),
410                ));
411            }
412            _ => ranked_quotes
413                .into_iter()
414                .filter_map(|candidates| candidates.into_iter().next())
415                .collect(),
416        };
417
418        // Encode solutions if encoding_options is set
419        if let Some(encoding_options) = request.options().encoding_options() {
420            let encode_start = Instant::now();
421            let encoded = self
422                .encoder
423                .encode(order_quotes, encoding_options.clone())
424                .await;
425            histogram!("encoding_duration_seconds").record(encode_start.elapsed().as_secs_f64());
426            order_quotes = match encoded {
427                Ok(quotes) => quotes,
428                Err(e) => {
429                    counter!("encoding_failures_total").increment(1);
430                    return Err(e);
431                }
432            };
433        }
434
435        // Calculate totals
436        let total_gas_estimate = order_quotes
437            .iter()
438            .map(|o| o.gas_estimate())
439            .fold(BigUint::ZERO, |acc, g| acc + g);
440
441        let solve_time_ms = start.elapsed().as_millis() as u64;
442
443        Ok(Quote::new(order_quotes, total_gas_estimate, solve_time_ms))
444    }
445
446    /// Solves a single order by fanning out to the worker pools allocated to it.
447    async fn solve_order(
448        &self,
449        order: Order,
450        params: SolveParams,
451        deadline: Instant,
452        min_responses: usize,
453        allocation: &Allocation<'_>,
454    ) -> OrderResponses {
455        let start_time = Instant::now();
456        let order_id = order.id().to_string();
457
458        let allocated: Vec<(&str, LiquidityScope)> = allocation
459            .worker_pools()
460            .iter()
461            .map(|worker_pool| (worker_pool.name(), worker_pool.liquidity_scope()))
462            .collect();
463        debug!(
464            order_id = %order_id,
465            worker_pools = ?allocated,
466            "dispatching order to allocated worker pools"
467        );
468
469        // Fan-out: send order to the allocated worker pools. Worker pools that do not serve this
470        // order were already dropped by `allocate`, so nothing here filters by access.
471        let mut pending: FuturesUnordered<_> = allocation
472            .worker_pools()
473            .iter()
474            .map(|worker_pool| {
475                let order_clone = order.clone();
476                let worker_pool_name = worker_pool.name().to_string();
477                let queue = worker_pool.queue().clone();
478                let task_params = params.clone();
479
480                async move {
481                    let result = queue
482                        .enqueue(order_clone, task_params)
483                        .await;
484                    (worker_pool_name, result)
485                }
486            })
487            .collect();
488
489        let mut quotes = Vec::new();
490        let mut failed_solvers: Vec<(String, SolveError)> = Vec::new();
491        let mut remaining_worker_pools: FxHashSet<String> = allocation
492            .worker_pools()
493            .iter()
494            .map(|p| p.name().to_string())
495            .collect();
496        let mut has_public_response = false;
497        let mut has_exclusive_access_response = false;
498
499        // Collect responses with timeout
500        loop {
501            let deadline_instant = tokio::time::Instant::from_std(deadline);
502
503            tokio::select! {
504                // Always checks timeout first, ensuring we respect the deadline
505                biased;
506
507                // Timeout reached
508                _ = tokio::time::sleep_until(deadline_instant) => {
509                    // Mark all remaining worker pools as timed out
510                    let elapsed_ms = deadline.saturating_duration_since(Instant::now())
511                        .as_millis() as u64;
512                    for worker_pool_name in remaining_worker_pools.drain() {
513                        failed_solvers.push((
514                            worker_pool_name,
515                            SolveError::Timeout { elapsed_ms },
516                        ));
517                    }
518                    break;
519                }
520
521                // Response received
522                result = pending.next() => {
523                    match result {
524                        Some((worker_pool_name, Ok(single_quote))) => {
525                            // Remove from remaining
526                            remaining_worker_pools.remove(&worker_pool_name);
527
528                            if allocation.is_exclusive(&worker_pool_name) {
529                                has_exclusive_access_response = true;
530                            } else {
531                                has_public_response = true;
532                            }
533
534                            quotes.push(WorkerPoolQuote {
535                                worker_pool: worker_pool_name.clone(),
536                                quote: single_quote.order().clone(),
537                                solve_time_ms: single_quote.solve_time_ms(),
538                            });
539
540                            // Scope-aware early return: when the allocation routes through
541                            // exclusive liquidity, only fire once we have ≥1 public AND ≥1
542                            // exclusive-access response (so the surplus overlay has both
543                            // inputs). Otherwise, use pure count-based gating (original
544                            // behaviour).
545                            let scope_ready = if allocation.exclusive_routing_active() {
546                                has_public_response && has_exclusive_access_response
547                            } else {
548                                true
549                            };
550                            if min_responses > 0
551                                && quotes.len() >= min_responses
552                                && scope_ready
553                            {
554                                debug!(
555                                    order_id = %order_id,
556                                    responses = quotes.len(),
557                                    min_responses,
558                                    "early return: min_responses reached"
559                                );
560                                counter!("worker_router_early_returns_total").increment(1);
561                                break;
562                            }
563                        }
564                        Some((worker_pool_name, Err(e))) => {
565                            remaining_worker_pools.remove(&worker_pool_name);
566                            // A failed exclusive-access worker pool still counts as "responded"
567                            // for gating — we know it won't produce a surplus quote, so the
568                            // public worker pools can early-return without waiting for a result
569                            // that will never come.
570                            if allocation.is_exclusive(&worker_pool_name) {
571                                has_exclusive_access_response = true;
572                            }
573                            debug!(
574                                pool = %worker_pool_name,
575                                order_id = %order_id,
576                                error = %e,
577                                "solver pool failed"
578                            );
579                            failed_solvers.push((worker_pool_name, e));
580                        }
581                        None => {
582                            // All futures completed
583                            break;
584                        }
585                    }
586                }
587            }
588        }
589
590        // Record metrics
591        let duration = start_time.elapsed().as_secs_f64();
592        histogram!("worker_router_solve_duration_seconds").record(duration);
593        histogram!("worker_router_solver_responses").record(quotes.len() as f64);
594
595        // Record failures by worker pool and error type
596        for (worker_pool_name, error) in &failed_solvers {
597            let error_type = solver_error_label(error);
598            counter!("worker_router_solver_failures_total", "pool" => worker_pool_name.clone(), "error_type" => error_type).increment(1);
599        }
600
601        if !failed_solvers.is_empty() {
602            let timeout_count = failed_solvers
603                .iter()
604                .filter(|(_, e)| matches!(e, SolveError::Timeout { .. }))
605                .count();
606            let other_count = failed_solvers.len() - timeout_count;
607            warn!(
608                order_id = %order_id,
609                timeout_count,
610                other_failures = other_count,
611                "some solver pools failed"
612            );
613        }
614
615        OrderResponses { order_id, quotes, failed_solvers }
616    }
617
618    /// Returns all valid quotes for an order, ranked by `amount_out_net_gas` descending.
619    ///
620    /// If no valid quotes exist, returns a single-element vec with a placeholder
621    /// (`NoRouteFound` or `Timeout`) so that downstream always has at least one
622    /// candidate per order.
623    fn rank_quotes(&self, responses: &OrderResponses, options: &QuoteOptions) -> Vec<OrderQuote> {
624        let mut valid_quotes: Vec<_> = responses
625            .quotes
626            .iter()
627            .filter(|wq| is_rankable(&wq.quote, options))
628            .collect();
629
630        // Sort descending by amount_out_net_gas
631        valid_quotes.sort_by(|a, b| {
632            b.quote
633                .amount_out_net_gas()
634                .cmp(a.quote.amount_out_net_gas())
635        });
636
637        if !valid_quotes.is_empty() {
638            counter!("worker_router_orders_total", "status" => "success").increment(1);
639            let best = valid_quotes[0];
640            counter!("worker_router_best_quote_pool", "pool" => best.worker_pool.clone())
641                .increment(1);
642            debug!(
643                order_id = %best.quote.order_id(),
644                number_of_candidates = valid_quotes.len(),
645                "ranked quotes"
646            );
647            return valid_quotes
648                .into_iter()
649                .map(|pq| pq.quote.clone())
650                .collect();
651        }
652
653        // No valid quote found - return a NoRouteFound response
654        // Try to get any response to extract block info, or create a placeholder
655        let fallback = if let Some(WorkerPoolQuote { quote: any_q, .. }) = responses.quotes.first()
656        {
657            counter!("worker_router_orders_total", "status" => "no_route").increment(1);
658            let mut fallback = OrderQuote::new(
659                responses.order_id.clone(),
660                QuoteStatus::NoRouteFound,
661                any_q.amount_in().clone(),
662                BigUint::ZERO,
663                BigUint::ZERO,
664                BigUint::ZERO,
665                any_q.block().clone(),
666                String::new(),
667                any_q.sender().clone(),
668                any_q.receiver().clone(),
669                any_q.solved_against().clone(),
670            );
671            // Only label the cause when a quote is actually over the request's
672            // max_gas, so a future filter or non-Success status cannot silently
673            // get attributed to the gas cap.
674            let over_max_gas = responses.quotes.iter().any(|pq| {
675                options
676                    .max_gas()
677                    .is_some_and(|max| pq.quote.gas_estimate() > max)
678            });
679            fallback.set_no_route_cause(over_max_gas.then_some(SolveError::MaxGasExceeded));
680            fallback
681        } else {
682            // No responses at all - determine status from failure types
683            let status = if responses.failed_solvers.is_empty() {
684                QuoteStatus::NoRouteFound
685            } else {
686                // If all failures are timeouts, report as Timeout
687                // Otherwise report as NoRouteFound (more general failure)
688                let all_timeouts = responses
689                    .failed_solvers
690                    .iter()
691                    .all(|(_, e)| matches!(e, SolveError::Timeout { .. }));
692                let all_not_ready = responses
693                    .failed_solvers
694                    .iter()
695                    .all(|(_, e)| matches!(e, SolveError::NotReady(_)));
696                if all_timeouts {
697                    QuoteStatus::Timeout
698                } else if all_not_ready {
699                    QuoteStatus::NotReady
700                } else {
701                    QuoteStatus::NoRouteFound
702                }
703            };
704
705            // Record status metric
706            let status_label = match status {
707                QuoteStatus::Timeout => "timeout",
708                QuoteStatus::NotReady => "not_ready",
709                _ => "no_route",
710            };
711            counter!("worker_router_orders_total", "status" => status_label).increment(1);
712
713            // No worker responded — use the requested label if set, otherwise "0"
714            // (we have no block context here since no worker completed).
715            let label = options
716                .state_label()
717                .cloned()
718                .unwrap_or_else(|| "0".to_string());
719            let mut fallback = OrderQuote::new(
720                responses.order_id.clone(),
721                status,
722                BigUint::ZERO,
723                BigUint::ZERO,
724                BigUint::ZERO,
725                BigUint::ZERO,
726                BlockInfo::new(0, String::new(), 0),
727                String::new(),
728                Bytes::default(),
729                Bytes::default(),
730                label,
731            );
732            fallback.set_no_route_cause(aggregate_no_route_cause(&responses.failed_solvers));
733            fallback
734        };
735        vec![fallback]
736    }
737
738    /// Returns the effective timeout for a request.
739    fn effective_timeout(&self, options: &QuoteOptions) -> Duration {
740        options
741            .timeout_ms()
742            .map(Duration::from_millis)
743            .unwrap_or(self.config.default_timeout())
744    }
745}
746
747/// Builds the final ranked quote list for one order by deciding whether an exclusive-access
748/// route should execute instead of the best public route.
749///
750/// Inputs: `public_ranked` is the ranking of public-worker-pool quotes from `rank_quotes`; its
751/// head is the public reference from which the committed amount is derived. `responses`
752/// additionally holds the candidates from `ExclusiveAccess`-scoped worker pools (routes that may
753/// use exclusive components).
754///
755/// There are two ways to set the commitment:
756/// - **Matched**: `public_ranked` starts with a successful public quote, and the commitment follows
757///   that quote, as shown below.
758/// - **Default fee**: no public quote succeeded in the solve timeout, so `public_ranked` holds only
759///   the `NoRouteFound` placeholder. The commitment is the route output minus
760///   `NO_PUBLIC_ROUTE_FEE_BPS` of the exclusive leg. The candidate is skipped if the fee leaves the
761///   user with nothing after gas.
762///
763/// In matched mode a candidate must at least match the public reference net of gas; what it
764/// produces on top, `improvement = exclusive_net − public_net`, is then split. The user's net
765/// target is `required_net = public_net + margin`, where `margin` is `user_share_bps` of
766/// `improvement` — so the mark-up is only as large as the route can afford, and a route with more
767/// to give quotes a better user price. Splitting the improvement rather than marking up the trade
768/// also puts the mark-up below a basis point of the trade whenever the improvement is small.
769///
770/// The committed amount is the larger of two lower bounds:
771/// `max(public_amount_out, required_net + exclusive_gas)`. The first guarantees the quoted
772/// `amount_out` is never below the public market's; the second guarantees the user — who pays
773/// the exclusive route's gas — nets at least `required_net`.
774///
775/// Which bound is larger depends on the gas comparison:
776/// - `exclusive_gas > public_gas`: committed = `required_net + exclusive_gas`, which exceeds
777///   `public_amount_out`. The user receives more tokens than the public quote and nets exactly
778///   `required_net`.
779/// - `exclusive_gas <= public_gas`: committed = `public_amount_out`. The user nets
780///   `public_amount_out − exclusive_gas`, which exceeds `public_net` by the gas difference. A
781///   commitment of `required_net + exclusive_gas` would still leave the user whole and capture
782///   more, but it is below `public_amount_out` — quoting less than the public market is ruled out,
783///   so the gas difference stays with the user.
784///
785/// If there is a commitment, this returns a new list. The head is the pinned surplus quote, and the
786/// entries of `public_ranked` follow it as price-guard fallbacks. The pinned quote is the winning
787/// candidate with:
788/// - `amount_out` set to the committed amount,
789/// - `Swap::committed_amount_out` set on each exclusive leg, and
790/// - an order-level `SurplusInfo`.
791///
792/// If there is no commitment, this returns `public_ranked` unchanged. In default-fee mode the
793/// placeholder is the only fallback, so the order answers "no route" if there is no commitment or
794/// the price guard rejects the exclusive quote.
795///
796/// In matched mode the user is never worse off than the public market, in quoted `amount_out` and
797/// net of gas. A candidate that equals the public reference is quoted with zero surplus, and
798/// executes on exclusive liquidity at the public price.
799///
800/// Per-leg attribution: the route's surplus over the committed amount (`realized − committed`)
801/// is deducted from the exclusive legs — each leg absorbs what it can, capped at its own output,
802/// in route order. Only exclusive legs can withhold output, so the whole
803/// surplus must come out of them; public branches pay out in full. If the exclusive legs cannot
804/// absorb all of it, the remainder is left with the user (who then receives more than the
805/// committed amount).
806///
807/// Exact-in orders only: the commitment, both gates, and the surplus are all denominated in
808/// `amount_out`. Exact-out support would invert the logic — fixed output, commitment and surplus
809/// on the input side — and needs its own treatment here.
810fn combine_with_surplus(
811    responses: &OrderResponses,
812    pool_scopes: &FxHashMap<String, LiquidityScope>,
813    options: &QuoteOptions,
814    public_ranked: Vec<OrderQuote>,
815    user_share_bps: u32,
816    chain: Chain,
817) -> Vec<OrderQuote> {
818    let Some(exclusive_candidate) =
819        best_exclusive_candidate(responses, pool_scopes, options, chain)
820    else {
821        return public_ranked;
822    };
823
824    let public_reference = public_ranked
825        .first()
826        .filter(|q| q.status() == QuoteStatus::Success);
827
828    let commitment = match public_reference {
829        Some(public_reference) => {
830            matched_commitment(public_reference, exclusive_candidate, user_share_bps)
831        }
832        None => default_fee_commitment(exclusive_candidate),
833    };
834    let Some(committed_amount_out) = commitment else {
835        return public_ranked;
836    };
837
838    // Only meaningful against a public reference: the user's improvement over what a plain
839    // public quote would have paid. The no-public-route fallback (default_fee_commitment) has
840    // no public amount_out to diff against.
841    if let Some(public_reference) = public_reference {
842        let user_savings = &committed_amount_out - public_reference.amount_out();
843        gauge!("exclusive_user_savings_amount").increment(user_savings.to_f64().unwrap_or(0.0));
844    }
845
846    let mut result = Vec::with_capacity(public_ranked.len() + 1);
847    result.push(pin_commitment(exclusive_candidate, committed_amount_out));
848    result.extend(public_ranked);
849    result
850}
851
852/// Whether a quote is eligible to be ranked against the others for this request.
853fn is_rankable(quote: &OrderQuote, options: &QuoteOptions) -> bool {
854    quote.status() == QuoteStatus::Success &&
855        options
856            .max_gas()
857            .map(|max| quote.gas_estimate() <= max)
858            .unwrap_or(true)
859}
860
861/// Returns the exclusive-access candidate with the highest output net of gas. The candidate must
862/// obey the request `max_gas` and the route shape rules of `has_valid_exclusive_route`.
863fn best_exclusive_candidate<'a>(
864    responses: &'a OrderResponses,
865    pool_scopes: &FxHashMap<String, LiquidityScope>,
866    options: &QuoteOptions,
867    chain: Chain,
868) -> Option<&'a OrderQuote> {
869    responses
870        .quotes
871        .iter()
872        .filter(|wq| pool_scopes.get(&wq.worker_pool) == Some(&LiquidityScope::IncludeExclusive))
873        .filter(|wq| wq.quote.status() == QuoteStatus::Success)
874        .filter(|wq| {
875            options
876                .max_gas()
877                .map(|max| wq.quote.gas_estimate() <= max)
878                .unwrap_or(true)
879        })
880        .filter(|wq| has_valid_exclusive_route(&wq.quote, chain))
881        .max_by(|a, b| {
882            a.quote
883                .amount_out_net_gas()
884                .cmp(b.quote.amount_out_net_gas())
885        })
886        .map(|wq| &wq.quote)
887}
888
889/// Returns the amount to commit against a successful public quote:
890/// `max(public_amount_out, required_net + gas)`.
891///
892/// Returns `None` if the candidate fails a gate. The candidate must match the public output net of
893/// gas, and it must produce at least the public output. `combine_with_surplus` gives the reason for
894/// each bound.
895fn matched_commitment(
896    public_reference: &OrderQuote,
897    exclusive_candidate: &OrderQuote,
898    user_share_bps: u32,
899) -> Option<BigUint> {
900    // The candidate route must match the public reference net-of-gas; anything it produces on
901    // top is the improvement, of which the user keeps a share (the margin).
902    let public_net_amount_out = public_reference.amount_out_net_gas();
903    if exclusive_candidate.amount_out_net_gas() < public_net_amount_out {
904        return None;
905    }
906    let improvement = exclusive_candidate.amount_out_net_gas() - public_net_amount_out;
907    let required_net_amount_out = public_net_amount_out + user_margin(&improvement, user_share_bps);
908
909    // Exact-in assumption: everything below compares and commits output amounts. For exact-out
910    // orders this comparison would have to run on amount_in instead (see function docs).
911    // We promise the user at least the public route's output. A private route that produces
912    // less can't keep that promise, so we skip it — even when its gas savings make it better net.
913    let public_amount_out = public_reference.amount_out();
914    if exclusive_candidate.amount_out() < public_amount_out {
915        return None;
916    }
917
918    // Gas the user pays to execute the exclusive route, in output-token terms.
919    let gas_cost = exclusive_candidate.amount_out() - exclusive_candidate.amount_out_net_gas();
920    Some((required_net_amount_out + gas_cost).max(public_amount_out.clone()))
921}
922
923/// Returns the amount to commit if the public worker pools find no route. The amount is the
924/// candidate output minus `NO_PUBLIC_ROUTE_FEE_BPS` of the exclusive leg output.
925///
926/// The fee applies to the exclusive leg only, not to the whole route. The public market prices the
927/// public branches of a split route, so the protocol takes nothing from them.
928///
929/// Returns `None` if the commitment does not cover the route gas, or if the route has no exclusive
930/// leg.
931fn default_fee_commitment(exclusive_candidate: &OrderQuote) -> Option<BigUint> {
932    let exclusive_leg_amount_out = exclusive_candidate
933        .route()?
934        .swaps()
935        .iter()
936        .find(|swap| is_exclusive(swap.protocol_component()))?
937        .amount_out();
938
939    let realized_amount_out = exclusive_candidate.amount_out();
940    let fee = exclusive_leg_amount_out * BigUint::from(NO_PUBLIC_ROUTE_FEE_BPS) /
941        BigUint::from(BPS_DENOMINATOR);
942    let committed_amount_out = realized_amount_out - fee;
943
944    let gas_cost = realized_amount_out - exclusive_candidate.amount_out_net_gas();
945    if committed_amount_out <= gas_cost {
946        return None;
947    }
948    Some(committed_amount_out)
949}
950
951/// Sets `exclusive_candidate` to `committed_amount_out`. This sets the committed amount on each
952/// leg, sets `amount_out` and `amount_out_net_gas`, and attaches the `SurplusInfo`.
953///
954/// The exclusive components capture all output above the commitment.
955fn pin_commitment(exclusive_candidate: &OrderQuote, committed_amount_out: BigUint) -> OrderQuote {
956    let exclusive_route_amount_out = exclusive_candidate.amount_out();
957    let exclusive_gas_cost = exclusive_route_amount_out - exclusive_candidate.amount_out_net_gas();
958    let surplus_amount = exclusive_route_amount_out - &committed_amount_out;
959
960    gauge!("exclusive_fee_amount").increment(surplus_amount.to_f64().unwrap_or(0.0));
961
962    let mut surplus_quote = exclusive_candidate.clone();
963
964    // Final output of each swap's path, walked backwards (a path's terminal output propagates
965    // to its chained predecessors). Converting captured surplus into a leg's token needs the
966    // realized downstream price `path_final_out / leg_out`; for terminal legs the ratio is 1.
967    let path_final_outs: Vec<BigUint> = surplus_quote
968        .route()
969        .map(|route| {
970            let swaps = route.swaps();
971            let mut finals = vec![BigUint::ZERO; swaps.len()];
972            let mut current_final = BigUint::ZERO;
973            for i in (0..swaps.len()).rev() {
974                let is_terminal =
975                    i == swaps.len() - 1 || swaps[i + 1].token_in() != swaps[i].token_out();
976                if is_terminal {
977                    current_final = swaps[i].amount_out().clone();
978                }
979                finals[i] = current_final.clone();
980            }
981            finals
982        })
983        .unwrap_or_default();
984
985    if let Some(route) = surplus_quote.route_mut() {
986        // Capture the route's surplus at the exclusive legs.
987        // Each leg absorbs up to its path's capacity; any remainder is left with the user.
988        let mut surplus = exclusive_route_amount_out - &committed_amount_out;
989        for (i, swap) in route.swaps_mut().iter_mut().enumerate() {
990            if is_exclusive(swap.protocol_component()) {
991                let Some(path_final_out) = path_final_outs.get(i) else {
992                    continue;
993                };
994                let captured = surplus
995                    .clone()
996                    .min(path_final_out.clone());
997                // Convert into the leg's own token, rounding down so any error is taken
998                // from the protocol, not the user. Today the leg is always the last hop of
999                // its path (validator), so path_final_out equals the leg's output and this
1000                // divides by itself — a plain subtraction. Once mid-path legs are allowed,
1001                // the "user gets at least the committed amount" guarantee also requires the
1002                // components after the leg to have diminishing returns.
1003                let captured_leg = if *path_final_out == BigUint::ZERO {
1004                    BigUint::ZERO
1005                } else {
1006                    &captured * swap.amount_out() / path_final_out
1007                };
1008                debug_assert!(
1009                    captured_leg <= *swap.amount_out(),
1010                    "captured amount ({captured_leg}) must not exceed the leg's output ({})",
1011                    swap.amount_out(),
1012                );
1013                let committed_leg = swap.amount_out() - &captured_leg;
1014                surplus -= captured;
1015                swap.set_committed_amount_out(committed_leg);
1016            }
1017        }
1018    }
1019
1020    surplus_quote.set_amount_out(committed_amount_out.clone());
1021
1022    // The user nets the committed amount minus the exclusive route's gas; in matched mode the
1023    // commitment is built so this is >= the public head's net, keeping the candidate list ranked
1024    // descending.
1025    surplus_quote.set_amount_out_net_gas(&committed_amount_out - &exclusive_gas_cost);
1026
1027    let surplus_info = SurplusInfo::new(surplus_amount, committed_amount_out);
1028    surplus_quote.with_surplus(surplus_info)
1029}
1030
1031/// Returns `true` only for routes carrying exactly one exclusive leg that produces the route's
1032/// output token, either itself or through a single native wrap leg.
1033///
1034/// Returns `false` for routes with no exclusive leg, more than one exclusive leg, an empty route,
1035/// no route at all, or an exclusive leg whose output reaches the route's output token any other
1036/// way.
1037///
1038/// The single-leg constraint is a v1 restriction that keeps per-leg surplus attribution
1039/// unambiguous: multiple exclusive legs make the per-component attribution non-unique.
1040///
1041/// A trailing wrap leg is allowed because it converts 1:1, so it needs no inverse simulation.
1042/// `pin_commitment` converts captured surplus into a leg's token with the ratio
1043/// `path_final_out / leg_out`, which a wrap leaves at exactly 1 — the same arithmetic a terminal
1044/// leg gets. Without this a pool quoting the native token could never serve a request for the
1045/// wrapped token, even though tycho streams the wrap as an ordinary component.
1046fn has_valid_exclusive_route(quote: &OrderQuote, chain: Chain) -> bool {
1047    let Some(route) = quote.route() else {
1048        return false;
1049    };
1050
1051    let swaps = route.swaps();
1052    if swaps.is_empty() {
1053        return false;
1054    }
1055
1056    let Some(output_token) = route.output_token() else {
1057        return false;
1058    };
1059
1060    let mut exclusive_count = 0;
1061
1062    for (index, swap) in swaps.iter().enumerate() {
1063        if !is_exclusive(swap.protocol_component()) {
1064            continue;
1065        }
1066
1067        // Only the immediately following leg is considered: a wrap run that ends at the output
1068        // token is always a single leg, and requiring adjacency keeps this validator in step with
1069        // the chaining rule `pin_commitment` uses to find a path's final output.
1070        let reaches_output = *swap.token_out() == output_token ||
1071            swaps
1072                .get(index + 1)
1073                .is_some_and(|next| {
1074                    next.token_in() == swap.token_out() &&
1075                        next.token_out() == &output_token &&
1076                        is_native_wrap(next, chain)
1077                });
1078        if !reaches_output {
1079            counter!("exclusive_route_invalid_shape_total").increment(1);
1080            return false;
1081        }
1082        exclusive_count += 1;
1083    }
1084
1085    exclusive_count == 1
1086}
1087
1088/// Returns `true` when the leg converts between the native token and its wrapped form, which tycho
1089/// streams as a 1:1 component.
1090///
1091/// An unregistered custom chain resolves no wrap pair, so no leg qualifies and the exclusive leg
1092/// must be terminal.
1093fn is_native_wrap(swap: &Swap, chain: Chain) -> bool {
1094    let (Ok(native), Ok(wrapped)) = (chain.try_native_token(), chain.try_wrapped_native_token())
1095    else {
1096        return false;
1097    };
1098
1099    let (token_in, token_out) = (swap.token_in(), swap.token_out());
1100    (*token_in == native.address && *token_out == wrapped.address) ||
1101        (*token_in == wrapped.address && *token_out == native.address)
1102}
1103
1104/// Shared-graph facts beat path-specific reasons (most specific first:
1105/// amount-too-small, no-scorable-paths, no-graph-path), which beat liquidity,
1106/// data, algorithm, and infrastructure errors; the first error wins within a
1107/// tier.
1108fn aggregate_no_route_cause(failed_solvers: &[(String, SolveError)]) -> Option<SolveError> {
1109    failed_solvers
1110        .iter()
1111        .map(|(_, error)| error)
1112        .min_by_key(|error| cause_tier(error))
1113        .cloned()
1114}
1115
1116fn cause_tier(error: &SolveError) -> u8 {
1117    use crate::algorithm::NoPathReason;
1118    match error {
1119        SolveError::NoRouteFound { reason: Some(reason), .. } => match reason {
1120            NoPathReason::SourceTokenNotInGraph | NoPathReason::DestinationTokenNotInGraph => 0,
1121            NoPathReason::AmountTooSmall => 1,
1122            NoPathReason::NoScorablePaths => 2,
1123            NoPathReason::NoGraphPath => 3,
1124        },
1125        SolveError::InsufficientLiquidity { .. } | SolveError::MaxGasExceeded => 2,
1126        SolveError::MissingData(_) |
1127        SolveError::MarketDataStale { .. } |
1128        SolveError::ComputationFailed(_) |
1129        SolveError::NotReady(_) => 3,
1130        SolveError::SimulationFailed(_) | SolveError::AlgorithmError(_) => 4,
1131        SolveError::Timeout { .. } |
1132        SolveError::QueueFull |
1133        SolveError::Internal(_) |
1134        SolveError::InvalidOrder(_) |
1135        SolveError::FailedEncoding(_) |
1136        SolveError::EncodingUnavailable(_) |
1137        SolveError::PriceCheckFailed { .. } => 5,
1138        SolveError::NoRouteFound { reason: None, .. } => 6,
1139    }
1140}
1141
1142fn refine_gas_estimates(
1143    order_responses: &mut Vec<OrderResponses>,
1144    encoding_options: &EncodingOptions,
1145) -> Result<(), SolveError> {
1146    for responses in order_responses {
1147        for WorkerPoolQuote { quote, .. } in &mut responses.quotes {
1148            if quote.status() != QuoteStatus::Success {
1149                continue;
1150            }
1151            let solution = Solution::try_from(&*quote)?
1152                .with_user_transfer_type(encoding_options.transfer_type().clone());
1153            let refined_gas = estimate_gas_usage(&solution, derive_strategy(quote));
1154            let naive_gas = quote.gas_estimate().clone();
1155            if naive_gas > BigUint::ZERO {
1156                let gas_cost_in_token_out = quote.amount_out() - quote.amount_out_net_gas();
1157                let new_gas_cost = &gas_cost_in_token_out * &refined_gas / &naive_gas;
1158                let new_net = if new_gas_cost <= *quote.amount_out() {
1159                    quote.amount_out() - &new_gas_cost
1160                } else {
1161                    BigUint::ZERO
1162                };
1163                quote.set_amount_out_net_gas(new_net);
1164                quote.set_gas_estimate(refined_gas);
1165            }
1166        }
1167    }
1168    Ok(())
1169}
1170
1171fn derive_strategy(quote: &OrderQuote) -> Strategy {
1172    let Some(route) = quote.route() else { return Strategy::Single };
1173    let swaps = route.swaps();
1174    if swaps.len() == 1 {
1175        Strategy::Single
1176    } else if swaps.iter().any(|s| *s.split() > 0.0) {
1177        Strategy::Split
1178    } else {
1179        Strategy::Sequential
1180    }
1181}
1182
1183#[cfg(test)]
1184mod tests {
1185    use std::sync::{
1186        atomic::{AtomicUsize, Ordering},
1187        Arc,
1188    };
1189
1190    use rstest::rstest;
1191    use tycho_execution::encoding::evm::swap_encoder::swap_encoder_registry::SwapEncoderRegistry;
1192    use tycho_simulation::{
1193        tycho_common::models::Chain,
1194        tycho_core::{
1195            models::{token::Token, Address, Chain as SimChain},
1196            Bytes,
1197        },
1198    };
1199
1200    use super::*;
1201    use crate::{
1202        algorithm::test_utils::{component, MockProtocolSim},
1203        feed::exclusivity::mark_exclusive,
1204        types::internal::SolveTask,
1205        EncodingOptions, OrderSide, Route, SingleOrderQuote, Swap,
1206    };
1207
1208    fn default_encoder() -> Encoder {
1209        let registry = SwapEncoderRegistry::new(Chain::Ethereum)
1210            .add_default_encoders(None)
1211            .expect("default encoders should always succeed");
1212        let encoder =
1213            Encoder::new(Chain::Ethereum, registry).expect("encoder creation should succeed");
1214        // Load fees so encoding can run; the fetcher supplies on-chain values in production.
1215        encoder
1216            .router_fees()
1217            .set(crate::encoding::router_fees::RouterFees::new(
1218                100_000_000,
1219                100_000,
1220                20_000_000,
1221                rustc_hash::FxHashMap::default(),
1222            ));
1223        encoder
1224    }
1225
1226    /// Builds a worker pool response with no recorded solve time, for the ranking tests that do
1227    /// not exercise timing. Use [`timed_worker_quote`] where the solve time is the point.
1228    fn worker_quote((worker_pool, quote): (String, OrderQuote)) -> WorkerPoolQuote {
1229        timed_worker_quote(&worker_pool, quote, 0)
1230    }
1231
1232    fn timed_worker_quote(
1233        worker_pool: &str,
1234        quote: OrderQuote,
1235        solve_time_ms: u64,
1236    ) -> WorkerPoolQuote {
1237        WorkerPoolQuote { worker_pool: worker_pool.to_string(), quote, solve_time_ms }
1238    }
1239
1240    /// A minimal successful quote for tests that only care about the net-of-gas amount.
1241    fn success_quote(net: u64) -> OrderQuote {
1242        OrderQuote::new(
1243            "o1".to_string(),
1244            QuoteStatus::Success,
1245            BigUint::from(1_000u64),
1246            BigUint::from(net + 10),
1247            BigUint::from(10u64),
1248            BigUint::from(net),
1249            BlockInfo::new(42, "0xabc".to_string(), 0),
1250            "algo".to_string(),
1251            Bytes::default(),
1252            Bytes::default(),
1253            "1".to_string(),
1254        )
1255    }
1256
1257    /// `public_only` must carry the order identity across, or the surplus path logs and ranks
1258    /// against a response set that has lost it.
1259    #[test]
1260    fn test_public_only_keeps_order_id_and_failures() {
1261        let responses = OrderResponses {
1262            order_id: "o1".to_string(),
1263            quotes: vec![
1264                timed_worker_quote("public", success_quote(1_000), 3),
1265                timed_worker_quote("excl", success_quote(900), 4),
1266            ],
1267            failed_solvers: vec![("c".to_string(), SolveError::QueueFull)],
1268        };
1269        let scopes = FxHashMap::from_iter([
1270            ("public".to_string(), LiquidityScope::PublicOnly),
1271            ("excl".to_string(), LiquidityScope::IncludeExclusive),
1272        ]);
1273        let public = responses.public_only(&scopes);
1274
1275        assert_eq!(public.order_id, "o1");
1276        assert_eq!(public.quotes.len(), 1);
1277        assert_eq!(public.quotes[0].worker_pool, "public");
1278        assert_eq!(public.quotes[0].solve_time_ms, 3);
1279        assert_eq!(public.failed_solvers.len(), 1);
1280    }
1281
1282    fn make_address(byte: u8) -> tycho_simulation::tycho_common::models::Address {
1283        Address::from([byte; 20])
1284    }
1285
1286    fn make_order() -> Order {
1287        Order::new(
1288            make_address(0x01),
1289            make_address(0x02),
1290            BigUint::from(1000u64),
1291            OrderSide::Sell,
1292            make_address(0xAA),
1293        )
1294        .with_id("test-order".to_string())
1295    }
1296
1297    fn make_single_quote(amount_out_net_gas: u64) -> SingleOrderQuote {
1298        let make_token = |addr: Address| Token {
1299            address: addr,
1300            symbol: "T".to_string(),
1301            decimals: 18,
1302            tax: Default::default(),
1303            gas: vec![],
1304            chain: SimChain::Ethereum,
1305            quality: 100,
1306        };
1307        let tin = make_address(0x01);
1308        let tout = make_address(0x02);
1309        let tin_token = make_token(tin.clone());
1310        let tout_token = make_token(tout.clone());
1311        let swap = Swap::new(
1312            "pool-1".to_string(),
1313            "uniswap_v2".to_string(),
1314            tin.clone(),
1315            tout.clone(),
1316            BigUint::from(1000u64),
1317            BigUint::from(990u64),
1318            BigUint::from(50_000u64),
1319            component(
1320                "0x0000000000000000000000000000000000000001",
1321                &[tin_token.clone(), tout_token.clone()],
1322            ),
1323            Box::new(MockProtocolSim::default()),
1324        );
1325        let mut tokens = FxHashMap::default();
1326        tokens.insert(tin, tin_token);
1327        tokens.insert(tout, tout_token);
1328        let quote = OrderQuote::new(
1329            "test-order".to_string(),
1330            QuoteStatus::Success,
1331            BigUint::from(1000u64),
1332            BigUint::from(990u64),
1333            BigUint::from(100_000u64),
1334            BigUint::from(amount_out_net_gas),
1335            BlockInfo::new(1, "0x123".to_string(), 1000),
1336            "test".to_string(),
1337            Bytes::from(make_address(0xAA).as_ref()),
1338            Bytes::from(make_address(0xAA).as_ref()),
1339            "1".to_string(),
1340        )
1341        .with_route(Route::new(vec![swap], tokens).expect("non-empty route"));
1342        SingleOrderQuote::new(quote, 5)
1343    }
1344
1345    // Helper to create a mock worker pool that responds with a given solution
1346    fn create_mock_worker_pool(
1347        name: &str,
1348        response: Result<SingleOrderQuote, SolveError>,
1349        delay_ms: u64,
1350    ) -> (SolverPoolHandle, tokio::task::JoinHandle<()>) {
1351        let (pool, worker, _) = create_counting_mock_worker_pool(name, response, delay_ms);
1352        (pool, worker)
1353    }
1354
1355    /// Mock worker pool that also counts the tasks it received, so a test can assert a worker pool
1356    /// was never dispatched to rather than only that its quote did not win.
1357    fn create_counting_mock_worker_pool(
1358        name: &str,
1359        response: Result<SingleOrderQuote, SolveError>,
1360        delay_ms: u64,
1361    ) -> (SolverPoolHandle, tokio::task::JoinHandle<()>, Arc<AtomicUsize>) {
1362        let (tx, rx) = async_channel::bounded::<SolveTask>(10);
1363        let handle = TaskQueueHandle::from_sender(tx);
1364        let received = Arc::new(AtomicUsize::new(0));
1365
1366        let worker = {
1367            let received = Arc::clone(&received);
1368            tokio::spawn(async move {
1369                while let Ok(task) = rx.recv().await {
1370                    received.fetch_add(1, Ordering::SeqCst);
1371                    if delay_ms > 0 {
1372                        tokio::time::sleep(Duration::from_millis(delay_ms)).await;
1373                    }
1374                    task.respond(response.clone());
1375                }
1376            })
1377        };
1378
1379        (SolverPoolHandle::new(name, handle), worker, received)
1380    }
1381
1382    #[test]
1383    fn test_config_default() {
1384        let config = WorkerPoolRouterConfig::default();
1385        assert_eq!(config.default_timeout(), Duration::from_secs(1));
1386        assert_eq!(config.min_responses(), 1);
1387    }
1388
1389    #[test]
1390    fn test_config_builder() {
1391        let config = WorkerPoolRouterConfig::default()
1392            .with_timeout(Duration::from_millis(500))
1393            .with_min_responses(2);
1394        assert_eq!(config.default_timeout(), Duration::from_millis(500));
1395        assert_eq!(config.min_responses(), 2);
1396    }
1397
1398    #[tokio::test]
1399    async fn test_router_no_pools() {
1400        let worker_router =
1401            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1402        let request = QuoteRequest::new(vec![make_order()], QuoteOptions::default());
1403
1404        let result = worker_router
1405            .quote(request, ExclusiveAccess::Denied)
1406            .await;
1407        assert!(matches!(result, Err(SolveError::Internal(_))));
1408    }
1409
1410    #[tokio::test]
1411    async fn test_router_single_pool_success() {
1412        let (pool, worker) = create_mock_worker_pool("pool_a", Ok(make_single_quote(900)), 0);
1413
1414        let worker_router =
1415            WorkerPoolRouter::new(vec![pool], WorkerPoolRouterConfig::default(), default_encoder());
1416        let options = QuoteOptions::default().with_encoding_options(EncodingOptions::new(0.01));
1417        let request = QuoteRequest::new(vec![make_order()], options);
1418
1419        let result = worker_router
1420            .quote(request, ExclusiveAccess::Denied)
1421            .await;
1422        assert!(result.is_ok());
1423
1424        let quote = result.unwrap();
1425        assert_eq!(quote.orders().len(), 1);
1426        assert_eq!(quote.orders()[0].status(), QuoteStatus::Success);
1427        // amount_out_net_gas is refined using estimate_gas_usage before ranking
1428        assert_eq!(*quote.orders()[0].amount_out_net_gas(), BigUint::from(873u64));
1429        assert!(!quote.orders()[0]
1430            .transaction()
1431            .unwrap()
1432            .data()
1433            .is_empty());
1434
1435        drop(worker_router);
1436        worker.abort();
1437    }
1438
1439    #[tokio::test]
1440    async fn test_router_selects_best_of_two() {
1441        // Pool A: worse quote (net gas = 800)
1442        let (pool_a, worker_a) = create_mock_worker_pool("pool_a", Ok(make_single_quote(800)), 0);
1443        // Pool B: better quote (net gas = 950)
1444        let (pool_b, worker_b) = create_mock_worker_pool("pool_b", Ok(make_single_quote(950)), 0);
1445
1446        // Wait for both responses to test best selection logic
1447        let config = WorkerPoolRouterConfig::default().with_min_responses(2);
1448        let worker_router = WorkerPoolRouter::new(vec![pool_a, pool_b], config, default_encoder());
1449        let options = QuoteOptions::default().with_encoding_options(EncodingOptions::new(0.01));
1450        let request = QuoteRequest::new(vec![make_order()], options);
1451
1452        let result = worker_router
1453            .quote(request, ExclusiveAccess::Denied)
1454            .await;
1455        assert!(result.is_ok());
1456
1457        let quote = result.unwrap();
1458        assert_eq!(quote.orders().len(), 1);
1459        // Pool B wins (higher refined amount_out_net_gas after estimate_gas_usage)
1460        assert_eq!(*quote.orders()[0].amount_out_net_gas(), BigUint::from(938u64));
1461        assert!(!quote.orders()[0]
1462            .transaction()
1463            .unwrap()
1464            .data()
1465            .is_empty());
1466
1467        drop(worker_router);
1468        worker_a.abort();
1469        worker_b.abort();
1470    }
1471
1472    #[tokio::test]
1473    async fn test_router_timeout() {
1474        // Pool that takes too long
1475        let (pool, worker) = create_mock_worker_pool("slow_pool", Ok(make_single_quote(900)), 500);
1476
1477        let config = WorkerPoolRouterConfig::default().with_timeout(Duration::from_millis(50));
1478        let worker_router = WorkerPoolRouter::new(vec![pool], config, default_encoder());
1479        let request = QuoteRequest::new(vec![make_order()], QuoteOptions::default());
1480
1481        let result = worker_router
1482            .quote(request, ExclusiveAccess::Denied)
1483            .await;
1484        assert!(result.is_ok());
1485
1486        let quote = result.unwrap();
1487        // Should timeout and return NoRouteFound or Timeout status
1488        assert_eq!(quote.orders().len(), 1);
1489        assert!(matches!(
1490            quote.orders()[0].status(),
1491            QuoteStatus::Timeout | QuoteStatus::NoRouteFound
1492        ));
1493
1494        drop(worker_router);
1495        worker.abort();
1496    }
1497
1498    #[tokio::test]
1499    async fn test_router_early_return_on_min_responses() {
1500        // Pool A: fast
1501        let (pool_a, worker_a) =
1502            create_mock_worker_pool("fast_pool", Ok(make_single_quote(800)), 0);
1503        // Pool B: slow (but we won't wait for it)
1504        let (pool_b, worker_b) =
1505            create_mock_worker_pool("slow_pool", Ok(make_single_quote(950)), 500);
1506
1507        let config = WorkerPoolRouterConfig::default()
1508            .with_timeout(Duration::from_millis(1000))
1509            .with_min_responses(1);
1510        let worker_router = WorkerPoolRouter::new(vec![pool_a, pool_b], config, default_encoder());
1511
1512        let start = Instant::now();
1513        let options = QuoteOptions::default().with_encoding_options(EncodingOptions::new(0.01));
1514        let request = QuoteRequest::new(vec![make_order()], options);
1515
1516        let result = worker_router
1517            .quote(request, ExclusiveAccess::Denied)
1518            .await;
1519        let elapsed = start.elapsed();
1520
1521        assert!(result.is_ok());
1522        // Should return quickly (not waiting for pool_b)
1523        assert!(elapsed < Duration::from_millis(200));
1524
1525        // Should have pool_a's quote
1526        let quote = result.unwrap();
1527        assert_eq!(quote.orders().len(), 1);
1528        assert_eq!(quote.orders()[0].status(), QuoteStatus::Success);
1529        // Should have encoding
1530        assert!(!quote.orders()[0]
1531            .transaction()
1532            .unwrap()
1533            .data()
1534            .is_empty());
1535
1536        drop(worker_router);
1537        worker_a.abort();
1538        worker_b.abort();
1539    }
1540
1541    /// The `denied_access` case sets `min_responses(2)`, which would hold a request with access
1542    /// until both pools answer; a denied request's allocation holds only the public pool, so it
1543    /// must return without waiting on the slow exclusive pool it will never use.
1544    #[rstest]
1545    #[case::pending_exclusive_pool(ExclusiveAccess::Granted, 0, Some(300), 1, true)]
1546    #[case::pending_public_pool(ExclusiveAccess::Granted, 300, Some(0), 1, true)]
1547    #[case::failed_exclusive_pool(ExclusiveAccess::Granted, 0, None, 1, false)]
1548    #[case::denied_access(ExclusiveAccess::Denied, 0, Some(500), 2, false)]
1549    #[tokio::test]
1550    async fn test_router_early_return_scope_gating(
1551        #[case] access: ExclusiveAccess,
1552        #[case] public_delay_ms: u64,
1553        #[case] exclusive_delay_ms: Option<u64>,
1554        #[case] min_responses: usize,
1555        #[case] expect_surplus: bool,
1556    ) {
1557        let (public_pool, public_worker) =
1558            create_mock_worker_pool("public_pool", Ok(make_single_quote(800)), public_delay_ms);
1559        let exclusive_response = match exclusive_delay_ms {
1560            Some(_) => Ok(make_exclusive_quote(1100)),
1561            None => {
1562                Err(SolveError::NoRouteFound { order_id: "test-order".to_string(), reason: None })
1563            }
1564        };
1565        let public_pool = public_pool.with_liquidity_scope(LiquidityScope::PublicOnly);
1566        let (exclusive_pool, exclusive_worker) = create_mock_worker_pool(
1567            "exclusive_pool",
1568            exclusive_response,
1569            exclusive_delay_ms.unwrap_or(0),
1570        );
1571        let exclusive_pool = exclusive_pool.with_liquidity_scope(LiquidityScope::IncludeExclusive);
1572
1573        let config = WorkerPoolRouterConfig::default()
1574            .with_timeout(Duration::from_millis(2000))
1575            .with_min_responses(min_responses);
1576        let worker_router =
1577            WorkerPoolRouter::new(vec![public_pool, exclusive_pool], config, default_encoder());
1578        let request = QuoteRequest::new(vec![make_order()], QuoteOptions::default());
1579
1580        let start = Instant::now();
1581        let result = worker_router
1582            .quote(request, access)
1583            .await
1584            .expect("quote should succeed");
1585        let elapsed = start.elapsed();
1586
1587        // Well under the 2s timeout: the gate releases as soon as every allocated scope has
1588        // responded.
1589        assert!(elapsed < Duration::from_millis(500), "took {elapsed:?}");
1590        let order = &result.orders()[0];
1591        assert_eq!(order.status(), QuoteStatus::Success);
1592        assert_eq!(*order.amount_out(), BigUint::from(990u64));
1593        assert_eq!(order.surplus_amount().is_some(), expect_surplus);
1594
1595        drop(worker_router);
1596        public_worker.abort();
1597        exclusive_worker.abort();
1598    }
1599
1600    /// The exclusive pool offers a strictly better route (net 1100 vs 800), so it wins whenever it
1601    /// is allocated. A request without access must never reach it: not merely lose to the public
1602    /// leg in ranking, but cost the exclusive pool no work at all.
1603    #[rstest]
1604    #[case::denied(ExclusiveAccess::Denied, false)]
1605    #[case::granted(ExclusiveAccess::Granted, true)]
1606    #[tokio::test]
1607    async fn test_router_exclusive_access_allocation(
1608        #[case] access: ExclusiveAccess,
1609        #[case] expect_exclusive_leg: bool,
1610    ) {
1611        let (public_pool, public_worker) =
1612            create_mock_worker_pool("public_pool", Ok(make_single_quote(800)), 0);
1613        let (exclusive_pool, exclusive_worker, exclusive_tasks) =
1614            create_counting_mock_worker_pool("exclusive_pool", Ok(make_exclusive_quote(1100)), 0);
1615        let exclusive_pool = exclusive_pool.with_liquidity_scope(LiquidityScope::IncludeExclusive);
1616
1617        let worker_router = WorkerPoolRouter::new(
1618            vec![public_pool, exclusive_pool],
1619            WorkerPoolRouterConfig::default().with_timeout(Duration::from_millis(2000)),
1620            default_encoder(),
1621        );
1622        let request = QuoteRequest::new(vec![make_order()], QuoteOptions::default());
1623
1624        let result = worker_router
1625            .quote(request, access)
1626            .await
1627            .expect("quote should succeed");
1628
1629        // The gate is the dispatch, not the ranking: a denied request costs the exclusive pool no
1630        // CPU and no latency.
1631        assert_eq!(
1632            exclusive_tasks.load(Ordering::SeqCst) > 0,
1633            expect_exclusive_leg,
1634            "exclusive pool dispatch"
1635        );
1636
1637        let order = &result.orders()[0];
1638        assert_eq!(order.status(), QuoteStatus::Success);
1639
1640        let routes_through_exclusive = order
1641            .route()
1642            .expect("successful quote has a route")
1643            .swaps()
1644            .iter()
1645            .any(|swap| is_exclusive(swap.protocol_component()));
1646        assert_eq!(routes_through_exclusive, expect_exclusive_leg);
1647        assert_eq!(order.surplus_amount().is_some(), expect_exclusive_leg);
1648
1649        // Either way the quoted output is the public reference, so denied access costs the
1650        // caller nothing they were promised.
1651        assert_eq!(*order.amount_out(), BigUint::from(990u64));
1652
1653        drop(worker_router);
1654        public_worker.abort();
1655        exclusive_worker.abort();
1656    }
1657
1658    #[rstest]
1659    #[case::under_limit(100, Some(200), true)]
1660    #[case::at_limit(200, Some(200), true)]
1661    #[case::over_limit(300, Some(200), false)]
1662    #[case::no_limit(500, None, true)]
1663    fn test_max_gas_constraint(
1664        #[case] gas_estimate: u64,
1665        #[case] max_gas: Option<u64>,
1666        #[case] should_pass: bool,
1667    ) {
1668        let responses = OrderResponses {
1669            order_id: "test".to_string(),
1670            quotes: vec![worker_quote((
1671                "pool".to_string(),
1672                OrderQuote::new(
1673                    "test".to_string(),
1674                    QuoteStatus::Success,
1675                    BigUint::from(1000u64),
1676                    BigUint::from(990u64),
1677                    BigUint::from(gas_estimate),
1678                    BigUint::from(900u64),
1679                    BlockInfo::new(1, "0x123".to_string(), 1000),
1680                    "test".to_string(),
1681                    Bytes::from(make_address(0xAA).as_ref()),
1682                    Bytes::from(make_address(0xAA).as_ref()),
1683                    "1".to_string(),
1684                ),
1685            ))],
1686            failed_solvers: vec![],
1687        };
1688
1689        let options = match max_gas {
1690            Some(gas) => QuoteOptions::default().with_max_gas(BigUint::from(gas)),
1691            None => QuoteOptions::default(),
1692        };
1693
1694        let worker_router =
1695            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1696        let result = worker_router.rank_quotes(&responses, &options);
1697
1698        if should_pass {
1699            assert_eq!(result[0].status(), QuoteStatus::Success);
1700        } else {
1701            assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1702        }
1703    }
1704
1705    #[tokio::test]
1706    async fn test_router_captures_solver_errors() {
1707        // Pool that returns an error
1708        let (pool, worker) =
1709            create_mock_worker_pool("error_pool", Err(SolveError::no_route_found("test-order")), 0);
1710
1711        let worker_router =
1712            WorkerPoolRouter::new(vec![pool], WorkerPoolRouterConfig::default(), default_encoder());
1713        let request = QuoteRequest::new(vec![make_order()], QuoteOptions::default());
1714
1715        let result = worker_router
1716            .quote(request, ExclusiveAccess::Denied)
1717            .await;
1718        assert!(result.is_ok());
1719
1720        let quote = result.unwrap();
1721        assert_eq!(quote.orders().len(), 1);
1722        // Should be NoRouteFound since the only solver returned an error
1723        assert_eq!(quote.orders()[0].status(), QuoteStatus::NoRouteFound);
1724
1725        drop(worker_router);
1726        worker.abort();
1727    }
1728
1729    #[test]
1730    fn test_rank_quotes_all_timeouts_returns_timeout_status() {
1731        let responses = OrderResponses {
1732            order_id: "test".to_string(),
1733            quotes: vec![],
1734            failed_solvers: vec![
1735                ("pool_a".to_string(), SolveError::Timeout { elapsed_ms: 100 }),
1736                ("pool_b".to_string(), SolveError::Timeout { elapsed_ms: 100 }),
1737            ],
1738        };
1739
1740        let worker_router =
1741            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1742        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1743
1744        assert_eq!(result.len(), 1);
1745        assert_eq!(result[0].status(), QuoteStatus::Timeout);
1746    }
1747
1748    #[test]
1749    fn test_rank_quotes_mixed_failures_returns_no_route_found() {
1750        let responses = OrderResponses {
1751            order_id: "test".to_string(),
1752            quotes: vec![],
1753            failed_solvers: vec![
1754                ("pool_a".to_string(), SolveError::Timeout { elapsed_ms: 100 }),
1755                ("pool_b".to_string(), SolveError::no_route_found("test")),
1756            ],
1757        };
1758
1759        let worker_router =
1760            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1761        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1762
1763        assert_eq!(result.len(), 1);
1764        assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1765    }
1766
1767    #[test]
1768    fn test_rank_quotes_no_failures_returns_no_route_found() {
1769        let responses =
1770            OrderResponses { order_id: "test".to_string(), quotes: vec![], failed_solvers: vec![] };
1771
1772        let worker_router =
1773            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1774        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1775
1776        assert_eq!(result.len(), 1);
1777        assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1778    }
1779
1780    #[test]
1781    fn rank_quotes_attaches_no_route_reason() {
1782        use crate::algorithm::NoPathReason;
1783        let responses = OrderResponses {
1784            order_id: "test".to_string(),
1785            quotes: vec![],
1786            failed_solvers: vec![
1787                (
1788                    "pool_a".to_string(),
1789                    SolveError::no_route_found_with_reason("test", NoPathReason::NoGraphPath),
1790                ),
1791                (
1792                    "pool_b".to_string(),
1793                    SolveError::no_route_found_with_reason(
1794                        "test",
1795                        NoPathReason::DestinationTokenNotInGraph,
1796                    ),
1797                ),
1798            ],
1799        };
1800        let worker_router =
1801            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1802        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1803        assert_eq!(result.len(), 1);
1804        assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1805        // Token-not-in-graph wins over no_graph_path regardless of pool order.
1806        assert_eq!(result[0].no_route_reason(), Some(NoPathReason::DestinationTokenNotInGraph));
1807    }
1808
1809    #[test]
1810    fn test_rank_quotes_no_route_reason_single_failure() {
1811        use crate::algorithm::NoPathReason;
1812        let responses = OrderResponses {
1813            order_id: "test".to_string(),
1814            quotes: vec![],
1815            failed_solvers: vec![(
1816                "pool_a".to_string(),
1817                SolveError::no_route_found_with_reason("test", NoPathReason::NoGraphPath),
1818            )],
1819        };
1820        let worker_router =
1821            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1822        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1823        assert_eq!(result[0].no_route_reason(), Some(NoPathReason::NoGraphPath));
1824    }
1825
1826    #[test]
1827    fn test_aggregate_cause_surfaces_amount_too_small() {
1828        use crate::algorithm::NoPathReason;
1829        let failed = vec![(
1830            "bf".to_string(),
1831            SolveError::no_route_found_with_reason("o1", NoPathReason::AmountTooSmall),
1832        )];
1833        assert!(matches!(
1834            aggregate_no_route_cause(&failed),
1835            Some(SolveError::NoRouteFound { reason: Some(NoPathReason::AmountTooSmall), .. })
1836        ));
1837    }
1838
1839    #[test]
1840    fn test_aggregate_no_route_cause_independent_of_pool_order() {
1841        use crate::algorithm::NoPathReason;
1842        let dust = || SolveError::no_route_found_with_reason("o1", NoPathReason::AmountTooSmall);
1843        let no_path = || SolveError::no_route_found_with_reason("o1", NoPathReason::NoGraphPath);
1844        let forward = vec![("bf1".to_string(), no_path()), ("bf3".to_string(), dust())];
1845        let reversed = vec![("bf3".to_string(), dust()), ("bf1".to_string(), no_path())];
1846        for failed in [forward, reversed] {
1847            assert!(matches!(
1848                aggregate_no_route_cause(&failed),
1849                Some(SolveError::NoRouteFound { reason: Some(NoPathReason::AmountTooSmall), .. })
1850            ));
1851        }
1852    }
1853
1854    #[test]
1855    fn test_aggregate_token_not_in_graph_wins_over_amount_too_small() {
1856        use crate::algorithm::NoPathReason;
1857        let failed = vec![
1858            (
1859                "bf".to_string(),
1860                SolveError::no_route_found_with_reason("o1", NoPathReason::AmountTooSmall),
1861            ),
1862            (
1863                "ml".to_string(),
1864                SolveError::no_route_found_with_reason(
1865                    "o2",
1866                    NoPathReason::DestinationTokenNotInGraph,
1867                ),
1868            ),
1869        ];
1870        assert!(matches!(
1871            aggregate_no_route_cause(&failed),
1872            Some(SolveError::NoRouteFound {
1873                reason: Some(NoPathReason::DestinationTokenNotInGraph),
1874                ..
1875            })
1876        ));
1877    }
1878
1879    #[test]
1880    fn test_aggregate_prefers_liquidity_over_infra() {
1881        let failed = vec![
1882            ("a".to_string(), SolveError::QueueFull),
1883            ("b".to_string(), SolveError::insufficient_liquidity(1u32.into(), 0u32.into())),
1884        ];
1885        assert!(matches!(
1886            aggregate_no_route_cause(&failed),
1887            Some(SolveError::InsufficientLiquidity { .. })
1888        ));
1889    }
1890
1891    #[test]
1892    fn test_rank_quotes_max_gas_filtered_sets_max_gas_exceeded() {
1893        let responses = OrderResponses {
1894            order_id: "o1".to_string(),
1895            quotes: vec![worker_quote((
1896                "pool".to_string(),
1897                OrderQuote::new(
1898                    "o1".to_string(),
1899                    QuoteStatus::Success,
1900                    BigUint::from(1_000u64),
1901                    BigUint::from(990u64),
1902                    BigUint::from(100_000u64),
1903                    BigUint::from(990u64),
1904                    BlockInfo::new(1, "0xabc".to_string(), 0),
1905                    "test".to_string(),
1906                    Bytes::default(),
1907                    Bytes::default(),
1908                    "1".to_string(),
1909                ),
1910            ))],
1911            failed_solvers: vec![],
1912        };
1913        let options = QuoteOptions::default().with_max_gas(BigUint::from(1u64));
1914        let worker_router =
1915            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1916        let result = worker_router.rank_quotes(&responses, &options);
1917        assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1918        assert!(matches!(result[0].no_route_cause(), Some(SolveError::MaxGasExceeded)));
1919    }
1920
1921    #[test]
1922    fn test_rank_quotes_no_cause_when_gas_within_max() {
1923        let responses = OrderResponses {
1924            order_id: "o1".to_string(),
1925            quotes: vec![worker_quote((
1926                "pool".to_string(),
1927                OrderQuote::new(
1928                    "o1".to_string(),
1929                    QuoteStatus::NoRouteFound,
1930                    BigUint::from(1_000u64),
1931                    BigUint::from(990u64),
1932                    BigUint::from(100u64),
1933                    BigUint::from(990u64),
1934                    BlockInfo::new(1, "0xabc".to_string(), 0),
1935                    "test".to_string(),
1936                    Bytes::default(),
1937                    Bytes::default(),
1938                    "1".to_string(),
1939                ),
1940            ))],
1941            failed_solvers: vec![],
1942        };
1943        let options = QuoteOptions::default().with_max_gas(BigUint::from(1_000u64));
1944        let worker_router =
1945            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1946        let result = worker_router.rank_quotes(&responses, &options);
1947        assert_eq!(result[0].status(), QuoteStatus::NoRouteFound);
1948        assert!(
1949            result[0].no_route_cause().is_none(),
1950            "gas within max_gas must not be labelled MaxGasExceeded: {:?}",
1951            result[0].no_route_cause()
1952        );
1953    }
1954
1955    #[test]
1956    fn test_all_timeout_fallback_carries_timeout_cause() {
1957        let responses = OrderResponses {
1958            order_id: "o1".to_string(),
1959            quotes: vec![],
1960            failed_solvers: vec![("pool".to_string(), SolveError::Timeout { elapsed_ms: 9 })],
1961        };
1962        let worker_router =
1963            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
1964        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
1965        assert_eq!(result[0].status(), QuoteStatus::Timeout);
1966        assert!(matches!(result[0].no_route_cause(), Some(SolveError::Timeout { .. })));
1967    }
1968
1969    #[test]
1970    fn test_rank_quotes_returns_sorted_candidates() {
1971        let responses = OrderResponses {
1972            order_id: "test".to_string(),
1973            quotes: vec![
1974                worker_quote((
1975                    "pool_a".to_string(),
1976                    OrderQuote::new(
1977                        "test".to_string(),
1978                        QuoteStatus::Success,
1979                        BigUint::from(1000u64),
1980                        BigUint::from(800u64),
1981                        BigUint::from(100_000u64),
1982                        BigUint::from(800u64),
1983                        BlockInfo::new(1, "0x123".to_string(), 1000),
1984                        "test".to_string(),
1985                        Bytes::from(make_address(0xAA).as_ref()),
1986                        Bytes::from(make_address(0xAA).as_ref()),
1987                        "1".to_string(),
1988                    ),
1989                )),
1990                worker_quote((
1991                    "pool_b".to_string(),
1992                    OrderQuote::new(
1993                        "test".to_string(),
1994                        QuoteStatus::Success,
1995                        BigUint::from(1000u64),
1996                        BigUint::from(950u64),
1997                        BigUint::from(100_000u64),
1998                        BigUint::from(950u64),
1999                        BlockInfo::new(1, "0x123".to_string(), 1000),
2000                        "test".to_string(),
2001                        Bytes::from(make_address(0xAA).as_ref()),
2002                        Bytes::from(make_address(0xAA).as_ref()),
2003                        "1".to_string(),
2004                    ),
2005                )),
2006            ],
2007            failed_solvers: vec![],
2008        };
2009
2010        let worker_router =
2011            WorkerPoolRouter::new(vec![], WorkerPoolRouterConfig::default(), default_encoder());
2012        let result = worker_router.rank_quotes(&responses, &QuoteOptions::default());
2013
2014        assert_eq!(result.len(), 2);
2015        assert_eq!(*result[0].amount_out_net_gas(), BigUint::from(950u64));
2016        assert_eq!(*result[1].amount_out_net_gas(), BigUint::from(800u64));
2017    }
2018
2019    /// Like `make_single_quote` but the swap uses an exclusive protocol component. The swap's own
2020    /// output is `leg_amount_out`, letting tests exercise per-leg attribution where the leg
2021    /// differs from the route total
2022    fn make_exclusive_quote_with_leg(
2023        amount_out: u64,
2024        amount_out_net_gas: u64,
2025        leg_amount_out: u64,
2026    ) -> SingleOrderQuote {
2027        let make_token = |addr: Address| Token {
2028            address: addr,
2029            symbol: "T".to_string(),
2030            decimals: 18,
2031            tax: Default::default(),
2032            gas: vec![],
2033            chain: SimChain::Ethereum,
2034            quality: 100,
2035        };
2036        let tin = make_address(0x01);
2037        let tout = make_address(0x02);
2038        let tin_token = make_token(tin.clone());
2039        let tout_token = make_token(tout.clone());
2040        let mut comp = component(
2041            "0x0000000000000000000000000000000000000002",
2042            &[tin_token.clone(), tout_token.clone()],
2043        );
2044        mark_exclusive(&mut comp);
2045        let swap = Swap::new(
2046            "pool-perm".to_string(),
2047            "vm:exclusive".to_string(),
2048            tin.clone(),
2049            tout.clone(),
2050            BigUint::from(1000u64),
2051            BigUint::from(leg_amount_out),
2052            BigUint::from(50_000u64),
2053            comp,
2054            Box::new(MockProtocolSim::default()),
2055        );
2056        let mut tokens = FxHashMap::default();
2057        tokens.insert(tin, tin_token);
2058        tokens.insert(tout, tout_token);
2059        let quote = OrderQuote::new(
2060            "test-order".to_string(),
2061            QuoteStatus::Success,
2062            BigUint::from(1000u64),
2063            BigUint::from(amount_out),
2064            BigUint::from(100_000u64),
2065            BigUint::from(amount_out_net_gas),
2066            BlockInfo::new(1, "0x123".to_string(), 1000),
2067            "test".to_string(),
2068            Bytes::from(make_address(0xAA).as_ref()),
2069            Bytes::from(make_address(0xAA).as_ref()),
2070            "1".to_string(),
2071        )
2072        .with_route(Route::new(vec![swap], tokens).expect("non-empty route"));
2073        SingleOrderQuote::new(quote, 5)
2074    }
2075
2076    fn make_exclusive_quote(amount_out: u64) -> SingleOrderQuote {
2077        make_exclusive_quote_with_leg(amount_out, amount_out, amount_out)
2078    }
2079
2080    /// Split route with two parallel branches (same token pair): a public component producing
2081    /// `public_leg_out` and an exclusive component producing `exclusive_leg_out`. Route output is
2082    /// the sum of the branches; zero gas cost.
2083    fn make_exclusive_split_quote(public_leg_out: u64, exclusive_leg_out: u64) -> SingleOrderQuote {
2084        let make_token = |addr: Address| Token {
2085            address: addr,
2086            symbol: "T".to_string(),
2087            decimals: 18,
2088            tax: Default::default(),
2089            gas: vec![],
2090            chain: SimChain::Ethereum,
2091            quality: 100,
2092        };
2093        let tin = make_address(0x01);
2094        let tout = make_address(0x02);
2095        let tin_token = make_token(tin.clone());
2096        let tout_token = make_token(tout.clone());
2097        let public_swap = Swap::new(
2098            "pool-pub".to_string(),
2099            "uniswap_v2".to_string(),
2100            tin.clone(),
2101            tout.clone(),
2102            BigUint::from(500u64),
2103            BigUint::from(public_leg_out),
2104            BigUint::from(50_000u64),
2105            component(
2106                "0x0000000000000000000000000000000000000001",
2107                &[tin_token.clone(), tout_token.clone()],
2108            ),
2109            Box::new(MockProtocolSim::default()),
2110        );
2111        let mut exclusive_comp = component(
2112            "0x0000000000000000000000000000000000000002",
2113            &[tin_token.clone(), tout_token.clone()],
2114        );
2115        mark_exclusive(&mut exclusive_comp);
2116        let exclusive_swap = Swap::new(
2117            "pool-perm".to_string(),
2118            "vm:exclusive".to_string(),
2119            tin.clone(),
2120            tout.clone(),
2121            BigUint::from(500u64),
2122            BigUint::from(exclusive_leg_out),
2123            BigUint::from(50_000u64),
2124            exclusive_comp,
2125            Box::new(MockProtocolSim::default()),
2126        );
2127        let mut tokens = FxHashMap::default();
2128        tokens.insert(tin, tin_token);
2129        tokens.insert(tout, tout_token);
2130        let total = public_leg_out + exclusive_leg_out;
2131        let quote = OrderQuote::new(
2132            "test-order".to_string(),
2133            QuoteStatus::Success,
2134            BigUint::from(1000u64),
2135            BigUint::from(total),
2136            BigUint::from(100_000u64),
2137            BigUint::from(total),
2138            BlockInfo::new(1, "0x123".to_string(), 1000),
2139            "test".to_string(),
2140            Bytes::from(make_address(0xAA).as_ref()),
2141            Bytes::from(make_address(0xAA).as_ref()),
2142            "1".to_string(),
2143        )
2144        .with_route(
2145            Route::new(vec![public_swap, exclusive_swap], tokens).expect("non-empty route"),
2146        );
2147        SingleOrderQuote::new(quote, 5)
2148    }
2149
2150    /// Like `make_single_quote` but with a configurable `amount_out_net_gas` so tests can
2151    /// express non-zero gas cost.
2152    fn make_public_quote_with_net(amount_out: u64, amount_out_net_gas: u64) -> SingleOrderQuote {
2153        let make_token = |addr: Address| Token {
2154            address: addr,
2155            symbol: "T".to_string(),
2156            decimals: 18,
2157            tax: Default::default(),
2158            gas: vec![],
2159            chain: SimChain::Ethereum,
2160            quality: 100,
2161        };
2162        let tin = make_address(0x01);
2163        let tout = make_address(0x02);
2164        let tin_token = make_token(tin.clone());
2165        let tout_token = make_token(tout.clone());
2166        let swap = Swap::new(
2167            "pool-1".to_string(),
2168            "uniswap_v2".to_string(),
2169            tin.clone(),
2170            tout.clone(),
2171            BigUint::from(1000u64),
2172            BigUint::from(amount_out),
2173            BigUint::from(50_000u64),
2174            component(
2175                "0x0000000000000000000000000000000000000001",
2176                &[tin_token.clone(), tout_token.clone()],
2177            ),
2178            Box::new(MockProtocolSim::default()),
2179        );
2180        let mut tokens = FxHashMap::default();
2181        tokens.insert(tin, tin_token);
2182        tokens.insert(tout, tout_token);
2183        let quote = OrderQuote::new(
2184            "test-order".to_string(),
2185            QuoteStatus::Success,
2186            BigUint::from(1000u64),
2187            BigUint::from(amount_out),
2188            BigUint::from(100_000u64),
2189            BigUint::from(amount_out_net_gas),
2190            BlockInfo::new(1, "0x123".to_string(), 1000),
2191            "test".to_string(),
2192            Bytes::from(make_address(0xAA).as_ref()),
2193            Bytes::from(make_address(0xAA).as_ref()),
2194            "1".to_string(),
2195        )
2196        .with_route(Route::new(vec![swap], tokens).expect("non-empty route"));
2197        SingleOrderQuote::new(quote, 5)
2198    }
2199
2200    fn make_public_quote_zero_gas(amount_out: u64) -> SingleOrderQuote {
2201        make_public_quote_with_net(amount_out, amount_out)
2202    }
2203
2204    /// Builds an `OrderResponses` with a public quote and an exclusive-access quote (zero gas
2205    /// cost).
2206    fn exclusive_access_responses(public_out: u64, exclusive_out: u64) -> OrderResponses {
2207        let public = make_public_quote_zero_gas(public_out)
2208            .order()
2209            .clone();
2210        let exclusive_access = make_exclusive_quote(exclusive_out)
2211            .order()
2212            .clone();
2213        OrderResponses {
2214            order_id: "test-order".to_string(),
2215            quotes: vec![
2216                worker_quote(("public_pool".to_string(), public)),
2217                worker_quote(("exclusive_access_pool".to_string(), exclusive_access)),
2218            ],
2219            failed_solvers: vec![],
2220        }
2221    }
2222
2223    fn exclusive_access_pool_scopes() -> FxHashMap<String, LiquidityScope> {
2224        FxHashMap::from_iter([
2225            ("public_pool".to_string(), LiquidityScope::PublicOnly),
2226            ("exclusive_access_pool".to_string(), LiquidityScope::IncludeExclusive),
2227        ])
2228    }
2229
2230    /// Head selection across the gate case matrix. Each route is given as `(gross, net)`, followed
2231    /// by the user's share of the improvement in bps; expected is
2232    /// `(head amount_out, head net, captured surplus)`. A surplus win prepends the pinned quote to
2233    /// the public fallbacks; otherwise the public ranking is returned unchanged.
2234    #[rstest]
2235    // A tenth of the 50 improvement goes to the user, the rest is captured.
2236    #[case::exclusive_beats_public((900, 900), (950, 950), 1_000, (905, 905, Some(45)))]
2237    #[case::exclusive_below_public((950, 950), (900, 900), 1_000, (950, 950, None))]
2238    // A tie is quoted at the public price, with nothing to capture.
2239    #[case::exclusive_ties_public_net((900, 900), (900, 900), 1_000, (900, 900, Some(0)))]
2240    // An improvement of 1 rounds the user's tenth up to the whole unit, leaving zero surplus.
2241    #[case::exclusive_marginally_better((999, 999), (1000, 1000), 1_000, (1000, 1000, Some(0)))]
2242    #[case::exclusive_cannot_cover_public_gross((1000, 950), (990, 980), 1_000, (1000, 950, None))]
2243    // Gas-heavier exclusive route: the committed amount rises to max(1000, 951 + 140) = 1091 so
2244    // the user nets the public 950 plus a tenth of the 10 improvement; the protocol captures 9.
2245    #[case::exclusive_with_higher_gas((1000, 950), (1100, 960), 1_000, (1091, 951, Some(9)))]
2246    // Gas-cheaper exclusive route: the improvement (110) includes the gas saving, so the user
2247    // nets 950 + 11 = 961 and the protocol captures 1100 - (961 + 40) = 99.
2248    #[case::exclusive_with_lower_gas((1000, 950), (1100, 1060), 1_000, (1001, 961, Some(99)))]
2249    // A 50 improvement on a 1_000_000 trade: the user's 5 is 0.05 bps of the trade, below what a
2250    // mark-up on the trade itself could express.
2251    #[case::sub_bps_markup((1_000_000, 1_000_000), (1_000_050, 1_000_050), 1_000,
2252        (1_000_005, 1_000_005, Some(45)))]
2253    // Zero share: the user is left at the public net and the protocol captures the improvement.
2254    #[case::zero_share((900, 900), (950, 950), 0, (900, 900, Some(50)))]
2255    // Full share: the whole improvement goes to the user and there is nothing left to capture.
2256    #[case::full_share((900, 900), (950, 950), 10_000, (950, 950, Some(0)))]
2257    fn test_combine_head_selection(
2258        #[case] public: (u64, u64),
2259        #[case] exclusive: (u64, u64),
2260        #[case] user_share_bps: u32,
2261        #[case] expected: (u64, u64, Option<u64>),
2262    ) {
2263        let (public_out, public_net) = public;
2264        let (exclusive_out, exclusive_net) = exclusive;
2265        let (expected_amount_out, expected_net, expected_surplus) = expected;
2266
2267        let responses = responses_with_gas(public_out, public_net, exclusive_out, exclusive_net);
2268        let public_ranked = vec![make_public_quote_with_net(public_out, public_net)
2269            .order()
2270            .clone()];
2271        let combined = combine_with_surplus(
2272            &responses,
2273            &exclusive_access_pool_scopes(),
2274            &QuoteOptions::default(),
2275            public_ranked,
2276            user_share_bps,
2277            SimChain::Ethereum,
2278        );
2279
2280        let expected_surplus = expected_surplus.map(BigUint::from);
2281        assert_eq!(combined.len(), if expected_surplus.is_some() { 2 } else { 1 });
2282        assert_eq!(*combined[0].amount_out(), BigUint::from(expected_amount_out));
2283        assert_eq!(*combined[0].amount_out_net_gas(), BigUint::from(expected_net));
2284        assert_eq!(combined[0].surplus_amount(), expected_surplus.as_ref());
2285        if expected_surplus.is_some() {
2286            assert_eq!(
2287                combined[0].committed_amount_out(),
2288                Some(&BigUint::from(expected_amount_out))
2289            );
2290        }
2291    }
2292
2293    #[rstest]
2294    #[case::over_max_gas(
2295        make_exclusive_quote(1100).order().clone(),
2296        Some(50_000)
2297    )]
2298    #[case::mid_route_exclusive_leg(
2299        make_route_quote(&[("vm:exclusive", 0x01, 0x02), ("uniswap_v2", 0x02, 0x03)]),
2300        None
2301    )]
2302    fn test_combine_filters_exclusive_candidate(
2303        #[case] exclusive_quote: OrderQuote,
2304        #[case] max_gas: Option<u64>,
2305    ) {
2306        let mut options = QuoteOptions::default();
2307        if let Some(max) = max_gas {
2308            options = options.with_max_gas(BigUint::from(max));
2309        }
2310        let responses = OrderResponses {
2311            order_id: "test-order".to_string(),
2312            quotes: vec![
2313                worker_quote((
2314                    "public_pool".to_string(),
2315                    make_public_quote_zero_gas(900)
2316                        .order()
2317                        .clone(),
2318                )),
2319                worker_quote(("exclusive_access_pool".to_string(), exclusive_quote)),
2320            ],
2321            failed_solvers: vec![],
2322        };
2323        let public_ranked = vec![make_public_quote_zero_gas(900)
2324            .order()
2325            .clone()];
2326        let combined = combine_with_surplus(
2327            &responses,
2328            &exclusive_access_pool_scopes(),
2329            &options,
2330            public_ranked,
2331            1_000,
2332            SimChain::Ethereum,
2333        );
2334
2335        assert_eq!(combined.len(), 1);
2336        assert_eq!(*combined[0].amount_out(), BigUint::from(900u64));
2337        assert_eq!(combined[0].surplus_amount(), None);
2338    }
2339
2340    /// The `NoRouteFound` placeholder `rank_quotes` returns when no public candidate succeeded.
2341    fn no_route_quote() -> OrderQuote {
2342        OrderQuote::new(
2343            "test-order".to_string(),
2344            QuoteStatus::NoRouteFound,
2345            BigUint::from(1000u64),
2346            BigUint::ZERO,
2347            BigUint::ZERO,
2348            BigUint::ZERO,
2349            BlockInfo::new(1, "0x123".to_string(), 1000),
2350            String::new(),
2351            Bytes::from(make_address(0xAA).as_ref()),
2352            Bytes::from(make_address(0xAA).as_ref()),
2353            "1".to_string(),
2354        )
2355    }
2356
2357    /// Builds an `OrderResponses` where the public worker pool failed and only the exclusive-access
2358    /// pool returned a quote.
2359    fn no_public_route_responses(exclusive: OrderQuote) -> OrderResponses {
2360        OrderResponses {
2361            order_id: "test-order".to_string(),
2362            quotes: vec![worker_quote(("exclusive_access_pool".to_string(), exclusive))],
2363            failed_solvers: vec![(
2364                "public_pool".to_string(),
2365                SolveError::NoRouteFound { order_id: "test-order".to_string(), reason: None },
2366            )],
2367        }
2368    }
2369
2370    #[test]
2371    fn test_combine_no_public_route_applies_default_fee() {
2372        // No public reference to match: 5 bps of the exclusive leg's 1_000_000 output is
2373        // withheld, so the user is committed 999_500 and the exclusive component captures 500.
2374        let responses = no_public_route_responses(
2375            make_exclusive_quote(1_000_000)
2376                .order()
2377                .clone(),
2378        );
2379        let combined = combine_with_surplus(
2380            &responses,
2381            &exclusive_access_pool_scopes(),
2382            &QuoteOptions::default(),
2383            vec![no_route_quote()],
2384            1_000,
2385            SimChain::Ethereum,
2386        );
2387
2388        assert_eq!(combined.len(), 2);
2389        assert_eq!(combined[0].status(), QuoteStatus::Success);
2390        assert_eq!(*combined[0].amount_out(), BigUint::from(999_500u64));
2391        assert_eq!(*combined[0].amount_out_net_gas(), BigUint::from(999_500u64));
2392        assert_eq!(combined[0].surplus_amount(), Some(&BigUint::from(500u64)));
2393
2394        let exclusive_leg = combined[0]
2395            .route()
2396            .expect("surplus quote should have a route")
2397            .swaps()
2398            .iter()
2399            .find(|s| is_exclusive(s.protocol_component()))
2400            .expect("should have an exclusive swap")
2401            .committed_amount_out()
2402            .cloned();
2403        assert_eq!(exclusive_leg, Some(BigUint::from(999_500u64)));
2404
2405        // The placeholder stays on as the price-guard fallback, so a rejected exclusive quote
2406        // still answers "no route".
2407        assert_eq!(combined[1].status(), QuoteStatus::NoRouteFound);
2408    }
2409
2410    #[test]
2411    fn test_combine_no_public_route_split_route_fee_on_exclusive_leg() {
2412        // Split route with no public reference: the 5 bps fee is charged on the exclusive leg's
2413        // 500_000 output only (250), not on the route's 1_100_000. The public branch pays out in
2414        // full, so the whole fee comes out of the exclusive leg: 500_000 − 250 = 499_750.
2415        let responses = no_public_route_responses(
2416            make_exclusive_split_quote(600_000, 500_000)
2417                .order()
2418                .clone(),
2419        );
2420        let combined = combine_with_surplus(
2421            &responses,
2422            &exclusive_access_pool_scopes(),
2423            &QuoteOptions::default(),
2424            vec![no_route_quote()],
2425            1_000,
2426            SimChain::Ethereum,
2427        );
2428
2429        assert_eq!(*combined[0].amount_out(), BigUint::from(1_099_750u64));
2430        assert_eq!(combined[0].surplus_amount(), Some(&BigUint::from(250u64)));
2431
2432        let route = combined[0]
2433            .route()
2434            .expect("surplus quote should have a route");
2435        let public_leg = route
2436            .swaps()
2437            .iter()
2438            .find(|s| !is_exclusive(s.protocol_component()))
2439            .expect("should have a public swap");
2440        assert_eq!(public_leg.committed_amount_out(), None);
2441
2442        let exclusive_leg = route
2443            .swaps()
2444            .iter()
2445            .find(|s| is_exclusive(s.protocol_component()))
2446            .expect("should have an exclusive swap");
2447        assert_eq!(exclusive_leg.committed_amount_out(), Some(&BigUint::from(499_750u64)));
2448    }
2449
2450    #[test]
2451    fn test_combine_no_public_route_fee_below_gas() {
2452        // Gas leaves 3 of the 10_000 output, so the 5 fee would put the user under water: the
2453        // candidate is dropped and the order still reports no route.
2454        let responses = no_public_route_responses(
2455            make_exclusive_quote_with_leg(10_000, 3, 10_000)
2456                .order()
2457                .clone(),
2458        );
2459        let combined = combine_with_surplus(
2460            &responses,
2461            &exclusive_access_pool_scopes(),
2462            &QuoteOptions::default(),
2463            vec![no_route_quote()],
2464            1_000,
2465            SimChain::Ethereum,
2466        );
2467
2468        assert_eq!(combined.len(), 1);
2469        assert_eq!(combined[0].status(), QuoteStatus::NoRouteFound);
2470    }
2471
2472    #[test]
2473    fn test_combine_stamps_per_leg_committed_amount_out() {
2474        let responses = exclusive_access_responses(900, 1000);
2475        let public_ranked = vec![make_public_quote_zero_gas(900)
2476            .order()
2477            .clone()];
2478        let combined = combine_with_surplus(
2479            &responses,
2480            &exclusive_access_pool_scopes(),
2481            &QuoteOptions::default(),
2482            public_ranked,
2483            1_000,
2484            SimChain::Ethereum,
2485        );
2486
2487        let surplus_quote = &combined[0];
2488        let route = surplus_quote
2489            .route()
2490            .expect("surplus quote should have a route");
2491        let perm_swap = route
2492            .swaps()
2493            .iter()
2494            .find(|s| is_exclusive(s.protocol_component()))
2495            .expect("should have an exclusive swap");
2496
2497        // committed_leg = leg.amount_out * committed_route_out / realized_route_out
2498        // = 1000 * 910 / 1000 = 910
2499        assert_eq!(perm_swap.committed_amount_out(), Some(&BigUint::from(910u64)),);
2500    }
2501
2502    #[test]
2503    fn test_combine_committed_leg_deduction() {
2504        // leg = 995, committed = 910, realized = 1000: the route's surplus (90) is deducted from
2505        // the exclusive leg in full — committed_leg = 995 − 90 = 905, exactly, no rounding.
2506        let responses = OrderResponses {
2507            order_id: "test-order".to_string(),
2508            quotes: vec![
2509                worker_quote((
2510                    "public_pool".to_string(),
2511                    make_public_quote_zero_gas(900)
2512                        .order()
2513                        .clone(),
2514                )),
2515                worker_quote((
2516                    "exclusive_access_pool".to_string(),
2517                    make_exclusive_quote_with_leg(1000, 1000, 995)
2518                        .order()
2519                        .clone(),
2520                )),
2521            ],
2522            failed_solvers: vec![],
2523        };
2524        let public_ranked = vec![make_public_quote_zero_gas(900)
2525            .order()
2526            .clone()];
2527        let combined = combine_with_surplus(
2528            &responses,
2529            &exclusive_access_pool_scopes(),
2530            &QuoteOptions::default(),
2531            public_ranked,
2532            1_000,
2533            SimChain::Ethereum,
2534        );
2535
2536        let route = combined[0]
2537            .route()
2538            .expect("surplus quote should have a route");
2539        let perm_swap = route
2540            .swaps()
2541            .iter()
2542            .find(|s| is_exclusive(s.protocol_component()))
2543            .expect("should have an exclusive swap");
2544        assert_eq!(perm_swap.committed_amount_out(), Some(&BigUint::from(905u64)));
2545    }
2546
2547    #[rstest]
2548    #[case::zero("0", Some(0))]
2549    #[case::whole_improvement("10000", Some(10_000))]
2550    #[case::padded(" 2500 ", Some(2_500))]
2551    #[case::above_whole_improvement("10001", None)]
2552    #[case::negative("-1", None)]
2553    #[case::fractional("0.5", None)]
2554    #[case::empty("", None)]
2555    fn test_parse_user_improvement_share_bps(#[case] raw: &str, #[case] expected: Option<u32>) {
2556        assert_eq!(parse_user_improvement_share_bps(raw), expected);
2557    }
2558
2559    #[test]
2560    fn test_combine_split_route_attribution() {
2561        // Split route: public branch 600 + exclusive branch 500 = 1100 realized vs 1010
2562        // committed (zero gas, a tenth of the 100 improvement to the user). Only the exclusive
2563        // leg is stamped; the public branch flows to the user untouched.
2564        let responses = OrderResponses {
2565            order_id: "test-order".to_string(),
2566            quotes: vec![
2567                worker_quote((
2568                    "public_pool".to_string(),
2569                    make_public_quote_zero_gas(1000)
2570                        .order()
2571                        .clone(),
2572                )),
2573                worker_quote((
2574                    "exclusive_access_pool".to_string(),
2575                    make_exclusive_split_quote(600, 500)
2576                        .order()
2577                        .clone(),
2578                )),
2579            ],
2580            failed_solvers: vec![],
2581        };
2582        let public_ranked = vec![make_public_quote_zero_gas(1000)
2583            .order()
2584            .clone()];
2585        let combined = combine_with_surplus(
2586            &responses,
2587            &exclusive_access_pool_scopes(),
2588            &QuoteOptions::default(),
2589            public_ranked,
2590            1_000,
2591            SimChain::Ethereum,
2592        );
2593
2594        assert_eq!(*combined[0].amount_out(), BigUint::from(1010u64));
2595        let route = combined[0]
2596            .route()
2597            .expect("surplus quote should have a route");
2598        let public_leg = route
2599            .swaps()
2600            .iter()
2601            .find(|s| !is_exclusive(s.protocol_component()))
2602            .expect("should have a public swap");
2603        assert_eq!(public_leg.committed_amount_out(), None);
2604
2605        // The public branch (600) pays out in full, so the entire surplus
2606        // (1100 − 1010 = 90) is deducted from the exclusive leg: committed_leg = 500 − 90 =
2607        // 410. The user receives 600 + 410 = 1010 (exactly the committed amount) and the
2608        // exclusive component captures all 90.
2609        let exclusive_leg = route
2610            .swaps()
2611            .iter()
2612            .find(|s| is_exclusive(s.protocol_component()))
2613            .expect("should have an exclusive swap");
2614        assert_eq!(exclusive_leg.committed_amount_out(), Some(&BigUint::from(410u64)));
2615    }
2616
2617    /// Builds an `OrderResponses` where both quotes carry explicit `amount_out_net_gas`.
2618    fn responses_with_gas(
2619        public_out: u64,
2620        public_net: u64,
2621        exclusive_out: u64,
2622        exclusive_net: u64,
2623    ) -> OrderResponses {
2624        OrderResponses {
2625            order_id: "test-order".to_string(),
2626            quotes: vec![
2627                worker_quote((
2628                    "public_pool".to_string(),
2629                    make_public_quote_with_net(public_out, public_net)
2630                        .order()
2631                        .clone(),
2632                )),
2633                worker_quote((
2634                    "exclusive_access_pool".to_string(),
2635                    make_exclusive_quote_with_leg(exclusive_out, exclusive_net, exclusive_out)
2636                        .order()
2637                        .clone(),
2638                )),
2639            ],
2640            failed_solvers: vec![],
2641        }
2642    }
2643
2644    /// Builds a Success quote whose route has one swap per `(protocol_system, token_in, token_out)`
2645    /// leg, for exercising `has_valid_exclusive_route` on multi-leg and multi-path route shapes.
2646    fn make_route_quote(legs: &[(&str, u8, u8)]) -> OrderQuote {
2647        let legs: Vec<(&str, Address, Address)> = legs
2648            .iter()
2649            .map(|(protocol_system, token_in, token_out)| {
2650                (*protocol_system, make_address(*token_in), make_address(*token_out))
2651            })
2652            .collect();
2653        make_route_quote_for_tokens(&legs)
2654    }
2655
2656    fn make_route_quote_for_tokens(legs: &[(&str, Address, Address)]) -> OrderQuote {
2657        let make_token = |addr: &Address| Token {
2658            address: addr.clone(),
2659            symbol: "T".to_string(),
2660            decimals: 18,
2661            tax: Default::default(),
2662            gas: vec![],
2663            chain: SimChain::Ethereum,
2664            quality: 100,
2665        };
2666        let mut tokens = FxHashMap::default();
2667        let mut swaps = Vec::new();
2668        for (protocol_system, tin, tout) in legs {
2669            let (tin, tout) = (tin.clone(), tout.clone());
2670            let tin_token = make_token(&tin);
2671            let tout_token = make_token(&tout);
2672            let mut comp = component(
2673                "0x0000000000000000000000000000000000000002",
2674                &[tin_token.clone(), tout_token.clone()],
2675            );
2676            comp.protocol_system = protocol_system.to_string();
2677            if *protocol_system == "vm:exclusive" {
2678                mark_exclusive(&mut comp);
2679            }
2680            swaps.push(Swap::new(
2681                format!("pool-{tin}-{tout}"),
2682                protocol_system.to_string(),
2683                tin.clone(),
2684                tout.clone(),
2685                BigUint::from(1000u64),
2686                BigUint::from(1000u64),
2687                BigUint::from(50_000u64),
2688                comp,
2689                Box::new(MockProtocolSim::default()),
2690            ));
2691            tokens.insert(tin, tin_token);
2692            tokens.insert(tout, tout_token);
2693        }
2694        OrderQuote::new(
2695            "test-order".to_string(),
2696            QuoteStatus::Success,
2697            BigUint::from(1000u64),
2698            BigUint::from(1000u64),
2699            BigUint::from(100_000u64),
2700            BigUint::from(1000u64),
2701            BlockInfo::new(1, "0x123".to_string(), 1000),
2702            "test".to_string(),
2703            Bytes::from(make_address(0xAA).as_ref()),
2704            Bytes::from(make_address(0xAA).as_ref()),
2705            "1".to_string(),
2706        )
2707        .with_route(Route::new(swaps, tokens).expect("non-empty route"))
2708    }
2709
2710    /// Route-shape validation: exactly one exclusive leg, terminal in its path.
2711    #[rstest]
2712    #[case::terminal_exclusive_leg(
2713        &[("uniswap_v2", 0x01, 0x02), ("vm:exclusive", 0x02, 0x03)], true)]
2714    #[case::mid_route_exclusive_leg(
2715        &[("vm:exclusive", 0x01, 0x02), ("uniswap_v2", 0x02, 0x03)], false)]
2716    // Split route in sequential representation: path 1 is a single exclusive hop 0x01→0x02;
2717    // path 2 is 0x01→0x03→0x02. The exclusive leg is terminal for its path because the next
2718    // swap starts over from 0x01.
2719    #[case::exclusive_leg_ending_its_path(
2720        &[("vm:exclusive", 0x01, 0x02), ("uniswap_v2", 0x01, 0x03), ("uniswap_v2", 0x03, 0x02)],
2721        true)]
2722    #[case::no_exclusive_leg(
2723        &[("uniswap_v2", 0x01, 0x02), ("uniswap_v2", 0x02, 0x03)], false)]
2724    // Two exclusive legs: out of scope for v1 (ambiguous per-component attribution).
2725    #[case::two_exclusive_legs(
2726        &[("vm:exclusive", 0x01, 0x02), ("vm:exclusive", 0x01, 0x02)], false)]
2727    // Diamond split: 0x01 splits into 0x01->0x02 (exclusive) and 0x01->0x03, both merging into
2728    // 0x02->0x04 and 0x03->0x04. The exclusive leg feeds the merge point, not the route's output
2729    // (0x04), so it's mid-path even though the next serialized swap starts a sibling branch.
2730    #[case::exclusive_leg_feeding_diamond_merge(
2731        &[
2732            ("vm:exclusive", 0x01, 0x02),
2733            ("uniswap_v2", 0x01, 0x03),
2734            ("uniswap_v2", 0x02, 0x04),
2735            ("uniswap_v2", 0x03, 0x04),
2736        ],
2737        false)]
2738    // Same diamond shape, reordered so the exclusive leg's real continuation is adjacent to it —
2739    // regression case for a prior bug where terminal-ness was inferred from adjacency alone.
2740    #[case::exclusive_leg_feeding_diamond_merge_reordered(
2741        &[
2742            ("vm:exclusive", 0x01, 0x02),
2743            ("uniswap_v2", 0x02, 0x04),
2744            ("uniswap_v2", 0x01, 0x03),
2745            ("uniswap_v2", 0x03, 0x04),
2746        ],
2747        false)]
2748    // Sibling branches sharing a prefix: 0x01->0x02->0x03->0x04 alongside
2749    // 0x01->0x02->0x05->0x04(exclusive). The exclusive leg is the terminal hop of its own branch
2750    // and produces the route's output token, so it's valid despite sharing 0x01->0x02 with the
2751    // other branch.
2752    #[case::exclusive_leg_on_sibling_branch(
2753        &[
2754            ("uniswap_v2", 0x01, 0x02),
2755            ("uniswap_v2", 0x02, 0x03),
2756            ("uniswap_v2", 0x03, 0x04),
2757            ("uniswap_v2", 0x02, 0x05),
2758            ("vm:exclusive", 0x05, 0x04),
2759        ],
2760        true)]
2761    fn test_exclusive_route_validation(#[case] legs: &[(&str, u8, u8)], #[case] expected: bool) {
2762        let quote = make_route_quote(legs);
2763        assert_eq!(has_valid_exclusive_route(&quote, SimChain::Ethereum), expected);
2764    }
2765
2766    /// Native token, its wrapped form, and an unrelated token. Read from the same chain config the
2767    /// validator uses, so the pair cannot drift from tycho's.
2768    fn wrap_tokens() -> (Address, Address, Address) {
2769        (
2770            SimChain::Ethereum
2771                .native_token()
2772                .address,
2773            SimChain::Ethereum
2774                .wrapped_native_token()
2775                .address,
2776            make_address(0x07),
2777        )
2778    }
2779
2780    #[test]
2781    fn test_exclusive_leg_wrapped_into_output() {
2782        let (native, wrapped, other) = wrap_tokens();
2783        let quote = make_route_quote_for_tokens(&[
2784            ("vm:exclusive", other, native.clone()),
2785            ("uniswap_v2", native, wrapped),
2786        ]);
2787
2788        assert!(has_valid_exclusive_route(&quote, SimChain::Ethereum));
2789    }
2790
2791    #[test]
2792    fn test_exclusive_leg_unwrapped_into_output() {
2793        let (native, wrapped, other) = wrap_tokens();
2794        let quote = make_route_quote_for_tokens(&[
2795            ("vm:exclusive", other, wrapped.clone()),
2796            ("uniswap_v2", wrapped, native),
2797        ]);
2798
2799        assert!(has_valid_exclusive_route(&quote, SimChain::Ethereum));
2800    }
2801
2802    #[test]
2803    fn test_exclusive_leg_followed_by_non_wrap() {
2804        let (native, _, other) = wrap_tokens();
2805        let quote = make_route_quote_for_tokens(&[
2806            ("vm:exclusive", other, native.clone()),
2807            ("uniswap_v2", native, make_address(0x08)),
2808        ]);
2809
2810        assert!(!has_valid_exclusive_route(&quote, SimChain::Ethereum));
2811    }
2812
2813    #[test]
2814    fn test_exclusive_leg_wrap_not_producing_output() {
2815        let (native, wrapped, other) = wrap_tokens();
2816        let quote = make_route_quote_for_tokens(&[
2817            ("vm:exclusive", other, native.clone()),
2818            ("uniswap_v2", native, wrapped.clone()),
2819            ("uniswap_v2", wrapped, make_address(0x08)),
2820        ]);
2821
2822        assert!(!has_valid_exclusive_route(&quote, SimChain::Ethereum));
2823    }
2824
2825    #[test]
2826    fn test_exclusive_leg_wrap_of_another_path() {
2827        let (native, wrapped, other) = wrap_tokens();
2828        let quote = make_route_quote_for_tokens(&[
2829            ("vm:exclusive", other, make_address(0x08)),
2830            ("uniswap_v2", native, wrapped),
2831        ]);
2832
2833        assert!(!has_valid_exclusive_route(&quote, SimChain::Ethereum));
2834    }
2835}