Skip to content

fix: address runtime and tooling findings #633

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rapids-bot merged 2 commits into from
Aug 3, 2026
Merged

fix: address runtime and tooling findings #633

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cbba5c0
fix: address runtime and tooling findings
willkill07 Aug 1, 2026
40db6d5
test(core): derive native compatibility requirement
willkill07 Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/core/src/api/runtime/scope_stack.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -452,7 +452,7 @@ pub fn create_scope_stack_from_propagation(
/// # Examples
///
/// ```no_run
/// # async fn example() -> nemo_relay::Result<()> {
/// # async fn example() -> nemo_relay::error::Result<()> {
/// use nemo_relay::api::runtime::{TASK_SCOPE_STACK, fork_scope_stack};
///
/// let stack = fork_scope_stack()?;
Expand Down
152 changes: 78 additions & 74 deletions crates/core/src/plugin/dynamic/native.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2218,80 +2218,11 @@ unsafe extern "C" fn native_async_next_invoke_stream(
return;
}
continuation_context
.run(async move {
let mut callback_guard = callback_guard;
let result = AssertUnwindSafe(async {
match next_fn(request).await {
Ok(mut stream) => {
while let Some(item) = stream.next().await {
match item {
Ok(chunk) => {
if let Some(chunk) = native_string_from_json(&chunk) {
let keep_going = unsafe {
cb(
user_data as *mut c_void,
chunk,
ptr::null(),
false,
)
};
unsafe {
native_string_free(chunk);
}
if !keep_going {
callback_guard.finish();
return;
}
} else {
break;
}
}
Err(error) => {
if let Some(message) =
native_string_from_str(&error.to_string())
{
unsafe {
let _ = cb(
user_data as *mut c_void,
ptr::null(),
message,
false,
);
native_string_free(message);
}
callback_guard.finish();
}
return;
}
}
}
unsafe {
let _ =
cb(user_data as *mut c_void, ptr::null(), ptr::null(), true);
}
callback_guard.finish();
}
Err(error) => {
if let Some(message) = native_string_from_str(&error.to_string()) {
unsafe {
let _ =
cb(user_data as *mut c_void, ptr::null(), message, false);
native_string_free(message);
}
callback_guard.finish();
}
}
}
})
.catch_unwind()
.await;
if let Err(payload) = result {
callback_guard.fail(&format!(
"native async stream continuation panicked: {}",
panic_payload_message(payload.as_ref())
));
}
})
.run(deliver_native_async_next_stream(
next_fn,
request,
callback_guard,
))
.await;
output_stream_for_cleanup
.downstream_aborts
Expand All @@ -2305,6 +2236,79 @@ unsafe extern "C" fn native_async_next_invoke_stream(
NemoRelayStatus::Ok
}

async fn deliver_native_async_next_stream(
next_fn: LlmStreamExecutionNextFn,
request: LlmRequest,
mut callback_guard: NativeAsyncStreamCallbackGuard,
) {
let result = AssertUnwindSafe(async {
match next_fn(request).await {
Ok(stream) => forward_native_async_next_stream(stream, &mut callback_guard).await,
Err(error) => callback_guard.fail(&error.to_string()),
}
})
.catch_unwind()
.await;
if let Err(payload) = result {
callback_guard.fail(&format!(
"native async stream continuation panicked: {}",
panic_payload_message(payload.as_ref())
));
}
}

async fn forward_native_async_next_stream(
stream: LlmJsonStream,
callback_guard: &mut NativeAsyncStreamCallbackGuard,
) {
forward_native_async_next_stream_with(stream, callback_guard, native_string_from_json).await;
}

async fn forward_native_async_next_stream_with(
mut stream: LlmJsonStream,
callback_guard: &mut NativeAsyncStreamCallbackGuard,
to_native_string: impl Fn(&Json) -> Option<*mut NemoRelayNativeString>,
) {
while let Some(item) = stream.next().await {
match item {
Ok(chunk) => {
let Some(chunk) = to_native_string(&chunk) else {
callback_guard.fail(
"failed to serialize or allocate native async stream continuation chunk",
);
return;
};
let keep_going = unsafe {
(callback_guard.cb)(
callback_guard.user_data as *mut c_void,
chunk,
ptr::null(),
false,
)
};
unsafe { native_string_free(chunk) };
if !keep_going {
callback_guard.finish();
return;
}
}
Err(error) => {
callback_guard.fail(&error.to_string());
return;
}
}
}
unsafe {
let _ = (callback_guard.cb)(
callback_guard.user_data as *mut c_void,
ptr::null(),
ptr::null(),
true,
);
}
callback_guard.finish();
}

fn wrap_native_async_tool_json(
instance: Arc<NativePluginInstance>,
cb: NemoRelayNativeAsyncMiddlewareCb,
Expand Down
Loading
Loading