feat(v1.1.9): queue:: Rhai SDK module

executor-core/sdk/queue.rs registers the queue:: namespace into the
per-call Rhai engine:

  queue::enqueue("name", message)               -> ()
  queue::enqueue("name", message, opts)         -> ()  // opts: Map
  queue::depth("name")                          -> i64
  queue::depth_pending("name")                  -> i64

No handle pattern (queues are the grouping unit, mirroring pubsub).
Opts map accepts optional delay_ms (i64) and max_attempts (u32);
unknown keys are silently ignored to match Rhai's Map semantics.

Bridge details:
- captures Arc<dyn QueueService> + Arc<SdkCallCx> per closure (cheap
  clone)
- block_on via TokioHandle::try_current() — same pattern pubsub/kv use
- message → JSON via a local message_to_json that base64-encodes Blobs
  at any depth (mirrors pubsub) and REJECTS FnPtr / closures with a
  clear error (queue messages must survive Postgres round-trip)
- depth / depth_pending clamp u64 → i64::MAX (Rhai's INT is i64)

register_all gains queue::register(...) positionally after pubsub.

Unit tests (executor-core --lib): blob → base64, nested-map round-trip,
FnPtr rejection, opts parsing (partial / max_attempts only / non-integer
delay rejection).

Integration tests (executor-core/tests/sdk_queue.rs, 6 binaries):
enqueue map payload, enqueue with opts threads through, blob base64
encodes, depth returns service value, FnPtr in payload throws, empty
queue_name throws.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/executor-core/src/sdk/mod.rs

@@ -20,6 +20,7 @@ pub mod files;
 pub mod http;
 pub mod kv;
 pub mod pubsub;
+pub mod queue;
 pub mod secrets;
 pub mod stdlib;
 pub mod users;
@@ -45,6 +46,7 @@ pub fn register_all(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCal
     http::register(engine, services, cx.clone());
     files::register(engine, services, cx.clone());
     pubsub::register(engine, services, cx.clone());
+    queue::register(engine, services, cx.clone());
     secrets::register(engine, services, cx.clone());
     email::register(engine, services, cx.clone());
     users::register(engine, services, cx);

crates/executor-core/src/sdk/queue.rs

@@ -0,0 +1,266 @@
+//! `queue::` Rhai bridge — durable per-app queue producer + inspection
+//! (v1.1.9).
+//!
+//! ```rhai
+//! queue::enqueue("payments.process", #{ order_id: 123, amount: 4999 });
+//! queue::enqueue("payments.process", msg, #{ delay_ms: 60000, max_attempts: 5 });
+//! let total = queue::depth("payments.process");
+//! let pending = queue::depth_pending("payments.process");
+//! ```
+//!
+//! No handle pattern (queues are the grouping unit, mirroring pubsub).
+//! No `peek` / `dequeue` / `purge` script-side surface — consumers are
+//! `queue:receive` triggers; exposing manual dequeue would force the
+//! platform to surface visibility-timeout semantics into script-land.
+//!
+//! `app_id` is derived from `cx.app_id` inside the service — it never
+//! appears in the script-side signature.
+
+use std::sync::Arc;
+
+use base64::engine::general_purpose::STANDARD;
+use base64::Engine as _;
+use picloud_shared::{EnqueueOpts, QueueError, QueueService, SdkCallCx, Services};
+use rhai::{Array, Dynamic, Engine as RhaiEngine, EvalAltResult, Map, Module};
+use serde_json::Value as Json;
+use tokio::runtime::Handle as TokioHandle;
+
+pub(super) fn register(engine: &mut RhaiEngine, services: &Services, cx: Arc<SdkCallCx>) {
+    let svc = services.queue.clone();
+    let mut module = Module::new();
+
+    // `queue::enqueue(name, message)` — default opts.
+    {
+        let svc = svc.clone();
+        let cx = cx.clone();
+        module.set_native_fn(
+            "enqueue",
+            move |name: &str, message: Dynamic| -> Result<(), Box<EvalAltResult>> {
+                let json = message_to_json(&message)?;
+                enqueue_blocking(&svc, &cx, name, json, EnqueueOpts::default()).map(|_| ())
+            },
+        );
+    }
+
+    // `queue::enqueue(name, message, opts)` — opts is a Rhai Map with
+    // optional `delay_ms` and `max_attempts` keys.
+    {
+        let svc = svc.clone();
+        let cx = cx.clone();
+        module.set_native_fn(
+            "enqueue",
+            move |name: &str, message: Dynamic, opts: Map| -> Result<(), Box<EvalAltResult>> {
+                let json = message_to_json(&message)?;
+                let opts = parse_opts(&opts)?;
+                enqueue_blocking(&svc, &cx, name, json, opts).map(|_| ())
+            },
+        );
+    }
+
+    // `queue::depth(name)` — total rows in the queue.
+    {
+        let svc = svc.clone();
+        let cx = cx.clone();
+        module.set_native_fn(
+            "depth",
+            move |name: &str| -> Result<i64, Box<EvalAltResult>> {
+                let svc = svc.clone();
+                let cx = cx.clone();
+                let name = name.to_string();
+                block_on_u64(async move { svc.depth(&cx, &name).await })
+            },
+        );
+    }
+
+    // `queue::depth_pending(name)` — only currently-claimable rows.
+    {
+        let svc = svc.clone();
+        let cx = cx.clone();
+        module.set_native_fn(
+            "depth_pending",
+            move |name: &str| -> Result<i64, Box<EvalAltResult>> {
+                let svc = svc.clone();
+                let cx = cx.clone();
+                let name = name.to_string();
+                block_on_u64(async move { svc.depth_pending(&cx, &name).await })
+            },
+        );
+    }
+
+    engine.register_static_module("queue", module.into());
+}
+
+/// Run the async enqueue inside the synchronous Rhai context. Mirrors
+/// `pubsub::block_on` but for the queue's `Result<_, QueueError>`.
+fn enqueue_blocking(
+    svc: &Arc<dyn QueueService>,
+    cx: &Arc<SdkCallCx>,
+    name: &str,
+    payload: Json,
+    opts: EnqueueOpts,
+) -> Result<(), Box<EvalAltResult>> {
+    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
+        EvalAltResult::ErrorRuntime(
+            format!("queue: no tokio runtime available: {e}").into(),
+            rhai::Position::NONE,
+        )
+        .into()
+    })?;
+    let svc = svc.clone();
+    let cx = cx.clone();
+    let name = name.to_string();
+    handle
+        .block_on(async move { svc.enqueue(&cx, &name, payload, opts).await })
+        .map(|_| ())
+        .map_err(|err| -> Box<EvalAltResult> {
+            EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
+        })
+}
+
+fn block_on_u64<F>(fut: F) -> Result<i64, Box<EvalAltResult>>
+where
+    F: std::future::Future<Output = Result<u64, QueueError>> + Send + 'static,
+{
+    let handle = TokioHandle::try_current().map_err(|e| -> Box<EvalAltResult> {
+        EvalAltResult::ErrorRuntime(
+            format!("queue: no tokio runtime available: {e}").into(),
+            rhai::Position::NONE,
+        )
+        .into()
+    })?;
+    let n = handle.block_on(fut).map_err(|err| -> Box<EvalAltResult> {
+        EvalAltResult::ErrorRuntime(format!("queue: {err}").into(), rhai::Position::NONE).into()
+    })?;
+    Ok(i64::try_from(n).unwrap_or(i64::MAX))
+}
+
+fn parse_opts(opts: &Map) -> Result<EnqueueOpts, Box<EvalAltResult>> {
+    let mut out = EnqueueOpts::default();
+    if let Some(v) = opts.get("delay_ms") {
+        out.delay_ms = Some(v.as_int().map_err(|_| -> Box<EvalAltResult> {
+            EvalAltResult::ErrorRuntime(
+                "queue::enqueue: opts.delay_ms must be an integer".into(),
+                rhai::Position::NONE,
+            )
+            .into()
+        })?);
+    }
+    if let Some(v) = opts.get("max_attempts") {
+        let n = v.as_int().map_err(|_| -> Box<EvalAltResult> {
+            EvalAltResult::ErrorRuntime(
+                "queue::enqueue: opts.max_attempts must be an integer".into(),
+                rhai::Position::NONE,
+            )
+            .into()
+        })?;
+        out.max_attempts = Some(u32::try_from(n).unwrap_or(0));
+    }
+    Ok(out)
+}
+
+/// Convert a Rhai `Dynamic` into JSON. Mirrors `pubsub::message_to_json`
+/// — blobs become base64, FnPtr / closures are rejected so a script
+/// can't accidentally enqueue something that won't survive a trip
+/// through Postgres + back through the bridge.
+fn message_to_json(value: &Dynamic) -> Result<Json, Box<EvalAltResult>> {
+    if value.is::<rhai::FnPtr>() {
+        return Err(EvalAltResult::ErrorRuntime(
+            "queue::enqueue: messages must not contain FnPtr / closures".into(),
+            rhai::Position::NONE,
+        )
+        .into());
+    }
+    if value.is_blob() {
+        let blob = value.clone().into_blob().unwrap_or_default();
+        return Ok(Json::String(STANDARD.encode(&blob)));
+    }
+    if value.is_unit() {
+        return Ok(Json::Null);
+    }
+    if let Ok(b) = value.as_bool() {
+        return Ok(Json::Bool(b));
+    }
+    if let Ok(i) = value.as_int() {
+        return Ok(Json::Number(i.into()));
+    }
+    if let Ok(f) = value.as_float() {
+        return Ok(serde_json::Number::from_f64(f).map_or(Json::Null, Json::Number));
+    }
+    if value.is_string() {
+        return Ok(Json::String(value.clone().into_string().unwrap_or_default()));
+    }
+    if let Some(arr) = value.clone().try_cast::<Array>() {
+        let mut out = Vec::with_capacity(arr.len());
+        for v in &arr {
+            out.push(message_to_json(v)?);
+        }
+        return Ok(Json::Array(out));
+    }
+    if let Some(map) = value.clone().try_cast::<Map>() {
+        let mut out = serde_json::Map::new();
+        for (k, v) in map {
+            out.insert(k.to_string(), message_to_json(&v)?);
+        }
+        return Ok(Json::Object(out));
+    }
+    Ok(Json::String(value.to_string()))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rhai::Dynamic;
+
+    #[test]
+    fn message_blob_encodes_to_base64_string() {
+        let blob: rhai::Blob = vec![0x01, 0x02, 0x03];
+        let d = Dynamic::from(blob);
+        let json = message_to_json(&d).unwrap();
+        // STANDARD base64 of [1,2,3] is "AQID"
+        assert_eq!(json, Json::String("AQID".into()));
+    }
+
+    #[test]
+    fn message_nested_map_round_trips_through_json() {
+        let mut m = Map::new();
+        m.insert("k".into(), Dynamic::from(42_i64));
+        let d = Dynamic::from(m);
+        let json = message_to_json(&d).unwrap();
+        let expected = serde_json::json!({ "k": 42 });
+        assert_eq!(json, expected);
+    }
+
+    #[test]
+    fn message_rejects_fnptr() {
+        // Build a Rhai FnPtr and confirm we reject it.
+        let f = rhai::FnPtr::new("anything").unwrap();
+        let d = Dynamic::from(f);
+        let err = message_to_json(&d).unwrap_err();
+        assert!(err.to_string().contains("FnPtr"));
+    }
+
+    #[test]
+    fn parse_opts_accepts_partial_overrides() {
+        let mut m = Map::new();
+        m.insert("delay_ms".into(), Dynamic::from(500_i64));
+        let opts = parse_opts(&m).unwrap();
+        assert_eq!(opts.delay_ms, Some(500));
+        assert_eq!(opts.max_attempts, None);
+    }
+
+    #[test]
+    fn parse_opts_max_attempts_only() {
+        let mut m = Map::new();
+        m.insert("max_attempts".into(), Dynamic::from(5_i64));
+        let opts = parse_opts(&m).unwrap();
+        assert_eq!(opts.delay_ms, None);
+        assert_eq!(opts.max_attempts, Some(5));
+    }
+
+    #[test]
+    fn parse_opts_rejects_non_integer_delay() {
+        let mut m = Map::new();
+        m.insert("delay_ms".into(), Dynamic::from("nope"));
+        assert!(parse_opts(&m).is_err());
+    }
+}

crates/executor-core/tests/sdk_queue.rs

@@ -0,0 +1,211 @@
+//! `queue::` SDK bridge integration tests — runs a real Rhai engine
+//! against an in-memory `QueueService` that records the enqueued
+//! `(queue_name, payload, opts)`. Verifies opts handling, blob
+//! base64 encoding, depth/depth_pending pass-through.
+
+use std::collections::BTreeMap;
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use picloud_executor_core::{Engine, ExecRequest, InvocationType, Limits};
+use picloud_shared::{
+    AppId, EnqueueOpts, ExecutionId, NoopDeadLetterService, NoopDocsService, NoopEventEmitter,
+    NoopFilesService, NoopHttpService, NoopKvService, NoopModuleSource, NoopPubsubService,
+    QueueError, QueueMessageId, QueueService, RequestId, ScriptId, ScriptSandbox, SdkCallCx,
+    Services,
+};
+use serde_json::{json, Value};
+
+#[derive(Default)]
+struct RecordingQueue {
+    enqueues: Mutex<Vec<(String, Value, EnqueueOpts)>>,
+    depth: Mutex<u64>,
+    depth_pending: Mutex<u64>,
+}
+
+#[async_trait]
+impl QueueService for RecordingQueue {
+    async fn enqueue(
+        &self,
+        _cx: &SdkCallCx,
+        queue_name: &str,
+        payload: Value,
+        opts: EnqueueOpts,
+    ) -> Result<QueueMessageId, QueueError> {
+        if queue_name.is_empty() {
+            return Err(QueueError::EmptyName);
+        }
+        self.enqueues
+            .lock()
+            .unwrap()
+            .push((queue_name.to_string(), payload, opts));
+        Ok(QueueMessageId::new())
+    }
+
+    async fn depth(&self, _cx: &SdkCallCx, _queue_name: &str) -> Result<u64, QueueError> {
+        Ok(*self.depth.lock().unwrap())
+    }
+
+    async fn depth_pending(
+        &self,
+        _cx: &SdkCallCx,
+        _queue_name: &str,
+    ) -> Result<u64, QueueError> {
+        Ok(*self.depth_pending.lock().unwrap())
+    }
+}
+
+fn make_engine(svc: Arc<RecordingQueue>) -> Arc<Engine> {
+    let services = Services::new(
+        Arc::new(NoopKvService),
+        Arc::new(NoopDocsService),
+        Arc::new(NoopDeadLetterService),
+        Arc::new(NoopEventEmitter),
+        Arc::new(NoopModuleSource),
+        Arc::new(NoopHttpService),
+        Arc::new(NoopFilesService),
+        Arc::new(NoopPubsubService),
+        Arc::new(picloud_shared::NoopSecretsService),
+        Arc::new(picloud_shared::NoopEmailService),
+        Arc::new(picloud_shared::NoopUsersService),
+        svc,
+    );
+    Arc::new(Engine::new(Limits::default(), services))
+}
+
+fn baseline_request(app_id: AppId) -> ExecRequest {
+    let execution_id = ExecutionId::new();
+    ExecRequest {
+        execution_id,
+        request_id: RequestId::new(),
+        script_id: ScriptId::new(),
+        script_name: "queue-test".into(),
+        invocation_type: InvocationType::Http,
+        path: "/queue-test".into(),
+        headers: BTreeMap::new(),
+        body: Value::Null,
+        params: BTreeMap::new(),
+        query: BTreeMap::new(),
+        rest: String::new(),
+        sandbox_overrides: ScriptSandbox::default(),
+        app_id,
+        principal: None,
+        trigger_depth: 0,
+        root_execution_id: execution_id,
+        is_dead_letter_handler: false,
+        event: None,
+    }
+}
+
+async fn run(engine: Arc<Engine>, src: &str, req: ExecRequest) {
+    let src = src.to_string();
+    tokio::task::spawn_blocking(move || engine.execute(&src, req))
+        .await
+        .expect("spawn_blocking should not panic")
+        .expect("script execution should succeed");
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn enqueue_map_payload() {
+    let svc = Arc::new(RecordingQueue::default());
+    let engine = make_engine(svc.clone());
+    run(
+        engine,
+        r#"queue::enqueue("jobs", #{ id: 1, name: "x" });"#,
+        baseline_request(AppId::new()),
+    )
+    .await;
+    let (qn, msg, opts) = svc.enqueues.lock().unwrap()[0].clone();
+    assert_eq!(qn, "jobs");
+    assert_eq!(msg, json!({ "id": 1, "name": "x" }));
+    assert!(opts.delay_ms.is_none());
+    assert!(opts.max_attempts.is_none());
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn enqueue_with_opts_threads_through() {
+    let svc = Arc::new(RecordingQueue::default());
+    let engine = make_engine(svc.clone());
+    run(
+        engine,
+        r#"queue::enqueue("jobs", #{ x: 1 }, #{ delay_ms: 60000, max_attempts: 5 });"#,
+        baseline_request(AppId::new()),
+    )
+    .await;
+    let (qn, _msg, opts) = svc.enqueues.lock().unwrap()[0].clone();
+    assert_eq!(qn, "jobs");
+    assert_eq!(opts.delay_ms, Some(60_000));
+    assert_eq!(opts.max_attempts, Some(5));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn enqueue_blob_base64_encodes() {
+    let svc = Arc::new(RecordingQueue::default());
+    let engine = make_engine(svc.clone());
+    run(
+        engine,
+        r#"
+            let data = base64::decode("aGVsbG8=");
+            queue::enqueue("blobs", #{ raw: data });
+        "#,
+        baseline_request(AppId::new()),
+    )
+    .await;
+    let (_qn, msg, _opts) = svc.enqueues.lock().unwrap()[0].clone();
+    assert_eq!(msg, json!({ "raw": "aGVsbG8=" }));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn depth_returns_service_value() {
+    let svc = Arc::new(RecordingQueue::default());
+    *svc.depth.lock().unwrap() = 99;
+    let engine = make_engine(svc.clone());
+    // Stash result via a known-shape map; the engine returns the eval value.
+    let src = r#"
+        let n = queue::depth("any");
+        #{ statusCode: 200, body: n }
+    "#
+    .to_string();
+    let req = baseline_request(AppId::new());
+    let resp = tokio::task::spawn_blocking({
+        let engine = engine.clone();
+        move || engine.execute(&src, req)
+    })
+    .await
+    .expect("spawn_blocking should not panic")
+    .expect("script execution should succeed");
+    assert_eq!(resp.body, json!(99));
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn enqueue_rejects_fnptr_in_payload() {
+    let svc = Arc::new(RecordingQueue::default());
+    let engine = make_engine(svc.clone());
+    let src = r#"
+        let f = |x| x + 1;
+        queue::enqueue("jobs", #{ closure: f });
+    "#
+    .to_string();
+    let req = baseline_request(AppId::new());
+    let res = tokio::task::spawn_blocking(move || engine.execute(&src, req))
+        .await
+        .expect("spawn_blocking should not panic");
+    let err = res.unwrap_err();
+    let msg = err.to_string();
+    assert!(
+        msg.contains("FnPtr") || msg.contains("closure"),
+        "expected FnPtr rejection, got: {msg}"
+    );
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn enqueue_empty_name_throws() {
+    let svc = Arc::new(RecordingQueue::default());
+    let engine = make_engine(svc.clone());
+    let src = r#"queue::enqueue("", 1);"#.to_string();
+    let req = baseline_request(AppId::new());
+    let res = tokio::task::spawn_blocking(move || engine.execute(&src, req))
+        .await
+        .expect("spawn_blocking should not panic");
+    assert!(res.is_err(), "empty queue_name should throw");
+}