(
&self,
prev: Option<&RustStepOutcome>,
item: &serde_json::Value,
)
| 5065 | |
| 5066 | impl NodePipelineStage { |
| 5067 | fn invoke( |
| 5068 | &self, |
| 5069 | prev: Option<&RustStepOutcome>, |
| 5070 | item: &serde_json::Value, |
| 5071 | ) -> Option<RustAgentStepSpec> { |
| 5072 | let previous = prev |
| 5073 | .map(|o| serde_json::to_value(o).unwrap_or(serde_json::Value::Null)) |
| 5074 | .unwrap_or(serde_json::Value::Null); |
| 5075 | let payload = serde_json::json!({ "previous": previous, "item": item }); |
| 5076 | |
| 5077 | let (tx, rx) = std::sync::mpsc::sync_channel::<Option<RustAgentStepSpec>>(1); |
| 5078 | self.tsfn.call_with_return_value( |
| 5079 | payload, |
| 5080 | napi::threadsafe_function::ThreadsafeFunctionCallMode::NonBlocking, |
| 5081 | move |ret: napi::JsUnknown| { |
| 5082 | // Fail-closed: a null/unreadable return stops this chain rather |
| 5083 | // than fabricating a step. |
| 5084 | let _ = tx.send(parse_js_step_spec(ret)); |
| 5085 | Ok(()) |
| 5086 | }, |
| 5087 | ); |
| 5088 | // Fail-closed on timeout/throw: under Fatal strategy a JS throw means |
| 5089 | // the return closure never fires, so the channel stays empty — treat |
| 5090 | // as None (stop this chain) instead of blocking forever. |
| 5091 | tokio::task::block_in_place(|| { |
| 5092 | rx.recv_timeout(std::time::Duration::from_millis(self.timeout_ms)) |
| 5093 | .unwrap_or(None) |
| 5094 | }) |
| 5095 | } |
| 5096 | } |
| 5097 | |
| 5098 | /// Parse a JS pipeline-stage return value into an `AgentStepSpec`, or `None` |
no test coverage detected