feat(plugin): dispatch plugin event subscribers through the RPC worker

Wires the addPlugin subscriber branch end to end: EventSubscriberInterface
and Capable become fallible and dyn-compatible (their sole implementor is
the PHP plugin proxy, which answers getSubscribedEvents over RPC), listeners
register as Callable::PhpMethod and are invoked with a per-call event handle,
and the R table now drops entries when a child-side stub destructs. The R
table keeps its IndexMap with monotonically increasing handles, so released
handles are never reused and no generation counter is needed.

Upstream has no subscriber-plugin test, so the path is covered by a
Shirabe-owned fixture exercising all three getSubscribedEvents shapes.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

1 files changed, 183 insertions, 10 deletions

diff --git a/crates/shirabe/src/plugin/php_plugin_proxy.rs b/crates/shirabe/src/plugin/php_plugin_proxy.rs
index 2b823fa5..143af8bb 100644
--- a/crates/shirabe/src/plugin/php_plugin_proxy.rs
+++ b/crates/shirabe/src/plugin/php_plugin_proxy.rs
@@ -7,11 +7,14 @@
 
 use crate::autoload::ClassLoader;
 use crate::composer::ComposerHandle;
+use crate::event_dispatcher::event_dispatcher::dispatch_event_method;
+use crate::event_dispatcher::{EventInterface, EventSubscriberInterface, SubscribedEventEntry};
 use crate::io::IOInterface;
 use crate::plugin::plugin_interface::PluginInterface;
 use indexmap::IndexMap;
 use shirabe_php_rpc::{
-    PhpThrow, PluginValue, RustMethodDispatcher, RustObjHandle, call_php_method, release_php_handle,
+    PhpObjHandle, PhpThrow, PluginValue, RustMethodDispatcher, RustObjHandle, call_php_method,
+    release_php_handle,
 };
 
 /// A Rust-side entity a PHP proxy stub points back to.
@@ -22,20 +25,39 @@ enum RustEntity {
 }
 
 thread_local! {
-    /// The R table. Entries are strong references kept for the worker's lifetime.
-    /// TODO(plugin): GC (dropping entries on ReleaseRustHandle) is not implemented yet;
-    /// until then entities registered here are intentionally never released.
+    /// The R table. Entries are strong references held while the child-side stub lives; a
+    /// ReleaseRustHandle notification (sent by the stub's `__destruct`) removes the entry.
+    /// Handles are monotonically increasing and never reused, so a released handle can never
+    /// be confused with a later entity (no generation counter is needed).
     /// TODO(plugin): thread-local while the worker and its stub intern table are
     /// process-global; the session lock serializes calls, and every dispatch currently runs on
     /// the thread that registered the handle, but a handle minted on one thread is invisible
     /// to another.
     static R_TABLE: std::cell::RefCell<IndexMap<u64, RustEntity>> =
         std::cell::RefCell::new(IndexMap::new());
+
+    static RELEASE_HOOK_INSTALLED: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
+}
+
+/// Installs the ReleaseRustHandle listener that drops R-table entries when the child-side stub
+/// is destructed. Idempotent; called by every entity registration.
+fn ensure_release_hook_installed() {
+    RELEASE_HOOK_INSTALLED.with(|installed| {
+        if !installed.get() {
+            installed.set(true);
+            shirabe_php_rpc::set_release_rust_handle_hook(|rhandle| {
+                R_TABLE.with(|table| {
+                    table.borrow_mut().shift_remove(&rhandle);
+                });
+            });
+        }
+    });
 }
 
 /// Registers the Composer instance in the R table, interned by shared-pointer identity so the
 /// same instance always crosses the boundary as the same handle (`===` in the child).
 pub(crate) fn register_composer_entity(composer: &ComposerHandle) -> u64 {
+    ensure_release_hook_installed();
     R_TABLE.with(|table| {
         let mut table = table.borrow_mut();
         let ptr = std::rc::Rc::as_ptr(composer.as_rc()) as *const () as usize;
@@ -54,6 +76,7 @@ pub(crate) fn register_composer_entity(composer: &ComposerHandle) -> u64 {
 
 /// Registers an IO instance in the R table, interned like [`register_composer_entity`].
 pub(crate) fn register_io_entity(io: &std::rc::Rc<std::cell::RefCell<dyn IOInterface>>) -> u64 {
+    ensure_release_hook_installed();
     R_TABLE.with(|table| {
         let mut table = table.borrow_mut();
         let ptr = std::rc::Rc::as_ptr(io) as *const () as usize;
@@ -118,10 +141,18 @@ pub(crate) fn find_file_in_registered_loaders(class: &str) -> Option<String> {
 /// Serves `CallRustMethod` requests from the child while a plugin-related call is in flight:
 /// rhandle 0 is the runtime service endpoint (autoload lookups), every other rhandle resolves
 /// through the R table. Unsupported methods are explicit errors, never silent fallbacks.
-#[derive(Debug)]
-pub(crate) struct PluginRpcDispatcher;
+#[derive(Debug, Default)]
+pub(crate) struct PluginRpcDispatcher<'a> {
+    /// At most one live event handle exposed per dispatched listener call, alongside the
+    /// persistent R table.
+    ///
+    /// TODO(plugin): a listener that stores the event stub beyond its own call observes an
+    /// unknown handle error afterwards; keeping events in the R table needs full proxying of
+    /// the object graph an event exposes, which does not exist yet.
+    pub(crate) event: Option<(u64, &'a dyn EventInterface)>,
+}
 
-impl RustMethodDispatcher for PluginRpcDispatcher {
+impl RustMethodDispatcher for PluginRpcDispatcher<'_> {
     fn dispatch(
         &mut self,
         rhandle: u64,
@@ -150,6 +181,12 @@ impl RustMethodDispatcher for PluginRpcDispatcher {
             )));
         }
 
+        if let Some((event_rhandle, event)) = self.event
+            && event_rhandle == rhandle
+        {
+            return dispatch_event_method(event, method_name);
+        }
+
         // The entity is cloned out so no table borrow is held while the handler runs (a
         // handler that re-enters register_*_entity would otherwise panic on the RefCell).
         let entity = R_TABLE.with(|table| table.borrow().get(&rhandle).cloned());
@@ -264,11 +301,18 @@ fn runtime_throw(message: String) -> PhpThrow {
 pub struct PhpPluginProxy {
     pub(crate) phandle: u64,
     pub(crate) class: String,
+    /// Every interface the entity's class implements (`class_implements` in the child), for
+    /// the `instanceof` checks PHP performs on plugin objects.
+    pub(crate) implements: Vec<String>,
 }
 
 impl PhpPluginProxy {
-    pub fn new(phandle: u64, class: String) -> Self {
-        Self { phandle, class }
+    pub fn new(phandle: u64, class: String, implements: Vec<String>) -> Self {
+        Self {
+            phandle,
+            class,
+            implements,
+        }
     }
 
     fn forward_lifecycle_call(
@@ -294,7 +338,12 @@ impl PhpPluginProxy {
                 snapshot: None,
             }),
         ];
-        let outcome = call_php_method(self.phandle, method, args, Some(&mut PluginRpcDispatcher))?;
+        let outcome = call_php_method(
+            self.phandle,
+            method,
+            args,
+            Some(&mut PluginRpcDispatcher::default()),
+        )?;
         match outcome {
             Ok(_) => Ok(()),
             // TODO(plugin): the original exception class is collapsed to RuntimeException on
@@ -359,11 +408,135 @@ impl PluginInterface for PhpPluginProxy {
         self.class.clone()
     }
 
+    fn as_event_subscriber(&self) -> Option<&dyn EventSubscriberInterface> {
+        if self
+            .implements
+            .iter()
+            .any(|interface| interface == "Composer\\EventDispatcher\\EventSubscriberInterface")
+        {
+            Some(self)
+        } else {
+            None
+        }
+    }
+
     fn __as_php_plugin_proxy(&self) -> Option<&PhpPluginProxy> {
         Some(self)
     }
 }
 
+impl EventSubscriberInterface for PhpPluginProxy {
+    fn get_subscribed_events(&self) -> anyhow::Result<IndexMap<String, SubscribedEventEntry>> {
+        // PHP: `$subscriber->getSubscribedEvents()` (an instance call of the static method).
+        let outcome = call_php_method(
+            self.phandle,
+            "getSubscribedEvents",
+            Vec::new(),
+            Some(&mut PluginRpcDispatcher::default()),
+        )?;
+        let value = match outcome {
+            Ok(value) => value,
+            // TODO(plugin): the original exception class is collapsed to RuntimeException on
+            // this side of the boundary.
+            Err(throw) => {
+                return Err(anyhow::anyhow!(shirabe_php_shim::RuntimeException {
+                    message: throw.message,
+                    code: throw.code,
+                }));
+            }
+        };
+        decode_subscribed_events(&self.class, value)
+    }
+
+    fn subscriber_handle(&self) -> PhpObjHandle {
+        PhpObjHandle {
+            phandle: self.phandle,
+            class: self.class.clone(),
+            implements: self.implements.clone(),
+        }
+    }
+}
+
+/// Decodes a `getSubscribedEvents()` wire value into the three shapes `addSubscriber`
+/// distinguishes: `'method'`, `['method', priority]`, and `[['method', priority], ...]`.
+/// A shape outside these is an explicit error, never a silently dropped listener.
+fn decode_subscribed_events(
+    class: &str,
+    value: PluginValue,
+) -> anyhow::Result<IndexMap<String, SubscribedEventEntry>> {
+    let entries = match value {
+        PluginValue::Array(entries) => entries,
+        PluginValue::List(items) => {
+            // A PHP list-shaped array (integer keys) cannot map event names; an empty one is
+            // the only legal case (an empty PHP array serializes as a list).
+            if items.is_empty() {
+                IndexMap::new()
+            } else {
+                return Err(subscribed_events_shape_error(
+                    class,
+                    &PluginValue::List(items),
+                ));
+            }
+        }
+        other => return Err(subscribed_events_shape_error(class, &other)),
+    };
+    let mut events: IndexMap<String, SubscribedEventEntry> = IndexMap::new();
+    for (event_name, params) in entries {
+        // TODO(phase-e): lossy UTF-8; event and method names are bytes in PHP.
+        let event_name = String::from_utf8_lossy(&event_name).into_owned();
+        let entry = match &params {
+            PluginValue::String(method) => {
+                SubscribedEventEntry::Method(String::from_utf8_lossy(method).into_owned())
+            }
+            PluginValue::List(items) => match items.first() {
+                Some(PluginValue::String(method)) => SubscribedEventEntry::MethodWithPriority(
+                    String::from_utf8_lossy(method).into_owned(),
+                    decode_listener_priority(class, items.get(1))?,
+                ),
+                _ => {
+                    let mut listeners = Vec::with_capacity(items.len());
+                    for listener in items {
+                        let PluginValue::List(pair) = listener else {
+                            return Err(subscribed_events_shape_error(class, &params));
+                        };
+                        let Some(PluginValue::String(method)) = pair.first() else {
+                            return Err(subscribed_events_shape_error(class, &params));
+                        };
+                        listeners.push((
+                            String::from_utf8_lossy(method).into_owned(),
+                            decode_listener_priority(class, pair.get(1))?,
+                        ));
+                    }
+                    SubscribedEventEntry::Methods(listeners)
+                }
+            },
+            _ => return Err(subscribed_events_shape_error(class, &params)),
+        };
+        events.insert(event_name, entry);
+    }
+    Ok(events)
+}
+
+fn decode_listener_priority(
+    class: &str,
+    value: Option<&PluginValue>,
+) -> anyhow::Result<Option<i64>> {
+    match value {
+        None => Ok(None),
+        Some(PluginValue::Int(priority)) => Ok(Some(*priority)),
+        Some(other) => Err(subscribed_events_shape_error(class, other)),
+    }
+}
+
+fn subscribed_events_shape_error(class: &str, value: &PluginValue) -> anyhow::Error {
+    anyhow::anyhow!(shirabe_php_shim::RuntimeException {
+        message: format!(
+            "{class}::getSubscribedEvents() returned an unsupported shape over RPC: {value:?}"
+        ),
+        code: 0,
+    })
+}
+
 impl Drop for PhpPluginProxy {
     fn drop(&mut self) {
         // A dead worker has nothing left to release.