feat(fifa17): wire async economy handlers into the host via a runtime bridge (unrouted)

Bridges the synchronous thread-per-connection host to the async
transfer-market/pile handlers WITHOUT flipping the classifier. classify()
is untouched; production still proxies every economy route to Python. The
new dispatch is exercised only by the integration harness via
Server::try_handle_economy — handler wiring, not authority cutover.

async_bridge.rs: AsyncBridge owns ONE process-lifetime multi-threaded Tokio
runtime, shared by every connection via Arc. block_on() runs a future from
the sync dispatch thread; if invoked from within an ambient runtime it
offloads onto its own runtime + a std channel instead of panicking
("cannot start a runtime from within a runtime"). 4 unit tests incl. the
nested-runtime-safety case and concurrent multi-thread drivers.

lib.rs: EconomyRoute + classify_economy (mirrors the Python route table:
credits, purchasegroup, store/transaction, purchased, item DELETE/PUT,
ut/delete match/item/trade, auctionhouse/transfermarket, tradePile, trade).
EconomyServices (Core econ transport + durable MarketStore/PileStore + the
bridge + the pack-content pool), attached via Server::with_economy (kept out
of `new`/`from_config` so existing tests build a DB-less Server; production
from_config attachment is the barrier step). Server::try_handle_economy
dispatches: sync handlers (credits/purchasegroup/store-buy/pack-open/
quick-sell/match) inline; async handlers (market list/query/buy/cancel,
move) on the bridge via owned `async move` blocks. build_content_pool
derives the resolvable FIFA∩Core candidate pool from Core content.

market.rs: FIX the load-bearing hazard the FakeEconomy tests missed — the
async market handlers call the BLOCKING reqwest Core client, which panics
(reqwest::blocking::wait::enter) when run while a Tokio runtime is entered.
off_runtime() hops each Core call to a fresh OS thread with no runtime
entered, so blocking is legal. handle_move_items resolver gains `+ Sync`
(future must be Send for the bridge).

tests/economy_integration.rs: economy_full_sequence_through_dispatch drives
the WHOLE cluster through the REAL Server dispatch + bridge against a live
in-process Core (seeded with fifa17 dev content: 100k coins + owned cards),
on a plain OS thread (direct bridge path), over the real blocking
HttpCoreClient — no fakes: Store BUY (pool draw + shape + debit 400 + mint 5),
credits, quick-sell (reverse-resolve + credit), match WIN (+400), market
list->query->buy->query(sold)->second-buy-fails(no double debit), cancel
(cancelled not buyable), move-items, then reopen the durable market/pile
stores from disk (sold + pile persist). Deterministic. start_core_seeded
loads fifa17 dev content so /collection renders real definitions.

Tests: host 68 lib (+4 bridge) + 2 economy_integration + 24 host_test, all
green; adapter unchanged-green; clippy -D warnings + fmt clean.

openfut-utas-host/src/async_bridge.rs

@@ -0,0 +1,151 @@
+//! Sync → async execution bridge for the economy handlers.
+//!
+//! The UTAS host is a **synchronous, thread-per-connection** server
+//! ([`crate::Server::serve_listener`] spawns one `std::thread` per socket and
+//! [`crate::Server::handle_with_ip`] runs to completion on that plain OS
+//! thread). Most handlers — `/club`, squad, credits, match reward, the Store
+//! writers — are fully synchronous (Core is reached over a **blocking** reqwest
+//! client). Only the durable transfer-market + pile handlers are `async`,
+//! because their store is `sqlx` with the `runtime-tokio` feature.
+//!
+//! Rather than sprinkle per-request runtimes, the host owns ONE long-lived
+//! multi-threaded Tokio runtime for the whole process lifetime and drives the
+//! few async handlers on it with [`AsyncBridge::block_on`].
+//!
+//! ## Execution contract
+//!
+//! * Exactly one runtime exists for the host lifetime; it is created once
+//!   (`Server::from_config`) and shared by every connection thread via `Arc`.
+//! * `block_on` is called from the synchronous dispatch thread, which is NOT a
+//!   Tokio worker, so the common path is a direct [`tokio::runtime::Runtime::block_on`].
+//! * **Nested-runtime safety:** should dispatch ever be invoked from inside a
+//!   Tokio context (it is not in production, but a future async caller or a test
+//!   might), calling `block_on`/`Handle::block_on` on the current thread would
+//!   panic (*"Cannot start a runtime from within a runtime"*). We detect an
+//!   ambient runtime with [`tokio::runtime::Handle::try_current`] and, in that
+//!   case, offload the future onto our own runtime and block the caller on a
+//!   std channel — no runtime is entered on the current thread, so it cannot
+//!   panic.
+//! * The blocking Core client used *inside* the async handlers is safe here: it
+//!   runs its own private runtime on a dedicated background thread, so calling
+//!   it from a runtime worker parks the worker on a channel rather than entering
+//!   a runtime. The host client is process-lifetime (`Arc`) and never dropped in
+//!   an async context.
+
+use std::future::Future;
+
+use tokio::runtime::{Builder, Handle, Runtime};
+
+/// A process-lifetime Tokio runtime plus a nested-safe blocking bridge.
+pub struct AsyncBridge {
+    rt: Runtime,
+}
+
+impl AsyncBridge {
+    /// Build the shared multi-threaded runtime. Multi-threaded (not
+    /// current-thread) so that a handler which parks a worker on the blocking
+    /// Core client still leaves other workers to drive `sqlx` IO to completion.
+    pub fn new() -> std::io::Result<Self> {
+        let rt = Builder::new_multi_thread()
+            .worker_threads(4)
+            .enable_all()
+            .thread_name("openfut-econ")
+            .build()?;
+        Ok(Self { rt })
+    }
+
+    /// A cloneable handle to the shared runtime (used by callers that want to
+    /// `spawn` onto the same executor rather than block).
+    pub fn handle(&self) -> Handle {
+        self.rt.handle().clone()
+    }
+
+    /// Drive `fut` to completion from a synchronous caller, returning its output.
+    ///
+    /// The `Send + 'static` bound lets the nested-safety fallback move the future
+    /// onto our runtime; dispatch satisfies it by building `async move` blocks
+    /// that own `Arc` clones of the shared services (never per-request state).
+    pub fn block_on<F>(&self, fut: F) -> F::Output
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        match Handle::try_current() {
+            // Common path: a plain OS dispatch thread, no ambient runtime.
+            Err(_) => self.rt.block_on(fut),
+            // Ambient runtime present: driving on the current thread would panic.
+            // Offload to our runtime and block on a std channel — the current
+            // thread never enters a runtime, so it is panic-free.
+            Ok(_) => {
+                let (tx, rx) = std::sync::mpsc::sync_channel(1);
+                self.rt.spawn(async move {
+                    let _ = tx.send(fut.await);
+                });
+                rx.recv().expect("AsyncBridge offloaded task panicked")
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn block_on_from_plain_thread_runs_future() {
+        let bridge = AsyncBridge::new().unwrap();
+        let out = bridge.block_on(async { 2 + 3 });
+        assert_eq!(out, 5);
+    }
+
+    #[test]
+    fn block_on_from_plain_thread_drives_async_io() {
+        // A future with an actual await point (timer) completes on our runtime.
+        let bridge = AsyncBridge::new().unwrap();
+        let out = bridge.block_on(async {
+            tokio::task::yield_now().await;
+            41 + 1
+        });
+        assert_eq!(out, 42);
+    }
+
+    #[test]
+    fn block_on_is_nested_runtime_safe() {
+        // Simulate the hazardous case: dispatch invoked from WITHIN a Tokio
+        // context. A naive `Runtime::block_on` here panics; the bridge must not.
+        let bridge = AsyncBridge::new().unwrap();
+        let ambient = Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()
+            .unwrap();
+        let out = ambient.block_on(async {
+            // We are now inside `ambient`. Calling the bridge must offload, not
+            // panic with "Cannot start a runtime from within a runtime".
+            bridge.block_on(async {
+                tokio::task::yield_now().await;
+                7 * 6
+            })
+        });
+        assert_eq!(out, 42);
+    }
+
+    #[test]
+    fn block_on_concurrently_from_many_threads() {
+        // The shared runtime is driven by many dispatch threads at once, exactly
+        // as thread-per-connection dispatch does.
+        let bridge = std::sync::Arc::new(AsyncBridge::new().unwrap());
+        let mut handles = Vec::new();
+        for i in 0..16i64 {
+            let b = bridge.clone();
+            handles.push(std::thread::spawn(move || {
+                b.block_on(async move {
+                    tokio::task::yield_now().await;
+                    i * 2
+                })
+            }));
+        }
+        let sum: i64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
+        assert_eq!(sum, (0..16i64).map(|i| i * 2).sum::<i64>());
+    }
+}