(
&self,
env: Env,
items: Vec<serde_json::Value>,
stages: Vec<napi::JsFunction>,
timeout_ms: Option<u32>,
)
| 3371 | ts_return_type = "Promise<Array<StepOutcomeObject | null>>" |
| 3372 | )] |
| 3373 | pub fn pipeline( |
| 3374 | &self, |
| 3375 | env: Env, |
| 3376 | items: Vec<serde_json::Value>, |
| 3377 | stages: Vec<napi::JsFunction>, |
| 3378 | timeout_ms: Option<u32>, |
| 3379 | ) -> napi::Result<napi::JsObject> { |
| 3380 | use napi::threadsafe_function::{ThreadSafeCallContext, ThreadsafeFunction}; |
| 3381 | // Single-object arg so the JS stage signature is `(ctx) => ...`. |
| 3382 | let single_obj = |ctx: ThreadSafeCallContext<serde_json::Value>| { |
| 3383 | Ok(vec![ctx.env.to_js_value(&ctx.value)?]) |
| 3384 | }; |
| 3385 | let timeout = timeout_ms.map(|t| t as u64).unwrap_or(30_000); |
| 3386 | |
| 3387 | // Build the thread-safe functions on the JS thread (JsFunction is not |
| 3388 | // Send), then wrap each as a synchronous PipelineStage the combinator |
| 3389 | // can call from the worker runtime. |
| 3390 | let rust_stages: Vec<RustPipelineStage<serde_json::Value>> = stages |
| 3391 | .into_iter() |
| 3392 | .map(|f| { |
| 3393 | let tsfn: ThreadsafeFunction< |
| 3394 | serde_json::Value, |
| 3395 | napi::threadsafe_function::ErrorStrategy::Fatal, |
| 3396 | > = f.create_threadsafe_function(0, single_obj)?; |
| 3397 | let stage = Arc::new(NodePipelineStage { |
| 3398 | tsfn, |
| 3399 | timeout_ms: timeout, |
| 3400 | }); |
| 3401 | let pipeline_stage: RustPipelineStage<serde_json::Value> = |
| 3402 | Arc::new(move |prev, item| stage.invoke(prev, item)); |
| 3403 | Ok::<_, napi::Error>(pipeline_stage) |
| 3404 | }) |
| 3405 | .collect::<napi::Result<Vec<_>>>()?; |
| 3406 | |
| 3407 | let session = self.inner.clone(); |
| 3408 | let (deferred, promise) = env.create_deferred::<Vec<Option<StepOutcomeObject>>, _>()?; |
| 3409 | get_runtime().spawn(async move { |
| 3410 | let executor = session.agent_executor(); |
| 3411 | let outcomes = execute_pipeline(executor, items, rust_stages, None).await; |
| 3412 | let mapped: Vec<Option<StepOutcomeObject>> = outcomes |
| 3413 | .into_iter() |
| 3414 | .map(|o| o.map(StepOutcomeObject::from)) |
| 3415 | .collect(); |
| 3416 | deferred.resolve(move |_env| Ok(mapped)); |
| 3417 | }); |
| 3418 | Ok(promise) |
| 3419 | } |
| 3420 | |
| 3421 | /// Send a prompt or request and get a streaming event iterator. |
| 3422 | /// |
no test coverage detected