feat(actor): resolve options from actor create input

crates/agentos-actor-plugin/src/config.rs

@@ -123,6 +123,14 @@ impl AgentOsConfigJson {
         serde_json::from_str(config_json).context("agent-os config JSON parse error")
     }
 
+    /// Decode the validated per-instance options returned by the TypeScript
+    /// host overlay. RivetKit transports these as CBOR bytes separately from
+    /// the actor's public create input.
+    pub(crate) fn parse_instance_options(bytes: &[u8]) -> Result<Self> {
+        ciborium::from_reader(std::io::Cursor::new(bytes))
+            .context("agent-os instance options CBOR parse error")
+    }
+
     /// Build a fresh [`AgentOsConfig`] (non-`Clone`, so rebuilt per bring-up).
     ///
     /// `fallback_pool` is the per-plugin-runtime sidecar pool used when the
@@ -206,6 +214,47 @@ impl AgentOsConfigJson {
     }
 }
 
+#[cfg(test)]
+mod tests {
+    use super::AgentOsConfigJson;
+
+    #[test]
+    fn decodes_per_instance_options_from_cbor() {
+        let mut bytes = Vec::new();
+        ciborium::into_writer(
+            &serde_json::json!({
+                "additionalInstructions": "tenant-specific",
+                "loopbackExemptPorts": [4100]
+            }),
+            &mut bytes,
+        )
+        .expect("encode options");
+
+        let config =
+            AgentOsConfigJson::parse_instance_options(&bytes).expect("decode per-instance options");
+        assert_eq!(
+            config.additional_instructions.as_deref(),
+            Some("tenant-specific")
+        );
+        assert_eq!(config.loopback_exempt_ports, vec![4100]);
+    }
+
+    #[test]
+    fn rejects_unknown_per_instance_options() {
+        let mut bytes = Vec::new();
+        ciborium::into_writer(
+            &serde_json::json!({ "notAnAgentOsOption": true }),
+            &mut bytes,
+        )
+        .expect("encode options");
+
+        let error = AgentOsConfigJson::parse_instance_options(&bytes)
+            .err()
+            .expect("unknown options must fail");
+        assert!(error.to_string().contains("instance options CBOR"));
+    }
+}
+
 /// Read a projected package into a [`SoftwareInfoDto`] (sans `commands`, which
 /// are filled from the live VM in the dispatch arm). Returns `None` if the
 /// package manifest is missing or malformed.

crates/agentos-actor-plugin/src/host_ctx.rs

@@ -1,185 +1,104 @@
-//! Plugin-side host bridge — the inverse of the RivetKit host vtable impl.
+//! Thin AgentOS-facing wrapper around RivetKit's portable actor context.
 //!
-//! Wraps the `HostVtable` the host hands to `rivet_actor_run` and exposes safe
-//! async/sync methods the actor run loop calls: durable storage (`db_*`), event
-//! pull (`next_event`), replies, broadcast. Each async op is a sync submit +
-//! completion callback bridged to an `await` via a oneshot (spec §5.4), driven
-//! on the plugin runtime. Depends only on `rivet-actor-plugin-abi` + `tokio`
-//! (no `agentos-client`), so it builds independently of the secure-exec layer.
+//! RivetKit owns the ABI bridge, completion channels, context refcounts, and
+//! pushed-event queue. AgentOS keeps only the small string/byte adaptations its
+//! persistence and action modules use.
 
 #![allow(dead_code)]
 
-use std::ffi::c_void;
+use anyhow::anyhow;
+use rivet_actor_plugin_abi::{DylibBackend, Event, PortableActorCtx, ReplyToken};
 
-use rivet_actor_plugin_abi as abi;
-use tokio::sync::oneshot;
-
-/// `Send` wrapper so an `AbiResult` (which holds raw pointers) can travel
-/// through the oneshot channel and the spawned actor future stays `Send`.
-struct SendResult(abi::AbiResult);
-unsafe impl Send for SendResult {}
-
-/// Refcounted handle to the host actor context. `Clone` bumps the host refcount
-/// (`ctx_clone`) so detached tasks can hold it; `Drop` releases it.
+#[derive(Clone)]
 pub(crate) struct HostCtx {
-    vtable: abi::HostVtable,
-}
-
-unsafe impl Send for HostCtx {}
-unsafe impl Sync for HostCtx {}
-
-impl Clone for HostCtx {
-    fn clone(&self) -> Self {
-        let mut v = self.vtable;
-        v.ctx = (self.vtable.ctx_clone)(self.vtable.ctx);
-        Self { vtable: v }
-    }
-}
-
-impl Drop for HostCtx {
-    fn drop(&mut self) {
-        (self.vtable.ctx_release)(self.vtable.ctx);
-    }
-}
-
-/// Completion callback the host invokes when an async op finishes: reclaims the
-/// boxed oneshot sender and delivers the result. Panic-firewalled.
-extern "C" fn complete(user_data: *mut c_void, result: abi::AbiResult) {
-    let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| unsafe {
-        let tx = Box::from_raw(user_data as *mut oneshot::Sender<SendResult>);
-        let _ = tx.send(SendResult(result));
-    }));
-}
-
-fn decode_result(result: abi::AbiResult) -> Result<Vec<u8>, String> {
-    match result.status {
-        abi::AbiStatus::Ok => Ok(unsafe { result.payload.into_vec() }),
-        _ => {
-            let bytes = unsafe { result.payload.into_vec() };
-            Err(String::from_utf8_lossy(&bytes).into_owned())
-        }
-    }
-}
-
-/// Decode a `next_event` payload: `[tag u32 LE][reply_token u64 LE][bytes]`.
-fn decode_event(bytes: &[u8]) -> Option<(u32, u64, Vec<u8>)> {
-    if bytes.len() < 12 {
-        return None;
-    }
-    let tag = u32::from_le_bytes(bytes[0..4].try_into().ok()?);
-    let token = u64::from_le_bytes(bytes[4..12].try_into().ok()?);
-    Some((tag, token, bytes[12..].to_vec()))
+    inner: PortableActorCtx,
 }
 
 impl HostCtx {
-    /// Adopt a strong ref to the host ctx handed in via `rivet_actor_run`'s
-    /// `HostVtable`. The host retains and releases its own ref independently
-    /// after run cleanup, so the plugin clone balances this handle's `Drop`.
-    pub(crate) fn from_vtable(vtable: abi::HostVtable) -> Self {
-        (vtable.ctx_clone)(vtable.ctx);
-        Self { vtable }
-    }
-
-    fn ctx(&self) -> *const c_void {
-        self.vtable.ctx
+    pub(crate) fn from_backend(backend: DylibBackend) -> Self {
+        Self {
+            inner: PortableActorCtx::new_dylib(backend),
+        }
     }
 
     pub(crate) async fn db_exec(&self, sql: Vec<u8>) -> Result<Vec<u8>, String> {
-        let (tx, rx) = oneshot::channel::<SendResult>();
-        let ud = Box::into_raw(Box::new(tx)) as *mut c_void;
-        (self.vtable.db_exec)(self.ctx(), abi::OwnedBuf::from_vec(sql), complete, ud);
-        decode_result(
-            rx.await
-                .map(|r| r.0)
-                .unwrap_or_else(|_| abi::AbiResult::channel_closed()),
-        )
+        let sql = std::str::from_utf8(&sql).map_err(|error| format!("sql utf8: {error}"))?;
+        self.inner
+            .db_exec(sql)
+            .await
+            .map_err(|error| format!("{error:#}"))
     }
 
     pub(crate) async fn db_query(&self, sql: Vec<u8>, params: Vec<u8>) -> Result<Vec<u8>, String> {
-        self.submit_sql(self.vtable.db_query, sql, params).await
+        let sql = std::str::from_utf8(&sql).map_err(|error| format!("sql utf8: {error}"))?;
+        self.inner
+            .db_query(sql, (!params.is_empty()).then_some(params))
+            .await
+            .map_err(|error| format!("{error:#}"))
     }
 
     pub(crate) async fn db_run(&self, sql: Vec<u8>, params: Vec<u8>) -> Result<Vec<u8>, String> {
-        self.submit_sql(self.vtable.db_run, sql, params).await
-    }
-
-    async fn submit_sql(
-        &self,
-        f: abi::DbSqlFn,
-        sql: Vec<u8>,
-        params: Vec<u8>,
-    ) -> Result<Vec<u8>, String> {
-        let (tx, rx) = oneshot::channel::<SendResult>();
-        let ud = Box::into_raw(Box::new(tx)) as *mut c_void;
-        f(
-            self.ctx(),
-            abi::OwnedBuf::from_vec(sql),
-            abi::OwnedBuf::from_vec(params),
-            complete,
-            ud,
-        );
-        decode_result(
-            rx.await
-                .map(|r| r.0)
-                .unwrap_or_else(|_| abi::AbiResult::channel_closed()),
-        )
+        let sql = std::str::from_utf8(&sql).map_err(|error| format!("sql utf8: {error}"))?;
+        self.inner
+            .db_run(sql, (!params.is_empty()).then_some(params))
+            .await
+            .map(|()| Vec::new())
+            .map_err(|error| format!("{error:#}"))
     }
 
-    /// Pull the next lifecycle event, or `None` when the stream is closed.
-    pub(crate) async fn next_event(&self) -> Option<(u32, u64, Vec<u8>)> {
-        let (tx, rx) = oneshot::channel::<SendResult>();
-        let ud = Box::into_raw(Box::new(tx)) as *mut c_void;
-        (self.vtable.next_event)(self.ctx(), complete, ud);
-        let result = rx
-            .await
-            .map(|r| r.0)
-            .unwrap_or_else(|_| abi::AbiResult::channel_closed());
-        match result.status {
-            abi::AbiStatus::Ok => {
-                let bytes = unsafe { result.payload.into_vec() };
-                decode_event(&bytes)
-            }
-            _ => {
-                unsafe { result.payload.free_self() };
+    pub(crate) async fn next_event(&self) -> Option<Event> {
+        match self.inner.next_event().await {
+            Ok(event) => event,
+            Err(error) => {
+                self.log_warn(&format!("native actor event stream failed: {error:#}"));
                 None
             }
         }
     }
 
     pub(crate) fn sql_is_enabled(&self) -> bool {
-        (self.vtable.sql_is_enabled)(self.ctx()) != 0
+        self.inner.sql_is_enabled()
     }
 
-    /// Signal actor startup to the host (required: the native-plugin factory is
-    /// built with manual startup-ready, so the host's `start()` caller awaits
-    /// this before the actor is considered live). `ok = false` reports a fatal
-    /// startup error with `msg`.
-    pub(crate) fn startup_ready(&self, ok: bool, msg: &str) {
-        let err = abi::BorrowedBuf::from_slice(msg.as_bytes());
-        (self.vtable.startup_ready)(self.ctx(), u8::from(ok), err);
+    pub(crate) fn startup_ready(&self, ok: bool, message: &str) {
+        let result = if ok {
+            Ok(())
+        } else {
+            Err(anyhow!(message.to_owned()))
+        };
+        if let Err(error) = self.inner.startup_ready(result) {
+            self.log_warn(&format!("failed to signal actor startup: {error:#}"));
+        }
     }
 
-    pub(crate) fn reply_ok(&self, token: u64, payload: Vec<u8>) -> abi::AbiStatus {
-        (self.vtable.reply_ok)(self.ctx(), token, abi::OwnedBuf::from_vec(payload))
+    pub(crate) fn reply_ok(&self, token: u64, payload: Vec<u8>) {
+        if let Err(error) = self.inner.reply_ok(ReplyToken(token), payload) {
+            self.log_warn(&format!("failed to complete actor reply: {error:#}"));
+        }
     }
 
-    pub(crate) fn reply_err(&self, token: u64, msg: &str) -> abi::AbiStatus {
-        (self.vtable.reply_err)(
-            self.ctx(),
-            token,
-            abi::OwnedBuf::from_vec(msg.as_bytes().to_vec()),
-        )
+    pub(crate) fn reply_err(&self, token: u64, message: &str) {
+        if let Err(error) = self.inner.reply_err(ReplyToken(token), message) {
+            self.log_warn(&format!("failed to complete actor error reply: {error:#}"));
+        }
     }
 
-    pub(crate) fn broadcast(&self, name: Vec<u8>, payload: Vec<u8>) -> abi::AbiStatus {
-        (self.vtable.broadcast)(
-            self.ctx(),
-            abi::OwnedBuf::from_vec(name),
-            abi::OwnedBuf::from_vec(payload),
-        )
+    pub(crate) fn broadcast(&self, name: Vec<u8>, payload: Vec<u8>) {
+        let name = match String::from_utf8(name) {
+            Ok(name) => name,
+            Err(error) => {
+                self.log_warn(&format!(
+                    "failed to broadcast non-UTF-8 event name: {error}"
+                ));
+                return;
+            }
+        };
+        if let Err(error) = self.inner.broadcast(name, payload) {
+            self.log_warn(&format!("failed to broadcast actor event: {error:#}"));
+        }
     }
 
-    pub(crate) fn log_warn(&self, msg: &str) {
-        (self.vtable.log)(self.ctx(), 3, abi::BorrowedBuf::from_slice(msg.as_bytes()));
+    pub(crate) fn log_warn(&self, message: &str) {
+        self.inner.log(3, message);
     }
 }