gemini_adk_fluent_rs/handoff.rs
1//! Warm handoff — transfer the conversation to a human with its context
2//! intact.
3//!
4//! The single UX bar for an escalation is that the person picking up never
5//! asks the caller to repeat themselves. What the receiving desk needs is
6//! not the raw session but a compact, serializable packet: who is calling,
7//! what has been said, what the governed flow has established, and — when a
8//! summarizer is available — two sentences of "what they want and what was
9//! already tried".
10//!
11//! [`HandoffRecorder`] accumulates the final transcripts as the call runs
12//! (redacted upstream when
13//! [`redaction`](gemini_adk_rs::live::redaction) is installed, so a packet
14//! can never leak what the router already scrubbed). [`HandoffPacket`] is
15//! the snapshot the connector delivers however its platform wants —
16//! a screen-pop payload, SIP headers, a CRM note. Assembly is transport-
17//! agnostic by design; delivering it is the connector's job.
18//!
19//! ```no_run
20//! # use gemini_adk_fluent_rs::prelude::*;
21//! # use gemini_adk_fluent_rs::handoff::HandoffRecorder;
22//! # use std::sync::Arc;
23//! # async fn run(handle: LiveHandle, flash_llm: Arc<dyn BaseLlm>) -> Result<(), Box<dyn std::error::Error>> {
24//! let recorder = HandoffRecorder::attach(&handle, 40);
25//! // … the call runs; escalation triggers …
26//! let mut packet = recorder.packet(&handle, &["telephony:caller", "intent", "verified"]);
27//! packet.summarize(&*flash_llm).await.ok(); // optional 2–3 sentence summary
28//! let json = serde_json::to_string(&packet)?; // hand this to the agent desktop
29//! # let _ = json; Ok(())
30//! # }
31//! ```
32
33use std::collections::{BTreeMap, VecDeque};
34use std::sync::Arc;
35
36use serde::{Deserialize, Serialize};
37use tokio::task::JoinHandle;
38
39use gemini_adk_rs::live::{LiveEvent, LiveHandle};
40use gemini_adk_rs::llm::{BaseLlm, LlmError, LlmRequest};
41
42/// One finalized turn of the conversation.
43#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
44pub struct HandoffTurn {
45 /// `"caller"` or `"agent"`.
46 pub speaker: String,
47 /// The final transcript of the turn (redacted upstream if redaction is
48 /// installed on the session).
49 pub text: String,
50}
51
52/// The context packet handed to the receiving human.
53#[derive(Debug, Clone, Serialize, Deserialize, Default)]
54pub struct HandoffPacket {
55 /// A short synthesized summary of what the caller wants and what has
56 /// been attempted — filled by [`summarize`](Self::summarize), `None`
57 /// until then.
58 #[serde(skip_serializing_if = "Option::is_none")]
59 pub summary: Option<String>,
60 /// The last N finalized turns, oldest first.
61 pub transcript: Vec<HandoffTurn>,
62 /// The requested state keys and their current values — authentication
63 /// status, captured intent, caller identity, whatever the deployment
64 /// selects. Keys absent from state are omitted.
65 pub state: BTreeMap<String, serde_json::Value>,
66 /// The governed flow's standing at handoff: steps done, steps active,
67 /// and requirements still unmet. `None` when the session is ungoverned.
68 #[serde(skip_serializing_if = "Option::is_none")]
69 pub flow: Option<HandoffFlowStatus>,
70}
71
72/// The flow's standing at the moment of handoff.
73#[derive(Debug, Clone, Serialize, Deserialize, Default)]
74pub struct HandoffFlowStatus {
75 /// Steps that have latched done.
76 pub done: Vec<String>,
77 /// Steps currently active.
78 pub active: Vec<String>,
79 /// Required steps not yet completed — what the human still has to do.
80 pub missing: Vec<String>,
81}
82
83impl HandoffPacket {
84 /// Fill [`summary`](Self::summary) with a 2–3 sentence synthesis of the
85 /// transcript, using any [`BaseLlm`]. The packet is useful without it;
86 /// call this only when the escalation path has the latency budget.
87 pub async fn summarize(&mut self, llm: &dyn BaseLlm) -> Result<(), LlmError> {
88 let mut conversation = String::new();
89 for turn in &self.transcript {
90 conversation.push_str(&format!("{}: {}\n", turn.speaker, turn.text));
91 }
92 let mut request = LlmRequest::from_text(conversation);
93 request.system_instruction = Some(
94 "Summarize this call for the human agent about to take it over, \
95 in 2-3 sentences: what the caller wants, and what has already \
96 been tried or established. No preamble."
97 .into(),
98 );
99 self.summary = Some(llm.generate(request).await?.text());
100 Ok(())
101 }
102}
103
104/// Accumulates the conversation as it happens, so a packet can be assembled
105/// at any moment without asking the caller to wait.
106pub struct HandoffRecorder {
107 turns: Arc<parking_lot::Mutex<VecDeque<HandoffTurn>>>,
108 task: JoinHandle<()>,
109}
110
111impl HandoffRecorder {
112 /// Start recording finalized transcripts from the session's event
113 /// stream, keeping the most recent `max_turns`.
114 pub fn attach(handle: &LiveHandle, max_turns: usize) -> HandoffRecorder {
115 let turns: Arc<parking_lot::Mutex<VecDeque<HandoffTurn>>> =
116 Arc::new(parking_lot::Mutex::new(VecDeque::new()));
117 let mut events = handle.events();
118 let store = turns.clone();
119 let task = tokio::spawn(async move {
120 loop {
121 let (speaker, text) = match events.recv().await {
122 Ok(LiveEvent::InputTranscript {
123 text,
124 is_final: true,
125 }) => ("caller", text),
126 Ok(LiveEvent::OutputTranscript {
127 text,
128 is_final: true,
129 }) => ("agent", text),
130 Ok(LiveEvent::TextComplete(text)) => ("agent", text),
131 Ok(_) => continue,
132 Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
133 Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
134 };
135 if text.trim().is_empty() {
136 continue;
137 }
138 let mut turns = store.lock();
139 turns.push_back(HandoffTurn {
140 speaker: speaker.into(),
141 text,
142 });
143 while turns.len() > max_turns {
144 turns.pop_front();
145 }
146 }
147 });
148 HandoffRecorder { turns, task }
149 }
150
151 /// Assemble the packet: recorded transcript, the requested state keys,
152 /// and the flow's current standing. Synchronous — callable from any
153 /// escalation path, including a tool handler.
154 pub fn packet(&self, handle: &LiveHandle, state_keys: &[&str]) -> HandoffPacket {
155 let state = handle.state();
156 let mut selected = BTreeMap::new();
157 for &key in state_keys {
158 if let Some(value) = state.get::<serde_json::Value>(key) {
159 selected.insert(key.to_string(), value);
160 }
161 }
162 let flow = handle.explain().map(|explanation| HandoffFlowStatus {
163 done: state.get::<Vec<String>>("flow:done").unwrap_or_default(),
164 active: explanation.active,
165 missing: explanation.missing_requirements,
166 });
167 HandoffPacket {
168 summary: None,
169 transcript: self.turns.lock().iter().cloned().collect(),
170 state: selected,
171 flow,
172 }
173 }
174
175 /// Stop recording. Packets already assembled are unaffected.
176 pub fn detach(self) {
177 self.task.abort();
178 }
179}
180
181#[cfg(test)]
182mod tests {
183 use super::*;
184
185 #[test]
186 fn packet_serializes_without_optional_sections() {
187 let packet = HandoffPacket {
188 summary: None,
189 transcript: vec![HandoffTurn {
190 speaker: "caller".into(),
191 text: "I want to change my booking".into(),
192 }],
193 state: BTreeMap::from([("verified".into(), serde_json::json!(true))]),
194 flow: None,
195 };
196 let json = serde_json::to_value(&packet).unwrap();
197 assert!(json.get("summary").is_none(), "None summary is omitted");
198 assert!(json.get("flow").is_none(), "ungoverned session omits flow");
199 assert_eq!(json["transcript"][0]["speaker"], "caller");
200 assert_eq!(json["state"]["verified"], true);
201 }
202}