feat: add async middleware C and Go APIs

crates/adaptive/src/acg_component.rs

@@ -582,26 +582,32 @@ pub(crate) fn create_acg_llm_request_intercept(
     provider: String,
     plugin: Arc<dyn ProviderPlugin>,
 ) -> LlmRequestInterceptFn {
-    Arc::new(move |_name: &str, request: LlmRequest, annotated| {
-        let input_content = request.content.clone();
-        let translated =
-            translate_request(&request, &agent_id, &provider, plugin.as_ref(), &hot_cache)
-                .unwrap_or(request);
-        if annotated.is_some() && translated.content != input_content {
-            let translated_annotated = build_semantic_request_view(&translated)
-                .map_err(|error| nemo_relay::error::FlowError::Internal(error.to_string()))?
-                .annotated_request;
-            return Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                LlmRequest {
-                    headers: translated.headers,
-                    content: input_content,
-                },
-                Some(translated_annotated),
-            ));
-        }
-        Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-            translated, annotated,
-        ))
+    Arc::new(move |_name: String, request: LlmRequest, annotated| {
+        let hot_cache = hot_cache.clone();
+        let agent_id = agent_id.clone();
+        let provider = provider.clone();
+        let plugin = plugin.clone();
+        Box::pin(async move {
+            let input_content = request.content.clone();
+            let translated =
+                translate_request(&request, &agent_id, &provider, plugin.as_ref(), &hot_cache)
+                    .unwrap_or(request);
+            if annotated.is_some() && translated.content != input_content {
+                let translated_annotated = build_semantic_request_view(&translated)
+                    .map_err(|error| nemo_relay::error::FlowError::Internal(error.to_string()))?
+                    .annotated_request;
+                return Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                    LlmRequest {
+                        headers: translated.headers,
+                        content: input_content,
+                    },
+                    Some(translated_annotated),
+                ));
+            }
+            Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                translated, annotated,
+            ))
+        })
     })
 }
 

crates/adaptive/src/adaptive_hints_intercept.rs

@@ -174,14 +174,14 @@ impl AdaptiveHintsIntercept {
     pub fn into_request_fn(self) -> LlmRequestInterceptFn {
         let this = Arc::new(self);
         Arc::new(
-            move |_name: &str,
+            move |_name: String,
                   mut request: LlmRequest,
                   mut annotated: Option<AnnotatedLlmRequest>| {
+                let this = this.clone();
                 let scope_path = extract_scope_path();
                 let manual_ls = read_manual_latency_sensitivity();
                 let scope_depth = scope_path.len();
                 let call_index = this.call_counter.fetch_add(1, Ordering::Relaxed);
-
                 let effective_agent_id = this.effective_agent_id();
                 let cached_hints =
                     this.load_hints(&scope_path, &effective_agent_id, call_index, scope_depth);
@@ -196,9 +196,9 @@ impl AdaptiveHintsIntercept {
                     inject_agent_hints(&mut request, &mut annotated, &hints);
                 }
 
-                Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                    request, annotated,
-                ))
+                let outcome =
+                    nemo_relay::api::llm::LlmRequestInterceptOutcome::new(request, annotated);
+                Box::pin(async move { Ok(outcome) })
             },
         )
     }

crates/adaptive/src/lib.rs

@@ -14,6 +14,10 @@
 pub(crate) static TEST_GLOBAL_CONTEXT_MUTEX: tokio::sync::Mutex<()> =
     tokio::sync::Mutex::const_new(());
 
+#[cfg(test)]
+#[path = "../tests/support/mod.rs"]
+pub(crate) mod test_support;
+
 pub mod acg;
 pub mod acg_component;
 pub mod acg_learner;

crates/adaptive/tests/integration/runtime_integration_tests.rs

@@ -605,6 +605,7 @@ async fn test_adaptive_plugin_registers_and_passes_calls_through() {
             content: json!({"messages": []}),
         },
     )
+    .await
     .unwrap();
     assert_eq!(request.request.content["messages"], json!([]));
 
@@ -739,9 +740,11 @@ impl Plugin for HeaderPlugin {
                 false,
                 Arc::new(|_name, mut request, annotated| {
                     request.headers.insert("x-plugin".into(), json!("set"));
-                    Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                        request, annotated,
-                    ))
+                    Box::pin(async move {
+                        Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                            request, annotated,
+                        ))
+                    })
                 }),
             )?;
             ctx.register_tool_request_intercept(
@@ -752,7 +755,7 @@ impl Plugin for HeaderPlugin {
                     if let Json::Object(ref mut map) = args {
                         map.insert("x-tool-plugin".into(), json!(true));
                     }
-                    Ok(args)
+                    Box::pin(async move { Ok(args) })
                 }),
             )?;
             ctx.register_llm_execution_intercept(
@@ -823,6 +826,7 @@ async fn test_top_level_plugin_registers_request_and_execution_intercepts() {
             content: json!({"messages": []}),
         },
     )
+    .await
     .unwrap();
     assert_eq!(request.request.headers.get("x-plugin"), Some(&json!("set")));
 

crates/adaptive/tests/support/mod.rs

@@ -0,0 +1,12 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::future::Future;
+
+pub(crate) fn block_on<F: Future>(future: F) -> F::Output {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("test runtime should build")
+        .block_on(future)
+}

crates/adaptive/tests/unit/acg_component_tests.rs

@@ -1087,11 +1087,11 @@ fn acg_component_request_intercept_passes_original_request_and_annotation_when_t
         plugin,
     );
 
-    let outcome = intercept(
-        "anthropic",
+    let outcome = crate::test_support::block_on(intercept(
+        "anthropic".to_string(),
         invalid_request.clone(),
         Some(annotated.clone()),
-    )
+    ))
     .expect("request intercept should pass through");
     let translated = outcome.request;
     let returned_annotated = outcome.annotated_request;
@@ -1335,7 +1335,12 @@ fn acg_component_request_intercept_rewrites_annotation_without_mutating_provider
         plugin,
     );
 
-    let outcome = intercept("anthropic", request, Some(original_annotation.clone())).unwrap();
+    let outcome = crate::test_support::block_on(intercept(
+        "anthropic".to_string(),
+        request,
+        Some(original_annotation.clone()),
+    ))
+    .unwrap();
 
     assert_eq!(outcome.request.content, original_content);
     let annotation = outcome

crates/adaptive/tests/unit/adaptive_hints_intercept_tests.rs

@@ -4,6 +4,7 @@
 //! Unit tests for adaptive hints intercept in the NeMo Relay adaptive crate.
 
 use super::*;
+
 use std::sync::{Mutex, OnceLock};
 
 use crate::trie::data_models::{LlmCallPrediction, PredictionMetrics};
@@ -196,14 +197,14 @@ fn test_adaptive_hints_intercept_injects_prediction_hints_and_manual_override()
         stream: None,
         extra: serde_json::Map::new(),
     };
-    let outcome = req_fn(
-        "model",
+    let outcome = crate::test_support::block_on(req_fn(
+        "model".to_string(),
         LlmRequest {
             headers: serde_json::Map::new(),
             content: serde_json::json!({}),
         },
         Some(annotated.clone()),
-    )
+    ))
     .unwrap();
     let request = outcome.request;
     let returned_annotated = outcome.annotated_request;
@@ -266,14 +267,14 @@ fn test_adaptive_hints_intercept_uses_defaults_and_ignores_poisoned_cache() {
     }));
     let req_fn =
         AdaptiveHintsIntercept::new(hot_cache, "fallback-agent".to_string()).into_request_fn();
-    let outcome = req_fn(
-        "model",
+    let outcome = crate::test_support::block_on(req_fn(
+        "model".to_string(),
         LlmRequest {
             headers: serde_json::Map::new(),
             content: serde_json::json!({}),
         },
         None,
-    )
+    ))
     .unwrap();
     let request = outcome.request;
     let annotated = outcome.annotated_request;
@@ -305,14 +306,14 @@ fn test_adaptive_hints_intercept_uses_defaults_and_ignores_poisoned_cache() {
     });
     let poisoned_req_fn =
         AdaptiveHintsIntercept::new(poisoned_cache, "fallback-agent".to_string()).into_request_fn();
-    let poisoned_outcome = poisoned_req_fn(
-        "model",
+    let poisoned_outcome = crate::test_support::block_on(poisoned_req_fn(
+        "model".to_string(),
         LlmRequest {
             headers: serde_json::Map::new(),
             content: serde_json::json!({"existing": true}),
         },
         None,
-    )
+    ))
     .unwrap();
     let poisoned_request = poisoned_outcome.request;
     assert!(

crates/adaptive/tests/unit/plugin_component_tests.rs

@@ -364,6 +364,7 @@ async fn adaptive_plugin_registers_runtime_and_rolls_back_registration() {
             content: json!({}),
         },
     )
+    .await
     .unwrap();
     assert!(request.request.headers.is_empty());
 

crates/adaptive/tests/unit/runtime_features_tests.rs

@@ -139,9 +139,11 @@ fn assert_llm_request_intercept_registered(name: &str) {
             i32::MAX,
             false,
             Arc::new(|_name, request, annotated| {
-                Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                    request, annotated,
-                ))
+                Box::pin(async move {
+                    Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                        request, annotated,
+                    ))
+                })
             }),
         ),
         name,
@@ -154,9 +156,11 @@ fn assert_llm_request_intercept_absent(name: &str) {
         i32::MAX,
         false,
         Arc::new(|_name, request, annotated| {
-            Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                request, annotated,
-            ))
+            Box::pin(async move {
+                Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                    request, annotated,
+                ))
+            })
         }),
     )
     .unwrap();
@@ -565,6 +569,7 @@ async fn adaptive_hints_feature_registers_request_intercept() {
             content: json!({}),
         },
     )
+    .await
     .unwrap();
     assert!(request.request.headers.contains_key(AGENT_HINTS_HEADER_KEY));
 
@@ -730,9 +735,11 @@ async fn registration_context_registers_all_supported_callback_types() {
         5,
         false,
         Arc::new(|_name, request, annotated| {
-            Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
-                request, annotated,
-            ))
+            Box::pin(async move {
+                Ok(nemo_relay::api::llm::LlmRequestInterceptOutcome::new(
+                    request, annotated,
+                ))
+            })
         }),
     )
     .unwrap();

crates/adaptive/tests/unit/runtime_tests.rs

@@ -637,6 +637,7 @@ async fn adaptive_runtime_bind_scope_requires_registration_and_passes_through_wi
     };
 
     let translated = llm_request_intercepts("anthropic", request.clone())
+        .await
         .expect("request intercept chain should pass through when no hot-cache state exists");
 
     assert_eq!(translated.request.content, request.content);

crates/cli/src/sessions/mod.rs

@@ -582,6 +582,10 @@ impl SessionManager {
                 .map_err(CliError::from)
             })
             .await?;
+        // Manual lifecycle events publish on the serial dispatcher. This
+        // test-only seam returns after the matching end event is observable so
+        // a subsequent synthetic provider call cannot overtake it.
+        nemo_relay::api::subscriber::flush_subscribers().map_err(CliError::from)?;
         let mut sessions = self.inner.lock().await;
         if let Some(session) = sessions.get_mut(&session_id) {
             session.record_completed_llm_response(response_for_hints, owner_subagent_id);
@@ -784,9 +788,9 @@ impl Session {
                     NormalizedEvent::SubagentStarted(event) => self.start_subagent(event).await,
                     NormalizedEvent::SubagentEnded(event) => self.end_subagent(event).await,
                     NormalizedEvent::LlmHint(event) => self.add_llm_hint(event),
-                    NormalizedEvent::LlmStarted(event) => self.start_hook_llm(event),
-                    NormalizedEvent::LlmEnded(event) => self.end_hook_llm(event),
-                    NormalizedEvent::ToolStarted(event) => self.start_tool(event),
+                    NormalizedEvent::LlmStarted(event) => self.start_hook_llm(event).await,
+                    NormalizedEvent::LlmEnded(event) => self.end_hook_llm(event).await,
+                    NormalizedEvent::ToolStarted(event) => self.start_tool(event).await,
                     NormalizedEvent::ToolEnded(event) => self.end_tool(event).await,
                     NormalizedEvent::PromptSubmitted(event) => self.start_turn(event).await,
                     NormalizedEvent::Compaction(event) => self.mark("compaction", event),
@@ -1142,8 +1146,8 @@ impl Session {
         if self.turn_scope.is_none() {
             return Ok(Vec::new());
         }
-        self.close_active_llms(reason)?;
-        self.close_active_tools(reason)?;
+        self.close_active_llms(reason).await?;
+        self.close_active_tools(reason).await?;
         let closed_subagents = self.close_active_subagents(reason).await?;
         let output = self.last_turn_llm_output.take().unwrap_or(output);
         self.clear_correlation_state();
@@ -1182,7 +1186,7 @@ impl Session {
     }
 
     // Ends all active hook-observed LLM calls before closing their containing scopes.
-    fn close_active_llms(&mut self, reason: &str) -> Result<(), CliError> {
+    async fn close_active_llms(&mut self, reason: &str) -> Result<(), CliError> {
         let active_llms: Vec<_> = self.llms.drain().map(|(_, handle)| handle).collect();
         for handle in active_llms {
             llm_call_end(
@@ -1198,7 +1202,7 @@ impl Session {
 
     // Ends all active tool calls with a synthetic close result before ending their containing scopes.
     // Draining first avoids holding mutable map state while the runtime emits lifecycle events.
-    fn close_active_tools(&mut self, reason: &str) -> Result<(), CliError> {
+    async fn close_active_tools(&mut self, reason: &str) -> Result<(), CliError> {
         let active_tools: Vec<_> = self
             .tools
             .drain()
@@ -1428,7 +1432,7 @@ impl Session {
     // ignored so repeated pre hooks do not create parallel handles for one provider call. Aliased
     // child-session LLMs carry their subagent owner in metadata and are resolved by
     // `hook_llm_owner`.
-    fn start_hook_llm(&mut self, event: LlmEvent) -> Result<(), CliError> {
+    async fn start_hook_llm(&mut self, event: LlmEvent) -> Result<(), CliError> {
         self.ensure_turn_started(event.metadata.clone())?;
         if self.llms.contains_key(&event.api_call_id) {
             return Ok(());
@@ -1454,7 +1458,7 @@ impl Session {
     // Ends a hook-observed LLM call, synthesizing a start if only the post hook arrives. The same
     // alias metadata recovery used by `start_hook_llm` keeps post-only aliased child LLMs under the
     // subagent instead of falling back to the root agent.
-    fn end_hook_llm(&mut self, event: LlmEvent) -> Result<(), CliError> {
+    async fn end_hook_llm(&mut self, event: LlmEvent) -> Result<(), CliError> {
         self.ensure_turn_started(event.metadata.clone())?;
         let (parent, metadata) = self.hook_llm_owner(event.metadata);
         let handle = match self.llms.remove(&event.api_call_id) {
@@ -1511,7 +1515,7 @@ impl Session {
     // Starts a tool call under an explicit subagent when available, otherwise under the turn
     // scope. Duplicate tool IDs are ignored so repeated pre-tool hooks do not create parallel
     // handles for one agent tool invocation.
-    fn start_tool(&mut self, event: ToolEvent) -> Result<(), CliError> {
+    async fn start_tool(&mut self, event: ToolEvent) -> Result<(), CliError> {
         self.ensure_turn_started(event.metadata.clone())?;
         if self.tools.contains_key(&event.tool_call_id) {
             return Ok(());
@@ -1529,7 +1533,7 @@ impl Session {
         let active_tool_arguments = arguments.clone();
         let active_tool_name = event.tool_name.clone();
         let active_tool_owner_subagent_id = owner.subagent_id.clone();
-        tool_conditional_execution(event.tool_name.as_str(), &arguments)?;
+        tool_conditional_execution(event.tool_name.as_str(), &arguments).await?;
         let metadata = tool_correlation_metadata(
             self.event_identity_metadata(event.metadata),
             owner.status,