From e2e5b6a0303e8b50cc0a2880421faf49c804af63 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Wed, 15 Jul 2026 00:31:16 -0700 Subject: [PATCH 1/3] docs(actor): plan native plugin host overlays --- .../native-plugin-host-overlay-plan.md | 347 ++++++++++++++++++ 1 file changed, 347 insertions(+) create mode 100644 docs/features/native-plugin-host-overlay-plan.md diff --git a/docs/features/native-plugin-host-overlay-plan.md b/docs/features/native-plugin-host-overlay-plan.md new file mode 100644 index 0000000000..44e6312715 --- /dev/null +++ b/docs/features/native-plugin-host-overlay-plan.md @@ -0,0 +1,347 @@ +# Native Plugin Host Overlay Implementation Plan + +Status: accepted for implementation + +Thread: `019f6472-e8f9-7e03-82ad-2cc0fb1ac2b2` + +## Outcome + +AgentOS remains a Rust-backed Rivet actor with direct SQLite and VM access, +while gaining the same TypeScript configuration, connection lifecycle, +authorization, custom-action, event-subscription, and callback facilities as a +normal RivetKit actor. + +The implementation must produce one actor, one `ActorFactory`, one database, +one connection set, and one lifecycle. TypeScript is an optional host overlay +around the native backend; it is not a second actor and does not proxy native +database or filesystem operations. + +Durable session resume is explicitly excluded. It is owned by a separate PR. + +## Current Architecture + +RivetKit currently chooses one of two mutually exclusive factory paths: + +```text +normal actor(...) + -> buildNativeFactory + -> NapiActorFactory(callbacks, config) + -> NAPI event loop invokes TypeScript + +nativeFactoryBuilder + -> createNativePluginFactory + -> NapiActorFactory::from_native_plugin + -> CallbackBindings::empty() + -> dylib pulls and handles every actor event +``` + +AgentOS constructs a minimal `actor({ actions: {} })` definition and attaches a +`nativeFactoryBuilder`. The marker bypasses normal callback construction, so +the public AgentOS config can accept callbacks that are never bound. + +The dylib ABI also makes the plugin own a pull-based event loop. The host moves +each non-cloneable actor reply into a token slab, the plugin pulls an event with +`next_event`, and later completes the token with `reply_ok` or `reply_err`. +Wrapping that flow in TypeScript would require proxy replies and duplicate +cancellation/lifecycle machinery. + +## Target Architecture + +RivetKit owns the universal actor event loop and pushes events into a native +backend: + +```text +ActorDefinition + -> host callback overlay (optional) + -> native backend descriptor (optional) + -> one composed Core ActorFactory + +ActorEvent + -> optional host pre-hook + -> host action when explicitly registered, otherwise native backend + -> optional host response hook + -> original actor reply +``` + +The generic native-plugin ABI remains independent of TypeScript and AgentOS. +NAPI adapts JavaScript thread-safe functions into generic Rust callback +closures. Existing native plugins may use an empty overlay. + +## Ownership Rules + +### RivetKit owns + +- Actor mailbox admission and event ordering. +- Original reply ownership and exactly-once completion. +- Callback timeouts, cancellation, and shutdown grace. +- Connection parameter validation and connection-state installation. +- Host action routing and native fallback. +- Host pre/post hooks and subscription authorization. +- Loading and ABI validation of native plugins. + +### The native AgentOS backend owns + +- VM creation and shutdown. +- AgentOS native action implementations. +- Direct actor SQLite access and AgentOS durable tables. +- Filesystem, processes, shells, previews, sessions, and cron behavior. +- Native background workers and event pumps. +- Native sleep/destroy cleanup. + +### The TypeScript AgentOS overlay owns + +- Per-instance `createOptions(c, input)` evaluation. +- Connection schemas and authentication hooks. +- Derived connection principal/state. +- Pre-action authorization. +- Custom TypeScript actions. +- Subscription policy. +- Server-side session and permission callbacks. +- JavaScript bindings invoked from the native backend. + +### Initially unsupported composition conflicts + +- Duplicate TypeScript/native action names are rejected. +- Native AgentOS remains the serialized actor-state owner. +- Native AgentOS remains the raw preview HTTP owner. +- A TypeScript `run`/workflow implementation is not composed into the AgentOS + backend in this change. +- Arbitrary JavaScript filesystem drivers remain unsupported across the native + boundary; native mount plugins remain supported. + +## Event Semantics + +| Event | Required order | +| --- | --- | +| Actor start | Validate input, evaluate `createOptions`, create native instance, signal ready | +| Connection preflight | Validate params, `onBeforeConnect`, `createConnState`, native preflight | +| Connection open | Native open, then `onConnect` | +| Connection close | Invoke both host and native cleanup; one failure must not skip the other | +| Subscription | Event `canSubscribe`, then native subscription handling | +| Action | `onBeforeAction`, TypeScript action or native fallback, `onBeforeActionResponse` | +| Sleep/destroy | Stop admission, cancel or drain in-flight work, run host cleanup and native cleanup within one grace deadline | +| Serialization | Native AgentOS owner for this change | + +Scheduled actions have no invoking connection. Authorization hooks receive +that absence explicitly and must choose a system-principal policy rather than +receiving a fabricated connection. + +## Native Plugin ABI Precursor + +### Single exported descriptor + +Replace independent `dlsym` lookups with one exported function returning a +versioned function table: + +```rust +#[repr(C)] +pub struct PluginApi { + pub abi_magic: u64, + pub abi_version: u64, + pub struct_size: usize, + pub plugin_init: PluginInitFn, + pub factory_new: FactoryNewFn, + pub factory_free: FactoryFreeFn, + pub instance_new: InstanceNewFn, + pub handle_event: HandleEventFn, + pub cancel_event: CancelEventFn, + pub shutdown: ShutdownFn, + pub instance_free: InstanceFreeFn, +} +``` + +The host must continue keeping loaded libraries alive for the process lifetime. +Unloading code while plugin-created tasks or copied function pointers remain is +unsafe. + +### Host-driven events + +Remove `next_event`, reply tokens, `reply_ok`, and `reply_err` from the plugin +contract. `handle_event` receives one encoded event and completes that event's +callback directly. Plugins may enqueue work internally and complete later. + +The host submits events in actor order but may have multiple events in flight. +Shutdown is a barrier: it closes admission, applies the actor grace deadline, +and guarantees that every admitted completion resolves exactly once. + +### Per-instance startup data + +The instance-start payload carries actor input and the validated result of +`createOptions`. Static package/plugin data may remain on the factory config. + +### Plugin-to-host calls + +Add one bounded asynchronous host-call primitive to the host vtable: + +```rust +host_call(ctx, name, payload, done, user_data) +``` + +Names are registered when the composed factory is built. Unknown names fail +with a typed error. Payload and response sizes, callback duration, and +concurrent calls are bounded. Host-call cancellation participates in actor +shutdown. + +This primitive backs AgentOS permission decisions, session-event observers, +bindings, agent stderr, limit warnings, and future native-originated callbacks. + +### ABI compatibility + +Any function-table or wire change bumps the exact-lockstep actor-plugin ABI. +The host rejects old plugins with a precise version mismatch. No fallback to an +older dispatch mode is added. + +## TypeScript API + +The AgentOS definition becomes generic over actor input, connection params, and +connection state: + +```ts +agentOS({ + createOptions(c, input) { + return { /* serializable AgentOS options */ }; + }, + connParamsSchema, + createConnState(c, params) { + return { userId, tenantId, roles }; + }, + onBeforeConnect(c, params) {}, + onConnect(c, conn) {}, + onDisconnect(c, conn) {}, + onBeforeAction(c, name, args) {}, + actions: { + customAction(c, value) {}, + }, + events: { + permissionRequest: event({ canSubscribe(c) { return canApprove(c); } }), + }, +}); +``` + +`createOptions` is actor-instance configuration. The one-shot +`AgentOs.create(options)` constructor continues accepting only concrete +`AgentOsOptions` and does not accept an option factory. + +Authentication secrets are validated at connection time and are not persisted +as connection state. `createConnState` stores a derived principal. The actor key +is the initial tenancy boundary: all admitted connections to one actor share +its native event domain unless an event subscription policy denies access. + +## Permission Callback Contract + +The server permission hook returns a structured decision: + +```ts +type PermissionDecision = + | { handled: true; reply: "once" | "always" | "reject" } + | { handled: false }; +``` + +`handled: false` forwards the request to authorized client subscribers. Exactly +one response wins. Duplicate or late responses fail deterministically rather +than being silently ignored. + +## Planned jj Revisions + +### RivetKit repository + +1. `refactor(plugin): load actor dylibs through one API descriptor` + - Introduce the single exported `PluginApi` descriptor. + - Preserve current pull dispatch temporarily. + - Add loader validation and descriptor lifetime tests. + +2. `feat(plugin): make native actor dispatch host-driven` + - Add per-instance start and pushed event handling. + - Remove the event pull/reply-token slab from the new ABI. + - Define concurrency, cancellation, startup, and shutdown behavior. + +3. `feat(actor): compose host callbacks with native backends` + - Build callbacks even when a definition has a native backend. + - Route lifecycle hooks, subscription policy, custom actions, and native + fallback through one actor factory. + - Preserve the existing standalone native-plugin path as an empty overlay. + +4. `feat(actor): add native pre-action and host-call callbacks` + - Add `onBeforeAction` to the generic actor API. + - Add bounded named plugin-to-host callbacks. + - Propagate caller connection and request metadata. + +### AgentOS repository + +5. `chore(actor): adopt the host-driven native plugin ABI` + - Port the AgentOS plugin exports and worker to pushed events. + - Preserve direct SQLite, VM, action, and event behavior. + +6. `feat(actor): resolve options from actor create input` + - Replace the obsolete pre-create shape with `createOptions(c, input)`. + - Add input/options validation and startup failure tests. + +7. `feat(actor): add connection and action authorization hooks` + - Add connection param schema/state generics and lifecycle hooks. + - Add `onBeforeAction` enforcement for every native action. + - Add event subscription authorization. + +8. `feat(actor): compose custom TypeScript actions with native actions` + - Route registered TypeScript actions through the host overlay. + - Fall back to the native action contract for other names. + - Reject collisions and preserve generated native action types. + +9. `feat(actor): route native callbacks and bindings through the host` + - Implement structured session and permission callbacks. + - Route JavaScript bindings through the generic host-call primitive. + - Add limits, timeout, error propagation, and shutdown tests. + +10. `docs(actor): document native overlays and authentication` + - Remove interim-runtime/stub language. + - Update authentication, approvals, multiplayer, bindings, architecture, + and API examples. + - Document the actor tenancy boundary and event visibility. + +Each revision must be independently described and leave its repository buildable +against the corresponding precursor revision. Cross-repository dependency pins +are updated only to published or otherwise reproducible RivetKit artifacts; +local absolute paths are never committed. + +## Validation + +### RivetKit + +- Actor-plugin ABI unit and loader tests. +- Native plugin lifecycle/action/connection integration tests. +- Existing NAPI actor hook and action tests. +- New composed native-backend tests covering empty and populated overlays. +- Cancellation, dropped-callback, timeout, and shutdown-grace tests. +- TypeScript package type checks and NAPI build. + +### AgentOS + +- `agentos-actor-plugin` unit and action-contract tests. +- AgentOS package type checks and config-schema tests. +- Authentication example type check and connection rejection/acceptance test. +- Native action authorization and custom TypeScript action integration tests. +- Permission hook handled/forwarded/timeout/duplicate-response tests. +- Binding invocation success, error, timeout, payload-limit, and shutdown tests. +- Existing filesystem, process, shell, preview, cron, persistence, and inspector + regressions. +- Website build after documentation changes. + +### Cross-repository handoff + +The AgentOS repository currently consumes published RivetKit preview packages +and an exact actor-plugin ABI crate. Implementation and local validation may use +a clean linked RivetKit build, but the final AgentOS dependency revision must +point to a reproducible published preview rather than a developer filesystem. + +## Completion Criteria + +- AgentOS config callbacks are either executed or rejected by validation; none + are silently dropped. +- Native actions retain direct Rust/SQLite/VM execution. +- `createOptions` is evaluated once per actor instance with actor input. +- Connection and action authorization receive the real invoking context. +- Custom TypeScript actions and native actions coexist under one client type. +- Permission callbacks have one structured and bounded response path. +- Bindings use the same generic host-call mechanism. +- Existing empty-overlay dylib actors remain supported. +- Documentation and checked examples describe the implemented behavior. +- Durable session resume code is untouched. From ec03318602c6c043d860377808f2426cfe89e837 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Wed, 15 Jul 2026 02:08:04 -0700 Subject: [PATCH 2/3] chore(actor): adopt the host-driven native plugin ABI --- crates/agentos-actor-plugin/src/host_ctx.rs | 213 +++----- crates/agentos-actor-plugin/src/lib.rs | 471 +++++++++++------- .../src/persistence_e2e.rs | 34 +- 3 files changed, 386 insertions(+), 332 deletions(-) diff --git a/crates/agentos-actor-plugin/src/host_ctx.rs b/crates/agentos-actor-plugin/src/host_ctx.rs index 314b12f0e6..538e836b8b 100644 --- a/crates/agentos-actor-plugin/src/host_ctx.rs +++ b/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); - let _ = tx.send(SendResult(result)); - })); -} - -fn decode_result(result: abi::AbiResult) -> Result, 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)> { - 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) -> Result, String> { - let (tx, rx) = oneshot::channel::(); - 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, params: Vec) -> Result, 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, params: Vec) -> Result, String> { - self.submit_sql(self.vtable.db_run, sql, params).await - } - - async fn submit_sql( - &self, - f: abi::DbSqlFn, - sql: Vec, - params: Vec, - ) -> Result, String> { - let (tx, rx) = oneshot::channel::(); - 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)> { - let (tx, rx) = oneshot::channel::(); - 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 { + 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) -> abi::AbiStatus { - (self.vtable.reply_ok)(self.ctx(), token, abi::OwnedBuf::from_vec(payload)) + pub(crate) fn reply_ok(&self, token: u64, payload: Vec) { + 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, payload: Vec) -> abi::AbiStatus { - (self.vtable.broadcast)( - self.ctx(), - abi::OwnedBuf::from_vec(name), - abi::OwnedBuf::from_vec(payload), - ) + pub(crate) fn broadcast(&self, name: Vec, payload: Vec) { + 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); } } diff --git a/crates/agentos-actor-plugin/src/lib.rs b/crates/agentos-actor-plugin/src/lib.rs index 5cadc455b5..f38bcd8f85 100644 --- a/crates/agentos-actor-plugin/src/lib.rs +++ b/crates/agentos-actor-plugin/src/lib.rs @@ -7,8 +7,8 @@ //! **unmodified** `agentos-client` to spawn + drive the sidecar, calling back //! into the host `HostVtable` for durable storage and events. //! -//! This file is the export/ABI/runtime skeleton (spec phase 4 foundation). The -//! actor run loop + the plugin-side host-vtable bridge are layered on top next. +//! The actor loop is implemented against RivetKit's portable context so the +//! dylib owns AgentOS behavior while the host owns lifecycle dispatch. #![allow(unsafe_op_in_unsafe_fn)] @@ -27,7 +27,10 @@ mod vm; #[cfg(test)] mod persistence_e2e; -use std::sync::Arc; +use std::io::Cursor; +use std::panic::AssertUnwindSafe; +use std::sync::{Arc, Mutex}; +use std::thread; /// Process-global plugin state created once per `dlopen` (spec §5.2): the /// plugin's own tokio runtime (`enable_all` — the time driver is required by @@ -48,18 +51,7 @@ fn write_err(out: *mut abi::OwnedBuf, msg: &str) { } } -#[no_mangle] -pub extern "C" fn rivet_actor_abi_magic() -> u64 { - abi::RIVET_ACTOR_ABI_MAGIC -} - -#[no_mangle] -pub extern "C" fn rivet_actor_abi_version() -> u64 { - abi::RIVET_ACTOR_ABI_VERSION -} - -#[no_mangle] -pub extern "C" fn rivet_actor_plugin_init(out_err: *mut abi::OwnedBuf) -> *mut c_void { +extern "C" fn plugin_init(out_err: *mut abi::OwnedBuf) -> *mut c_void { // Debug-only tracing to stderr, gated on AGENTOS_PLUGIN_LOG (RUST_LOG // syntax). Without a subscriber every tracing::warn! from the embedded // agentos-client (e.g. a failed fire-and-forget shell write) is silently @@ -105,8 +97,7 @@ struct Factory { pool: String, } -#[no_mangle] -pub extern "C" fn rivet_actor_factory_new( +extern "C" fn factory_new( plugin: *mut c_void, config_json: abi::BorrowedBuf, sidecar_path: abi::BorrowedBuf, @@ -136,85 +127,102 @@ pub extern "C" fn rivet_actor_factory_new( })) as *mut c_void } -/// Send wrapper for the host completion `user_data` so it can move into the -/// spawned actor task. struct SendUserData(*mut c_void); unsafe impl Send for SendUserData {} -struct RunGuard { +struct CompletionTarget { done: abi::CompletionFn, - ud: SendUserData, - fired: bool, + user_data: SendUserData, } -impl RunGuard { - fn finish(&mut self, result: abi::AbiResult) { - if !self.fired { - self.fired = true; - (self.done)(self.ud.0, result); +impl CompletionTarget { + fn finish(self, result: abi::AbiResult) { + (self.done)(self.user_data.0, result); + } + + fn finish_outcome(self, outcome: &Result<(), String>) { + match outcome { + Ok(()) => self.finish(abi::AbiResult::ok(abi::OwnedBuf::empty())), + Err(message) => self.finish(abi::AbiResult::err(abi::OwnedBuf::from_vec( + message.clone().into_bytes(), + ))), } } } -impl Drop for RunGuard { - fn drop(&mut self) { - self.finish(abi::AbiResult::status_only(abi::AbiStatus::Cancelled)); - } +struct Instance { + inner: Arc, } -/// Run one actor instance: build the `HostCtx` bridge over the host vtable, -/// spawn the actor loop on the plugin runtime, and signal completion when the -/// event stream closes (host cancel). The VM-dispatch layer (decode actions + -/// drive the sidecar via `agentos-client`) slots into `actor_loop` next. -#[no_mangle] -#[allow(clippy::not_unsafe_ptr_arg_deref)] -pub extern "C" fn rivet_actor_run( +struct InstanceInner { + bridge: abi::DylibEventBridge, + join: Mutex>>, + shutdown: Mutex, + cancel: CancellationToken, +} + +#[derive(Default)] +struct ShutdownState { + started: bool, + outcome: Option>, + waiters: Vec, +} + +extern "C" fn instance_new( factory: *mut c_void, host: *const abi::HostVtable, - done: abi::CompletionFn, + start: abi::BorrowedBuf, + out_err: *mut abi::OwnedBuf, + terminal_done: abi::CompletionFn, user_data: *mut c_void, ) -> *mut c_void { - let instance = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| unsafe { - let factory = &*(factory as *const Factory); - let host_ctx = host_ctx::HostCtx::from_vtable(*host); - let sidecar_path = factory.sidecar_path.clone(); - let config = factory.config.clone(); - let pool = factory.pool.clone(); - let ud = SendUserData(user_data); - let cancel = CancellationToken::new(); - let run_cancel = cancel.clone(); - let handle = factory.runtime.spawn(async move { - let mut guard = RunGuard { - done, - ud, - fired: false, - }; - actor_loop(host_ctx, sidecar_path, config, pool, run_cancel).await; - guard.finish(abi::AbiResult::ok(abi::OwnedBuf::empty())); - }); - Instance { - abort: Some(handle.abort_handle()), - cancel, - } - })) - .unwrap_or_else(|_| { - done( - user_data, - abi::AbiResult::status_only(abi::AbiStatus::Panic), - ); - Instance { - abort: None, - cancel: CancellationToken::new(), + if factory.is_null() || host.is_null() { + write_err(out_err, "null factory or host handle"); + return std::ptr::null_mut(); + } + let _: abi::InstanceStart = + match ciborium::from_reader(Cursor::new(unsafe { start.as_slice() })) { + Ok(start) => start, + Err(error) => { + write_err(out_err, &format!("decode instance start: {error}")); + return std::ptr::null_mut(); + } + }; + let factory = unsafe { &*(factory as *const Factory) }; + let (backend, bridge) = unsafe { abi::DylibBackend::from_host_vtable(&*host) }; + let host_ctx = host_ctx::HostCtx::from_backend(backend); + let runtime = factory.runtime.clone(); + let sidecar_path = factory.sidecar_path.clone(); + let config = factory.config.clone(); + let pool = factory.pool.clone(); + let cancel = CancellationToken::new(); + let run_cancel = cancel.clone(); + let terminal = CompletionTarget { + done: terminal_done, + user_data: SendUserData(user_data), + }; + let join = thread::spawn(move || { + let outcome = std::panic::catch_unwind(AssertUnwindSafe(|| { + runtime.block_on(actor_loop(host_ctx, sidecar_path, config, pool, run_cancel)); + })); + match outcome { + Ok(()) => terminal.finish(abi::AbiResult::ok(abi::OwnedBuf::empty())), + Err(_) => terminal.finish(abi::AbiResult::status_only(abi::AbiStatus::Panic)), } }); - Box::into_raw(Box::new(instance)) as *mut c_void + Box::into_raw(Box::new(Instance { + inner: Arc::new(InstanceInner { + bridge, + join: Mutex::new(Some(join)), + shutdown: Mutex::new(ShutdownState::default()), + cancel, + }), + })) as *mut c_void } -/// Plugin-side actor run loop (ported from `rivetkit-agent-os::run`): drains -/// host lifecycle events via the `HostCtx` bridge, brings the VM up lazily on -/// the first action, tears it down on Sleep/Destroy, and ends when the host -/// closes the stream. Action dispatch (decode + drive the VM) is the remaining -/// layer; until it lands, actions reply with a clear not-yet-ported error. +/// Plugin-side actor run loop: drains host lifecycle events through the +/// portable bridge, brings the VM up lazily on the first action, tears it down +/// on Sleep/Destroy, and ends when the host closes the stream. async fn actor_loop( host: host_ctx::HostCtx, sidecar_path: String, @@ -222,10 +230,16 @@ async fn actor_loop( pool: String, cancel: CancellationToken, ) { - // Ensure the agent-os schema exists before handling events (best-effort; - // mirrors rivetkit-agent-os run.rs). + // Ensure the durable schema exists before accepting work. Starting with a + // half-migrated actor would turn every later filesystem/session failure + // into a misleading action error. if host.sql_is_enabled() { - let _ = persistence::migrate(&host).await; + if let Err(error) = persistence::migrate(&host).await { + let message = format!("agent-os schema migration failed: {error:#}"); + host.log_warn(&message); + host.startup_ready(false, &message); + return; + } } // The VM handle + actor vars live on a dedicated worker task that drains // stateful jobs (Action/Http/Sleep/Destroy) serially in submission order. @@ -237,7 +251,7 @@ async fn actor_loop( // `ensure_vm().await` ran inline in this loop, so the first action starved // ConnOpen, the actor websocket connection setup timed out at 5000ms, and // live `sessionEvent` streaming silently delivered zero events. - let (job_tx, job_rx) = tokio::sync::mpsc::unbounded_channel::(); + let (job_tx, job_rx) = tokio::sync::mpsc::channel::(MAX_PENDING_ACTOR_JOBS); let worker = tokio::spawn(actor_worker( host.clone(), sidecar_path, @@ -256,27 +270,40 @@ async fn actor_loop( _ = cancel.cancelled() => break, event = host.next_event() => event, }; - let Some((tag, token, payload)) = event else { + let Some(event) = event else { break; }; - match abi::AbiEventTag::from_u32(tag) { - Some(abi::AbiEventTag::Action) => { - if job_tx.send(ActorJob::Action { token, payload }).is_err() { - let _ = host.reply_err(token, "agent-os actor worker unavailable"); - } + match event { + abi::Event::Action { + name, args, reply, .. + } => { + enqueue_job( + &host, + &job_tx, + ActorJob::Action { + token: reply.0, + name, + args, + }, + ); } - Some(abi::AbiEventTag::Http) => { - if job_tx.send(ActorJob::Http { token, payload }).is_err() { - let _ = host.reply_err(token, "agent-os actor worker unavailable"); - } + abi::Event::Http { request, reply } => { + enqueue_job( + &host, + &job_tx, + ActorJob::Http { + token: reply.0, + request, + }, + ); } - Some(abi::AbiEventTag::ConnPreflight) => { - let _ = host.reply_ok(token, Vec::new()); + abi::Event::ConnPreflight { reply, .. } => { + let _ = host.reply_ok(reply.0, Vec::new()); } - Some(abi::AbiEventTag::ConnOpen) => { - let _ = host.reply_ok(token, Vec::new()); + abi::Event::ConnOpen { reply, .. } => { + let _ = host.reply_ok(reply.0, Vec::new()); } - Some(abi::AbiEventTag::Subscribe) => { + abi::Event::Subscribe { reply, .. } => { // Accept the connection's event subscription (e.g. `sessionEvent`) // so RivetKit registers it and routes matching broadcasts to this // connection. Subscription state is tracked by RivetKit core; the @@ -285,43 +312,94 @@ async fn actor_loop( // subscription — so the connection was never registered and every // broadcast (sessionEvent, vmBooted, ...) was silently dropped, // making live `sessionEvent` streaming deliver nothing. - let _ = host.reply_ok(token, Vec::new()); + let _ = host.reply_ok(reply.0, Vec::new()); } - Some(abi::AbiEventTag::SerializeState) => { + abi::Event::SerializeState { reply } => { // Agent OS persists durable data through SQLite and rebuilds VM/process // state after wake, so there is no opaque actor state payload to return. - let _ = host.reply_ok(token, Vec::new()); + let _ = host.reply_ok(reply.0, Vec::new()); } - Some(abi::AbiEventTag::Sleep) => { - if job_tx.send(ActorJob::Sleep { token }).is_err() { - let _ = host.reply_ok(token, Vec::new()); - } + abi::Event::Sleep { reply } => { + enqueue_job(&host, &job_tx, ActorJob::Sleep { token: reply.0 }); } - Some(abi::AbiEventTag::Destroy) => { - if job_tx.send(ActorJob::Destroy { token }).is_err() { - let _ = host.reply_ok(token, Vec::new()); - } + abi::Event::Destroy { reply } => { + enqueue_job(&host, &job_tx, ActorJob::Destroy { token: reply.0 }); } - Some(t) if t.needs_reply() => { - let _ = host.reply_err(token, "event not supported by agent-os actor"); + abi::Event::QueueSend { reply, .. } | abi::Event::WebSocketOpen { reply, .. } => { + let _ = host.reply_err(reply.0, "event not supported by agent-os actor"); } - _ => {} + abi::Event::ConnClosed { .. } => {} } } // Stream closed (cancel/teardown): stop accepting new jobs and let the // worker drain in-flight work + shut the VM down before we return. drop(job_tx); - let _ = worker.await; + if let Err(error) = worker.await { + host.log_warn(&format!("agent-os actor worker task failed: {error}")); + } } /// Stateful actor jobs serviced by the worker task in submission order. Keeping /// VM-touching work (bring-up, action dispatch, HTTP proxy, shutdown) off the /// event loop is what lets the loop answer connection-lifecycle events promptly. enum ActorJob { - Action { token: u64, payload: Vec }, - Http { token: u64, payload: Vec }, - Sleep { token: u64 }, - Destroy { token: u64 }, + Action { + token: u64, + name: String, + args: Vec, + }, + Http { + token: u64, + request: Vec, + }, + Sleep { + token: u64, + }, + Destroy { + token: u64, + }, +} + +const MAX_PENDING_ACTOR_JOBS: usize = 64; +const PENDING_ACTOR_JOBS_WARN_REMAINING: usize = 13; + +impl ActorJob { + fn token(&self) -> u64 { + match self { + Self::Action { token, .. } + | Self::Http { token, .. } + | Self::Sleep { token } + | Self::Destroy { token } => *token, + } + } +} + +fn enqueue_job( + host: &host_ctx::HostCtx, + sender: &tokio::sync::mpsc::Sender, + job: ActorJob, +) { + let token = job.token(); + if sender.capacity() <= PENDING_ACTOR_JOBS_WARN_REMAINING { + host.log_warn(&format!( + "agent-os actor job queue is near its limit: remaining={} limit={MAX_PENDING_ACTOR_JOBS}", + sender.capacity() + )); + } + match sender.try_send(job) { + Ok(()) => {} + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + let _ = host.reply_err( + token, + &format!( + "agent-os actor pending-job limit {MAX_PENDING_ACTOR_JOBS} exceeded; raise MAX_PENDING_ACTOR_JOBS in the plugin build" + ), + ); + } + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + let _ = host.reply_err(token, "agent-os actor worker unavailable"); + } + } } /// Owns the VM handle + actor vars and processes `ActorJob`s serially. Runs as a @@ -332,7 +410,7 @@ async fn actor_worker( sidecar_path: String, config: Arc, pool: String, - mut job_rx: tokio::sync::mpsc::UnboundedReceiver, + mut job_rx: tokio::sync::mpsc::Receiver, cancel: CancellationToken, ) { let mut vm: Option = None; @@ -346,7 +424,11 @@ async fn actor_worker( }, }; match job { - ActorJob::Action { token, payload } => { + ActorJob::Action { + token, + name, + args: action_args, + } => { if let Err(error) = vm::ensure_vm(&host, &sidecar_path, &config, &pool, &mut vm).await { @@ -358,30 +440,23 @@ async fn actor_worker( let _ = host.reply_err(token, "vm unavailable after bring-up"); continue; }; - match abi::decode_action_payload(&payload) { - Ok((name, action_args)) => { - tracing::debug!(action = %name, "agent-os action start"); - actions::dispatch( - &host, - vm_ref, - &config, - &mut vars, - &name, - &action_args, - token, - ) - .await; - tracing::debug!(action = %name, "agent-os action done"); - } - Err(_) => { - let _ = host.reply_err(token, "malformed action event payload"); - } - } + tracing::debug!(action = %name, "agent-os action start"); + actions::dispatch( + &host, + vm_ref, + &config, + &mut vars, + &name, + &action_args, + token, + ) + .await; + tracing::debug!(action = %name, "agent-os action done"); } - ActorJob::Http { token, payload } => { + ActorJob::Http { token, request } => { // Preview proxy: do NOT bring the VM up for HTTP (matches r6's // run loop, which passes `vm.as_ref()`); no VM → 404. - let response = http::proxy_preview(&host, vm.as_ref(), &payload).await; + let response = http::proxy_preview(&host, vm.as_ref(), &request).await; let _ = host.reply_ok(token, response); } ActorJob::Sleep { token } => { @@ -401,59 +476,111 @@ async fn actor_worker( vm::shutdown_vm(&host, &mut vm, "error").await; } -struct Instance { - abort: Option, - cancel: CancellationToken, +extern "C" fn handle_event( + instance: *mut c_void, + event_id: u64, + event: abi::AbiEvent, + done: abi::CompletionFn, + user_data: *mut c_void, +) { + let instance = unsafe { &*(instance as *const Instance) }; + instance + .inner + .bridge + .handle_event(event_id, event, done, user_data); } -#[no_mangle] -pub extern "C" fn rivet_actor_cancel(instance: *mut c_void) { - if instance.is_null() { - return; - } - let _ = std::panic::catch_unwind(|| unsafe { - let inst = &*(instance as *const Instance); - inst.cancel.cancel(); - }); +extern "C" fn cancel_event(instance: *mut c_void, event_id: u64) { + let instance = unsafe { &*(instance as *const Instance) }; + instance.inner.bridge.cancel_event(event_id); } -#[no_mangle] -pub extern "C" fn rivet_actor_grace_deadline(instance: *mut c_void) { - if instance.is_null() { +extern "C" fn shutdown( + instance: *mut c_void, + force: u8, + done: abi::CompletionFn, + user_data: *mut c_void, +) { + let instance = unsafe { &*(instance as *const Instance) }; + let inner = instance.inner.clone(); + let completion = CompletionTarget { + done, + user_data: SendUserData(user_data), + }; + if force != 0 { + inner.cancel.cancel(); + } + inner.bridge.close(force != 0); + + let mut shutdown = inner.shutdown.lock().expect("shutdown lock"); + if let Some(outcome) = &shutdown.outcome { + completion.finish_outcome(outcome); + return; + } + shutdown.waiters.push(completion); + if shutdown.started { return; } - let _ = std::panic::catch_unwind(|| unsafe { - let inst = &*(instance as *const Instance); - inst.cancel.cancel(); - if let Some(abort) = inst.abort.as_ref() { - abort.abort(); + shutdown.started = true; + drop(shutdown); + + thread::spawn(move || { + let outcome = inner + .join + .lock() + .expect("instance join lock") + .take() + .map(|join| { + join.join() + .map_err(|_| "agent-os actor thread panicked".to_owned()) + }) + .transpose() + .map(|_| ()); + inner.bridge.finish_shutdown(); + let waiters = { + let mut shutdown = inner.shutdown.lock().expect("shutdown lock"); + shutdown.outcome = Some(outcome.clone()); + std::mem::take(&mut shutdown.waiters) + }; + for waiter in waiters { + waiter.finish_outcome(&outcome); } }); } -#[no_mangle] -pub extern "C" fn rivet_actor_instance_free(instance: *mut c_void) { +extern "C" fn instance_free(instance: *mut c_void) { if !instance.is_null() { - let _ = std::panic::catch_unwind(|| unsafe { - drop(Box::from_raw(instance as *mut Instance)); - }); + unsafe { drop(Box::from_raw(instance as *mut Instance)) }; } } -#[no_mangle] -pub extern "C" fn rivet_actor_factory_free(factory: *mut c_void) { +extern "C" fn factory_free(factory: *mut c_void) { if !factory.is_null() { - let _ = std::panic::catch_unwind(|| unsafe { - drop(Box::from_raw(factory as *mut Factory)); - }); + unsafe { drop(Box::from_raw(factory as *mut Factory)) }; } } -#[no_mangle] -pub extern "C" fn rivet_actor_plugin_shutdown(plugin: *mut c_void) { +extern "C" fn plugin_shutdown(plugin: *mut c_void) { if !plugin.is_null() { - let _ = std::panic::catch_unwind(|| unsafe { - drop(Box::from_raw(plugin as *mut Plugin)); - }); + unsafe { drop(Box::from_raw(plugin as *mut Plugin)) }; } } + +static PLUGIN_API: abi::PluginApi = abi::PluginApi::new( + plugin_init, + factory_new, + factory_free, + instance_new, + handle_event, + cancel_event, + shutdown, + instance_free, + plugin_shutdown, +); + +/// AgentOS exports one process-lifetime descriptor. RivetKit validates and +/// copies this table while retaining the loaded library for process lifetime. +#[no_mangle] +pub extern "C" fn rivet_actor_plugin_api() -> *const abi::PluginApi { + &PLUGIN_API +} diff --git a/crates/agentos-actor-plugin/src/persistence_e2e.rs b/crates/agentos-actor-plugin/src/persistence_e2e.rs index 40236fb9ed..f52d97e03f 100644 --- a/crates/agentos-actor-plugin/src/persistence_e2e.rs +++ b/crates/agentos-actor-plugin/src/persistence_e2e.rs @@ -39,15 +39,6 @@ extern "C" fn sql_is_enabled(_ctx: *const c_void) -> u8 { } // Unused-by-persistence vtable stubs. -extern "C" fn next_event(_ctx: *const c_void, done: abi::CompletionFn, ud: *mut c_void) { - done(ud, abi::AbiResult::channel_closed()); -} -extern "C" fn reply_ok(_c: *const c_void, _t: u64, _p: abi::OwnedBuf) -> abi::AbiStatus { - abi::AbiStatus::Ok -} -extern "C" fn reply_err(_c: *const c_void, _t: u64, _p: abi::OwnedBuf) -> abi::AbiStatus { - abi::AbiStatus::Ok -} extern "C" fn startup_ready(_c: *const c_void, _ok: u8, _e: abi::BorrowedBuf) {} extern "C" fn broadcast(_c: *const c_void, _n: abi::OwnedBuf, _p: abi::OwnedBuf) -> abi::AbiStatus { abi::AbiStatus::Ok @@ -128,6 +119,24 @@ extern "C" fn async_unavailable( )), ); } +extern "C" fn host_call_unavailable( + _c: *const c_void, + name: abi::OwnedBuf, + request: abi::OwnedBuf, + done: abi::CompletionFn, + ud: *mut c_void, +) { + unsafe { + name.free_self(); + request.free_self(); + } + done( + ud, + abi::AbiResult::err(abi::OwnedBuf::from_vec( + b"host callbacks are not available in persistence tests".to_vec(), + )), + ); +} extern "C" fn hibernatable_ws_ack( _c: *const c_void, gateway_id: abi::OwnedBuf, @@ -273,6 +282,7 @@ fn mock_host_ctx(host: &MockHost) -> HostCtx { db_exec, db_query, db_run, + host_call: host_call_unavailable, sql_is_enabled, state_get, state_set, @@ -303,14 +313,12 @@ fn mock_host_ctx(host: &MockHost) -> HostCtx { conn_disconnect: async_unavailable, hibernatable_ws_ack, conn_send, - next_event, - reply_ok, - reply_err, startup_ready, broadcast, log, }; - HostCtx::from_vtable(vtable) + let (backend, _event_bridge) = unsafe { abi::DylibBackend::from_host_vtable(&vtable) }; + HostCtx::from_backend(backend) } #[tokio::test] From d2fcfdacd7847f24a76289ed1ebc21008305f46b Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Wed, 15 Jul 2026 02:22:24 -0700 Subject: [PATCH 3/3] feat(actor): resolve options from actor create input --- crates/agentos-actor-plugin/src/config.rs | 49 ++++++++ crates/agentos-actor-plugin/src/lib.rs | 13 ++- packages/agentos/src/actor.ts | 123 +++++++++++++-------- packages/agentos/src/config.ts | 58 ++++++---- packages/agentos/src/index.ts | 32 ++++-- packages/agentos/tests/actor.test.ts | 129 +++++++++++++++++++++- 6 files changed, 321 insertions(+), 83 deletions(-) diff --git a/crates/agentos-actor-plugin/src/config.rs b/crates/agentos-actor-plugin/src/config.rs index be2d19398f..99e646b936 100644 --- a/crates/agentos-actor-plugin/src/config.rs +++ b/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 { + 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. diff --git a/crates/agentos-actor-plugin/src/lib.rs b/crates/agentos-actor-plugin/src/lib.rs index f38bcd8f85..847cdd43b6 100644 --- a/crates/agentos-actor-plugin/src/lib.rs +++ b/crates/agentos-actor-plugin/src/lib.rs @@ -180,7 +180,7 @@ extern "C" fn instance_new( write_err(out_err, "null factory or host handle"); return std::ptr::null_mut(); } - let _: abi::InstanceStart = + let start: abi::InstanceStart = match ciborium::from_reader(Cursor::new(unsafe { start.as_slice() })) { Ok(start) => start, Err(error) => { @@ -189,11 +189,20 @@ extern "C" fn instance_new( } }; let factory = unsafe { &*(factory as *const Factory) }; + let config = match start.instance_options { + Some(options) => match config::AgentOsConfigJson::parse_instance_options(&options) { + Ok(config) => Arc::new(config), + Err(error) => { + write_err(out_err, &format!("decode instance options: {error:#}")); + return std::ptr::null_mut(); + } + }, + None => factory.config.clone(), + }; let (backend, bridge) = unsafe { abi::DylibBackend::from_host_vtable(&*host) }; let host_ctx = host_ctx::HostCtx::from_backend(backend); let runtime = factory.runtime.clone(); let sidecar_path = factory.sidecar_path.clone(); - let config = factory.config.clone(); let pool = factory.pool.clone(); let cancel = CancellationToken::new(); let run_cancel = cancel.clone(); diff --git a/packages/agentos/src/actor.ts b/packages/agentos/src/actor.ts index f741bba5d7..4e7333ef23 100644 --- a/packages/agentos/src/actor.ts +++ b/packages/agentos/src/actor.ts @@ -22,12 +22,14 @@ import { type CoreRuntime, type DatabaseProvider, type NapiNativePluginOptions, + type NativeFactoryBuilderOptions, type RawAccess, } from "rivetkit"; import { type AgentOsActorConfig, type AgentOsActorConfigInput, agentOsActorConfigSchema, + type NativeAgentOsOptions, nativeAgentOsOptionsSchema, } from "./config.js"; import type { AgentOsActions } from "./generated/actor-actions.generated.js"; @@ -136,11 +138,11 @@ function normalizedPackageRefs(software: unknown[]): NormalizedPackageRef[] { return refs; } -export function buildConfigJson( - parsed: AgentOsActorConfig, -): string { +function buildConfigEnvelope( + optionsInput: NativeAgentOsOptions | undefined, +): Record { const options = nativeAgentOsOptionsSchema.parse( - parsed.options ?? {}, + optionsInput ?? {}, ) as Record; const softwareInput = Array.isArray(options.software) ? options.software : []; const defaultSoftwareEnabled = options.defaultSoftware !== false; @@ -150,21 +152,29 @@ export function buildConfigJson( const packages = packageRefs.map((ref) => ({ packagePath: ref.path })); const mounts = serializeNativeMounts(options.mounts); const sidecar = serializeSidecar(options.sidecar); - return JSON.stringify({ - // The actor forwards ONLY package dirs; the sidecar resolves each agent from - // the projected `/opt/agentos//current/agentos-package.json` (no - // client-side adapter-entrypoint resolution — see root CLAUDE.md). - packages, - packagesMountAt: OPT_AGENTOS_ROOT, - additionalInstructions: options.additionalInstructions, - loopbackExemptPorts: options.loopbackExemptPorts, - allowedNodeBuiltins: options.allowedNodeBuiltins, - permissions: options.permissions, - rootFilesystem: options.rootFilesystem, - mounts, - limits: options.limits, - sidecar, - }); + return JSON.parse( + JSON.stringify({ + // The actor forwards ONLY package dirs; the sidecar resolves each agent from + // the projected `/opt/agentos//current/agentos-package.json` (no + // client-side adapter-entrypoint resolution — see root CLAUDE.md). + packages, + packagesMountAt: OPT_AGENTOS_ROOT, + additionalInstructions: options.additionalInstructions, + loopbackExemptPorts: options.loopbackExemptPorts, + allowedNodeBuiltins: options.allowedNodeBuiltins, + permissions: options.permissions, + rootFilesystem: options.rootFilesystem, + mounts, + limits: options.limits, + sidecar, + }), + ) as Record; +} + +export function buildConfigJson( + parsed: AgentOsActorConfig, +): string { + return JSON.stringify(buildConfigEnvelope(parsed.options)); } function serializeNativeMounts(input: unknown): NativeMountLike[] | undefined { @@ -227,11 +237,13 @@ function serializeSidecar(input: unknown): { pool?: string } | undefined { return typeof record.pool === "string" ? { pool: record.pool } : {}; } -function buildNativeFactoryBuilder( - parsed: AgentOsActorConfig, - actorOptions: Record, -): (runtime: CoreRuntime) => ActorFactoryHandle { - return (runtime) => { +function buildNativeFactoryBuilder( + parsed: AgentOsActorConfig, +): ( + runtime: CoreRuntime, + options?: NativeFactoryBuilderOptions, +) => ActorFactoryHandle { + return (runtime, builderOptions) => { if (runtime.kind !== "napi") { throw new Error( `agentOS() is only supported on the native NAPI runtime (current runtime kind: ${runtime.kind})`, @@ -248,9 +260,6 @@ function buildNativeFactoryBuilder( pluginPath: getPluginPath(), // Opaque config envelope the plugin parses (config.rs::AgentOsConfigJson). configJson: buildConfigJson(parsed), - // RivetKit's native-plugin bridge owns the runtime ActorConfig for - // cdylib actors, so forward the merged per-actor lifecycle options too. - actorOptions, // Resolve the prebuilt sidecar binary from the npm package so the plugin // spawns the bundled binary rather than relying on `agentos-sidecar` // being on PATH. @@ -264,10 +273,34 @@ function buildNativeFactoryBuilder( // native-plugin actors.) inspectorTabs: AGENTOS_INSPECTOR_CONFIG.tabs, } as NapiNativePluginOptions & { - actorOptions?: Record; inspectorTabs: typeof AGENTOS_INSPECTOR_CONFIG.tabs; }; - return runtime.createNativePluginFactory(options); + const callbacks = { + ...(builderOptions?.callbacks ?? {}), + } as Record; + if (parsed.createOptions) { + if (!builderOptions) { + throw new Error( + "agentOS() createOptions requires RivetKit's composed native factory builder", + ); + } + callbacks.createNativeOptions = + builderOptions.createNativeOptionsCallback( + async (c, input: TInput | undefined) => { + const instanceOptions = await parsed.createOptions?.(c, input); + const merged = nativeAgentOsOptionsSchema.parse({ + ...(parsed.options ?? {}), + ...(instanceOptions ?? {}), + }) as NativeAgentOsOptions; + return buildConfigEnvelope(merged); + }, + ); + } + return runtime.createNativePluginFactory( + options, + builderOptions ? callbacks : undefined, + builderOptions?.config, + ); }; } @@ -279,10 +312,13 @@ function buildNativeFactoryBuilder( * That is what gives `createClient()` a fully-typed handle * (e.g. `handle.exec()` returns `ExecResult`, not `unknown`). */ -export type AgentOsActorDefinition = ActorDefinition< +export type AgentOsActorDefinition< + TInput = undefined, + TConnParams = undefined, +> = ActorDefinition< AgentOsActorState, TConnParams, - undefined, + TInput, AgentOsActorVars, undefined, DatabaseProvider, @@ -385,12 +421,13 @@ const AGENTOS_INSPECTOR_CONFIG = { ], }; -export function createAgentOS( - config: AgentOsActorConfigInput, -): AgentOsActorDefinition { - const parsed = agentOsActorConfigSchema.parse( - config, - ) as AgentOsActorConfig; +export function createAgentOS( + config: AgentOsActorConfigInput, +): AgentOsActorDefinition { + const parsed = agentOsActorConfigSchema.parse(config) as AgentOsActorConfig< + TInput, + TConnParams + >; // Construct a minimal definition through the existing actor() helper, then // attach the Rust factory builder marker. The actions block stays empty @@ -411,12 +448,10 @@ export function createAgentOS( // rivetkit tabs) so the dashboard renders the agent-os UI. Without this // the shipped tab assets are never surfaced. inspector: AGENTOS_INSPECTOR_CONFIG, - } as Parameters< - typeof actor - >[0]) as unknown as AgentOsActorDefinition; - definition.nativeFactoryBuilder = buildNativeFactoryBuilder( - parsed, - actorOptions, - ); + } as Parameters[0]) as unknown as AgentOsActorDefinition< + TInput, + TConnParams + >; + definition.nativeFactoryBuilder = buildNativeFactoryBuilder(parsed); return definition; } diff --git a/packages/agentos/src/config.ts b/packages/agentos/src/config.ts index 06fb85a872..6cf1e35369 100644 --- a/packages/agentos/src/config.ts +++ b/packages/agentos/src/config.ts @@ -58,6 +58,7 @@ export const agentOsActorConfigSchema = z }) .strict() .prefault(() => ({})), + createOptions: zFunction().optional(), onBeforeConnect: zFunction().optional(), onSessionEvent: zFunction().optional(), onPermissionRequest: zFunction().optional(), @@ -87,49 +88,64 @@ export type NativeAgentOsOptions = Pick< sidecar?: { kind: "shared"; pool?: string }; }; -type AgentOsActorContext = ActorContext< +export type AgentOsActorContext = ActorContext< AgentOsActorState, TConnParams, undefined, AgentOsActorVars, - undefined, + TInput, any >; -interface AgentOsActorConfigCallbacks { +interface AgentOsActorConfigCallbacks { + createOptions?: ( + c: AgentOsActorContext, + input: TInput | undefined, + ) => NativeAgentOsOptions | Promise; onBeforeConnect?: ( - c: BeforeConnectContext< - AgentOsActorState, - AgentOsActorVars, - undefined, - any - >, + c: BeforeConnectContext, params: TConnParams, ) => void | Promise; onSessionEvent?: ( - c: AgentOsActorContext, + c: AgentOsActorContext, sessionId: string, event: JsonRpcNotification, ) => void | Promise; onPermissionRequest?: ( - c: AgentOsActorContext, + c: AgentOsActorContext, sessionId: string, request: PermissionRequest, ) => void | Promise; } // Parsed config (after Zod defaults/transforms applied). -export type AgentOsActorConfig = Omit< +export type AgentOsActorConfig< + TInput = undefined, + TConnParams = undefined, +> = Omit< z.infer, - "options" | "onBeforeConnect" | "onSessionEvent" | "onPermissionRequest" -> & - { options?: NativeAgentOsOptions } & - AgentOsActorConfigCallbacks; + | "options" + | "createOptions" + | "onBeforeConnect" + | "onSessionEvent" + | "onPermissionRequest" +> & { options?: NativeAgentOsOptions } & AgentOsActorConfigCallbacks< + TInput, + TConnParams + >; // Input config (what users pass in before Zod transforms). -export type AgentOsActorConfigInput = Omit< +export type AgentOsActorConfigInput< + TInput = undefined, + TConnParams = undefined, +> = Omit< z.input, - "options" | "onBeforeConnect" | "onSessionEvent" | "onPermissionRequest" -> & - { options?: NativeAgentOsOptions } & - AgentOsActorConfigCallbacks; + | "options" + | "createOptions" + | "onBeforeConnect" + | "onSessionEvent" + | "onPermissionRequest" +> & { options?: NativeAgentOsOptions } & AgentOsActorConfigCallbacks< + TInput, + TConnParams + >; diff --git a/packages/agentos/src/index.ts b/packages/agentos/src/index.ts index d745dd1706..81386baf10 100644 --- a/packages/agentos/src/index.ts +++ b/packages/agentos/src/index.ts @@ -91,13 +91,23 @@ export type { VmShutdownPayload, } from "./types.js"; -export type AgentOSActorConfigInput = - NativeAgentOsOptions & Omit, "options">; +export type AgentOSActorConfigInput< + TInput = undefined, + TConnParams = undefined, +> = NativeAgentOsOptions & + Omit, "options">; -export type AgentOSConfigInput = AgentOsOptions & { - preview?: AgentOsActorConfigInput["preview"]; - actorOptions?: AgentOsActorConfigInput["actorOptions"]; - onBeforeConnect?: AgentOsActorConfigInput["onBeforeConnect"]; +export type AgentOSConfigInput< + TInput = undefined, + TConnParams = undefined, +> = AgentOsOptions & { + preview?: AgentOsActorConfigInput["preview"]; + actorOptions?: AgentOsActorConfigInput["actorOptions"]; + createOptions?: AgentOsActorConfigInput["createOptions"]; + onBeforeConnect?: AgentOsActorConfigInput< + TInput, + TConnParams + >["onBeforeConnect"]; onSessionEvent?: ( sessionId: string, event: JsonRpcNotification, @@ -108,12 +118,13 @@ export type AgentOSConfigInput = AgentOsOptions & { ) => void | Promise; }; -export function agentOS( - config: AgentOSConfigInput = {}, -): AgentOsActorDefinition { +export function agentOS( + config: AgentOSConfigInput = {}, +): AgentOsActorDefinition { const { preview, actorOptions, + createOptions, onBeforeConnect, onSessionEvent, onPermissionRequest, @@ -124,6 +135,7 @@ export function agentOS( options, actorOptions, preview, + createOptions, onBeforeConnect, onSessionEvent: onSessionEvent ? (_ctx, sessionId, event) => onSessionEvent(sessionId, event) @@ -131,5 +143,5 @@ export function agentOS( onPermissionRequest: onPermissionRequest ? (_ctx, sessionId, request) => onPermissionRequest(sessionId, request) : undefined, - } as AgentOsActorConfigInput); + } as AgentOsActorConfigInput); } diff --git a/packages/agentos/tests/actor.test.ts b/packages/agentos/tests/actor.test.ts index b7326c51f2..3d8f89aed4 100644 --- a/packages/agentos/tests/actor.test.ts +++ b/packages/agentos/tests/actor.test.ts @@ -8,6 +8,7 @@ import type { ActorFactoryHandle, CoreRuntime, NapiNativePluginOptions, + NativeFactoryBuilderOptions, } from "rivetkit"; import type { createClient } from "rivetkit/client"; import { setupTest } from "rivetkit/test"; @@ -207,15 +208,34 @@ describe.sequential("@rivet-dev/agentos actor plugin package bridge", () => { "native-factory", ) as unknown as ActorFactoryHandle; const calls: NapiNativePluginOptions[] = []; + let receivedConfig: unknown; const runtime = { kind: "napi", - createNativePluginFactory(options: NapiNativePluginOptions) { + createNativePluginFactory( + options: NapiNativePluginOptions, + _callbacks?: object, + config?: unknown, + ) { calls.push(options); + receivedConfig = config; return expectedHandle; }, } as CoreRuntime; - const handle = definition.nativeFactoryBuilder?.(runtime); + const handle = definition.nativeFactoryBuilder?.(runtime, { + callbacks: {}, + config: { + actionTimeoutMs: 3_600_000, + sleepTimeoutMs: 500, + sleepGracePeriodMs: 1_000, + }, + createNativeOptionsCallback() { + throw new Error("not used"); + }, + createNativeHostCallCallback() { + throw new Error("not used"); + }, + } as unknown as NativeFactoryBuilderOptions); expect(handle).toBe(expectedHandle); expect(calls).toHaveLength(1); @@ -239,11 +259,108 @@ describe.sequential("@rivet-dev/agentos actor plugin package bridge", () => { }, ], }); - expect((calls[0] as any).actorOptions).toMatchObject({ - actionTimeout: 3_600_000, - sleepTimeout: 500, - sleepGracePeriod: 1_000, + expect(receivedConfig).toMatchObject({ + actionTimeoutMs: 3_600_000, + sleepTimeoutMs: 500, + sleepGracePeriodMs: 1_000, + }); + }); + + test("resolves validated native options from actor create input", async () => { + const definition = agentOS<{ tenantId: string }>({ + defaultSoftware: false, + software: [], + additionalInstructions: "static instructions", + createOptions: async (c, input) => { + expect(c).toEqual({ actorId: "actor-1" }); + return { + defaultSoftware: false, + software: [], + additionalInstructions: `tenant:${input?.tenantId}`, + loopbackExemptPorts: [4100], + }; + }, }); + let resolveOptions: + | ((ctx: unknown, input: unknown, isNew: boolean) => Promise) + | undefined; + const wrappedCallback = Symbol("create-native-options"); + const nativeCalls: Array<{ + callbacks?: Record; + }> = []; + const runtime = { + kind: "napi", + createNativePluginFactory( + _options: NapiNativePluginOptions, + callbacks?: object, + ) { + nativeCalls.push({ + callbacks: callbacks as Record | undefined, + }); + return Symbol("native-factory") as unknown as ActorFactoryHandle; + }, + } as CoreRuntime; + const builderOptions = { + callbacks: {}, + config: {}, + createNativeOptionsCallback(handler: (...args: unknown[]) => unknown) { + resolveOptions = async (ctx, input, isNew) => + await handler(ctx, input, isNew); + return wrappedCallback; + }, + createNativeHostCallCallback() { + throw new Error("not used"); + }, + } as unknown as NativeFactoryBuilderOptions; + + definition.nativeFactoryBuilder?.(runtime, builderOptions); + + expect(nativeCalls[0].callbacks?.createNativeOptions).toBe(wrappedCallback); + expect(resolveOptions).toEqual(expect.any(Function)); + const resolved = await resolveOptions?.( + { actorId: "actor-1" }, + { tenantId: "acme" }, + true, + ); + expect(resolved).toMatchObject({ + additionalInstructions: "tenant:acme", + loopbackExemptPorts: [4100], + packages: [], + packagesMountAt: "/opt/agentos", + }); + }); + + test("rejects invalid per-instance options before plugin startup", async () => { + const definition = agentOS<{ tenantId: string }>({ + defaultSoftware: false, + software: [], + createOptions: async () => ({ notAnAgentOsOption: true }) as never, + }); + let resolveOptions: + | ((ctx: unknown, input: unknown, isNew: boolean) => Promise) + | undefined; + const runtime = { + kind: "napi", + createNativePluginFactory() { + return Symbol("native-factory") as unknown as ActorFactoryHandle; + }, + } as CoreRuntime; + definition.nativeFactoryBuilder?.(runtime, { + callbacks: {}, + config: {}, + createNativeOptionsCallback(handler: (...args: unknown[]) => unknown) { + resolveOptions = async (ctx, input, isNew) => + await handler(ctx, input, isNew); + return (() => {}) as never; + }, + createNativeHostCallCallback() { + throw new Error("not used"); + }, + } as unknown as NativeFactoryBuilderOptions); + + await expect( + resolveOptions?.({}, { tenantId: "acme" }, true), + ).rejects.toThrow(/notAnAgentOsOption/); }); test("agentOS flat config keeps callbacks outside native VM options", () => {