refactor(curl-downloader): rewrite as a single async fn, drop Job/tick

Replaces the Job-table + tick()-driven polling loop with one async
download() that sends, decides (retry/redirect/fail/succeed via a new
decide() extracted from the former run_job), and loops until it
resolves — no more resolve/reject callbacks. The client switches from
reqwest::blocking::Client to the non-blocking reqwest::Client, with
body streaming now via tokio::fs.

Because real async I/O needs a live tokio reactor and none runs yet at
the process level (sync_executor::block_on is a no-reactor busy-spin
executor that only works when awaited futures resolve synchronously),
HttpDownloader::start_job drives CurlDownloader::download() through a
dedicated temporary current_thread Runtime (curl_runtime(), marked
TODO(phase-e)) instead. This keeps concurrency characteristics
unchanged for now — start_job still resolves one job at a time — real
parallel I/O lands once HttpDownloader/Loop are rearchitected on top of
FuturesUnordered.

count_active_jobs' curl.tick() polling and the Job.settled/curl_id
plumbing are removed as dead weight now that start_job settles curl
jobs synchronously, same as the rfs path already did.

abort_request is dropped: it had no caller (the PHP Promise-cancellation
flow it backs was never ported), and the job table it operated on no
longer exists.

Verified manually against real network I/O (sandbox disabled): `shirabe
show -a` (JSON metadata, in-memory body) and `shirabe create-project`
(actual dist zip download + extraction) both complete correctly with no
hang. Two unrelated pre-existing bugs surfaced during manual testing
(an event-dispatcher subscriber wiring gap during `require`, and a
RefCell reentrancy panic in `diagnose`) reproduce identically on the
pre-change code and are out of scope here.

3 files changed, 363 insertions, 517 deletions

diff --git a/crates/shirabe/src/util/http/curl_downloader.rs b/crates/shirabe/src/util/http/curl_downloader.rs
index a91455c1..93661d86 100644
--- a/crates/shirabe/src/util/http/curl_downloader.rs
+++ b/crates/shirabe/src/util/http/curl_downloader.rs
@@ -1,15 +1,22 @@
 //! ref: composer/src/Composer/Util/Http/CurlDownloader.php
 //!
-//! reqwest-based re-implementation. The libcurl multi-handle event loop of the PHP original is
-//! replaced by a blocking reqwest client: `download()` queues a job, and `tick()` drives one job to
-//! completion with a synchronous HTTP request. A blocking client is used (rather than the async
-//! client + a per-downloader tokio runtime) so the request never starts a tokio runtime from within
-//! another one when invoked through the codebase's sync bridges (see util/sync_executor.rs).
+//! reqwest-based re-implementation. Unlike the libcurl multi-handle event loop of the PHP
+//! original, `download()` is a single `async fn`: it sends the request, hands the result to
+//! `decide()` (retry / redirect / fail / succeed), and loops until it resolves. There is no job
+//! table or `tick()` driver anymore; the caller `.await`s `download()` directly and gets the
+//! final `Response` (or `TransportException`) back.
 //!
 //! The PHP control flow (insecure-URL check, redirect following, transport/status retries,
 //! authenticated-retry detection, max_file_size enforcement, atomic rename of the `~` temp file)
 //! is preserved. Per-request TLS/proxy/IP-resolve settings that reqwest only exposes per-Client
 //! are simplified to a single default Client; see the TODOs below.
+//!
+//! `abortRequest()` (PHP `CurlDownloader::abortRequest`, called from
+//! `HttpDownloader.php:275` when a React\Promise consumer cancels a download) has no equivalent
+//! here: shirabe has never ported the Promise/canceler machinery (`HttpDownloader::STATUS_ABORTED`
+//! is likewise unused), and there is no job table left to cancel now that `download()` runs to
+//! completion in one `.await`. Re-adding cancellation support belongs to whichever future task
+//! ports that Promise-cancellation flow.
 
 use crate::config::Config;
 use crate::downloader::MaxFileSizeExceededException;
@@ -30,46 +37,12 @@ use shirabe_php_shim::{
 };
 use std::sync::atomic::{AtomicBool, Ordering};
 
-/// resolve callback supplied by `HttpDownloader`. Receives the final `Response` on success.
-pub type ResolveCallback = Box<dyn Fn(Response) + Send + Sync>;
-/// reject callback supplied by `HttpDownloader`. Receives the recoverable error on failure.
-pub type RejectCallback = Box<dyn Fn(anyhow::Error) + Send + Sync>;
-
-/// One in-flight download. PHP stored this as a loosely-typed `array` and additionally kept the
-/// header/body stream resources and the resolve/reject callables out-of-band. Here a typed struct
-/// holds everything, which is what the PHP `Job` array modelled.
-struct CurlJob {
-    url: String,
-    origin: String,
-    attributes: IndexMap<String, PhpMixed>,
-    /// `options` after defaults/auth/stream-context have been merged in.
-    options: IndexMap<String, PhpMixed>,
-    /// Destination path when copying to a file (PHP `filename`), `None` for in-memory downloads.
-    filename: Option<String>,
-    resolve: ResolveCallback,
-    reject: RejectCallback,
-}
-
-impl std::fmt::Debug for CurlJob {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("CurlJob")
-            .field("url", &self.url)
-            .field("origin", &self.origin)
-            .field("attributes", &self.attributes)
-            .field("options", &self.options)
-            .field("filename", &self.filename)
-            .finish()
-    }
-}
-
 #[derive(Debug)]
 pub struct CurlDownloader {
     /// Connection pool / cookie / TLS-session sharing — reqwest::Client handles this internally,
     /// replacing the PHP multiHandle + shareHandle. Redirects are disabled because we follow them
     /// manually (to control auth-header re-attachment), matching CURLOPT_FOLLOWLOCATION = false.
-    client: reqwest::blocking::Client,
-    jobs: IndexMap<i64, CurlJob>,
-    next_id: i64,
+    client: reqwest::Client,
     io: std::rc::Rc<std::cell::RefCell<dyn IOInterface>>,
     config: std::rc::Rc<std::cell::RefCell<Config>>,
     auth_helper: std::rc::Rc<std::cell::RefCell<AuthHelper>>,
@@ -80,6 +53,15 @@ pub struct CurlDownloader {
 /// Function-static `$timeoutWarning` from `tick()`.
 static TIMEOUT_WARNING: AtomicBool = AtomicBool::new(false);
 
+/// The outcome of one send attempt, decided by `decide()`. Replaces the PHP promise
+/// resolve/reject callbacks (and this crate's former restart_job/resolve_job/reject_job side
+/// effects): `download()`'s loop matches on this instead.
+enum Decision {
+    Retry { url: String, delay_ms: Option<u64> },
+    Done(Response),
+    Failed(TransportException),
+}
+
 impl CurlDownloader {
     /// @param mixed[] $options
     pub fn new(
@@ -98,7 +80,7 @@ impl CurlDownloader {
         // (one HttpDownloader owns one CurlDownloader) but not pooled across them.
         // TODO: cookie sharing (CURL_LOCK_DATA_COOKIE) would need reqwest's `cookies` feature
         // (.cookie_store(true)); omitted as it is not required for package downloads.
-        let client = reqwest::blocking::Client::builder()
+        let client = reqwest::Client::builder()
             .pool_max_idle_per_host(8)
             .redirect(reqwest::redirect::Policy::none())
             .build()
@@ -113,8 +95,6 @@ impl CurlDownloader {
 
         Self {
             client,
-            jobs: IndexMap::new(),
-            next_id: 1,
             io,
             config,
             auth_helper,
@@ -126,17 +106,24 @@ impl CurlDownloader {
     /// @param mixed[]  $options
     /// @param non-empty-string $url
     ///
-    /// @return int internal job id
-    pub fn download(
-        &mut self,
-        resolve: ResolveCallback,
-        reject: RejectCallback,
+    /// Runs the request through the redirect/retry/status state machine until it resolves,
+    /// mirroring what the PHP promise resolver + `tick()` loop used to do together.
+    pub async fn download(
+        &self,
         origin: &str,
         url: &str,
         mut options: IndexMap<String, PhpMixed>,
         copy_to: Option<&str>,
-    ) -> anyhow::Result<Result<i64, TransportException>> {
-        let mut attributes: IndexMap<String, PhpMixed> = IndexMap::new();
+    ) -> anyhow::Result<Result<Response, TransportException>> {
+        let mut attributes: IndexMap<String, PhpMixed> = {
+            let mut m = IndexMap::new();
+            m.insert("retryAuthFailure".to_string(), PhpMixed::Bool(true));
+            m.insert("redirects".to_string(), PhpMixed::Int(0));
+            m.insert("retries".to_string(), PhpMixed::Int(0));
+            m.insert("storeAuth".to_string(), PhpMixed::Bool(false));
+            m.insert("ipResolve".to_string(), PhpMixed::Null);
+            m
+        };
         if options.contains_key("retry-auth-failure") {
             attributes.insert(
                 "retryAuthFailure".to_string(),
@@ -148,37 +135,6 @@ impl CurlDownloader {
             options.shift_remove("retry-auth-failure");
         }
 
-        self.init_download(resolve, reject, origin, url, options, copy_to, attributes)
-    }
-
-    #[allow(clippy::too_many_arguments, reason = "to keep PHP signature")]
-    fn init_download(
-        &mut self,
-        resolve: ResolveCallback,
-        reject: RejectCallback,
-        origin: &str,
-        url: &str,
-        options: IndexMap<String, PhpMixed>,
-        copy_to: Option<&str>,
-        attributes: IndexMap<String, PhpMixed>,
-    ) -> anyhow::Result<Result<i64, TransportException>> {
-        let defaults: IndexMap<String, PhpMixed> = {
-            let mut m = IndexMap::new();
-            m.insert("retryAuthFailure".to_string(), PhpMixed::Bool(true));
-            m.insert("redirects".to_string(), PhpMixed::Int(0));
-            m.insert("retries".to_string(), PhpMixed::Int(0));
-            m.insert("storeAuth".to_string(), PhpMixed::Bool(false));
-            m.insert("ipResolve".to_string(), PhpMixed::Null);
-            m
-        };
-        let mut attributes: IndexMap<String, PhpMixed> = {
-            let mut m = defaults;
-            for (k, v) in attributes {
-                m.insert(k, v);
-            }
-            m
-        };
-
         if attributes
             .get("ipResolve")
             .map(|v| v.is_null())
@@ -208,8 +164,8 @@ impl CurlDownloader {
         }
 
         // PHP added the auth options and ran StreamContextFactory::initOptions here, and would fail
-        // up-front if the body temp file could not be opened. reqwest opens no file until tick(),
-        // so the auth/stream-context merge happens once at send time (see send_once).
+        // up-front if the body temp file could not be opened. reqwest opens no file until
+        // send_once(), so the auth/stream-context merge happens once per attempt inside the loop.
 
         let header_strings = Self::header_list(&options);
         let if_modified = if shirabe_php_shim::stripos(
@@ -231,307 +187,297 @@ impl CurlDownloader {
             .map_err(|e| anyhow::anyhow!(e.message))?
             .and_then(|p| p.get_status(Some(" using proxy (%s)")).ok())
             .unwrap_or_default();
-        if attributes.get("redirects").and_then(|v| v.as_int()) == Some(0)
-            && attributes.get("retries").and_then(|v| v.as_int()) == Some(0)
-        {
-            self.io.write_error3(
-                &format!(
-                    "Downloading {}{}{}",
-                    Url::sanitize(url.to_string()),
-                    using_proxy,
-                    if_modified
-                ),
-                true,
-                crate::io::DEBUG,
-            );
-        }
-
-        let id = self.next_id;
-        self.next_id += 1;
-        self.jobs.insert(
-            id,
-            CurlJob {
-                url: url.to_string(),
-                origin: origin.to_string(),
-                attributes,
-                options,
-                filename: copy_to.map(|s| s.to_string()),
-                resolve,
-                reject,
-            },
+        // `attributes.redirects == 0 && attributes.retries == 0` in PHP is always true here since
+        // this runs exactly once, before the retry loop below has a chance to advance either.
+        self.io.write_error3(
+            &format!(
+                "Downloading {}{}{}",
+                Url::sanitize(url.to_string()),
+                using_proxy,
+                if_modified
+            ),
+            true,
+            crate::io::DEBUG,
         );
 
-        Ok(Ok(id))
-    }
-
-    pub fn abort_request(&mut self, id: i64) {
-        if let Some(job) = self.jobs.shift_remove(&id)
-            && let Some(filename) = &job.filename
-        {
-            unlink_silent(&format!("{}~", filename));
-        }
-    }
-
-    pub fn tick(&mut self) -> anyhow::Result<()> {
-        if self.jobs.is_empty() {
-            return Ok(());
-        }
-
-        // Drive every queued job to completion. The PHP multi-handle progressed all easy handles
-        // a little per tick(); here each tick() fully resolves one job (one blocking request),
-        // which is observationally equivalent for the sync wait_id() loop that calls tick().
-        let ids: Vec<i64> = self.jobs.keys().copied().collect();
-        for id in ids {
-            self.run_job(id)?;
-        }
-        Ok(())
-    }
-
-    /// Runs a single job through the redirect/retry/status state machine until it resolves or
-    /// rejects, invoking the stored resolve/reject callback. Mirrors the body of PHP `tick()`.
-    fn run_job(&mut self, id: i64) -> anyhow::Result<()> {
+        let mut current_url = url.to_string();
+        let mut current_origin = origin.to_string();
         loop {
-            let (url, origin, filename, options, attributes) = {
-                let job = match self.jobs.get(&id) {
-                    Some(j) => j,
-                    None => return Ok(()),
-                };
-                (
-                    job.url.clone(),
-                    job.origin.clone(),
-                    job.filename.clone(),
-                    job.options.clone(),
-                    job.attributes.clone(),
-                )
-            };
-
             // PHP merges auth options + stream-context options at curl_setopt time. We need the
             // resulting header/method/content/timeout/ssl/max_file_size, so do it here per send.
             let send_options = self.auth_helper.borrow_mut().add_authentication_options(
                 options.clone(),
-                &origin,
-                &url,
+                &current_origin,
+                &current_url,
             )?;
             let send_options =
-                crate::util::StreamContextFactory::init_options(&url, send_options, true)
+                crate::util::StreamContextFactory::init_options(&current_url, send_options, true)
                     .map_err(|e| anyhow::anyhow!(e.message))?;
 
-            let send_result = self.send_once(&url, &send_options, filename.as_deref(), &attributes);
-
-            let response = match send_result {
-                Ok(resp) => resp,
-                Err(transport_err) => {
-                    // CURLE_OPERATION_TIMEDOUT one-time warning (errno 28). reqwest cannot report
-                    // the curl errno, so this fires on the is_timeout() branch instead.
-                    if transport_err.was_timeout && !TIMEOUT_WARNING.load(Ordering::Relaxed) {
-                        TIMEOUT_WARNING.store(true, Ordering::Relaxed);
-                        self.io.write_error3(
-                            "<warning>A connection timeout was encountered. If you intend to run Composer without connecting to the internet, run the command again prefixed with COMPOSER_DISABLE_NETWORK=1 to make Composer run in offline mode.</warning>",
-                            true,
-                            crate::io::NORMAL,
-                        );
-                    }
+            let send_result = self
+                .send_once(&current_url, &send_options, copy_to, &attributes)
+                .await;
 
-                    // PHP retried on a set of curl errnos (7/16/92/6/28 and 56/35 with "Connection
-                    // reset by peer"); reqwest does not expose those errnos, so approximate with
-                    // is_connect/is_timeout/is_request on GET requests.
-                    let retries = attributes
-                        .get("retries")
-                        .and_then(|v| v.as_int())
-                        .unwrap_or(0);
-                    if transport_err.retryable
-                        && Self::method_is_get(&options)
-                        && retries < self.max_retries
-                    {
-                        let mut new_attrs = attributes.clone();
-                        new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1));
-                        // CURLE_COULDNT_CONNECT analogue: force IPv4 if no IP stack chosen.
-                        if transport_err.is_connect && !attributes.contains_key("ipResolve") {
-                            new_attrs.insert("ipResolve".to_string(), PhpMixed::Int(4));
-                        }
-                        self.io.write_error3(
-                            &format!(
-                                "Retrying ({}) {} due to connection error",
-                                retries + 1,
-                                Url::sanitize(url.clone())
-                            ),
-                            true,
-                            crate::io::DEBUG,
-                        );
-                        self.restart_job_with_delay(id, &url, new_attrs);
-                        continue;
-                    }
-                    // PHP throws a MaxFileSizeExceededException (a TransportException subclass) with
-                    // the raw "Maximum allowed download size reached..." message; preserve it
-                    // verbatim rather than wrapping it in the generic curl-error text.
-                    let message = if transport_err.is_max_file_size {
-                        MaxFileSizeExceededException(TransportException::new(
-                            transport_err.message.clone(),
-                            0,
-                        ))
-                        .0
-                        .message
-                    } else {
-                        TransportException::new(
-                            format!(
-                                "curl error while downloading {}: {}",
-                                Url::sanitize(url.clone()),
-                                transport_err.message
-                            ),
-                            0,
-                        )
-                        .message
-                    };
-                    self.reject_job(id, anyhow::anyhow!(message));
-                    return Ok(());
-                }
-            };
-
-            let status_code = response.status;
-
-            let curl_response = response.into_curl_response(&url);
-
-            self.io.write_error3(
-                &format!("[{}] {}", status_code, Url::sanitize(url.clone())),
-                true,
-                crate::io::DEBUG,
-            );
-
-            // Output JSON warnings (PHP HttpDownloader::outputWarnings) for >= 300 JSON bodies.
-            if curl_response.inner.get_status_code() >= 300
-                && curl_response.inner.get_header("content-type").as_deref()
-                    == Some("application/json")
-                && let Some(body) = curl_response.inner.get_body()
+            match self
+                .decide(
+                    send_result,
+                    &current_url,
+                    &current_origin,
+                    copy_to,
+                    &options,
+                    &mut attributes,
+                )
+                .await?
             {
-                let decoded = shirabe_php_shim::json_decode(body, true)?;
-                if let PhpMixed::Array(a) = decoded {
-                    HttpDownloader::output_warnings(self.io.clone(), &origin, &a)?;
-                }
-            }
-
-            // Authenticated-retry detection (401/403, Bitbucket login page, GitLab archive 404).
-            let auth_result = self.is_authenticated_retry_needed(
-                &url,
-                &origin,
-                filename.as_deref(),
-                &attributes,
-                &curl_response,
-            )?;
-            match auth_result {
-                Ok(prompt) if prompt.retry => {
-                    let mut new_attrs = attributes.clone();
-                    new_attrs.insert(
-                        "storeAuth".to_string(),
-                        match prompt.store_auth {
-                            StoreAuth::Bool(b) => PhpMixed::Bool(b),
-                            StoreAuth::Prompt => PhpMixed::String("prompt".to_string()),
-                        },
-                    );
-                    let retries = attributes
-                        .get("retries")
-                        .and_then(|v| v.as_int())
-                        .unwrap_or(0);
-                    new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1));
-                    self.restart_job(id, &url, new_attrs);
+                Decision::Retry {
+                    url: new_url,
+                    delay_ms,
+                } => {
+                    if let Some(ms) = delay_ms {
+                        tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
+                    }
+                    current_url = new_url;
+                    current_origin = Url::get_origin(&self.config.borrow(), &current_url);
                     continue;
                 }
-                Ok(_) => {}
-                Err(e) => {
-                    self.reject_job(id, anyhow::anyhow!(e.message));
-                    return Ok(());
-                }
+                Decision::Done(response) => return Ok(Ok(response)),
+                Decision::Failed(e) => return Ok(Err(e)),
             }
+        }
+    }
 
-            // Handle 3xx redirects, 304 Not Modified excluded.
-            let redirects = attributes
-                .get("redirects")
-                .and_then(|v| v.as_int())
-                .unwrap_or(0);
-            if (300..=399).contains(&status_code)
-                && status_code != 304
-                && redirects < self.max_redirects
-            {
-                match self.handle_redirect(&url, &attributes, &curl_response)? {
-                    Ok(location) if !location.is_empty() => {
-                        let mut new_attrs = attributes.clone();
-                        new_attrs.insert("redirects".to_string(), PhpMixed::Int(redirects + 1));
-                        // The redirect target becomes the new url; origin is recomputed in restart.
-                        self.restart_job(id, &location, new_attrs);
-                        continue;
-                    }
-                    Ok(_) => {}
-                    Err(e) => {
-                        self.reject_job(id, anyhow::anyhow!(e.message));
-                        return Ok(());
-                    }
+    /// Decides what to do with one send attempt's result: retry (transport error, auth-retry,
+    /// redirect, retryable status code), fail, or succeed. Mirrors the body of the former
+    /// `run_job` loop, minus the resolve/reject/restart side effects, which are now expressed as
+    /// the returned `Decision`.
+    async fn decide(
+        &self,
+        send_result: Result<RawResponse, TransportError>,
+        url: &str,
+        origin: &str,
+        filename: Option<&str>,
+        options: &IndexMap<String, PhpMixed>,
+        attributes: &mut IndexMap<String, PhpMixed>,
+    ) -> anyhow::Result<Decision> {
+        let response = match send_result {
+            Ok(resp) => resp,
+            Err(transport_err) => {
+                // CURLE_OPERATION_TIMEDOUT one-time warning (errno 28). reqwest cannot report
+                // the curl errno, so this fires on the is_timeout() branch instead.
+                if transport_err.was_timeout && !TIMEOUT_WARNING.load(Ordering::Relaxed) {
+                    TIMEOUT_WARNING.store(true, Ordering::Relaxed);
+                    self.io.write_error3(
+                        "<warning>A connection timeout was encountered. If you intend to run Composer without connecting to the internet, run the command again prefixed with COMPOSER_DISABLE_NETWORK=1 to make Composer run in offline mode.</warning>",
+                        true,
+                        crate::io::NORMAL,
+                    );
                 }
-            }
 
-            // Fail 4xx and 5xx responses (some are retried on GET).
-            if (400..=599).contains(&status_code) {
+                // PHP retried on a set of curl errnos (7/16/92/6/28 and 56/35 with "Connection
+                // reset by peer"); reqwest does not expose those errnos, so approximate with
+                // is_connect/is_timeout/is_request on GET requests.
                 let retries = attributes
                     .get("retries")
                     .and_then(|v| v.as_int())
                     .unwrap_or(0);
-                if Self::method_is_get(&options)
-                    && in_array(
-                        PhpMixed::Int(status_code),
-                        &PhpMixed::List(
-                            [423, 425, 500, 502, 503, 504, 507, 510]
-                                .iter()
-                                .map(|c| PhpMixed::Int(*c))
-                                .collect(),
-                        ),
-                        true,
-                    )
+                if transport_err.retryable
+                    && Self::method_is_get(options)
                     && retries < self.max_retries
                 {
+                    attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1));
+                    // CURLE_COULDNT_CONNECT analogue: force IPv4 if no IP stack chosen.
+                    if transport_err.is_connect && !attributes.contains_key("ipResolve") {
+                        attributes.insert("ipResolve".to_string(), PhpMixed::Int(4));
+                    }
                     self.io.write_error3(
                         &format!(
-                            "Retrying ({}) {} due to status code {}",
+                            "Retrying ({}) {} due to connection error",
                             retries + 1,
-                            Url::sanitize(url.clone()),
-                            status_code
+                            Url::sanitize(url.to_string())
                         ),
                         true,
                         crate::io::DEBUG,
                     );
-                    let mut new_attrs = attributes.clone();
-                    new_attrs.insert("retries".to_string(), PhpMixed::Int(retries + 1));
-                    self.restart_job_with_delay(id, &url, new_attrs);
-                    continue;
+                    if let Some(filename) = filename {
+                        unlink_silent(&format!("{}~", filename));
+                    }
+                    return Ok(Decision::Retry {
+                        url: url.to_string(),
+                        delay_ms: Self::retry_delay_ms(retries + 1),
+                    });
                 }
 
-                let status_msg = curl_response.inner.get_status_message().unwrap_or_default();
-                let e = self.fail_response(&url, filename.as_deref(), &curl_response, &status_msg);
-                self.reject_job_with_response(id, e, &curl_response);
-                return Ok(());
+                if let Some(filename) = filename {
+                    unlink_silent(&format!("{}~", filename));
+                }
+                // PHP throws a MaxFileSizeExceededException (a TransportException subclass) with
+                // the raw "Maximum allowed download size reached..." message verbatim rather than
+                // wrapping it in the generic curl-error text.
+                if transport_err.is_max_file_size {
+                    return Ok(Decision::Failed(
+                        MaxFileSizeExceededException(TransportException::new(
+                            transport_err.message,
+                            0,
+                        ))
+                        .0,
+                    ));
+                }
+                return Ok(Decision::Failed(TransportException::new(
+                    format!(
+                        "curl error while downloading {}: {}",
+                        Url::sanitize(url.to_string()),
+                        transport_err.message
+                    ),
+                    0,
+                )));
             }
+        };
+
+        let status_code = response.status;
+        let curl_response = response.into_curl_response(url);
 
-            // storeAuth on success.
-            let store_auth = attributes.get("storeAuth").cloned();
-            if !matches!(store_auth, Some(PhpMixed::Bool(false))) {
-                let store_auth = match store_auth {
-                    Some(PhpMixed::String(ref s)) if s == "prompt" => StoreAuth::Prompt,
-                    Some(PhpMixed::Bool(b)) => StoreAuth::Bool(b),
-                    _ => StoreAuth::Bool(false),
-                };
-                self.auth_helper.borrow().store_auth(&origin, store_auth)?;
+        self.io.write_error3(
+            &format!("[{}] {}", status_code, Url::sanitize(url.to_string())),
+            true,
+            crate::io::DEBUG,
+        );
+
+        // Output JSON warnings (PHP HttpDownloader::outputWarnings) for >= 300 JSON bodies.
+        if curl_response.inner.get_status_code() >= 300
+            && curl_response.inner.get_header("content-type").as_deref() == Some("application/json")
+            && let Some(body) = curl_response.inner.get_body()
+        {
+            let decoded = shirabe_php_shim::json_decode(body, true)?;
+            if let PhpMixed::Array(a) = decoded {
+                HttpDownloader::output_warnings(self.io.clone(), origin, &a)?;
             }
+        }
 
-            // Atomic rename of the `~` temp file to its final name (file mode).
-            if let Some(filename) = &filename {
-                rename(format!("{}~", filename), filename);
+        // Authenticated-retry detection (401/403, Bitbucket login page, GitLab archive 404).
+        let auth_result =
+            self.is_authenticated_retry_needed(url, origin, filename, attributes, &curl_response)?;
+        match auth_result {
+            Ok(prompt) if prompt.retry => {
+                attributes.insert(
+                    "storeAuth".to_string(),
+                    match prompt.store_auth {
+                        StoreAuth::Bool(b) => PhpMixed::Bool(b),
+                        StoreAuth::Prompt => PhpMixed::String("prompt".to_string()),
+                    },
+                );
+                let retries = attributes
+                    .get("retries")
+                    .and_then(|v| v.as_int())
+                    .unwrap_or(0);
+                attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1));
+                if let Some(filename) = filename {
+                    unlink_silent(&format!("{}~", filename));
+                }
+                return Ok(Decision::Retry {
+                    url: url.to_string(),
+                    delay_ms: None,
+                });
             }
+            Ok(_) => {}
+            Err(e) => return Ok(Decision::Failed(e)),
+        }
 
-            self.resolve_job(id, curl_response.inner);
-            return Ok(());
+        // Handle 3xx redirects, 304 Not Modified excluded.
+        let redirects = attributes
+            .get("redirects")
+            .and_then(|v| v.as_int())
+            .unwrap_or(0);
+        if (300..=399).contains(&status_code)
+            && status_code != 304
+            && redirects < self.max_redirects
+        {
+            match self.handle_redirect(url, attributes, &curl_response)? {
+                Ok(location) if !location.is_empty() => {
+                    attributes.insert("redirects".to_string(), PhpMixed::Int(redirects + 1));
+                    if let Some(filename) = filename {
+                        unlink_silent(&format!("{}~", filename));
+                    }
+                    return Ok(Decision::Retry {
+                        url: location,
+                        delay_ms: None,
+                    });
+                }
+                Ok(_) => {}
+                Err(e) => return Ok(Decision::Failed(e)),
+            }
         }
+
+        // Fail 4xx and 5xx responses (some are retried on GET).
+        if (400..=599).contains(&status_code) {
+            let retries = attributes
+                .get("retries")
+                .and_then(|v| v.as_int())
+                .unwrap_or(0);
+            if Self::method_is_get(options)
+                && in_array(
+                    PhpMixed::Int(status_code),
+                    &PhpMixed::List(
+                        [423, 425, 500, 502, 503, 504, 507, 510]
+                            .iter()
+                            .map(|c| PhpMixed::Int(*c))
+                            .collect(),
+                    ),
+                    true,
+                )
+                && retries < self.max_retries
+            {
+                self.io.write_error3(
+                    &format!(
+                        "Retrying ({}) {} due to status code {}",
+                        retries + 1,
+                        Url::sanitize(url.to_string()),
+                        status_code
+                    ),
+                    true,
+                    crate::io::DEBUG,
+                );
+                attributes.insert("retries".to_string(), PhpMixed::Int(retries + 1));
+                if let Some(filename) = filename {
+                    unlink_silent(&format!("{}~", filename));
+                }
+                return Ok(Decision::Retry {
+                    url: url.to_string(),
+                    delay_ms: Self::retry_delay_ms(retries + 1),
+                });
+            }
+
+            let status_msg = curl_response.inner.get_status_message().unwrap_or_default();
+            // Enrich with the response headers/status/body, mirroring PHP's catch block that
+            // calls setHeaders/setStatusCode/setResponse before reject().
+            let mut e = self.fail_response(url, filename, &curl_response, &status_msg);
+            e.set_headers(curl_response.inner.get_headers().clone());
+            e.set_status_code(Some(curl_response.inner.get_status_code()));
+            e.set_response(curl_response.inner.get_body().map(|s| s.to_string()));
+            return Ok(Decision::Failed(e));
+        }
+
+        // storeAuth on success.
+        let store_auth = attributes.get("storeAuth").cloned();
+        if !matches!(store_auth, Some(PhpMixed::Bool(false))) {
+            let store_auth = match store_auth {
+                Some(PhpMixed::String(ref s)) if s == "prompt" => StoreAuth::Prompt,
+                Some(PhpMixed::Bool(b)) => StoreAuth::Bool(b),
+                _ => StoreAuth::Bool(false),
+            };
+            self.auth_helper.borrow().store_auth(origin, store_auth)?;
+        }
+
+        // Atomic rename of the `~` temp file to its final name (file mode).
+        if let Some(filename) = filename {
+            rename(format!("{}~", filename), filename);
+        }
+
+        Ok(Decision::Done(curl_response.inner))
     }
 
-    /// Performs one blocking HTTP request via reqwest, enforcing max_file_size and streaming the
-    /// body to the `~` temp file when in file mode. Replaces PHP's curl_setopt block + curl I/O.
-    fn send_once(
+    /// Performs one non-blocking HTTP request via reqwest, enforcing max_file_size and streaming
+    /// the body to the `~` temp file when in file mode. Replaces PHP's curl_setopt block + curl I/O.
+    async fn send_once(
         &self,
         url: &str,
         options: &IndexMap<String, PhpMixed>,
@@ -584,7 +530,7 @@ impl CurlDownloader {
             builder = builder.body(body);
         }
 
-        let resp = builder.send().map_err(|e| TransportError {
+        let resp = builder.send().await.map_err(|e| TransportError {
             message: e.to_string(),
             retryable: e.is_timeout() || e.is_connect() || e.is_request(),
             is_connect: e.is_connect(),
@@ -611,15 +557,15 @@ impl CurlDownloader {
             headers_out.push(format!("{}: {}", k, v.to_str().unwrap_or("")));
         }
 
-        let body = Self::read_body_with_limit(resp, max_file_size, filename).map_err(
-            |(message, is_max_file_size)| TransportError {
+        let body = Self::read_body_with_limit(resp, max_file_size, filename)
+            .await
+            .map_err(|(message, is_max_file_size)| TransportError {
                 message,
                 retryable: false,
                 is_connect: false,
                 was_timeout: false,
                 is_max_file_size,
-            },
-        )?;
+            })?;
 
         Ok(RawResponse {
             status,
@@ -630,35 +576,34 @@ impl CurlDownloader {
 
     /// Reads the body, enforcing max_file_size, writing to the `~` temp file when in file mode.
     /// The `bool` in the error is `true` when the failure is a max_file_size violation.
-    fn read_body_with_limit(
-        resp: reqwest::blocking::Response,
+    async fn read_body_with_limit(
+        mut resp: reqwest::Response,
         max_file_size: Option<u64>,
         filename: Option<&str>,
     ) -> Result<Body, (String, bool)> {
-        use std::io::{Read as _, Write as _};
+        use tokio::io::AsyncWriteExt as _;
 
-        let mut stream = resp;
         let mut written: u64 = 0;
         enum Sink {
-            File(std::fs::File),
+            File(tokio::fs::File),
             Memory(Vec<u8>),
         }
         let mut sink = match filename {
             Some(f) => Sink::File(
-                std::fs::File::create(format!("{}~", f)).map_err(|e| (e.to_string(), false))?,
+                tokio::fs::File::create(format!("{}~", f))
+                    .await
+                    .map_err(|e| (e.to_string(), false))?,
             ),
             None => Sink::Memory(Vec::new()),
         };
 
-        let mut buffer = [0u8; 16 * 1024];
         loop {
-            let n = match stream.read(&mut buffer) {
-                Ok(0) => break,
-                Ok(n) => n,
+            let chunk = match resp.chunk().await {
+                Ok(Some(chunk)) => chunk,
+                Ok(None) => break,
                 Err(e) => return Err((e.to_string(), false)),
             };
-            let chunk = &buffer[..n];
-            written += n as u64;
+            written += chunk.len() as u64;
             if let Some(max) = max_file_size
                 && written > max
             {
@@ -671,8 +616,11 @@ impl CurlDownloader {
                 ));
             }
             match &mut sink {
-                Sink::File(f) => f.write_all(chunk).map_err(|e| (e.to_string(), false))?,
-                Sink::Memory(buf) => buf.extend_from_slice(chunk),
+                Sink::File(f) => f
+                    .write_all(&chunk)
+                    .await
+                    .map_err(|e| (e.to_string(), false))?,
+                Sink::Memory(buf) => buf.extend_from_slice(&chunk),
             }
         }
 
@@ -755,7 +703,7 @@ impl CurlDownloader {
     }
 
     fn is_authenticated_retry_needed(
-        &mut self,
+        &self,
         url: &str,
         origin: &str,
         filename: Option<&str>,
@@ -856,55 +804,17 @@ impl CurlDownloader {
         }))
     }
 
-    fn restart_job(&mut self, id: i64, url: &str, attributes: IndexMap<String, PhpMixed>) {
-        let filename = match self.jobs.get(&id) {
-            Some(job) => job.filename.clone(),
-            None => return,
-        };
-        if let Some(filename) = &filename {
-            unlink_silent(&format!("{}~", filename));
-        }
-
-        // Merge the new attributes over the job's existing ones.
-        let merged = {
-            let mut m = match self.jobs.get(&id) {
-                Some(job) => job.attributes.clone(),
-                None => return,
-            };
-            for (k, v) in attributes {
-                m.insert(k, v);
-            }
-            m
-        };
-        let origin = Url::get_origin(&self.config.borrow(), url);
-
-        // options/filename/resolve/reject are preserved across the restart, mirroring PHP forwarding
-        // the original job's resolve/reject into the restarted download; only url/origin/attributes
-        // change.
-        if let Some(job) = self.jobs.get_mut(&id) {
-            job.url = url.to_string();
-            job.origin = origin;
-            job.attributes = merged;
-        }
-    }
-
-    fn restart_job_with_delay(
-        &mut self,
-        id: i64,
-        url: &str,
-        attributes: IndexMap<String, PhpMixed>,
-    ) {
-        let retries = attributes
-            .get("retries")
-            .and_then(|v| v.as_int())
-            .unwrap_or(0);
+    /// The delay `restart_job_with_delay` used to sleep before restarting a retried job: half a
+    /// second from the 3rd retry onward, 100ms for the 2nd, none for the 1st. `retries` is the
+    /// post-increment retry count.
+    fn retry_delay_ms(retries: i64) -> Option<u64> {
         if retries >= 3 {
-            shirabe_php_shim::usleep(500000); // half a second delay for 3rd retry and beyond
+            Some(500)
         } else if retries >= 2 {
-            shirabe_php_shim::usleep(100000); // 100ms delay for 2nd retry
-        } // no sleep for the first retry
-
-        self.restart_job(id, url, attributes);
+            Some(100)
+        } else {
+            None
+        }
     }
 
     fn fail_response(
@@ -955,40 +865,6 @@ impl CurlDownloader {
         )
     }
 
-    /// Invokes the stored resolve callback and removes the job.
-    fn resolve_job(&mut self, id: i64, response: Response) {
-        if let Some(job) = self.jobs.shift_remove(&id) {
-            (job.resolve)(response);
-        }
-    }
-
-    /// Invokes the stored reject callback and removes the job, deleting the temp file.
-    fn reject_job(&mut self, id: i64, e: anyhow::Error) {
-        if let Some(job) = self.jobs.shift_remove(&id) {
-            if let Some(filename) = &job.filename {
-                unlink_silent(&format!("{}~", filename));
-            }
-            (job.reject)(e);
-        }
-    }
-
-    /// Reject after enriching the TransportException with the response headers/status/body, mirroring
-    /// PHP's catch block that calls setHeaders/setStatusCode/setResponse before reject().
-    fn reject_job_with_response(
-        &mut self,
-        id: i64,
-        mut e: TransportException,
-        response: &CurlResponse,
-    ) {
-        e.set_headers(response.inner.get_headers().clone());
-        e.set_status_code(Some(response.inner.get_status_code()));
-        e.set_response(response.inner.get_body().map(|s| s.to_string()));
-        let msg = e.message.clone();
-        // Carry the enriched exception through anyhow; the typed payload is reconstructed from the
-        // message on the HttpDownloader side. TransportException is the recoverable error here.
-        self.reject_job(id, anyhow::Error::new(e).context(msg));
-    }
-
     fn method_is_get(options: &IndexMap<String, PhpMixed>) -> bool {
         let method = options
             .get("http")
diff --git a/crates/shirabe/src/util/http_downloader.rs b/crates/shirabe/src/util/http_downloader.rs
index 448c7019..c2c83b85 100644
--- a/crates/shirabe/src/util/http_downloader.rs
+++ b/crates/shirabe/src/util/http_downloader.rs
@@ -97,34 +97,15 @@ impl Default for HttpDownloaderMockHandler {
     }
 }
 
+#[derive(Debug)]
 struct Job {
     id: i64,
     status: i64,
     request: Request,
     sync: bool,
     origin: String,
-    curl_id: Option<i64>,
     response: Option<Response>,
     exception: Option<anyhow::Error>,
-    /// Completion slot written by the curl resolve/reject closures (driven by `curl.tick()`)
-    /// and read by `count_active_jobs`. Uses `Arc<Mutex>` because `CurlDownloader::download`
-    /// requires `Send + Sync` callbacks.
-    settled: std::sync::Arc<std::sync::Mutex<Option<anyhow::Result<Response>>>>,
-}
-
-impl std::fmt::Debug for Job {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("Job")
-            .field("id", &self.id)
-            .field("status", &self.status)
-            .field("request", &self.request)
-            .field("sync", &self.sync)
-            .field("origin", &self.origin)
-            .field("curl_id", &self.curl_id)
-            .field("response", &self.response)
-            .field("exception", &self.exception)
-            .finish()
-    }
 }
 
 #[derive(Debug, Clone)]
@@ -134,6 +115,24 @@ struct Request {
     copy_to: Option<String>,
 }
 
+/// A single-threaded tokio Runtime used only to drive `CurlDownloader::download()` (which needs a
+/// real reactor now that it uses the non-blocking `reqwest::Client`, unlike `sync_executor::block_on`
+/// which assumes every awaited future resolves synchronously). `current_thread` is used because
+/// `block_on` (unlike `spawn`) has no `Send` bound, and `download()`'s future closes over
+/// `Rc<RefCell<...>>` handles that are not `Send`.
+///
+/// TODO(phase-e): remove this once `HttpDownloader::add`/`get` are driven by `Loop::wait`'s
+/// `FuturesUnordered` under a single top-level Runtime (see the async re-architecture design).
+fn curl_runtime() -> &'static tokio::runtime::Runtime {
+    static RT: std::sync::LazyLock<tokio::runtime::Runtime> = std::sync::LazyLock::new(|| {
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("failed to build the temporary CurlDownloader bridge runtime")
+    });
+    &RT
+}
+
 impl HttpDownloader {
     const STATUS_QUEUED: i64 = 1;
     const STATUS_STARTED: i64 = 2;
@@ -397,10 +396,8 @@ impl HttpDownloader {
             request: request.clone(),
             sync,
             origin,
-            curl_id: None,
             response: None,
             exception: None,
-            settled: std::sync::Arc::new(std::sync::Mutex::new(None)),
         };
         let can_use_curl = self.can_use_curl(&job);
         self.jobs.insert(id, job);
@@ -560,38 +557,20 @@ impl HttpDownloader {
             return;
         }
 
-        // curl branch: register the request with the curl multi handle. Completion is delivered
-        // by curl.tick() into the job's `settled` slot (read by count_active_jobs). The resolve
-        // callback stores the Response, reject stores the error; this mirrors PHP's promise
-        // resolve/reject firing during tick(). PHP catches any exception from download() and
-        // rejects the job.
-        let settled = self.jobs.get(&id).unwrap().settled.clone();
-        let settled_for_reject = settled.clone();
-        let resolve: Box<dyn Fn(Response) + Send + Sync> = Box::new(move |response: Response| {
-            *settled.lock().unwrap() = Some(Ok(response));
-        });
-        let reject: Box<dyn Fn(anyhow::Error) + Send + Sync> =
-            Box::new(move |error: anyhow::Error| {
-                *settled_for_reject.lock().unwrap() = Some(Err(error));
-            });
-
-        let download_result = {
-            let curl = self.curl.as_mut().unwrap();
-            curl.download(resolve, reject, &origin, &url, options, copy_to.as_deref())
-        };
-        match download_result {
-            Ok(Ok(curl_id)) => {
-                if let Some(job) = self.jobs.get_mut(&id) {
-                    job.curl_id = Some(curl_id);
-                }
-            }
-            Ok(Err(e)) => {
-                self.settle_job(id, Err(e.into()));
-            }
-            Err(e) => {
-                self.settle_job(id, Err(e));
+        // curl branch: `CurlDownloader::download` now runs the whole redirect/retry/auth state
+        // machine to completion itself and returns the settled result directly, so this drives
+        // it through the temporary bridge runtime instead of PHP's promise resolve/reject firing
+        // during tick(). PHP catches any exception from download() and rejects the job.
+        let result: anyhow::Result<Response> = {
+            let curl = self.curl.as_ref().unwrap();
+            match curl_runtime().block_on(curl.download(&origin, &url, options, copy_to.as_deref()))
+            {
+                Ok(Ok(response)) => Ok(response),
+                Ok(Err(transport_exception)) => Err(transport_exception.into()),
+                Err(e) => Err(e),
             }
-        }
+        };
+        self.settle_job(id, result);
     }
 
     fn mark_job_done(&mut self) {
@@ -637,24 +616,9 @@ impl HttpDownloader {
             }
         }
 
-        if let Some(curl) = self.curl.as_mut() {
-            curl.tick()?;
-        }
-
-        // Apply completions delivered by curl.tick() into each started job's `settled` slot.
-        // This reproduces the effect of PHP's resolve/reject callbacks firing during tick().
-        let started_ids: Vec<i64> = self
-            .jobs
-            .values()
-            .filter(|j| j.status == Self::STATUS_STARTED)
-            .map(|j| j.id)
-            .collect();
-        for id in started_ids {
-            let settled = self.jobs.get(&id).unwrap().settled.lock().unwrap().take();
-            if let Some(result) = settled {
-                self.settle_job(id, result);
-            }
-        }
+        // Unlike the old tick()-driven curl path, `start_job` now settles curl jobs synchronously
+        // (via the temporary bridge runtime), so no job is ever left lingering in STATUS_STARTED
+        // by the time we get here — nothing left to poll or collect.
 
         if let Some(index) = index {
             return Ok(
diff --git a/crates/shirabe/src/util/sync_executor.rs b/crates/shirabe/src/util/sync_executor.rs
index b7512c2d..ee0995fc 100644
--- a/crates/shirabe/src/util/sync_executor.rs
+++ b/crates/shirabe/src/util/sync_executor.rs
@@ -1,10 +1,16 @@
 //! Minimal synchronous future executor used as a drop-in for the `tokio::runtime::Runtime::new()`
 //! + `block_on` sync bridges scattered across the codebase (repository / installer / downloader).
 //!
-//! Those bridges drive `async fn`s whose `.await` points all resolve synchronously: the only async
-//! I/O (reqwest in `CurlDownloader`) is performed through a blocking client, so no reactor is
-//! required. Nesting `tokio` runtimes is forbidden ("Cannot start a runtime from within a runtime"),
-//! which is why this no-reactor executor exists; it can be nested freely.
+//! Those bridges drive `async fn`s whose `.await` points all resolve synchronously — this relies
+//! on the invariant below. `CurlDownloader` no longer qualifies: it now performs real async I/O
+//! via a non-blocking `reqwest::Client`, so `HttpDownloader` drives it through its own dedicated
+//! `tokio::runtime::Runtime` instead (see `http_downloader::curl_runtime`), not through this
+//! module. The other call sites (`file_downloader.rs`, `version_guesser.rs`,
+//! `installation_manager.rs`, `composer_repository.rs`, `sync_helper.rs`) still rely on this
+//! module because none of their awaited futures actually park on a reactor.
+//!
+//! Nesting `tokio` runtimes is forbidden ("Cannot start a runtime from within a runtime"), which is
+//! why this no-reactor executor exists for those remaining call sites; it can be nested freely.
 //!
 //! TODO(phase-e): remove this module once the async bridges are either made genuinely synchronous or
 //! consolidated onto a single shared runtime driven from `main`.