use crate::agentic::{generic_type_in_agent_method_error, generic_type_in_agent_return_type_error};
use crate::rpc_client_common::{
FunctionOutputInfo, collect_kept_args, collect_typed_params, decode_result_value,
encode_value_only_carrier, find_generic_param_in_inputs, find_generic_param_in_return,
is_principal_param, is_static_method, positional_record_schema_value,
};
use quote::{format_ident, quote};
use std::collections::{HashMap, HashSet};
use syn::ItemTrait;
pub fn get_remote_client(
item_trait: &ItemTrait,
constructor_data_value_param_defs: &[proc_macro2::TokenStream],
constructor_data_value_param_idents: &[proc_macro2::Ident],
constructor_agent_config_param_defs: &[proc_macro2::TokenStream],
constructor_agent_config_param_idents: &[proc_macro2::Ident],
agent_type_parameter_names: &[String],
agent_is_durable: bool,
) -> proc_macro2::TokenStream {
get_remote_client_for_type(
item_trait,
&item_trait.ident.to_string(),
constructor_data_value_param_defs,
constructor_data_value_param_idents,
constructor_agent_config_param_defs,
constructor_agent_config_param_idents,
agent_type_parameter_names,
agent_is_durable,
false,
)
}
pub fn get_remote_client_for_type(
item_trait: &ItemTrait,
type_name: &str,
constructor_data_value_param_defs: &[proc_macro2::TokenStream],
constructor_data_value_param_idents: &[proc_macro2::Ident],
constructor_agent_config_param_defs: &[proc_macro2::TokenStream],
constructor_agent_config_param_idents: &[proc_macro2::Ident],
agent_type_parameter_names: &[String],
agent_is_durable: bool,
construction_fallible: bool,
) -> proc_macro2::TokenStream {
let remote_client_type_name = format_ident!("{}Client", item_trait.ident);
let constructor_schema_type_name = format_ident!("__{}Constructor", item_trait.ident);
let remote_client_with_scope_card_type_name =
format_ident!("{}ClientWithScopeCard", item_trait.ident);
let RemoteAgentMethodsInfo {
methods_impl,
scoped_methods_impl,
mut method_names,
} = get_remote_agent_methods_info(
item_trait,
agent_type_parameter_names,
agent_is_durable,
construction_fallible,
);
let get_method_ident = agent_is_durable.then(|| {
if method_names.contains("get") {
method_names.fresh_ident("get_")
} else {
method_names.fresh_ident("get")
}
});
let new_phantom_method_ident = method_names.fresh_ident("new_phantom");
let get_phantom_method_ident = method_names.fresh_ident("get_phantom");
let get_with_config_method_ident = (agent_is_durable
&& !constructor_agent_config_param_defs.is_empty())
.then(|| method_names.fresh_ident("get_with_config"));
let new_phantom_with_config_method_ident = (!constructor_agent_config_param_defs.is_empty())
.then(|| method_names.fresh_ident("new_phantom_with_config"));
let get_phantom_with_config_method_ident = (!constructor_agent_config_param_defs.is_empty())
.then(|| method_names.fresh_ident("get_phantom_with_config"));
let phantom_id_accessor_ident =
agent_is_durable.then(|| method_names.fresh_ident("phantom_id"));
let get_agent_id_accessor_ident =
agent_is_durable.then(|| method_names.fresh_ident("get_agent_id"));
let agent_id_helper_ident = method_names.fresh_ident("agent_id");
let bind_helper_ident = agent_is_durable.then(|| method_names.fresh_ident("bind"));
let with_scope_card_ident = method_names.fresh_ident("with_scope_card");
let set_default_scope_card_ident = method_names.fresh_ident("set_default_scope_card");
let clear_default_scope_card_ident = method_names.fresh_ident("clear_default_scope_card");
let constructor_param_idents = constructor_data_value_param_idents
.iter()
.chain(constructor_agent_config_param_idents)
.cloned()
.collect::<Vec<_>>();
let phantom_id_param_ident = fresh_param_ident(&constructor_param_idents, "phantom_id");
let rpc_config_params_ident =
fresh_param_ident(&constructor_param_idents, "__golem_rpc_config_params");
let phantom_uuid_ident = fresh_param_ident(&constructor_param_idents, "phantom_uuid");
let constructor_value_ident = fresh_param_ident(&constructor_param_idents, "constructor_value");
let agent_id_ident = fresh_param_ident(&constructor_param_idents, "agent_id");
let encode_constructor = generate_constructor_data_value_params_encoding(
constructor_data_value_param_idents,
&constructor_value_ident,
construction_fallible,
);
let constructor_schema_fields = constructor_data_value_param_defs;
let constructor_schema_definition =
(construction_fallible && agent_type_parameter_names.is_empty()).then(|| {
quote! {
#[doc(hidden)]
#[derive(golem_rust::IntoSchema)]
struct #constructor_schema_type_name {
#(#constructor_schema_fields,)*
}
}
});
let agent_id_body = quote! {{
#encode_constructor
#constructor_value_ident
}};
let agent_config_params_as_rpc_param = {
let add_rpc_params_entries = constructor_agent_config_param_idents.iter().map(|param_ident|
quote! { #rpc_config_params_ident.append(&mut ::golem_rust::agentic::IntoRpcConfigParam::into_rpc_param(#param_ident, &[])); }
);
quote! {
{
let mut #rpc_config_params_ident = Vec::new();
#(#add_rpc_params_entries)*
#rpc_config_params_ident
}
}
};
let constructor_return_type = if construction_fallible {
quote! { Result<#remote_client_type_name, golem_rust::GolemReflectError> }
} else {
quote! { #remote_client_type_name }
};
let build_constructor_body =
|prelude: proc_macro2::TokenStream,
phantom_wire: proc_macro2::TokenStream,
phantom_struct: proc_macro2::TokenStream,
config: proc_macro2::TokenStream| {
if construction_fallible {
quote! {
#encode_constructor
#prelude
let __golem_constructor = golem_rust::encode_schema_value(&#constructor_value_ident)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?;
let #agent_id_ident = golem_rust::golem_agentic::golem::agent::host::make_agent_id(
#type_name,
__golem_constructor,
#phantom_wire,
)
.map_err(golem_rust::agentic::reflection::agentic_agent_error_to_reflect)?;
let __golem_constructor = golem_rust::encode_schema_value(&#constructor_value_ident)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?;
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::create(
#type_name,
__golem_constructor,
#phantom_wire,
#config,
)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)?;
Ok(#remote_client_type_name {
agent_id: #agent_id_ident,
phantom_id: #phantom_struct,
wasm_rpc,
__golem_default_scope_card: ::core::default::Default::default(),
})
}
} else {
quote! {
#encode_constructor
#prelude
let #agent_id_ident = golem_rust::golem_agentic::golem::agent::host::make_agent_id(
#type_name,
#constructor_value_ident(),
#phantom_wire,
)
.expect("Internal Error: Failed to make agent id");
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::create(
#type_name,
#constructor_value_ident(),
#phantom_wire,
#config,
)
.unwrap_or_else(|error| panic!("Internal Error: Agent type not registered: {error:?}"));
#remote_client_type_name {
agent_id: #agent_id_ident,
phantom_id: #phantom_struct,
wasm_rpc,
__golem_default_scope_card: ::core::default::Default::default(),
}
}
}
};
let optional_get_with_config_impl = if agent_is_durable
&& !constructor_agent_config_param_defs.is_empty()
{
let get_with_config_method_ident = get_with_config_method_ident
.as_ref()
.expect("durable agents with config allocate get_with_config");
let body = build_constructor_body(
quote! {},
quote! { None },
quote! { None },
agent_config_params_as_rpc_param.clone(),
);
quote! {
pub fn #get_with_config_method_ident(#(#constructor_data_value_param_defs,)* #(#constructor_agent_config_param_defs,)*) -> #constructor_return_type {
#body
}
}
} else {
quote! {}
};
let new_phantom_doc = if agent_is_durable {
"Creates a new agent instance with a fresh random phantom id."
} else {
"Creates a local logical proxy; each invocation receives a fresh final identity."
};
let optional_new_phantom_with_config_impl = if !constructor_agent_config_param_defs.is_empty() {
let new_phantom_with_config_method_ident = new_phantom_with_config_method_ident
.as_ref()
.expect("agents with config allocate new_phantom_with_config");
let body = if agent_is_durable {
build_constructor_body(
quote! { let #phantom_uuid_ident = golem_rust::Uuid::new_v4(); },
quote! { Some(#phantom_uuid_ident.into()) },
quote! { Some(#phantom_uuid_ident) },
agent_config_params_as_rpc_param.clone(),
)
} else {
build_ephemeral_constructor_body(
&remote_client_type_name,
type_name,
&encode_constructor,
&constructor_value_ident,
quote! { None },
agent_config_params_as_rpc_param.clone(),
construction_fallible,
)
};
quote! {
#[doc = #new_phantom_doc]
pub fn #new_phantom_with_config_method_ident(#(#constructor_data_value_param_defs,)* #(#constructor_agent_config_param_defs,)*) -> #constructor_return_type {
#body
}
}
} else {
quote! {}
};
let optional_get_phantom_with_config_impl = if !constructor_agent_config_param_defs.is_empty() {
let get_phantom_with_config_method_ident = get_phantom_with_config_method_ident
.as_ref()
.expect("agents with config allocate get_phantom_with_config");
let body = if agent_is_durable {
build_constructor_body(
quote! {},
quote! { Some(#phantom_id_param_ident.into()) },
quote! { Some(#phantom_id_param_ident) },
agent_config_params_as_rpc_param.clone(),
)
} else {
build_ephemeral_constructor_body(
&remote_client_type_name,
type_name,
&encode_constructor,
&constructor_value_ident,
quote! { Some(#phantom_id_param_ident.into()) },
agent_config_params_as_rpc_param.clone(),
construction_fallible,
)
};
quote! {
pub fn #get_phantom_with_config_method_ident(#phantom_id_param_ident: golem_rust::Uuid, #(#constructor_data_value_param_defs,)* #(#constructor_agent_config_param_defs,)*) -> #constructor_return_type {
#body
}
}
} else {
quote! {}
};
let get_impl = if agent_is_durable {
let get_method_ident = get_method_ident
.as_ref()
.expect("durable agents allocate get");
let body = build_constructor_body(
quote! {},
quote! { None },
quote! { None },
quote! { Vec::new() },
);
quote! {
pub fn #get_method_ident(#(#constructor_data_value_param_defs,)*) -> #constructor_return_type {
#body
}
}
} else {
quote! {}
};
let new_phantom_body = if agent_is_durable {
build_constructor_body(
quote! { let #phantom_uuid_ident = golem_rust::Uuid::new_v4(); },
quote! { Some(#phantom_uuid_ident.into()) },
quote! { Some(#phantom_uuid_ident) },
quote! { Vec::new() },
)
} else {
build_ephemeral_constructor_body(
&remote_client_type_name,
type_name,
&encode_constructor,
&constructor_value_ident,
quote! { None },
quote! { Vec::new() },
construction_fallible,
)
};
let get_phantom_body = if agent_is_durable {
build_constructor_body(
quote! {},
quote! { Some(#phantom_id_param_ident.into()) },
quote! { Some(#phantom_id_param_ident) },
quote! { Vec::new() },
)
} else {
build_ephemeral_constructor_body(
&remote_client_type_name,
type_name,
&encode_constructor,
&constructor_value_ident,
quote! { Some(#phantom_id_param_ident.into()) },
quote! { Vec::new() },
construction_fallible,
)
};
let durable_fields = agent_is_durable.then(|| {
quote! {
agent_id: String,
phantom_id: Option<golem_rust::Uuid>,
}
});
let get_phantom_impl = quote! {
pub fn #get_phantom_method_ident(#phantom_id_param_ident: golem_rust::Uuid, #(#constructor_data_value_param_defs,)*) -> #constructor_return_type { #get_phantom_body }
};
let accessors = agent_is_durable.then(|| {
quote! {
pub fn #phantom_id_accessor_ident(&self) -> Option<golem_rust::Uuid> { self.phantom_id }
pub fn #get_agent_id_accessor_ident(&self) -> String { self.agent_id.clone() }
}
});
let create_bound_transport = if construction_fallible {
quote! {
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::create(
#type_name,
golem_rust::encode_schema_value(&parts.constructor_value)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?,
parts.phantom_id.map(Into::into),
Vec::new(),
)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)?;
}
} else {
quote! {
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::new(
#type_name,
golem_rust::encode_schema_value(&parts.constructor_value)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?,
parts.phantom_id.map(Into::into),
Vec::new(),
);
}
};
let bind_impl = (construction_fallible
&& agent_is_durable
&& agent_type_parameter_names.is_empty())
.then(|| {
quote! {
pub fn #bind_helper_ident(agent_id: &golem_rust::ParsedAgentId)
-> Result<Self, golem_rust::GolemReflectError>
{
let parts = agent_id.parts()?;
if parts.type_name != #type_name {
return Err(golem_rust::GolemReflectError::InvalidType(format!(
"full client expects `{}`, identity is `{}`",
#type_name,
parts.type_name,
)));
}
let expected = golem_rust::schema::try_into_schema_graph::<#constructor_schema_type_name>()
.map_err(|error| golem_rust::GolemReflectError::InvalidType(error.to_string()))?;
golem_rust::SchemaRef::new(expected).validate_value(&parts.constructor_value)?;
#create_bound_transport
Ok(Self {
agent_id: agent_id.as_str().to_string(),
phantom_id: parts.phantom_id,
wasm_rpc,
__golem_default_scope_card: ::core::default::Default::default(),
})
}
}
});
let agent_id_helper_impl = construction_fallible.then(|| quote! {
pub fn #agent_id_helper_ident(
#(#constructor_data_value_param_defs,)*
#phantom_id_param_ident: Option<golem_rust::Uuid>,
) -> Result<golem_rust::ParsedAgentId, golem_rust::GolemReflectError> {
let constructor = #agent_id_body;
golem_rust::ParsedAgentId::from_value(#type_name, constructor, #phantom_id_param_ident)
}
});
quote! {
#constructor_schema_definition
pub struct #remote_client_type_name {
#durable_fields
wasm_rpc: golem_rust::golem_agentic::golem::agent::host::WasmRpc,
__golem_default_scope_card: golem_rust::permissions::__DefaultScopeCard,
}
impl #remote_client_type_name {
#agent_id_helper_impl
#bind_impl
#get_impl
#optional_get_with_config_impl
#[doc = #new_phantom_doc]
pub fn #new_phantom_method_ident(#(#constructor_data_value_param_defs,)*) -> #constructor_return_type {
#new_phantom_body
}
#optional_new_phantom_with_config_impl
#get_phantom_impl
#optional_get_phantom_with_config_impl
#accessors
pub fn #with_scope_card_ident<'a>(
&'a self,
scope_card: &'a golem_rust::permissions::ScopeCard,
) -> #remote_client_with_scope_card_type_name<'a> {
#remote_client_with_scope_card_type_name {
client: self,
scope_card,
}
}
pub fn #set_default_scope_card_ident(
&mut self,
scope_card: &golem_rust::permissions::ScopeCard,
) {
self.__golem_default_scope_card.set(scope_card);
}
pub fn #clear_default_scope_card_ident(&mut self) {
self.__golem_default_scope_card.clear();
}
#methods_impl
}
pub struct #remote_client_with_scope_card_type_name<'a> {
client: &'a #remote_client_type_name,
scope_card: &'a golem_rust::permissions::ScopeCard,
}
impl<'a> #remote_client_with_scope_card_type_name<'a> {
#scoped_methods_impl
}
}
}
fn build_ephemeral_constructor_body(
client: &syn::Ident,
type_name: &str,
encode_constructor: &proc_macro2::TokenStream,
constructor_value: &syn::Ident,
phantom_id: proc_macro2::TokenStream,
config: proc_macro2::TokenStream,
fallible: bool,
) -> proc_macro2::TokenStream {
if fallible {
quote! {
#encode_constructor
let __golem_constructor = golem_rust::encode_schema_value(&#constructor_value)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?;
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::create(
#type_name,
__golem_constructor,
#phantom_id,
#config,
)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)?;
Ok(#client { wasm_rpc, __golem_default_scope_card: ::core::default::Default::default() })
}
} else {
quote! {
#encode_constructor
let wasm_rpc = golem_rust::golem_agentic::golem::agent::host::WasmRpc::create(
#type_name,
#constructor_value(),
#phantom_id,
#config,
)
.unwrap_or_else(|error| panic!("Internal Error: Agent type not registered: {error:?}"));
#client { wasm_rpc, __golem_default_scope_card: ::core::default::Default::default() }
}
}
}
#[cfg(test)]
mod tests;
fn generate_constructor_data_value_params_encoding(
param_idents: &[proc_macro2::Ident],
constructor_value_ident: &proc_macro2::Ident,
fallible: bool,
) -> proc_macro2::TokenStream {
if !fallible {
let constructor_record = encode_parameters(param_idents, false, true);
return quote! {
let #constructor_value_ident = || #constructor_record;
};
}
let constructor_record =
positional_record_schema_value(param_idents, "Failed to convert constructor parameter");
let validate = if fallible {
quote! {
golem_rust::agentic::__reject_quota_tokens_in_agent_constructor(&#constructor_value_ident)
.map_err(golem_rust::GolemReflectError::InvalidInput)?;
}
} else {
quote! {
golem_rust::agentic::__reject_quota_tokens_in_agent_constructor(&#constructor_value_ident)
.unwrap_or_else(|err| panic!("Invalid agent constructor parameters: {err}"));
}
};
quote! {
let #constructor_value_ident = #constructor_record;
#validate
}
}
fn encode_parameters(
idents: &[syn::Ident],
asynchronous: bool,
constructor: bool,
) -> proc_macro2::TokenStream {
let resources = fresh_param_ident(idents, "__golem_resources");
let writer = fresh_param_ident(idents, "__golem_writer");
let fields = fresh_param_ident(idents, "__golem_fields");
let preflight = if asynchronous {
quote! { golem_rust::schema::wit::direct::WirePreflight::asynchronous() }
} else {
quote! { golem_rust::schema::wit::direct::WirePreflight::default() }
};
let prepare = asynchronous.then(|| quote! {
#((&golem_rust::agentic::AgentArgument(&#idents)).prepare_parameter().await.expect("Failed to prepare parameter"));*;
});
let reject_quota = constructor.then(|| quote! {
#resources.reject_quota_tokens().expect("quota tokens are not allowed in agent constructor parameters");
});
quote! {{
use golem_rust::agentic::WriteAgentParameter as _;
let mut #resources = #preflight;
#((&golem_rust::agentic::AgentArgument(&#idents)).preflight_parameter(&mut #resources).expect("Failed to preflight parameter"));*;
#reject_quota
#prepare
let mut #writer = golem_rust::schema::wit::direct::WireWriter::default();
let #fields: Vec<Option<golem_rust::schema::wit::wire::ValueNodeIndex>> = vec![#((&golem_rust::agentic::AgentArgument(&#idents)).write_parameter(&mut #writer).expect("Failed to encode parameter")),*];
let root = #writer.push(golem_rust::schema::wit::wire::SchemaValueNode::RecordValue(#fields.into_iter().flatten().collect()));
#writer.finish(root)
}}
}
fn get_remote_agent_methods_info(
tr: &ItemTrait,
type_parameter_names: &[String],
agent_is_durable: bool,
fallible: bool,
) -> RemoteAgentMethodsInfo {
let user_method_names = tr
.items
.iter()
.filter_map(extract_method)
.filter(|method| !is_static_method(&method.sig))
.map(|method| method.sig.ident.to_string());
let mut agent_method_names = AgentClientMethodNames::new(user_method_names);
let method_impls = tr
.items
.iter()
.filter_map(|trait_item| {
let method = extract_method(trait_item)?;
if is_static_method(&method.sig) {
return None;
}
if let Some(violation) = find_generic_param_in_return(&method.sig, type_parameter_names)
{
return Some(GeneratedMethod::error(
generic_type_in_agent_return_type_error(violation.span, &violation.type_name),
));
}
if let Some(violation) = find_generic_param_in_inputs(&method.sig, type_parameter_names)
{
return Some(GeneratedMethod::error(generic_type_in_agent_method_error(
violation.span,
&violation.type_name,
)));
}
let keep = |pat_type: &syn::PatType| !is_principal_param(pat_type);
let input_defs = collect_kept_args(&method.sig, keep);
let input_idents: Vec<syn::Ident> = collect_typed_params(&method.sig, keep)
.into_iter()
.map(|param| param.ident)
.collect();
let method_name = &method.sig.ident;
let trigger_name = agent_method_names.fresh_ident(format!("trigger_{method_name}"));
let pending_name = agent_method_names.fresh_ident(format!("pending_{method_name}"));
let schedule_name = agent_method_names.fresh_ident(format!("schedule_{method_name}"));
let schedule_cancelable_name =
agent_method_names.fresh_ident(format!("schedule_cancelable_{method_name}"));
let awaited = AwaitedMethod {
method_name,
impl_name: agent_method_names
.fresh_ident(format!("__golem_{method_name}_with_scope_card")),
pending_impl_name: agent_method_names
.fresh_ident(format!("__golem_pending_{method_name}_with_scope_card")),
input_defs: &input_defs,
input_idents: &input_idents,
scope_card_param: fresh_param_ident(&input_idents, "__golem_scope_card"),
};
Some(generate_method_code(
&awaited,
&trigger_name,
&pending_name,
&schedule_name,
&schedule_cancelable_name,
&input_defs,
&input_idents,
&method.sig,
agent_is_durable,
fallible,
))
})
.collect::<Vec<_>>();
let client_impls = method_impls.iter().map(|method| &method.client);
let scoped_impls = method_impls.iter().map(|method| &method.scoped);
let code = quote! { #(#client_impls)* };
let scoped_code = quote! { #(#scoped_impls)* };
RemoteAgentMethodsInfo::new(code, scoped_code, agent_method_names)
}
fn extract_method(item: &syn::TraitItem) -> Option<&syn::TraitItemFn> {
if let syn::TraitItem::Fn(m) = item {
Some(m)
} else {
None
}
}
fn generate_method_code(
awaited: &AwaitedMethod,
trigger_name: &syn::Ident,
pending_name: &syn::Ident,
schedule_name: &syn::Ident,
schedule_cancelable_name: &syn::Ident,
input_defs: &[&syn::FnArg],
input_idents: &[syn::Ident],
sig: &syn::Signature,
agent_is_durable: bool,
fallible: bool,
) -> GeneratedMethod {
let method_name = awaited.method_name;
if !fallible {
return generate_direct_method_code(
awaited,
trigger_name,
schedule_name,
schedule_cancelable_name,
input_defs,
input_idents,
sig,
agent_is_durable,
);
}
let remote_method_name = method_name.to_string();
let remote_token = quote! { #remote_method_name };
let fn_output_info = FunctionOutputInfo::from_signature(sig);
let return_type = match &sig.output {
syn::ReturnType::Type(_, ty) => quote! { #ty },
syn::ReturnType::Default => quote! { () },
};
let stream_types = sig
.inputs
.iter()
.filter_map(|arg| match arg {
syn::FnArg::Typed(param) => Some(param.ty.as_ref()),
syn::FnArg::Receiver(_) => None,
})
.chain(match &sig.output {
syn::ReturnType::Type(_, ty) => Some(ty.as_ref()),
syn::ReturnType::Default => None,
})
.collect::<Vec<_>>();
let reject_non_awaited_stream_invocation = quote! {
if false #(|| <#stream_types as golem_rust::agentic::Schema>::contains_stream())* {
let _ = (#(&#input_idents),*);
panic!("live streams cannot cross remote or scheduled agent invocation boundaries")
}
};
let process_invoke_result = match &sig.output {
syn::ReturnType::Type(_, ty) if !fn_output_info.is_unit => decode_result_value(
ty,
quote! { rpc_result_ok.expect("remote method returned no value") },
),
_ => quote! {},
};
let input_record = positional_record_schema_value(input_idents, "Failed to encode parameter");
let encoded_input = encode_value_only_carrier(input_record.clone());
let encode_input = quote! { let input = #encoded_input; };
let encode_input_async = quote! {
let input_value = #input_record;
let input = golem_rust::encode_schema_value_async(&input_value)
.await
.expect("Failed to encode parameters");
};
let scheduled_time_param = fresh_param_ident(input_idents, "scheduled_time");
if fallible {
let reject_non_awaited_stream_invocation = quote! {
if false #(|| <#stream_types as golem_rust::agentic::Schema>::contains_stream())* {
let _ = (#(&#input_idents),*);
return Err(golem_rust::GolemReflectError::InvalidType(
"live streams cannot cross trigger or scheduled agent invocation boundaries".to_string(),
));
}
};
let decode_value = match &sig.output {
syn::ReturnType::Type(_, ty) if !fn_output_info.is_unit => quote! {
let __golem_value = rpc_result_ok.ok_or_else(||
golem_rust::GolemReflectError::MalformedRemoteOutput(
format!("method `{}` returned unit instead of a value", #remote_token),
)
)?;
static __GOLEM_RPC_GRAPH_CACHE: ::std::sync::OnceLock<golem_rust::SchemaGraph> =
::std::sync::OnceLock::new();
let __golem_graph = __GOLEM_RPC_GRAPH_CACHE.get_or_init(|| {
<#ty as golem_rust::agentic::Schema>::get_type()
.get_schema_graph()
.expect("rpc result type must have a concrete schema graph")
});
<#ty as golem_rust::agentic::Schema>::from_schema_value(
__golem_value,
golem_rust::agentic::StructuredSchema::Default(__golem_graph.clone()),
)
.map_err(golem_rust::GolemReflectError::MalformedRemoteOutput)?
},
_ => quote! {
if rpc_result_ok.is_some() {
return Err(golem_rust::GolemReflectError::MalformedRemoteOutput(
format!("method `{}` returned a value instead of unit", #remote_token),
));
}
()
},
};
let pending_decode = match &sig.output {
syn::ReturnType::Type(_, ty) if !fn_output_info.is_unit => quote! {
|rpc_result_ok| {
let __golem_value = rpc_result_ok.ok_or_else(||
golem_rust::GolemReflectError::MalformedRemoteOutput(
format!("method `{}` returned unit instead of a value", #remote_token),
)
)?;
static __GOLEM_RPC_GRAPH_CACHE: ::std::sync::OnceLock<golem_rust::SchemaGraph> =
::std::sync::OnceLock::new();
let __golem_graph = __GOLEM_RPC_GRAPH_CACHE.get_or_init(|| {
<#ty as golem_rust::agentic::Schema>::get_type()
.get_schema_graph()
.expect("rpc result type must have a concrete schema graph")
});
<#ty as golem_rust::agentic::Schema>::from_schema_value(
__golem_value,
golem_rust::agentic::StructuredSchema::Default(__golem_graph.clone()),
)
.map_err(golem_rust::GolemReflectError::MalformedRemoteOutput)
}
},
_ => quote! {
|rpc_result_ok| {
if rpc_result_ok.is_some() {
return Err(golem_rust::GolemReflectError::MalformedRemoteOutput(
format!("method `{}` returned a value instead of unit", #remote_token),
));
}
Ok(())
}
},
};
let encode_input_async = quote! {
let input_value = #input_record;
let input = golem_rust::encode_schema_value_async(&input_value)
.await
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?;
};
let encode_input = quote! {
let input_value = #input_record;
let input = golem_rust::encode_schema_value(&input_value)
.map_err(|error| golem_rust::GolemReflectError::SchemaEncode(error.to_string()))?;
};
let invoke = awaited.invoke(&remote_token);
let scope_card_param = &awaited.scope_card_param;
let GeneratedMethod {
client: pending_client,
scoped: pending_scoped,
} = awaited.generate_pending(
pending_name,
quote! { Result<golem_rust::CallerPendingInvocation<#return_type>, golem_rust::GolemReflectError> },
quote! {
#reject_non_awaited_stream_invocation
let input_value = #input_record;
golem_rust::agentic::reflection::start_caller_pending(
&self.wasm_rpc,
#remote_token,
input_value,
#scope_card_param.as_ref(),
#pending_decode,
)
},
);
if agent_is_durable {
return awaited.generate(
quote! { Result<#return_type, golem_rust::GolemReflectError> },
quote! {
#encode_input_async
let rpc_result_future = #invoke.future;
let rpc_result_ok = golem_rust::agentic::await_invoke_schema_value_result(rpc_result_future)
.await
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)?;
Ok({ #decode_value })
},
quote! {
pub fn #trigger_name(#(#input_defs),*)
-> Result<(), golem_rust::GolemReflectError>
{
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.invoke(#remote_token, input, None)
.map(|_| ())
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
#pending_client
pub fn #schedule_name(
#(#input_defs,)*
#scheduled_time_param: golem_rust::ScheduledTime,
) -> Result<(), golem_rust::GolemReflectError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(
#scheduled_time_param,
#remote_token,
input,
None,
)
.map(|_| ())
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
pub fn #schedule_cancelable_name(
#(#input_defs,)*
#scheduled_time_param: golem_rust::ScheduledTime,
) -> Result<golem_rust::golem_agentic::golem::agent::host::CancellationToken, golem_rust::GolemReflectError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(
#scheduled_time_param,
#remote_token,
input,
None,
)
.map(|receipt| receipt.cancellation_token)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
},
)
.with_scoped(pending_scoped);
}
return awaited.generate(
quote! { Result<golem_rust::agentic::EphemeralInvocationResult<#return_type>, golem_rust::GolemReflectError> },
quote! {
#encode_input_async
let invocation = #invoke;
let metadata = invocation.metadata;
let rpc_result_ok = golem_rust::agentic::await_invoke_schema_value_result(invocation.future)
.await
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)?;
let value = { #decode_value };
Ok(golem_rust::agentic::EphemeralInvocationResult { metadata, value })
},
quote! {
pub fn #trigger_name(#(#input_defs),*)
-> Result<golem_rust::golem_agentic::golem::agent::host::InvocationMetadata, golem_rust::GolemReflectError>
{
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.invoke(#remote_token, input, None)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
#pending_client
pub fn #schedule_name(
#(#input_defs,)*
#scheduled_time_param: golem_rust::ScheduledTime,
) -> Result<golem_rust::golem_agentic::golem::agent::host::InvocationMetadata, golem_rust::GolemReflectError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(#scheduled_time_param, #remote_token, input, None)
.map(|receipt| receipt.metadata)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
pub fn #schedule_cancelable_name(
#(#input_defs,)*
#scheduled_time_param: golem_rust::ScheduledTime,
) -> Result<golem_rust::golem_agentic::golem::agent::host::CancelableScheduledInvocationReceipt, golem_rust::GolemReflectError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(#scheduled_time_param, #remote_token, input, None)
.map_err(golem_rust::agentic::reflection::agentic_rpc_error_to_reflect)
}
},
)
.with_scoped(pending_scoped);
}
let invoke = awaited.invoke(&remote_token);
if agent_is_durable {
return awaited.generate(
quote! { #return_type },
quote! {
#encode_input_async
let rpc_result_future = #invoke.future;
let rpc_result: Result<Option<golem_rust::SchemaValue>, golem_rust::golem_agentic::golem::agent::host::RpcError> =
golem_rust::agentic::await_invoke_schema_value_result(rpc_result_future).await;
let rpc_result_ok =
rpc_result.unwrap_or_else(|e| panic!("rpc call to {} failed: {:?}", #remote_token, e));
#process_invoke_result
},
quote! {
pub fn #trigger_name(#(#input_defs),*) {
#reject_non_awaited_stream_invocation
#encode_input
let rpc_result: Result<(), golem_rust::golem_agentic::golem::agent::host::RpcError> =
self.wasm_rpc.invoke(#remote_token, input, None).map(|_| ());
rpc_result.unwrap_or_else(|e| panic!("rpc call to trigger {} failed: {:?}", #remote_token, e));
}
pub fn #schedule_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<(), golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(
#scheduled_time_param,
#remote_token,
input,
None
).map(|_| ())
}
pub fn #schedule_cancelable_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::CancellationToken, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(
#scheduled_time_param,
#remote_token,
input,
None
).map(|receipt| receipt.cancellation_token)
}
},
);
}
awaited.generate(
quote! { golem_rust::agentic::EphemeralInvocationResult<#return_type> },
quote! {
#encode_input_async
let invocation = #invoke;
let metadata = invocation.metadata;
let rpc_result: Result<Option<golem_rust::SchemaValue>, golem_rust::golem_agentic::golem::agent::host::RpcError> =
golem_rust::agentic::await_invoke_schema_value_result(invocation.future).await;
let rpc_result_ok = rpc_result.unwrap_or_else(|e| panic!("rpc call to {} failed: {:?}", #remote_token, e));
let value = { #process_invoke_result };
golem_rust::agentic::EphemeralInvocationResult { metadata, value }
},
quote! {
pub fn #trigger_name(#(#input_defs),*) -> golem_rust::golem_agentic::golem::agent::host::InvocationMetadata {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.invoke(#remote_token, input, None)
.unwrap_or_else(|e| panic!("rpc call to trigger {} failed: {:?}", #remote_token, e))
}
pub fn #schedule_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::InvocationMetadata, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(#scheduled_time_param, #remote_token, input, None)
.map(|receipt| receipt.metadata)
}
pub fn #schedule_cancelable_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::CancelableScheduledInvocationReceipt, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(#scheduled_time_param, #remote_token, input, None)
}
},
)
}
fn generate_direct_method_code(
awaited: &AwaitedMethod,
trigger_name: &syn::Ident,
schedule_name: &syn::Ident,
schedule_cancelable_name: &syn::Ident,
input_defs: &[&syn::FnArg],
input_idents: &[syn::Ident],
sig: &syn::Signature,
agent_is_durable: bool,
) -> GeneratedMethod {
let remote_method_name = awaited.method_name.to_string();
let remote_token = quote! { #remote_method_name };
let return_type = match &sig.output {
syn::ReturnType::Type(_, ty) => quote! { #ty },
syn::ReturnType::Default => quote! { () },
};
let stream_types = sig
.inputs
.iter()
.filter_map(|arg| match arg {
syn::FnArg::Typed(param) => Some(param.ty.as_ref()),
syn::FnArg::Receiver(_) => None,
})
.chain(match &sig.output {
syn::ReturnType::Type(_, ty) => Some(ty.as_ref()),
syn::ReturnType::Default => None,
})
.collect::<Vec<_>>();
let reject_non_awaited_stream_invocation = quote! {
use golem_rust::agentic::AgentParameterStreams as _;
if false #(|| (&golem_rust::agentic::AgentParameterProbe::<#stream_types>(::std::marker::PhantomData)).parameter_contains_stream())* {
let _ = (#(&#input_idents),*);
panic!("live streams cannot cross remote or scheduled agent invocation boundaries")
}
};
let process_invoke_result = quote! {
golem_rust::schema::wit::direct::decode_result_payload::<#return_type>(rpc_result_ok)
.expect("Failed to deserialize rpc result to return type")
};
let encoded_input = encode_parameters(input_idents, false, false);
let encode_input = quote! { let input = #encoded_input; };
let encoded_input_async = encode_parameters(input_idents, true, false);
let encode_input_async = quote! { let input = #encoded_input_async; };
let scheduled_time_param = fresh_param_ident(input_idents, "scheduled_time");
let invoke = awaited.invoke(&remote_token);
if agent_is_durable {
return awaited.generate(
quote! { #return_type },
quote! {
#encode_input_async
let rpc_result_future = #invoke.future;
let rpc_result = rpc_result_future.get().await;
let rpc_result_ok =
rpc_result.unwrap_or_else(|e| panic!("rpc call to {} failed: {:?}", #remote_token, e));
#process_invoke_result
},
quote! {
pub fn #trigger_name(#(#input_defs),*) {
#reject_non_awaited_stream_invocation
#encode_input
let rpc_result: Result<(), golem_rust::golem_agentic::golem::agent::host::RpcError> =
self.wasm_rpc.invoke(#remote_token, input, None).map(|_| ());
rpc_result.unwrap_or_else(|e| panic!("rpc call to trigger {} failed: {:?}", #remote_token, e));
}
pub fn #schedule_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<(), golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(
#scheduled_time_param,
#remote_token,
input,
None
).map(|_| ())
}
pub fn #schedule_cancelable_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::CancellationToken, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(
#scheduled_time_param,
#remote_token,
input,
None
).map(|receipt| receipt.cancellation_token)
}
},
);
}
awaited.generate(
quote! { golem_rust::agentic::EphemeralInvocationResult<#return_type> },
quote! {
#encode_input_async
let invocation = #invoke;
let metadata = invocation.metadata;
let rpc_result = invocation.future.get().await;
let rpc_result_ok = rpc_result.unwrap_or_else(|e| panic!("rpc call to {} failed: {:?}", #remote_token, e));
let value = { #process_invoke_result };
golem_rust::agentic::EphemeralInvocationResult { metadata, value }
},
quote! {
pub fn #trigger_name(#(#input_defs),*) -> golem_rust::golem_agentic::golem::agent::host::InvocationMetadata {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.invoke(#remote_token, input, None)
.unwrap_or_else(|e| panic!("rpc call to trigger {} failed: {:?}", #remote_token, e))
}
pub fn #schedule_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::InvocationMetadata, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_invocation(#scheduled_time_param, #remote_token, input, None)
.map(|receipt| receipt.metadata)
}
pub fn #schedule_cancelable_name(#(#input_defs,)* #scheduled_time_param: golem_rust::ScheduledTime) -> Result<golem_rust::golem_agentic::golem::agent::host::CancelableScheduledInvocationReceipt, golem_rust::golem_agentic::golem::agent::host::RpcError> {
#reject_non_awaited_stream_invocation
#encode_input
self.wasm_rpc.schedule_cancelable_invocation(#scheduled_time_param, #remote_token, input, None)
}
},
)
}
fn fresh_param_ident(occupied: &[syn::Ident], preferred_name: &str) -> syn::Ident {
let occupied = occupied
.iter()
.map(|ident| syn::ext::IdentExt::unraw(ident).to_string())
.collect::<HashSet<_>>();
if !occupied.contains(preferred_name) {
return format_ident!("{}", preferred_name);
}
let mut suffix = 1usize;
loop {
let candidate = format!("{preferred_name}{suffix}");
if !occupied.contains(&candidate) {
return format_ident!("{}", candidate);
}
suffix += 1;
}
}
struct RemoteAgentMethodsInfo {
methods_impl: proc_macro2::TokenStream,
scoped_methods_impl: proc_macro2::TokenStream,
method_names: AgentClientMethodNames,
}
impl RemoteAgentMethodsInfo {
fn new(
methods_impl: proc_macro2::TokenStream,
scoped_methods_impl: proc_macro2::TokenStream,
method_names: AgentClientMethodNames,
) -> Self {
Self {
methods_impl,
scoped_methods_impl,
method_names,
}
}
}
struct GeneratedMethod {
client: proc_macro2::TokenStream,
scoped: proc_macro2::TokenStream,
}
impl GeneratedMethod {
fn error(error: proc_macro2::TokenStream) -> Self {
Self {
client: error,
scoped: quote! {},
}
}
fn with_scoped(mut self, scoped: proc_macro2::TokenStream) -> Self {
self.scoped.extend(scoped);
self
}
}
struct AwaitedMethod<'a> {
method_name: &'a syn::Ident,
impl_name: syn::Ident,
pending_impl_name: syn::Ident,
input_defs: &'a [&'a syn::FnArg],
input_idents: &'a [syn::Ident],
scope_card_param: syn::Ident,
}
impl AwaitedMethod<'_> {
fn generate_pending(
&self,
pending_name: &syn::Ident,
return_type: proc_macro2::TokenStream,
body: proc_macro2::TokenStream,
) -> GeneratedMethod {
let Self {
pending_impl_name,
input_defs,
input_idents,
scope_card_param,
..
} = self;
let typed_input_defs = input_defs
.iter()
.filter(|arg| matches!(arg, syn::FnArg::Typed(_)))
.collect::<Vec<_>>();
GeneratedMethod {
client: quote! {
fn #pending_impl_name(
&self,
#(#typed_input_defs,)*
#scope_card_param: Option<&golem_rust::permissions::ScopeCard>,
) -> #return_type {
let #scope_card_param = self.__golem_default_scope_card.resolve(#scope_card_param);
#body
}
pub fn #pending_name(#(#input_defs),*) -> #return_type {
self.#pending_impl_name(#(#input_idents,)* None)
}
},
scoped: quote! {
pub fn #pending_name(&self, #(#typed_input_defs),*) -> #return_type {
self.client.#pending_impl_name(#(#input_idents,)* Some(self.scope_card))
}
},
}
}
fn invoke(&self, remote_token: &proc_macro2::TokenStream) -> proc_macro2::TokenStream {
let scope_card_param = &self.scope_card_param;
quote! {
golem_rust::permissions::__async_invoke_and_await(
&self.wasm_rpc,
#remote_token,
input,
#scope_card_param.as_ref(),
)
}
}
fn generate(
&self,
return_type: proc_macro2::TokenStream,
body: proc_macro2::TokenStream,
rest: proc_macro2::TokenStream,
) -> GeneratedMethod {
let Self {
method_name,
impl_name,
pending_impl_name: _,
input_defs,
input_idents,
scope_card_param,
} = self;
let typed_input_defs = input_defs
.iter()
.filter(|arg| matches!(arg, syn::FnArg::Typed(_)))
.collect::<Vec<_>>();
GeneratedMethod {
client: quote! {
async fn #impl_name(
&self,
#(#typed_input_defs,)*
#scope_card_param: Option<&golem_rust::permissions::ScopeCard>,
) -> #return_type {
let #scope_card_param = self.__golem_default_scope_card.resolve(#scope_card_param);
#body
}
pub async fn #method_name(#(#input_defs),*) -> #return_type {
self.#impl_name(#(#input_idents,)* None).await
}
#rest
},
scoped: quote! {
pub async fn #method_name(&self, #(#typed_input_defs),*) -> #return_type {
self.client.#impl_name(#(#input_idents,)* Some(self.scope_card)).await
}
},
}
}
}
#[derive(Debug)]
struct AgentClientMethodNames {
method_names: HashSet<String>,
next_suffix_by_name: HashMap<String, usize>,
}
impl AgentClientMethodNames {
fn new(names: impl IntoIterator<Item = String>) -> Self {
Self {
method_names: names.into_iter().collect(),
next_suffix_by_name: HashMap::new(),
}
}
fn fresh_ident(&mut self, preferred_name: impl Into<String>) -> syn::Ident {
let preferred_name = preferred_name.into();
let name = if self.method_names.insert(preferred_name.clone()) {
self.next_suffix_by_name
.entry(preferred_name.clone())
.or_insert(1);
preferred_name
} else {
let next_suffix = self
.next_suffix_by_name
.entry(preferred_name.clone())
.or_insert(1);
loop {
let candidate = format!("{preferred_name}{next_suffix}");
*next_suffix += 1;
if self.method_names.insert(candidate.clone()) {
break candidate;
}
}
};
format_ident!("{}", name)
}
fn contains(&self, name: &str) -> bool {
self.method_names.contains(name)
}
}