feat(sidecar): stream extension events live during dispatch (#132)

Author: NathanFlurry

CLAUDE.md

@@ -42,7 +42,8 @@ Every bound that protects a shared resource — memory/heap, CPU/wall-clock, fd/
 
 ## Project Boundaries
 
-- Keep this repo Agent OS-agnostic: no ACP, agents, sessions, `agentos-protocol`, `agentos-client`, or `agentos-sidecar` dependencies in secure-exec code.
+- Keep the secure-exec runtime Agent OS-agnostic: no ACP, sessions, `agentos-protocol`, `agentos-client`, or `agentos-sidecar` dependencies in runtime code.
+- Packaged agent definitions/adapters live in `registry/agent/*`; generic VM software lives in `registry/software/*`.
 - `crates/bridge/` is the browser/native portability seam. Shared contracts belong there.
 - `crates/vfs/` is generic filesystem infrastructure: POSIX-style in-memory/overlay/mount/root engines plus generic chunked/object engines and in-memory stores. It must stay free of secure-exec sidecar, bridge, S3, SQLite, and host-disk coupling.
 - `crates/secure-exec-vfs/` contains concrete secure-exec filesystem backends: S3 adapters, host-disk SQLite/file stores, and bridge/callback-backed metadata stores. Policy decisions, config validation, and mount descriptor parsing stay in sidecar plugins.
@@ -80,6 +81,12 @@ Every bound that protects a shared resource — memory/heap, CPU/wall-clock, fd/
 
 ## Development
 
+### Release Tracks
+
+- **secure-exec runtime** — `@secure-exec/*` npm packages and `secure-exec-*` crates; releases keep npm/crates in sync, previews are npm-only. See "Preview-publishing" and "Publishing" for details.
+- **`@agentos-software/*` registry packages** — generic VM software from secure-exec `registry/software/*` plus agent adapters from secure-exec `registry/agent/*`; versioned independently of secure-exec runtime packages.
+- **agent-os product/API** — `@rivet-dev/agentos*`, AgentOs APIs, sidecar wrapper, docs, quickstarts, and examples; see agent-os `CLAUDE.md` for its pinning workflow.
+
 ### Preview-publishing
 
 Dispatch `.github/workflows/publish.yaml` (workflow_dispatch) with no version input to cut a **preview** (debug sidecar build, npm-only, dist-tag = sanitized branch name) — for handing a build to a downstream (agent-os) or external project. **Preview-publish is for previews ONLY; never cut a release with it.** Caveats: WASM-bearing packages (`@secure-exec/core`, `@agentos-software/*`) publish MANUALLY (see Publishing), and the crates.io job is skipped on preview — a *crate* change only reaches consumers locally (path dep / `[patch]`) or via a real release.
@@ -91,7 +98,7 @@ Dispatch `.github/workflows/publish.yaml` (workflow_dispatch) with no version in
 
 ## Publishing
 
-- **The `@secure-exec/*` npm packages and the `secure-exec-*` Cargo crates are always published at the same version** (npm and crates stay in sync), so a downstream pins both to one `<v>`. The `@agentos-software/*` registry software packages are on a **separate** version track.
+- **The `@secure-exec/*` npm packages and the `secure-exec-*` Cargo crates are always published at the same version** (npm and crates stay in sync), so a downstream pins both to one `<v>`. See "Release Tracks" for how this differs from `@agentos-software/*` and agent-os releases.
 - CI (`.github/workflows/publish.yaml`) does NOT build or publish the WASM command binaries. There is no `build-commands` job and nothing restores a `wasm-commands` artifact — the workflow only builds/publishes the sidecar binary and the pure-TS packages.
 - WASM-bearing packages are ALWAYS published MANUALLY: `@secure-exec/core` (which vendors `registry/native` commands into `packages/core/commands` via `copy-wasm-commands.mjs`, guarded by its `prepack --require`) and the `@agentos-software/*` registry software. `@secure-exec/core` is in `EXCLUDED` in `scripts/publish/src/lib/packages.ts`, so CI never publishes it.
 - Manual core flow: build the commands locally (`make -C registry/native wasm`), then `npm publish` (not `pnpm publish`) `@secure-exec/core` at the **same version** CI used for that release so dependents resolving `@secure-exec/core@<version>` succeed. `prepack` vendors the commands and fails loud if they are absent.

crates/sidecar/src/extension.rs

@@ -8,7 +8,7 @@ use crate::protocol::{
     ProcessKilledResponse, ProcessStartedResponse, SidecarRequestPayload, SidecarResponsePayload,
     StdinClosedResponse, StdinWrittenResponse, WriteStdinRequest,
 };
-use crate::state::{SharedSidecarRequestClient, SidecarError};
+use crate::state::{SharedEventSink, SharedSidecarRequestClient, SidecarError};
 
 pub type ExtensionFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, SidecarError>> + 'a>>;
 
@@ -151,6 +151,7 @@ pub struct ExtensionSnapshot {
     namespace: String,
     ownership: OwnershipScope,
     sidecar_requests: SharedSidecarRequestClient,
+    event_sink: SharedEventSink,
 }
 
 pub struct ExtensionContext<'a> {
@@ -163,11 +164,13 @@ impl ExtensionSnapshot {
         namespace: String,
         ownership: OwnershipScope,
         sidecar_requests: SharedSidecarRequestClient,
+        event_sink: SharedEventSink,
     ) -> Self {
         Self {
             namespace,
             ownership,
             sidecar_requests,
+            event_sink,
         }
     }
 
@@ -196,6 +199,28 @@ impl ExtensionSnapshot {
         crate::wire::event_frame_from_compat(self.ext_event(payload)).map_err(wire_protocol_error)
     }
 
+    /// Emit a wire event frame to the host the instant it is produced, rather
+    /// than collecting it into the dispatch result and waiting for the whole
+    /// request to resolve. Returns `Ok(None)` when delivered live, or
+    /// `Ok(Some(event))` when no live sink is configured (in-process sidecar) so
+    /// the caller can fall back to returning it in the dispatch's batch.
+    pub fn emit_event_wire(
+        &self,
+        event: crate::wire::EventFrame,
+    ) -> Result<Option<crate::wire::EventFrame>, SidecarError> {
+        self.event_sink.try_emit(event)
+    }
+
+    /// Build a namespaced ext-event from `payload` and emit it live (see
+    /// [`Self::emit_event_wire`]).
+    pub fn emit_ext_event(
+        &self,
+        payload: Vec<u8>,
+    ) -> Result<Option<crate::wire::EventFrame>, SidecarError> {
+        let event = self.ext_event_wire(payload)?;
+        self.emit_event_wire(event)
+    }
+
     pub fn invoke_callback(
         &self,
         payload: Vec<u8>,
@@ -241,6 +266,24 @@ impl<'a> ExtensionContext<'a> {
         self.snapshot.ext_event_wire(payload)
     }
 
+    /// Emit a wire event frame to the host live (see
+    /// [`ExtensionSnapshot::emit_event_wire`]).
+    pub fn emit_event_wire(
+        &self,
+        event: crate::wire::EventFrame,
+    ) -> Result<Option<crate::wire::EventFrame>, SidecarError> {
+        self.snapshot.emit_event_wire(event)
+    }
+
+    /// Build a namespaced ext-event from `payload` and emit it live (see
+    /// [`ExtensionSnapshot::emit_ext_event`]).
+    pub fn emit_ext_event(
+        &self,
+        payload: Vec<u8>,
+    ) -> Result<Option<crate::wire::EventFrame>, SidecarError> {
+        self.snapshot.emit_ext_event(payload)
+    }
+
     pub fn invoke_callback(
         &self,
         payload: Vec<u8>,
@@ -601,3 +644,84 @@ pub trait Extension: Send + Sync {
         Box::pin(async { Ok(()) })
     }
 }
+
+#[cfg(test)]
+mod live_event_tests {
+    use super::*;
+    use crate::state::EventSinkTransport;
+    use std::sync::{Arc, Mutex};
+
+    /// Records every event handed to the live sink, standing in for the stdio
+    /// `FrameEventTransport` that writes `ProtocolFrame::EventFrame`s to stdout.
+    #[derive(Default)]
+    struct RecordingEventSink {
+        events: Arc<Mutex<Vec<crate::wire::EventFrame>>>,
+    }
+
+    impl EventSinkTransport for RecordingEventSink {
+        fn emit_event(&self, event: crate::wire::EventFrame) -> Result<(), SidecarError> {
+            self.events.lock().unwrap().push(event);
+            Ok(())
+        }
+    }
+
+    fn snapshot_with_sink(event_sink: SharedEventSink) -> ExtensionSnapshot {
+        ExtensionSnapshot::new(
+            String::from("dev.rivet.test.live-event"),
+            OwnershipScope::session("conn-live", "sess-live"),
+            SharedSidecarRequestClient::default(),
+            event_sink,
+        )
+    }
+
+    // With a transport configured (the stdio path), an ext event is emitted live
+    // and `emit_ext_event` reports nothing left to batch.
+    #[test]
+    fn emit_ext_event_streams_live_when_sink_configured() {
+        let recorded = Arc::new(Mutex::new(Vec::new()));
+        let mut sink = SharedEventSink::default();
+        sink.set_transport(Arc::new(RecordingEventSink {
+            events: recorded.clone(),
+        }));
+
+        let snapshot = snapshot_with_sink(sink);
+        let leftover = snapshot
+            .emit_ext_event(b"live-update".to_vec())
+            .expect("emit must succeed");
+
+        assert!(
+            leftover.is_none(),
+            "a configured sink consumes the event live, leaving nothing to batch"
+        );
+        let recorded = recorded.lock().unwrap();
+        assert_eq!(recorded.len(), 1, "the event must reach the live transport");
+        // The live frame round-trips back to the namespaced ext payload.
+        let compat = crate::wire::event_frame_to_compat(recorded[0].clone())
+            .expect("recorded frame converts back to compat");
+        match compat.payload {
+            EventPayload::Ext(envelope) => {
+                assert_eq!(envelope.namespace, "dev.rivet.test.live-event");
+                assert_eq!(envelope.payload, b"live-update");
+            }
+            other => panic!("unexpected live event payload: {other:?}"),
+        }
+    }
+
+    // With no transport (an in-process sidecar), the event is handed back so the
+    // caller can fall back to the dispatch-result batch — preserving delivery.
+    #[test]
+    fn emit_ext_event_falls_back_to_batch_without_sink() {
+        let snapshot = snapshot_with_sink(SharedEventSink::default());
+        let leftover = snapshot
+            .emit_ext_event(b"batched-update".to_vec())
+            .expect("emit must succeed");
+
+        let frame = leftover.expect("without a sink the event is returned for batching");
+        let compat = crate::wire::event_frame_to_compat(frame)
+            .expect("returned frame converts back to compat");
+        match compat.payload {
+            EventPayload::Ext(envelope) => assert_eq!(envelope.payload, b"batched-update"),
+            other => panic!("unexpected batched event payload: {other:?}"),
+        }
+    }
+}

crates/sidecar/src/lib.rs

@@ -28,6 +28,7 @@ pub use extension::{
     ExtensionInterruptResponse, ExtensionResponse,
 };
 pub use service::{DispatchResult, NativeSidecar, NativeSidecarConfig, SidecarError};
+pub use state::EventSinkTransport;
 pub use state::SidecarRequestTransport;
 
 use wire::{DEFAULT_MAX_FRAME_BYTES, PROTOCOL_NAME, PROTOCOL_VERSION};

crates/sidecar/src/service.rs

@@ -30,8 +30,8 @@ use crate::protocol::{
     VmLifecycleState, WriteStdinRequest,
 };
 use crate::state::{
-    ActiveExecutionEvent, BridgeError, ConnectionState, JavascriptSocketFamily,
-    JavascriptSocketPathContext, ProcessEventEnvelope, SessionState, SharedBridge,
+    ActiveExecutionEvent, BridgeError, ConnectionState, EventSinkTransport, JavascriptSocketFamily,
+    JavascriptSocketPathContext, ProcessEventEnvelope, SessionState, SharedBridge, SharedEventSink,
     SharedSidecarRequestClient, SidecarRequestTransport, VmState, EXECUTION_DRIVER_NAME,
 };
 use crate::tools::register_host_callbacks;
@@ -870,6 +870,7 @@ pub struct NativeSidecar<B> {
     pub(crate) pending_sidecar_responses_gauge: Arc<QueueGauge>,
     pub(crate) outbound_sidecar_requests_gauge: Arc<QueueGauge>,
     pub(crate) sidecar_requests: SharedSidecarRequestClient,
+    pub(crate) event_sink: SharedEventSink,
     pub(crate) extensions: BTreeMap<String, Arc<dyn Extension>>,
     pub(crate) extension_sessions: BTreeMap<(String, String), ExtensionSessionResources>,
     pub(crate) extension_process_output_buffers:
@@ -980,6 +981,7 @@ where
                 MAX_OUTBOUND_SIDECAR_REQUESTS,
             ),
             sidecar_requests: SharedSidecarRequestClient::default(),
+            event_sink: SharedEventSink::default(),
             extensions: BTreeMap::new(),
             extension_sessions: BTreeMap::new(),
             extension_process_output_buffers: BTreeMap::new(),
@@ -1162,6 +1164,10 @@ where
         self.sidecar_requests.set_transport(transport);
     }
 
+    pub fn set_event_transport(&mut self, transport: Arc<dyn EventSinkTransport>) {
+        self.event_sink.set_transport(transport);
+    }
+
     pub fn register_extension(
         &mut self,
         extension: Box<dyn Extension>,
@@ -1488,6 +1494,7 @@ where
             namespace.clone(),
             request.ownership.clone(),
             self.sidecar_requests.clone(),
+            self.event_sink.clone(),
         );
         let ctx = ExtensionContext::new(snapshot, self);
         let response = extension.handle_request(ctx, envelope.payload).await?;
@@ -1898,6 +1905,7 @@ where
                 extension.namespace().to_owned(),
                 ownership.clone(),
                 self.sidecar_requests.clone(),
+                self.event_sink.clone(),
             );
             if let Err(error) = extension.on_session_disposed(snapshot).await {
                 if first_error.is_none() {

crates/sidecar/src/state.rs

@@ -198,6 +198,45 @@ impl SharedSidecarRequestClient {
     }
 }
 
+/// Fire-and-forget live event sink. Lets an extension emit a `session/update`
+/// (or any other) event frame to the host *mid-dispatch*, instead of having to
+/// return it from the dispatch and wait for the whole request to resolve before
+/// the stdio loop flushes it. Mirrors `SidecarRequestTransport`, but events have
+/// no response, no request id, and no timeout — they are written to the same
+/// outbound stdout channel the batch path uses.
+pub trait EventSinkTransport: Send + Sync {
+    fn emit_event(&self, event: crate::wire::EventFrame) -> Result<(), SidecarError>;
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct SharedEventSink {
+    transport: Option<Arc<dyn EventSinkTransport>>,
+}
+
+impl SharedEventSink {
+    pub(crate) fn set_transport(&mut self, transport: Arc<dyn EventSinkTransport>) {
+        self.transport = Some(transport);
+    }
+
+    /// Emit `event` live if a transport is configured (the stdio path). Returns
+    /// `Ok(None)` when the event was handed to the live transport, or
+    /// `Ok(Some(event))` when no transport is configured (e.g. an in-process
+    /// `NativeSidecar` with no stdout loop) so the caller can fall back to the
+    /// batch path and still deliver the event when the dispatch resolves.
+    pub(crate) fn try_emit(
+        &self,
+        event: crate::wire::EventFrame,
+    ) -> Result<Option<crate::wire::EventFrame>, SidecarError> {
+        match self.transport.as_ref() {
+            Some(transport) => {
+                transport.emit_event(event)?;
+                Ok(None)
+            }
+            None => Ok(Some(event)),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Bridge wrapper
 // ---------------------------------------------------------------------------

crates/sidecar/src/stdio.rs

@@ -4,8 +4,8 @@ use crate::wire::{
     SidecarResponseFrame, WireDispatchResult, WireFrameCodec,
 };
 use crate::{
-    Extension, ExtensionInterruptRequest, NativeSidecar, NativeSidecarConfig, SidecarError,
-    SidecarRequestTransport,
+    EventSinkTransport, Extension, ExtensionInterruptRequest, NativeSidecar, NativeSidecarConfig,
+    SidecarError, SidecarRequestTransport,
 };
 use secure_exec_bridge::queue_tracker::{tracked_sync_channel, TrackedLimit, TrackedSyncSender};
 use secure_exec_bridge::{
@@ -155,6 +155,12 @@ async fn run_async(extensions: Vec<Box<dyn Extension>>) -> Result<(), Box<dyn Er
     }));
     let callback_transport = Arc::new(FrameSidecarRequestTransport::new(write_tx.clone()));
     sidecar.set_sidecar_request_transport(callback_transport.clone());
+    // Live event sink: lets an extension stream `session/update` (and other)
+    // events to stdout mid-dispatch instead of batching them until the request
+    // resolves. Shares the same outbound `write_tx` channel as the batch path, so
+    // ordering and backpressure are identical.
+    let event_transport = Arc::new(FrameEventTransport::new(write_tx.clone()));
+    sidecar.set_event_transport(event_transport);
     let mut event_pump = time::interval(EVENT_PUMP_INTERVAL);
     let writer_codec = codec.clone();
     let reader_codec = codec.clone();
@@ -1381,6 +1387,28 @@ impl SessionScope {
     }
 }
 
+/// Live event sink backed by the outbound stdout channel. Writes each event as a
+/// `ProtocolFrame::EventFrame` immediately, using the same blocking
+/// backpressure semantics as the batch event path (`send_output_frame`): a full
+/// queue parks the producer until the writer thread drains stdout rather than
+/// tearing down the process.
+struct FrameEventTransport {
+    writer: TrackedSyncSender<ProtocolFrame>,
+}
+
+impl FrameEventTransport {
+    fn new(writer: TrackedSyncSender<ProtocolFrame>) -> Self {
+        Self { writer }
+    }
+}
+
+impl EventSinkTransport for FrameEventTransport {
+    fn emit_event(&self, event: crate::wire::EventFrame) -> Result<(), SidecarError> {
+        send_output_frame(&self.writer, ProtocolFrame::EventFrame(event))
+            .map_err(|error| SidecarError::Bridge(error.to_string()))
+    }
+}
+
 struct FrameSidecarRequestTransport {
     writer: TrackedSyncSender<ProtocolFrame>,
     pending: Arc<Mutex<BTreeMap<RequestId, mpsc::SyncSender<SidecarResponseFrame>>>>,

package.json

@@ -19,5 +19,11 @@
 		"turbo": "^2.5.6",
 		"typescript": "^5.9.2",
 		"vitest": "^2.1.8"
+	},
+	"pnpm": {
+		"overrides": {
+			"@mariozechner/pi-tui": "0.60.0",
+			"@mariozechner/pi-agent-core": "0.60.0"
+		}
 	}
 }