use super::handle::MobHandle;
use super::provisioner::MobProvisioner;
use super::turn_executor::{
ActorTurnTicket, FlowTurnExecutor, FlowTurnOutcome, FlowTurnTicket, TimeoutDisposition,
};
use crate::error::MobError;
use crate::event::MemberRef;
use crate::ids::{AgentIdentity, RunId, StepId};
use crate::machines::mob_machine as mob_dsl;
#[cfg(target_arch = "wasm32")]
use crate::tokio;
use async_trait::async_trait;
use futures::FutureExt;
use meerkat_core::EventEnvelope;
use meerkat_core::event::{AgentEvent, ScopedAgentEvent, StreamScopeFrame};
use meerkat_core::service::{StartTurnRequest, TurnToolOverlay};
use meerkat_core::types::{ContentInput, SessionId};
use std::panic::AssertUnwindSafe;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Mutex, mpsc, oneshot};
#[derive(Clone)]
pub struct ActorFlowTurnExecutor {
handle: MobHandle,
provisioner: Arc<dyn MobProvisioner>,
}
impl ActorFlowTurnExecutor {
pub fn new(handle: MobHandle, provisioner: Arc<dyn MobProvisioner>) -> Self {
Self {
handle,
provisioner,
}
}
fn actor_ticket(ticket: FlowTurnTicket) -> Arc<ActorTurnTicket> {
match ticket {
FlowTurnTicket::Actor(ticket) => ticket,
}
}
fn project_member_ref_session_binding(
member_ref: &MemberRef,
current_bridge_session_id: Option<SessionId>,
) -> Option<MemberRef> {
match member_ref {
MemberRef::Session { .. } => {
current_bridge_session_id.map(MemberRef::from_bridge_session_id)
}
MemberRef::BackendPeer {
peer_id,
address,
pubkey,
bootstrap_token,
..
} => Some(MemberRef::BackendPeer {
peer_id: peer_id.clone(),
address: address.clone(),
pubkey: *pubkey,
bootstrap_token: bootstrap_token.clone(),
session_id: current_bridge_session_id,
}),
}
}
fn spawn_subscription_bridge(
events: tokio::sync::mpsc::Receiver<AgentEvent>,
completion_tx: oneshot::Sender<FlowTurnOutcome>,
scoped_event_tx: Option<mpsc::Sender<ScopedAgentEvent>>,
scoped_frame: Option<StreamScopeFrame>,
) -> tokio::task::JoinHandle<()> {
Self::spawn_subscription_bridge_impl(
events,
completion_tx,
scoped_event_tx,
scoped_frame,
|payload| payload,
)
}
fn spawn_subscription_bridge_enveloped(
events: tokio::sync::mpsc::Receiver<EventEnvelope<AgentEvent>>,
completion_tx: oneshot::Sender<FlowTurnOutcome>,
scoped_event_tx: Option<mpsc::Sender<ScopedAgentEvent>>,
scoped_frame: Option<StreamScopeFrame>,
) -> tokio::task::JoinHandle<()> {
Self::spawn_subscription_bridge_impl(
events,
completion_tx,
scoped_event_tx,
scoped_frame,
|event| event.payload,
)
}
fn spawn_subscription_bridge_impl<E, F>(
mut events: tokio::sync::mpsc::Receiver<E>,
completion_tx: oneshot::Sender<FlowTurnOutcome>,
scoped_event_tx: Option<mpsc::Sender<ScopedAgentEvent>>,
scoped_frame: Option<StreamScopeFrame>,
mut extract_payload: F,
) -> tokio::task::JoinHandle<()>
where
E: Send + 'static,
F: FnMut(E) -> AgentEvent + Send + 'static,
{
tokio::spawn(async move {
let mut completion_tx = Some(completion_tx);
let mut pending_completed_run: Option<(String, Option<serde_json::Value>)> = None;
while let Some(event) = events.recv().await {
let payload = extract_payload(event);
if let (Some(tx), Some(frame)) = (&scoped_event_tx, &scoped_frame) {
let scoped = ScopedAgentEvent::new(vec![frame.clone()], payload.clone());
let _ = tx.send(scoped).await;
}
match payload {
AgentEvent::RunCompleted {
result,
structured_output,
extraction_required,
..
} => {
if extraction_required {
pending_completed_run = Some((result, structured_output));
continue;
}
if let Some(tx) = completion_tx.take() {
let _ = tx.send(FlowTurnOutcome::Completed {
output: result,
structured_output,
});
}
return;
}
AgentEvent::ExtractionSucceeded {
structured_output, ..
} => {
if let Some(tx) = completion_tx.take() {
let (output, _) = pending_completed_run
.take()
.unwrap_or_else(|| (structured_output.to_string(), None));
let _ = tx.send(FlowTurnOutcome::Completed {
output,
structured_output: Some(structured_output),
});
}
return;
}
AgentEvent::ExtractionFailed {
last_output,
attempts,
reason,
..
} => {
if let Some(tx) = completion_tx.take() {
let (output, _) =
pending_completed_run.take().unwrap_or((last_output, None));
let _ = tx.send(FlowTurnOutcome::Failed {
reason: format!(
"structured output extraction failed after {attempts} attempt(s): {reason}; last_output={output:?}"
),
});
}
return;
}
AgentEvent::InteractionComplete {
result,
structured_output,
..
} => {
if let Some(tx) = completion_tx.take() {
let _ = tx.send(FlowTurnOutcome::Completed {
output: result,
structured_output,
});
}
return;
}
AgentEvent::InteractionCallbackPending {
tool_name, args, ..
} => {
if let Some(tx) = completion_tx.take() {
let _ = tx.send(FlowTurnOutcome::Failed {
reason: format!("callback pending for tool '{tool_name}': {args}"),
});
}
return;
}
AgentEvent::RunFailed { error_report, .. } => {
if let Some(tx) = completion_tx.take() {
let _ = tx.send(FlowTurnOutcome::Failed {
reason: error_report.message,
});
}
return;
}
AgentEvent::InteractionFailed { reason, .. } => {
if let Some(tx) = completion_tx.take() {
let _ = tx.send(FlowTurnOutcome::Failed {
reason: reason.to_string(),
});
}
return;
}
_ => {}
}
}
if let Some(tx) = completion_tx {
let _ = tx.send(FlowTurnOutcome::Failed {
reason: "turn event stream closed before terminal outcome".to_string(),
});
}
})
}
async fn reconcile_detached_turn(bridge_handle: tokio::task::JoinHandle<()>, run_id: RunId) {
let reconcile = async {
match bridge_handle.await {
Ok(()) => {
tracing::debug!(run_id = %run_id, "detached timed-out turn finished");
}
Err(error) => {
tracing::warn!(
run_id = %run_id,
error = %error,
"detached timed-out turn ended abnormally"
);
}
}
};
if AssertUnwindSafe(reconcile).catch_unwind().await.is_err() {
tracing::error!(
run_id = %run_id,
"panic while joining detached timed-out turn"
);
}
}
}
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
impl FlowTurnExecutor for ActorFlowTurnExecutor {
async fn dispatch(
&self,
run_id: &RunId,
step_id: &StepId,
target: &AgentIdentity,
message: ContentInput,
flow_tool_overlay: Option<TurnToolOverlay>,
) -> Result<FlowTurnTicket, MobError> {
let entry = self
.handle
.get_member(target)
.await?
.ok_or_else(|| MobError::MemberNotFound(target.clone()))?;
let (completion_tx, completion_rx) = oneshot::channel::<FlowTurnOutcome>();
let scoped_event_tx = self.handle.flow_streams.lock().await.get(run_id).cloned();
let scope_frame = StreamScopeFrame::MobMember {
flow_run_id: run_id.to_string(),
agent_identity: entry.agent_identity.to_string(),
agent_runtime_id: None,
fence_token: None,
generation: None,
};
let bridge_handle = match entry.runtime_mode {
crate::MobRuntimeMode::AutonomousHost => {
if flow_tool_overlay.is_some() {
return Err(MobError::Internal(format!(
"flow tool overlay cannot be enforced for autonomous host member '{target}'; \
use turn_driven runtime mode for steps with allowed_tools/blocked_tools"
)));
}
let bridge_session_id = self
.handle
.resolve_bridge_session_id(&AgentIdentity::from(target.as_str()))
.await
.ok_or_else(|| {
MobError::Internal(format!(
"autonomous flow dispatch requires MobMachine session binding for '{target}'"
))
})?;
let injector = self
.provisioner
.interaction_event_injector(&bridge_session_id)
.await
.ok_or_else(|| MobError::MissingMemberCapability {
member_id: target.clone(),
capability: crate::error::MobMemberCapability::InteractionEventInjector,
context: "autonomous flow turn delivery",
})?;
let subscription = injector
.inject_with_subscription(
message,
meerkat_core::PlainEventSource::Rpc,
meerkat_core::types::HandlingMode::Queue,
None,
)
.map_err(|error| {
MobError::Internal(format!(
"autonomous flow dispatch inject failed for '{target}': {error}"
))
})?;
Self::spawn_subscription_bridge(
subscription.events,
completion_tx,
scoped_event_tx.clone(),
Some(scope_frame.clone()),
)
}
crate::MobRuntimeMode::TurnDriven => {
let bridge_session_id = self
.handle
.resolve_bridge_session_id(&AgentIdentity::from(target.as_str()))
.await;
let member_ref = match Self::project_member_ref_session_binding(
&entry.member_ref,
bridge_session_id,
) {
Some(member_ref) => member_ref,
None => {
return Err(MobError::Internal(format!(
"turn-driven flow dispatch requires MobMachine session binding for '{target}'"
)));
}
};
let (event_tx, event_rx) = mpsc::channel::<EventEnvelope<AgentEvent>>(8);
let bridge_handle = Self::spawn_subscription_bridge_enveloped(
event_rx,
completion_tx,
scoped_event_tx,
Some(scope_frame),
);
if let Err(error) = self
.provisioner
.start_flow_step(
&member_ref,
run_id,
step_id,
StartTurnRequest {
injected_context: Vec::new(),
prompt: message,
system_prompt: None,
event_tx: Some(event_tx),
runtime: meerkat_core::service::StartTurnRuntimeSemantics::new(
meerkat_core::types::HandlingMode::Queue,
flow_tool_overlay,
Vec::new(),
None,
),
},
)
.await
{
bridge_handle.abort();
return Err(error);
}
bridge_handle
}
};
Ok(FlowTurnTicket::Actor(Arc::new(ActorTurnTicket {
run_id: run_id.clone(),
completion_rx: Mutex::new(Some(completion_rx)),
bridge_handle: Mutex::new(Some(bridge_handle)),
})))
}
async fn await_terminal(
&self,
ticket: FlowTurnTicket,
timeout: Duration,
) -> Result<FlowTurnOutcome, MobError> {
let ticket = Self::actor_ticket(ticket);
let completion_rx = {
let mut lock = ticket.completion_rx.lock().await;
lock.take().ok_or_else(|| {
MobError::Internal("flow turn ticket cannot be awaited twice".to_string())
})?
};
match tokio::time::timeout(timeout, completion_rx).await {
Ok(Ok(outcome)) => {
if let Some(handle) = ticket.bridge_handle.lock().await.take() {
let _ = handle.await;
}
Ok(outcome)
}
Ok(Err(_)) => {
if let Some(handle) = ticket.bridge_handle.lock().await.take() {
let _ = handle.await;
}
Ok(FlowTurnOutcome::Failed {
reason: "turn completion channel closed".to_string(),
})
}
Err(_) => Err(MobError::FlowTurnTimedOut),
}
}
async fn on_timeout(&self, ticket: FlowTurnTicket) -> Result<TimeoutDisposition, MobError> {
let ticket = Self::actor_ticket(ticket);
let Some(bridge_handle) = ticket.bridge_handle.lock().await.take() else {
return Ok(TimeoutDisposition::Canceled);
};
let run_id = ticket.run_id.clone();
let effects = self
.handle
.apply_machine_input_effects(mob_dsl::MobMachineInput::ClassifyTurnTimeoutDisposition {
timed_out_run_id: mob_dsl::RunId::from(run_id.to_string()),
retryable: false,
})
.await?;
let disposition = effects
.into_iter()
.find_map(|effect| match effect {
mob_dsl::MobMachineEffect::TurnTimeoutDispositionClassified {
disposition, ..
} => Some(disposition),
_ => None,
})
.ok_or_else(|| {
MobError::Internal(
"MobMachine accepted turn-timeout observation but emitted no disposition"
.into(),
)
})?;
match disposition {
mob_dsl::TurnTimeoutDisposition::Detached => {
tokio::spawn(async move {
Self::reconcile_detached_turn(bridge_handle, run_id).await;
});
Ok(TimeoutDisposition::Detached)
}
mob_dsl::TurnTimeoutDisposition::Canceled
| mob_dsl::TurnTimeoutDisposition::Retryable => {
bridge_handle.abort();
Ok(TimeoutDisposition::Canceled)
}
}
}
}
#[cfg(test)]
mod tests {
use super::ActorFlowTurnExecutor;
use crate::ids::RunId;
use crate::runtime::FlowTurnOutcome;
use meerkat_core::AgentEvent;
use meerkat_core::event::{AgentErrorClass, AgentErrorReason, AgentErrorReport};
use meerkat_core::interaction::InteractionId;
use tokio::sync::oneshot;
#[tokio::test]
async fn test_reconcile_detached_turn_joins_bridge_even_when_bridge_panics() {
let run_id = RunId::new();
let bridge_handle = tokio::spawn(async move {
panic!("detached bridge panic");
});
ActorFlowTurnExecutor::reconcile_detached_turn(bridge_handle, run_id).await;
}
#[tokio::test]
async fn test_subscription_bridge_preserves_interaction_structured_output() {
let (events_tx, events_rx) = tokio::sync::mpsc::channel(1);
let (completion_tx, completion_rx) = oneshot::channel();
let bridge =
ActorFlowTurnExecutor::spawn_subscription_bridge(events_rx, completion_tx, None, None);
events_tx
.send(AgentEvent::InteractionComplete {
interaction_id: InteractionId(uuid::Uuid::new_v4()),
result: "{\"answer\":42}".to_string(),
structured_output: Some(serde_json::json!({"answer": 42})),
})
.await
.expect("send event");
match completion_rx.await.expect("completion outcome") {
FlowTurnOutcome::Completed {
output,
structured_output,
} => {
assert_eq!(output, "{\"answer\":42}");
assert_eq!(structured_output, Some(serde_json::json!({"answer": 42})));
}
other => panic!("expected completed outcome, got {other:?}"),
}
bridge.await.expect("bridge exits after terminal event");
}
#[tokio::test]
async fn test_subscription_bridge_preserves_run_structured_output() {
let (events_tx, events_rx) = tokio::sync::mpsc::channel(1);
let (completion_tx, completion_rx) = oneshot::channel();
let bridge =
ActorFlowTurnExecutor::spawn_subscription_bridge(events_rx, completion_tx, None, None);
events_tx
.send(AgentEvent::RunCompleted {
session_id: meerkat_core::SessionId::new(),
result: "{\"answer\":42}".to_string(),
structured_output: Some(serde_json::json!({"answer": 42})),
extraction_required: false,
usage: meerkat_core::Usage::default(),
terminal_cause_kind: None,
})
.await
.expect("send event");
match completion_rx.await.expect("completion outcome") {
FlowTurnOutcome::Completed {
output,
structured_output,
} => {
assert_eq!(output, "{\"answer\":42}");
assert_eq!(structured_output, Some(serde_json::json!({"answer": 42})));
}
other => panic!("expected completed outcome, got {other:?}"),
}
bridge.await.expect("bridge exits after terminal event");
}
#[tokio::test]
async fn test_subscription_bridge_keeps_run_failed_metadata_display_only() {
let (events_tx, events_rx) = tokio::sync::mpsc::channel(1);
let (completion_tx, completion_rx) = oneshot::channel();
let bridge =
ActorFlowTurnExecutor::spawn_subscription_bridge(events_rx, completion_tx, None, None);
let report = AgentErrorReport {
class: AgentErrorClass::Terminal,
reason: Some(AgentErrorReason::TurnTerminalCause {
outcome: meerkat_core::TurnTerminalOutcome::Failed,
cause_kind: meerkat_core::TurnTerminalCauseKind::LlmFailure,
}),
message: "LLM failure terminal turn".to_string(),
};
events_tx
.send(AgentEvent::RunFailed {
session_id: meerkat_core::SessionId::new(),
terminal_cause_kind: Some(meerkat_core::TurnTerminalCauseKind::LlmFailure),
error_report: report.clone(),
})
.await
.expect("send event");
match completion_rx.await.expect("completion outcome") {
FlowTurnOutcome::Failed { reason } => {
assert_eq!(reason, "LLM failure terminal turn");
}
other => panic!("expected failed outcome, got {other:?}"),
}
bridge.await.expect("bridge exits after terminal event");
}
#[tokio::test]
async fn test_subscription_bridge_waits_for_required_extraction_success() {
let (events_tx, events_rx) = tokio::sync::mpsc::channel(2);
let (completion_tx, completion_rx) = oneshot::channel();
let bridge =
ActorFlowTurnExecutor::spawn_subscription_bridge(events_rx, completion_tx, None, None);
let session_id = meerkat_core::SessionId::new();
events_tx
.send(AgentEvent::RunCompleted {
session_id: session_id.clone(),
result: "main answer".to_string(),
structured_output: None,
extraction_required: true,
usage: meerkat_core::Usage::default(),
terminal_cause_kind: None,
})
.await
.expect("send run completed event");
events_tx
.send(AgentEvent::ExtractionSucceeded {
session_id,
structured_output: serde_json::json!({"answer": 42}),
schema_warnings: None,
})
.await
.expect("send extraction succeeded event");
match completion_rx.await.expect("completion outcome") {
FlowTurnOutcome::Completed {
output,
structured_output,
} => {
assert_eq!(output, "main answer");
assert_eq!(structured_output, Some(serde_json::json!({"answer": 42})));
}
other => panic!("expected completed outcome, got {other:?}"),
}
bridge.await.expect("bridge exits after extraction event");
}
#[tokio::test]
async fn test_subscription_bridge_treats_extraction_failure_as_failed_step() {
let (events_tx, events_rx) = tokio::sync::mpsc::channel(2);
let (completion_tx, completion_rx) = oneshot::channel();
let bridge =
ActorFlowTurnExecutor::spawn_subscription_bridge(events_rx, completion_tx, None, None);
let session_id = meerkat_core::SessionId::new();
events_tx
.send(AgentEvent::RunCompleted {
session_id: session_id.clone(),
result: "main answer".to_string(),
structured_output: None,
extraction_required: true,
usage: meerkat_core::Usage::default(),
terminal_cause_kind: None,
})
.await
.expect("send run completed event");
events_tx
.send(AgentEvent::ExtractionFailed {
session_id,
last_output: "main answer".to_string(),
attempts: 2,
reason: "Invalid JSON".to_string(),
})
.await
.expect("send extraction failed event");
match completion_rx.await.expect("completion outcome") {
FlowTurnOutcome::Failed { reason } => {
assert!(
reason.contains("structured output extraction failed after 2 attempt(s)"),
"reason should preserve extraction attempts: {reason}"
);
assert!(
reason.contains("Invalid JSON"),
"reason should preserve extraction failure reason: {reason}"
);
assert!(
reason.contains("main answer"),
"reason should preserve main output context: {reason}"
);
}
other => panic!("expected failed outcome, got {other:?}"),
}
bridge.await.expect("bridge exits after extraction event");
}
}