use crate::CallOption;
use crate::call::{ActiveCallRef, ActiveCallType, Command};
use crate::event::EventReceiver;
use anyhow::{Result, anyhow};
use serde_json::json;
use std::time::Duration;
use tracing::{error, info, warn};
use super::{Playbook, PlaybookConfig, dialogue::DialogueHandler, handler::LlmHandler};
pub struct PlaybookRunner {
handler: Box<dyn DialogueHandler>,
call: ActiveCallRef,
config: PlaybookConfig,
event_receiver: EventReceiver,
}
impl PlaybookRunner {
pub fn with_handler(
handler: Box<dyn DialogueHandler>,
call: ActiveCallRef,
config: PlaybookConfig,
) -> Self {
let event_receiver = call.event_sender.subscribe();
Self {
handler,
call,
config,
event_receiver,
}
}
pub fn new(playbook: Playbook, call: ActiveCallRef) -> Result<Self> {
let event_receiver = call.event_sender.subscribe();
if let Ok(mut state) = call.call_state.try_write() {
if state.option.is_none() {
state.option = Some(CallOption::default());
}
if let Some(option) = state.option.as_mut() {
apply_playbook_config(option, &playbook.config);
}
}
let handler: Box<dyn DialogueHandler> = if let Some(llm_config) = &playbook.config.llm {
let mut llm_config = llm_config.clone();
if let Some(greeting) = playbook.config.greeting.clone() {
llm_config.greeting = Some(greeting);
}
let interruption_config = playbook.config.interruption.clone().unwrap_or_default();
let dtmf_config = playbook.config.dtmf.clone();
let dtmf_collectors = playbook.config.dtmf_collectors.clone();
let mut llm_handler = LlmHandler::new(
llm_config,
interruption_config,
playbook.config.follow_up,
playbook.scenes.clone(),
dtmf_config,
dtmf_collectors,
playbook.initial_scene_id.clone(),
playbook.config.sip.clone(),
);
llm_handler.set_event_sender(call.event_sender.clone());
llm_handler.set_call(call.clone());
Box::new(llm_handler)
} else {
return Err(anyhow!(
"No valid dialogue handler configuration found (e.g. missing 'llm')"
));
};
Ok(Self {
handler,
call,
config: playbook.config,
event_receiver,
})
}
pub async fn run(mut self) {
info!(
"PlaybookRunner started for session {}",
self.call.session_id
);
let mut answered = {
let state = self.call.call_state.read().await;
state.answer_time.is_some()
};
let wait_for_media_ready =
matches!(self.call.call_type, ActiveCallType::Sip | ActiveCallType::B2bua);
let mut media_ready = !wait_for_media_ready;
if let Ok(commands) = self.handler.on_start().await {
for cmd in commands {
let is_media = matches!(cmd, Command::Tts { .. } | Command::Play { .. });
if is_media && (!answered || !media_ready) {
info!(
wait_for_media_ready,
"Waiting for call media readiness before executing media command..."
);
while let Ok(event) = self.event_receiver.recv().await {
match &event {
crate::event::SessionEvent::Answer { .. } => {
info!("Call established");
answered = true;
}
crate::event::SessionEvent::MediaReady { .. } => {
info!("Call media ready");
media_ready = true;
}
crate::event::SessionEvent::Hangup { .. } => {
info!("Call hung up before media command, stopping");
return;
}
_ => {}
}
if answered && media_ready {
info!("Proceeding to execute media command");
break;
}
}
}
if let Err(e) = self.call.enqueue_command(cmd).await {
error!("Failed to enqueue start command: {}", e);
}
}
}
if !answered {
info!("Waiting for call establishment...");
while let Ok(event) = self.event_receiver.recv().await {
match &event {
crate::event::SessionEvent::Answer { .. } => {
info!("Call established, proceeding to playbook handles");
break;
}
crate::event::SessionEvent::Hangup { .. } => {
info!("Call hung up before established, stopping");
return;
}
_ => {}
}
}
}
while let Ok(event) = self.event_receiver.recv().await {
if let Ok(commands) = self.handler.on_event(&event).await {
for cmd in commands {
if let Err(e) = self.call.enqueue_command(cmd).await {
error!("Failed to enqueue command: {}", e);
}
}
}
match &event {
crate::event::SessionEvent::Hangup { .. } => {
info!("Call hung up, stopping playbook");
break;
}
_ => {}
}
}
if let Some(posthook) = self.config.posthook.clone() {
let mut handler = self.handler;
let session_id = self.call.session_id.clone();
drop(self.call);
crate::spawn(async move {
info!("Executing posthook for session {}", session_id);
let posthook_timeout = Duration::from_secs(
posthook.timeout.unwrap_or(30) as u64
);
let posthook_task = async {
let summary = if let Some(summary_type) = &posthook.summary {
match handler.summarize(summary_type.prompt()).await {
Ok(s) => Some(s),
Err(e) => {
error!("Failed to generate summary: {}", e);
None
}
}
} else {
None
};
let history = if posthook.include_history.unwrap_or(true) {
Some(handler.get_history().await)
} else {
None
};
let payload = json!({
"sessionId": session_id,
"summary": summary,
"history": history,
"timestamp": chrono::Utc::now().to_rfc3339(),
});
let client = reqwest::Client::new();
let method = posthook
.method
.as_deref()
.unwrap_or("POST")
.parse::<reqwest::Method>()
.unwrap_or(reqwest::Method::POST);
let mut request = client.request(method, &posthook.url).json(&payload);
if let Some(headers) = posthook.headers {
for (k, v) in headers {
request = request.header(k, v);
}
}
match request.send().await {
Ok(resp) => {
if resp.status().is_success() {
info!("Posthook sent successfully");
} else {
warn!("Posthook failed with status: {}", resp.status());
}
}
Err(e) => {
error!("Failed to send posthook: {}", e);
}
}
};
if tokio::time::timeout(posthook_timeout, posthook_task).await.is_err() {
error!("Posthook timed out for session {}", session_id);
}
});
}
}
}
pub fn apply_playbook_config(option: &mut CallOption, config: &PlaybookConfig) {
let api_key = config.llm.as_ref().and_then(|llm| llm.api_key.clone());
if let Some(mut asr) = config.asr.clone() {
if asr.secret_key.is_none() {
asr.secret_key = api_key.clone();
}
option.asr = Some(asr);
}
if let Some(mut tts) = config.tts.clone() {
if tts.secret_key.is_none() {
tts.secret_key = api_key.clone();
}
option.tts = Some(tts);
}
if let Some(vad) = config.vad.clone() {
option.vad = Some(vad);
}
if let Some(denoise) = config.denoise {
option.denoise = Some(denoise);
}
if let Some(agc) = config.agc.clone() {
option.agc = Some(agc);
}
if let Some(ambiance) = config.ambiance.clone() {
option.ambiance = Some(ambiance);
}
if let Some(recorder) = config.recorder.clone() {
option.recorder = Some(recorder);
}
if let Some(extra) = config.extra.clone() {
option.extra = Some(extra);
}
if let Some(mut realtime) = config.realtime.clone() {
if realtime.secret_key.is_none() {
realtime.secret_key = api_key.clone();
}
option.realtime = Some(realtime);
}
if let Some(mut eou) = config.eou.clone() {
if eou.secret_key.is_none() {
eou.secret_key = api_key;
}
option.eou = Some(eou);
}
if let Some(sip) = config.sip.clone() {
option.sip = Some(sip);
}
if let Some(ringback) = config.ringback_detection.clone() {
option.ringback_detection = Some(ringback);
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
EouOption, media::recorder::RecorderOption, media::vad::VADOption,
synthesis::SynthesisOption, transcription::TranscriptionOption,
};
use std::collections::HashMap;
#[test]
fn apply_playbook_config_sets_fields() {
let mut option = CallOption::default();
let mut extra = HashMap::new();
extra.insert("k".to_string(), "v".to_string());
let config = PlaybookConfig {
asr: Some(TranscriptionOption::default()),
tts: Some(SynthesisOption::default()),
vad: Some(VADOption::default()),
denoise: Some(true),
recorder: Some(RecorderOption::default()),
extra: Some(extra.clone()),
eou: Some(EouOption {
r#type: Some("test".to_string()),
endpoint: None,
secret_key: Some("key".to_string()),
secret_id: Some("id".to_string()),
timeout: Some(123),
extra: None,
}),
..Default::default()
};
apply_playbook_config(&mut option, &config);
assert!(option.asr.is_some());
assert!(option.tts.is_some());
assert!(option.vad.is_some());
assert_eq!(option.denoise, Some(true));
assert!(option.recorder.is_some());
assert_eq!(option.extra, Some(extra));
assert!(option.eou.is_some());
}
#[test]
fn apply_playbook_config_propagates_api_key() {
let mut option = CallOption::default();
let config = PlaybookConfig {
llm: Some(super::super::LlmConfig {
api_key: Some("test-key".to_string()),
..Default::default()
}),
asr: Some(TranscriptionOption::default()),
tts: Some(SynthesisOption::default()),
eou: Some(EouOption::default()),
..Default::default()
};
apply_playbook_config(&mut option, &config);
assert_eq!(
option.asr.as_ref().unwrap().secret_key,
Some("test-key".to_string())
);
assert_eq!(
option.tts.as_ref().unwrap().secret_key,
Some("test-key".to_string())
);
assert_eq!(
option.eou.as_ref().unwrap().secret_key,
Some("test-key".to_string())
);
}
#[test]
fn posthook_config_timeout_default() {
use crate::playbook::PostHookConfig;
let config = PostHookConfig {
url: "http://example.com".to_string(),
..Default::default()
};
assert_eq!(config.timeout, None);
assert_eq!(config.timeout.unwrap_or(30), 30);
let config = PostHookConfig {
url: "http://example.com".to_string(),
timeout: Some(60),
..Default::default()
};
assert_eq!(config.timeout, Some(60));
assert_eq!(config.timeout.unwrap_or(30), 60);
}
#[test]
fn posthook_config_serde_with_timeout() {
use crate::playbook::PostHookConfig;
let json = r#"{"url": "http://example.com", "timeout": 45}"#;
let config: PostHookConfig = serde_json::from_str(json).unwrap();
assert_eq!(config.timeout, Some(45));
assert_eq!(config.url, "http://example.com");
let json = r#"{"url": "http://example.com"}"#;
let config: PostHookConfig = serde_json::from_str(json).unwrap();
assert_eq!(config.timeout, None);
}
}