use super::*;
use serde_json::{json, Value};
use std::sync::atomic::AtomicUsize;
use tokio_util::sync::CancellationToken;
#[derive(Default)]
struct Host {
calls: AtomicUsize,
revoked: AtomicUsize,
}
#[async_trait::async_trait]
impl RuntimeActionHost for Host {
async fn send_message(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn send_user_message(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn append_entry(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn set_session_name(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn get_active_tools(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn set_active_tools(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn set_model(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn get_thinking_level(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn set_thinking_level(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn compact(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn get_system_prompt(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn new_session(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn fork(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn navigate_tree(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn switch_session(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn reload(&self, _: Value) -> Result<Value, String> {
unreachable!()
}
async fn child_agent(&self, owner: u64, args: Value) -> Result<Value, String> {
self.calls.fetch_add(1, Ordering::SeqCst);
assert!(
rpi_agent::tool_execution_scope::invocation_signal().is_some(),
"dispatch must retain the originating invocation's cancellation"
);
assert!(
args.get("__rpiRuntimeKey").is_none(),
"capabilities must not reach action payloads"
);
Ok(json!({"owner":owner}))
}
fn cancel_child_agents(&self, _: u64) {
self.revoked.fetch_add(1, Ordering::SeqCst);
}
}
fn call(id: usize, args: Value) -> (i32, Value) {
let mut out = StbString::empty();
let text = args.to_string();
let rc = trampoline_runtime_action(
RuntimeActionId::ChildAgent.into(),
StbStringRef::from_str(&text),
&mut out,
id as *mut c_void,
);
let owned = unsafe {
rpi_plugin_sdk::callbacks::OwnedString::from_raw(out, Some(crate::host_free_string))
};
(rc, serde_json::from_str(&owned.to_string_lossy()).unwrap())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn child_actions_require_a_live_parent_invocation_and_its_capability() {
let host = Arc::new(Host::default());
let root = ActionBridge::new(Handle::current(), host.clone());
let plugin = root.for_plugin("research");
let parent = plugin.for_invocation(false, CancellationToken::new());
let child = plugin.for_invocation(true, CancellationToken::new());
assert_eq!(
call(plugin.user_data() as usize, json!({"op":"start"})).0,
1
);
assert_eq!(
call(parent.user_data() as usize, json!({"op":"start"})).0,
1
);
assert_eq!(
call(
parent.user_data() as usize,
json!({"op":"start","__rpiRuntimeKey":"guessed"})
)
.0,
1
);
assert_eq!(
call(
child.user_data() as usize,
json!({"op":"start","__rpiRuntimeKey":child.invocation_key()})
)
.0,
1
);
{
let _guard = child.enter_invocation();
assert_eq!(
call(
parent.user_data() as usize,
json!({"op":"start","__rpiRuntimeKey":parent.invocation_key()})
)
.0,
1
);
}
let accepted = call(
parent.user_data() as usize,
json!({"op":"start","__rpiRuntimeKey":parent.invocation_key()}),
);
assert_eq!(accepted.0, 0);
assert_eq!(
accepted.1["owner"],
plugin.plugin_identity.as_ref().unwrap().0
);
assert_eq!(host.calls.load(Ordering::SeqCst), 1);
parent.invocation.as_ref().unwrap().2.cancel();
assert_eq!(
call(
parent.user_data() as usize,
json!({"op":"start","__rpiRuntimeKey":parent.invocation_key()})
)
.0,
1
);
}
#[tokio::test]
async fn dropped_candidate_and_unknown_handles_are_rejected_without_pointer_dereference() {
let host = Arc::new(Host::default());
let old = ActionBridge::new(Handle::current(), host.clone());
let old_id = old.user_data() as usize;
let candidate = ActionBridge::new(Handle::current(), host);
let candidate_id = candidate.user_data() as usize;
drop(candidate); assert!(ActionBridge::resolve_handle(old_id).is_some());
assert!(ActionBridge::resolve_handle(candidate_id).is_none());
assert_eq!(
call(candidate_id, json!({})).1["error"],
"expired runtime handle"
);
assert_eq!(
call(usize::MAX, json!({})).1["error"],
"expired runtime handle"
);
let replacement = ActionBridge::new(Handle::current(), Arc::new(Host::default()));
assert_ne!(replacement.user_data() as usize, candidate_id);
}
#[tokio::test]
async fn finishing_an_invocation_does_not_unload_its_plugin_owner() {
let host = Arc::new(Host::default());
let root = ActionBridge::new(Handle::current(), host.clone());
let plugin = root.for_plugin("research");
let invocation = plugin.for_invocation(false, CancellationToken::new());
let id = invocation.user_data() as usize;
drop(invocation);
assert!(ActionBridge::resolve_handle(id).is_none());
assert_eq!(host.revoked.load(Ordering::SeqCst), 0);
root.invalidate();
assert_eq!(host.revoked.load(Ordering::SeqCst), 1);
}
mod cancel_wait;
mod cleanup;
mod driver;
mod pacing;
mod streaming;