feat(plugin): serve HttpDownloader's async surface

add() and addCopy() answer with a promise, and enableAsync(), wait() and
countActiveJobs() answer alongside them. The Rust future runs to completion
before the promise is handed over, so requests a plugin starts together run
one after another rather than overlapping; overlapping them needs a promise
representation that crosses the boundary unresolved.

Everything else the surface does is preserved. add() still refuses a
downloader outside a Loop, and it does so by throwing out of the call the way
PHP does, where a failed request instead arrives as a rejection the caller
handles — __shirabe_rejected_promise is the failure half of the resolved-
promise helper. wait() and countActiveJobs() answer for a downloader with no
outstanding job, which, once every request settles before its call returns,
it never has.

This is where HttpDownloader parts company with ProcessExecutor, whose async
surface stays an explicit error: executeAsync() resolves its promise with a
Symfony Process, whose state is the proc_open() resource of whichever process
called start(), where a request resolves its promise with a Response.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

6 files changed, 145 insertions, 14 deletions

diff --git a/crates/shirabe-php-rpc/php/worker.php b/crates/shirabe-php-rpc/php/worker.php
index defa4fb5..b2323eb8 100644
--- a/crates/shirabe-php-rpc/php/worker.php
+++ b/crates/shirabe-php-rpc/php/worker.php
@@ -841,6 +841,21 @@ ShirabeRpcRuntime::$dispatch = [
         }
         return \React\Promise\resolve($args[0]);
     },
+    // An already-rejected promise for a Rust-side call whose PHP signature declares
+    // PromiseInterface. A failed request surfaces to the caller as a rejection it handles, the
+    // way it does under Composer, rather than as a throw out of the call that started it.
+    '__shirabe_rejected_promise' => static function ($args) {
+        if (!function_exists('React\\Promise\\reject')) {
+            throw new RuntimeException(
+                'react/promise is not loaded in the plugin process, so a PromiseInterface cannot be built'
+            );
+        }
+        [$class, $message, $code, $properties] = $args;
+
+        return \React\Promise\reject(
+            \Shirabe\MaterializedThrowable::revive($class, $message, (int) $code, $properties)
+        );
+    },
     // Drains a promise a plugin returned to the Rust side. React settles promises
     // synchronously, so an already-settled one runs these handlers during then(); one that is
     // still pending is an explicit error rather than a silently dropped continuation.
diff --git a/crates/shirabe/src/plugin/php_plugin_proxy.rs b/crates/shirabe/src/plugin/php_plugin_proxy.rs
index c966897b..0465eb6d 100644
--- a/crates/shirabe/src/plugin/php_plugin_proxy.rs
+++ b/crates/shirabe/src/plugin/php_plugin_proxy.rs
@@ -33,7 +33,7 @@ use shirabe_php_rpc::{
     PhpObjHandle, PhpThrow, PluginValue, RustMethodDispatcher, RustObjHandle,
     call_function_with_dispatcher, call_php_method, new_object, release_php_handle,
 };
-use shirabe_php_shim::{AnyThrowable, Catch as _, PhpClass as _, PhpMixed};
+use shirabe_php_shim::{AnyThrowable, Catch as _, LogicException, PhpClass as _, PhpMixed};
 use shirabe_symfony_console::command::Command;
 use shirabe_symfony_console::input::InputInterface;
 use shirabe_symfony_console::input::InputValue;
@@ -886,6 +886,35 @@ fn resolved_promise(value: PluginValue) -> Result<PluginValue, PhpThrow> {
     }
 }
 
+/// A `\React\Promise\PromiseInterface` already rejected with the exception `throw` describes,
+/// built in the worker. A failed request reaches the plugin as a rejection it handles, the way it
+/// does under Composer, rather than as a throw out of the call that started it.
+fn rejected_promise(throw: PhpThrow) -> Result<PluginValue, PhpThrow> {
+    let properties = PluginValue::Array(
+        throw
+            .properties
+            .iter()
+            .map(|(name, value)| (name.clone().into_bytes(), value.clone()))
+            .collect(),
+    );
+    match call_function_with_dispatcher(
+        "__shirabe_rejected_promise",
+        vec![
+            PluginValue::string(throw.exception_class),
+            PluginValue::string(throw.message),
+            PluginValue::Int(throw.code),
+            properties,
+        ],
+        Some(&mut PluginRpcDispatcher::default()),
+    ) {
+        Ok(Ok(promise)) => Ok(promise),
+        Ok(Err(throw)) => Err(throw),
+        Err(error) => Err(runtime_throw(format!(
+            "creating a rejected promise in the plugin process failed: {error:#}"
+        ))),
+    }
+}
+
 fn dispatch_filesystem_method(
     fs: &std::rc::Rc<std::cell::RefCell<crate::util::Filesystem>>,
     method_name: &str,
@@ -1198,12 +1227,46 @@ fn dispatch_http_downloader_method(
                 .map_err(|error| error_throw("copy failed", &error))?;
             Ok(response_to_wire(&response))
         }
-        // TODO(plugin,async): the async surface resolves its promises with a Response the wire
-        // cannot carry, and driving it needs a promise representation that crosses the boundary
-        // unresolved. Neither exists yet.
-        "add" | "addCopy" | "wait" | "enableAsync" | "countActiveJobs" => Err(runtime_throw(
-            format!("Shirabe does not support HttpDownloader::{method_name}() from a plugin yet"),
-        )),
+        // TODO(async): the future runs to completion here, so the promise the plugin receives
+        // is already settled and requests it starts together run one after another instead of
+        // overlapping. Deferring the settlement needs a promise representation that crosses the
+        // boundary unresolved.
+        "add" | "addCopy" => {
+            let url = arg::<String>(method_name, args, 0)?;
+            let outcome = if method_name == "add" {
+                let options =
+                    arg_or::<IndexMap<String, PhpMixed>>(method_name, args, 1, IndexMap::new())?;
+                crate::util::sync_executor::block_on(downloader.borrow().add(&url, options))
+            } else {
+                let to = arg::<String>(method_name, args, 1)?;
+                let options =
+                    arg_or::<IndexMap<String, PhpMixed>>(method_name, args, 2, IndexMap::new())?;
+                crate::util::sync_executor::block_on(
+                    downloader.borrow().add_copy(&url, &to, options),
+                )
+            };
+            match outcome {
+                Ok(response) => resolved_promise(response_to_wire(&response)),
+                // PHP raises the empty-url check and the async gate out of `add()`/`addCopy()`
+                // itself, before the promise exists, and only what the job's resolver raises
+                // becomes a rejection. Those two are the port's only `LogicException`s here, and
+                // a failed request is a `TransportException`.
+                Err(error) if error.is_instanceof::<LogicException>() => {
+                    Err(error_throw(&format!("{method_name} failed"), &error))
+                }
+                Err(error) => {
+                    rejected_promise(error_throw(&format!("{method_name} failed"), &error))
+                }
+            }
+        }
+        "enableAsync" => {
+            downloader.borrow_mut().enable_async();
+            Ok(PluginValue::Null)
+        }
+        // TODO(async): every request settles before the call that started it returns, so the
+        // downloader never holds a queued or started job for these two to report on.
+        "wait" => Ok(PluginValue::Null),
+        "countActiveJobs" => Ok(PluginValue::Int(0)),
         other => Err(runtime_throw(format!(
             "unknown HttpDownloader method `{other}`"
         ))),
diff --git a/crates/shirabe/tests/plugin/e2e_http_downloader_test.rs b/crates/shirabe/tests/plugin/e2e_http_downloader_test.rs
index 42471b61..f5fdfa97 100644
--- a/crates/shirabe/tests/plugin/e2e_http_downloader_test.rs
+++ b/crates/shirabe/tests/plugin/e2e_http_downloader_test.rs
@@ -76,6 +76,11 @@ get=ok
 response class=\"Composer\\\\Util\\\\Http\\\\Response\" body=\"{\\\"probe\\\":true,\\\"n\\\":42}\" headers=[]
 getHeader=null
 copy=ok file=\"{\\\"probe\\\":true,\\\"n\\\":42}\"
+add-before-enable=LogicException: You must use the HttpDownloader instance which is part of a Composer\\Loop instance to be able to run async http requests
+add=ok body=\"{\\\"probe\\\":true,\\\"n\\\":42}\"
+add-missing=ok rejected=\"Composer\\\\Downloader\\\\TransportException\"
+addCopy=ok file=\"{\\\"probe\\\":true,\\\"n\\\":42}\"
+countActiveJobs=0 wait=ok
 collect=ok
 ",
         upstream.trace
diff --git a/crates/shirabe/tests/plugin/fixtures/e2e-http-downloader/plugin/src/Plugin.php b/crates/shirabe/tests/plugin/fixtures/e2e-http-downloader/plugin/src/Plugin.php
index 054a47ca..6e91ad00 100644
--- a/crates/shirabe/tests/plugin/fixtures/e2e-http-downloader/plugin/src/Plugin.php
+++ b/crates/shirabe/tests/plugin/fixtures/e2e-http-downloader/plugin/src/Plugin.php
@@ -98,6 +98,39 @@ class Plugin implements PluginInterface, EventSubscriberInterface
             $downloader->copy('file://' . $payload, $target);
         }) . ' file=' . json_encode(@file_get_contents($target));
 
+        // A downloader that is not part of a Loop refuses async requests, so the probe records
+        // both sides of the gate.
+        $lines[] = 'add-before-enable=' . $this->describe(static function () use ($downloader, $payload): void {
+            $downloader->add('file://' . $payload);
+        });
+        $downloader->enableAsync();
+
+        $added = null;
+        $lines[] = 'add=' . $this->describe(static function () use ($downloader, $payload, &$added): void {
+            $downloader->add('file://' . $payload)->then(static function ($result) use (&$added): void {
+                $added = $result;
+            });
+        }) . ' body=' . json_encode($added === null ? null : $added->getBody());
+
+        // A path that cannot exist keeps the rejection reason free of the temporary directory.
+        $rejection = null;
+        $lines[] = 'add-missing=' . $this->describe(static function () use ($downloader, &$rejection): void {
+            $downloader->add('file:///shirabe-probe-missing.json')->then(null, static function ($error) use (&$rejection): void {
+                $rejection = $error;
+            });
+        }) . ' rejected=' . json_encode($rejection === null ? null : \get_class($rejection));
+
+        $lines[] = 'addCopy=' . $this->describe(static function () use ($downloader, $payload): void {
+            $downloader->addCopy('file://' . $payload, getcwd() . '/probe-async-copy.json')->then(null, static function ($error): void {
+                throw $error;
+            });
+        }) . ' file=' . json_encode(@file_get_contents(getcwd() . '/probe-async-copy.json'));
+
+        $lines[] = 'countActiveJobs=' . json_encode($downloader->countActiveJobs())
+            . ' wait=' . $this->describe(static function () use ($downloader): void {
+                $downloader->wait();
+            });
+
         // collect() unsets the response's own properties, so the object is spent afterwards and
         // nothing may read it again.
         $lines[] = 'collect=' . $this->describe(static function () use ($response): void {
diff --git a/docs/dev/php-rpc.md b/docs/dev/php-rpc.md
index 6aa1d737..d2ce4ae5 100644
--- a/docs/dev/php-rpc.md
+++ b/docs/dev/php-rpc.md
@@ -167,9 +167,12 @@ Notable internal helpers:
   `selfDir`/`installedIsLocalDir` restore) into the worker; skipped only when the class is not
   even autoloadable there, i.e. no Composer PHP runtime and therefore no observer code.
 - `__shirabe_resolved_promise` — wraps a value in `\React\Promise\resolve()`, so a Rust method
-  whose PHP signature declares `PromiseInterface` (the `DownloadManager` surface) can answer
-  with the object type the caller expects. The Rust future has already run to completion by
-  then; deferred resolution across the boundary does not exist yet.
+  whose PHP signature declares `PromiseInterface` (the `DownloadManager` and `HttpDownloader`
+  surfaces) can answer with the object type the caller expects. The Rust future has already run
+  to completion by then; deferred resolution across the boundary does not exist yet.
+- `__shirabe_rejected_promise` — the failure half of the same: `\React\Promise\reject()` over
+  the exception a `Throw` frame's four fields describe, so a failed request reaches the caller
+  as a rejection it handles rather than as a throw out of the call that started it.
 - `__shirabe_settle_promise` — the inverse: drains a promise a plugin returned to Rust. React
   settles synchronously, so an already-settled promise yields its value here (a rejection is
   re-thrown as the Throw reply); one that is still pending is an explicit error.
diff --git a/docs/dev/plugin-class-classification.md b/docs/dev/plugin-class-classification.md
index a383f7c6..5cb8fd6e 100644
--- a/docs/dev/plugin-class-classification.md
+++ b/docs/dev/plugin-class-classification.md
@@ -415,10 +415,22 @@ its options, its TLS defaults and the authentication it collects into the
 run's IO are state the two worlds have to share, and a plugin-`new`ed one
 allocates a Rust-side entity rather than a second downloader the graph knows
 nothing about. `get()` and `copy()` answer with a
-`Composer\Util\Http\Response` the child holds as a value (see below); the
-async surface — `add()`, `addCopy()`, `wait()`, `enableAsync()`,
-`countActiveJobs()` — is still an explicit error, because driving it needs a
-promise representation that crosses the boundary unresolved.
+`Composer\Util\Http\Response` the child holds as a value (see below), and the
+async surface answers too, with the future driven to completion before the
+promise is handed over. Requests a plugin starts together therefore run one
+after another rather than overlapping; overlapping them needs a promise
+representation that crosses the boundary unresolved. Everything else the async
+surface does is preserved: `add()` still refuses a downloader outside a `Loop`,
+a failed request still arrives as a rejection rather than as a throw, and
+`wait()` and `countActiveJobs()` still answer for a downloader that holds no
+outstanding job — which, once every request settles before its call returns, it
+never does.
+
+This is where `ProcessExecutor` and `HttpDownloader` part company. The
+executor's async surface stays an explicit error because `executeAsync()`
+resolves its promise with a `Symfony\Component\Process\Process`, whose state
+is the `proc_open()` resource of whichever process called `start()`; a request
+resolves its promise with a `Response`, which is data.
 
 `Loop` remains guarded, so `Composer::getLoop()` is still an explicit error
 and neither the graph's own executor nor its downloader is reachable through