3 files changed, 81 insertions, 9 deletions

diff --git a/crates/shirabe-php-rpc/src/env.rs b/crates/shirabe-php-rpc/src/env.rs
new file mode 100644
index 00000000..1655168f
--- /dev/null
+++ b/crates/shirabe-php-rpc/src/env.rs
@@ -0,0 +1,57 @@
+//! Propagation of environment mutations into the PHP worker.
+//!
+//! The worker is a long-lived child: it holds the environment it was handed at spawn, so a
+//! `putenv()` or a `$_ENV`/`$_SERVER` write the Rust side makes afterwards is invisible to the
+//! PHP code running in it — `@putenv` in a `scripts` entry, the bin dir the event dispatcher
+//! prepends to `PATH`, `COMPOSER_DEV_MODE`. The shim records every such write; this module
+//! replays the ones the worker has not seen yet, before the next call crosses the boundary.
+//!
+//! TODO(php-runtime): the reverse direction is missing — a write PHP code makes to its own
+//! environment is not reflected back into the Rust-side storages.
+
+use crate::PluginValue;
+use shirabe_php_shim::EnvStorageKind;
+use std::os::unix::ffi::OsStrExt as _;
+
+/// How many recorded mutations the worker has already been told about.
+static SYNCED: std::sync::Mutex<usize> = std::sync::Mutex::new(0);
+
+/// The name `__shirabe_sync_env` reads each mutation's target storage under.
+fn storage_name(storage: EnvStorageKind) -> &'static str {
+    match storage {
+        EnvStorageKind::Process => "process",
+        EnvStorageKind::Env => "_ENV",
+        EnvStorageKind::Server => "_SERVER",
+    }
+}
+
+/// Replays the environment mutations recorded since the last flush into the worker.
+///
+/// Called from the outermost `rpc_call` only, so the `__shirabe_sync_env` call it issues re-enters
+/// the session instead of recursing into another flush.
+pub(crate) fn flush() -> anyhow::Result<()> {
+    let mut synced = SYNCED.lock().unwrap();
+    let (cursor, mutations) = shirabe_php_shim::env_mutations_since(*synced);
+    if mutations.is_empty() {
+        return Ok(());
+    }
+
+    let mutations = mutations
+        .into_iter()
+        .map(|mutation| {
+            PluginValue::List(vec![
+                PluginValue::string(storage_name(mutation.storage)),
+                PluginValue::String(mutation.key.as_bytes().to_vec()),
+                match mutation.value {
+                    Some(value) => PluginValue::String(value.as_bytes().to_vec()),
+                    None => PluginValue::Null,
+                },
+            ])
+        })
+        .collect();
+    let outcome = crate::call_function("__shirabe_sync_env", vec![PluginValue::List(mutations)])?;
+    // The worker only writes the three storages; nothing it does there can throw.
+    outcome.unwrap_or_else(|throw| panic!("PHP RPC: __shirabe_sync_env threw: {}", throw.message));
+    *synced = cursor;
+    Ok(())
+}
diff --git a/crates/shirabe-php-rpc/src/lib.rs b/crates/shirabe-php-rpc/src/lib.rs
index d818b364..47304ba8 100644
--- a/crates/shirabe-php-rpc/src/lib.rs
+++ b/crates/shirabe-php-rpc/src/lib.rs
@@ -1,6 +1,7 @@
 //! Rust-to-PHP RPC over a Unix domain socket. See `docs/dev/php-rpc.md`.
 
 pub mod composer_runtime;
+mod env;
 pub mod frame;
 pub mod session;
 pub mod value;
@@ -780,7 +781,10 @@ fn rpc_call(
 ) -> anyhow::Result<Result<PluginValue, PhpThrow>> {
     // Held for the whole logical call session; nested calls from the same thread (issued by a
     // dispatcher handler) re-enter immediately, other threads are serialized.
-    let _session = session::SessionGuard::enter();
+    let session = session::SessionGuard::enter();
+    if session.is_outermost() {
+        env::flush()?;
+    }
     let my_id = NEXT_CORR_ID.fetch_add(2, Ordering::Relaxed);
     send_frame(&request(my_id))?;
     loop {
diff --git a/crates/shirabe-php-rpc/src/session.rs b/crates/shirabe-php-rpc/src/session.rs
index 29bd89f3..3e1a817d 100644
--- a/crates/shirabe-php-rpc/src/session.rs
+++ b/crates/shirabe-php-rpc/src/session.rs
@@ -16,18 +16,19 @@ struct SessionLock {
 }
 
 impl SessionLock {
-    fn acquire(&self) {
+    /// Returns whether this acquisition opened the session rather than re-entering one.
+    fn acquire(&self) -> bool {
         let mut owner = self.owner.lock().unwrap();
         let me = std::thread::current().id();
         loop {
             match *owner {
                 None => {
                     *owner = Some((me, 1));
-                    return;
+                    return true;
                 }
                 Some((tid, depth)) if tid == me => {
                     *owner = Some((me, depth + 1));
-                    return;
+                    return false;
                 }
                 Some(_) => {
                     owner = self.cvar.wait(owner).unwrap();
@@ -60,12 +61,20 @@ static SESSION: LazyLock<SessionLock> = LazyLock::new(|| SessionLock {
 
 /// RAII guard; acquired once at the outermost `rpc_call`, re-entered (depth += 1, no blocking)
 /// by nested calls from the same thread.
-pub struct SessionGuard;
+pub struct SessionGuard {
+    outermost: bool,
+}
 
 impl SessionGuard {
     pub fn enter() -> Self {
-        SESSION.acquire();
-        SessionGuard
+        let outermost = SESSION.acquire();
+        SessionGuard { outermost }
+    }
+
+    /// Whether this guard opened the session. Work that must happen once per logical call
+    /// session, before anything else crosses the boundary, keys off this.
+    pub fn is_outermost(&self) -> bool {
+        self.outermost
     }
 }
 
@@ -81,8 +90,10 @@ mod tests {
 
     #[test]
     fn same_thread_reenters_without_blocking() {
-        let _outer = SessionGuard::enter();
-        let _inner = SessionGuard::enter();
+        let outer = SessionGuard::enter();
+        let inner = SessionGuard::enter();
+        assert!(outer.is_outermost());
+        assert!(!inner.is_outermost());
     }
 
     #[test]