gemini_adk_fluent_rs/
simulation.rs

1//! Model-free conversation simulation.
2//!
3//! A deterministic harness that drives a [`CompiledConversation`] without any live
4//! API: a **fake user** supplies utterances (run through the conversation's
5//! recognizers) or sets slots directly, tools succeed on demand or after a
6//! latency, and the [`FlowStack`] advances turn by turn. Everything is driven by
7//! `State` + guards, so motifs, repair, policies, and digressions become testable
8//! in CI — "a flow SDK with simulation is infra; without it, a demo framework".
9//!
10//! ```no_run
11//! # use gemini_adk_fluent_rs::prelude::*;
12//! # use gemini_adk_fluent_rs::conversation::Conversation;
13//! # use gemini_adk_fluent_rs::simulation::Sim;
14//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
15//! let convo = Conversation::new("booking")
16//!     .stage("check").collect(["party_size", "slot"])
17//!         .next("confirm", Guard::captured(["party_size", "slot"]))
18//!     .stage("confirm").commit("book", Guard::is_true("user_confirmed"))
19//!         .next("done", Guard::called_ok("book"))
20//!     .stage("done").terminal()
21//!     .require(["done"])
22//!     .compile()?;
23//! let mut sim = Sim::new(&convo, Enforcement::Enforce);
24//! sim.user("a table for 4 tomorrow at 7pm").await;
25//! assert!(sim.active().contains(&"check".to_string()));
26//! assert!(!sim.allowed("book"));            // not confirmed yet
27//! sim.set("user_confirmed", true);
28//! sim.tool_ok("book");
29//! assert!(sim.is_complete());
30//! # Ok(())
31//! # }
32//! ```
33
34use std::collections::BTreeMap;
35use std::sync::Arc;
36use std::time::Instant;
37
38use serde::{Deserialize, Serialize};
39use serde_json::Value;
40
41use gemini_adk_rs::flow::{Enforcement, FlowExplanation};
42use gemini_adk_rs::live::{TranscriptTurn, TurnExtractor};
43use gemini_adk_rs::state::State;
44
45use crate::conversation::{CompiledConversation, FlowStack};
46
47struct BoundExtractor {
48    extractor: Arc<dyn TurnExtractor>,
49    /// (field name, state key) — how to promote the returned record into `State`.
50    fields: Vec<(String, String)>,
51}
52
53/// A deterministic, model-free driver for a compiled conversation.
54pub struct Sim {
55    stack: FlowStack,
56    extractors: Vec<BoundExtractor>,
57    state: State,
58    turn_no: u32,
59    /// Tools scheduled to succeed at a future turn — models tool latency.
60    pending_tools: Vec<(String, u32)>,
61}
62
63impl Sim {
64    /// Build a simulator over `convo` in the given enforcement mode.
65    pub fn new(convo: &CompiledConversation, mode: Enforcement) -> Self {
66        let extractors = convo
67            .all_extractors()
68            .into_iter()
69            .map(|e| BoundExtractor {
70                fields: e.field_state_keys(),
71                extractor: e.into_extractor(),
72            })
73            .collect();
74        Self {
75            stack: convo.stack(mode),
76            extractors,
77            state: State::new(),
78            turn_no: 0,
79            pending_tools: Vec::new(),
80        }
81    }
82
83    /// Set a state value directly (information a recognizer can't supply, or a
84    /// scripted shortcut). Does not advance a turn.
85    pub fn set(&self, key: impl Into<String>, value: impl Serialize) -> &Self {
86        let _ = self.state.set(key, value);
87        self
88    }
89
90    /// The fake user speaks: run the conversation's extractors over the utterance
91    /// to fill slots (respecting validators), then advance a turn.
92    pub async fn user(&mut self, utterance: &str) -> &mut Self {
93        let window = [TranscriptTurn {
94            turn_number: self.turn_no,
95            user: utterance.to_string(),
96            model: String::new(),
97            tool_calls: Vec::new(),
98            timestamp: Instant::now(),
99        }];
100        for bound in &self.extractors {
101            if let Ok(Value::Object(obj)) = bound
102                .extractor
103                .extract_with_state(&window, &self.state)
104                .await
105            {
106                for (name, key) in &bound.fields {
107                    if let Some(v) = obj.get(name)
108                        && !v.is_null()
109                    {
110                        let _ = self.state.set(key.clone(), v.clone());
111                    }
112                }
113            }
114        }
115        self.advance();
116        self
117    }
118
119    /// Advance a turn with no new user input (e.g. waiting on a tool/resolver).
120    pub fn turn(&mut self) -> &mut Self {
121        self.advance();
122        self
123    }
124
125    /// A tool succeeds now; records it and advances a turn (processing any
126    /// digression resume).
127    pub fn tool_ok(&mut self, tool: &str) -> &mut Self {
128        self.stack.on_tool_ok(tool, &self.state);
129        self.advance();
130        self
131    }
132
133    /// A tool fails or times out. Counts toward the active stage's
134    /// `escalate_after_tool_failures`; does not advance a turn.
135    pub fn tool_failed(&mut self, tool: &str) -> &mut Self {
136        self.stack.observe_tool(tool, false, &self.state);
137        self
138    }
139
140    /// The user barges in on the model. Counts toward the active stage's
141    /// `escalate_after_interruptions`; does not advance a turn.
142    pub fn interrupt(&mut self) -> &mut Self {
143        self.stack.on_interrupted(&self.state);
144        self
145    }
146
147    /// Schedule a tool to succeed `after` turns — models tool latency.
148    pub fn schedule_tool(&mut self, tool: impl Into<String>, after: u32) -> &mut Self {
149        self.pending_tools
150            .push((tool.into(), self.turn_no + after.max(1)));
151        self
152    }
153
154    fn advance(&mut self) {
155        self.turn_no += 1;
156        // Fire any tools whose latency has elapsed.
157        let due: Vec<String> = self
158            .pending_tools
159            .iter()
160            .filter(|(_, at)| *at <= self.turn_no)
161            .map(|(t, _)| t.clone())
162            .collect();
163        self.pending_tools.retain(|(_, at)| *at > self.turn_no);
164        for tool in due {
165            self.stack.on_tool_ok(&tool, &self.state);
166        }
167        self.stack.on_turn(&self.state);
168    }
169
170    /// Apply one scripted step: drive the conversation, or check an
171    /// expectation. `Err` carries why an expectation did not hold. This is
172    /// what [`Scenario::run`] does for each step, exposed so another driver
173    /// (an interactive session, a binding) shares the exact semantics.
174    pub async fn apply(&mut self, step: &SimStep) -> Result<(), String> {
175        match step {
176            SimStep::User(text) => {
177                self.user(text).await;
178            }
179            SimStep::Set { key, value } => {
180                self.set(key.clone(), value.clone());
181            }
182            SimStep::Remove { key } => {
183                let _ = self.state.remove(key);
184            }
185            SimStep::ToolOk(tool) => {
186                self.tool_ok(tool);
187            }
188            SimStep::ToolFailed(tool) => {
189                self.tool_failed(tool);
190            }
191            SimStep::Interrupt => {
192                self.interrupt();
193            }
194            SimStep::ToolResult { tool, ok } => {
195                self.stack.observe_tool(tool, *ok, &self.state);
196            }
197            SimStep::ScheduleTool { tool, after } => {
198                self.schedule_tool(tool.clone(), *after);
199            }
200            SimStep::Turn => {
201                self.turn();
202            }
203            SimStep::ExpectActive(expected) => {
204                let active = self.active();
205                for e in expected {
206                    if !active.contains(e) {
207                        return Err(format!("expected active '{e}', got {active:?}"));
208                    }
209                }
210            }
211            SimStep::ExpectDenied(tool) => {
212                if self.allowed(tool) {
213                    return Err(format!("expected '{tool}' denied, but it was admitted"));
214                }
215            }
216            SimStep::ExpectAllowed(tool) => {
217                if !self.allowed(tool) {
218                    let why = self.denied().get(tool).cloned().unwrap_or_default();
219                    return Err(format!("expected '{tool}' allowed, but denied: {why}"));
220                }
221            }
222            SimStep::ExpectSlot { key, value } => {
223                let got = self.state().get_raw(key);
224                if got.as_ref() != Some(value) {
225                    return Err(format!("expected slot '{key}' = {value}, got {got:?}"));
226                }
227            }
228            SimStep::ExpectComplete => {
229                if !self.is_complete() {
230                    return Err("expected conversation complete".into());
231                }
232            }
233        }
234        Ok(())
235    }
236
237    /// Active step ids in the currently-driving layer.
238    pub fn active(&self) -> Vec<String> {
239        self.stack.explain(&self.state).active
240    }
241
242    /// The active digression, if one is suspending the main flow.
243    pub fn active_overlay(&self) -> Option<&str> {
244        self.stack.active_overlay()
245    }
246
247    /// The instructions (stage postures) projected to the model this turn.
248    ///
249    /// On the turn a digression completes, these are its *closing* stage's —
250    /// the safety hand-off's "hand off to a human now", say — which is what a
251    /// live session would send. Assert on these to test that a digression is
252    /// heard, not merely that it fired.
253    pub fn postures(&self) -> Vec<String> {
254        self.stack.active_postures(&self.state)
255    }
256
257    /// Whether the conversation has ended because a `Resume::Terminate`
258    /// digression ran (as opposed to the main flow finishing).
259    pub fn is_terminated(&self) -> bool {
260        self.stack.is_terminated()
261    }
262
263    /// Whether `tool` is admitted right now.
264    pub fn allowed(&self, tool: &str) -> bool {
265        self.stack.admits_tool(tool, &self.state).is_ok()
266    }
267
268    /// Currently-blocked tools, mapped to the reason.
269    pub fn denied(&self) -> BTreeMap<String, String> {
270        self.stack.explain(&self.state).blocked_tools
271    }
272
273    /// Whether the conversation is complete.
274    pub fn is_complete(&self) -> bool {
275        self.stack.is_complete()
276    }
277
278    /// Read a slot value.
279    pub fn slot<T: serde::de::DeserializeOwned>(&self, key: &str) -> Option<T> {
280        self.state.get(key)
281    }
282
283    /// The active layer's control-plane explanation.
284    pub fn explain(&self) -> FlowExplanation {
285        self.stack.explain(&self.state)
286    }
287
288    /// The simulation state (for custom assertions / slot evidence).
289    pub fn state(&self) -> &State {
290        &self.state
291    }
292}
293
294/// One step in a serializable [`Scenario`].
295#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
296#[serde(rename_all = "snake_case")]
297pub enum SimStep {
298    /// The fake user speaks (recognizers fill slots), then a turn advances.
299    User(String),
300    /// Set a state value directly.
301    Set {
302        /// State key.
303        key: String,
304        /// Value to store.
305        value: Value,
306    },
307    /// Remove a state value, as the recorded runtime did.
308    Remove {
309        /// State key.
310        key: String,
311    },
312    /// A tool succeeds now.
313    ToolOk(String),
314    /// A tool fails (or times out) now; counts toward the active stage's
315    /// repair policy. Does not advance a turn.
316    ToolFailed(String),
317    /// The user barges in on the model; counts toward the active stage's
318    /// repair policy. Does not advance a turn.
319    Interrupt,
320    /// A tool call completes, successfully or not, where the live runtime
321    /// records it: the flow observes it without advancing a turn (unlike
322    /// [`ToolOk`](Self::ToolOk)). This is what a scenario extracted from a
323    /// recording uses.
324    ToolResult {
325        /// Tool name.
326        tool: String,
327        /// Whether it succeeded.
328        ok: bool,
329    },
330    /// Schedule a tool to succeed after N turns (latency).
331    ScheduleTool {
332        /// Tool name.
333        tool: String,
334        /// Turns to wait.
335        after: u32,
336    },
337    /// Advance a turn with no input.
338    Turn,
339    /// Assert these step ids are active.
340    ExpectActive(Vec<String>),
341    /// Assert a tool is currently blocked.
342    ExpectDenied(String),
343    /// Assert a tool is currently admitted.
344    ExpectAllowed(String),
345    /// Assert a slot equals a value.
346    ExpectSlot {
347        /// State key.
348        key: String,
349        /// Expected value.
350        value: Value,
351    },
352    /// Assert the conversation is complete.
353    ExpectComplete,
354}
355
356/// A serializable simulation script — a deterministic, model-free test case that
357/// can be authored in code or loaded from YAML/JSON.
358#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
359pub struct Scenario {
360    /// Scenario name (for diagnostics).
361    pub name: String,
362    /// The steps to execute, in order.
363    pub steps: Vec<SimStep>,
364}
365
366/// State keys the runtime owns. A scenario extracted from a recording skips
367/// them: the simulator recomputes them rather than being told them.
368const RUNTIME_PREFIXES: &[&str] = &[
369    "session:",
370    "flow:",
371    "derived:",
372    "repair:",
373    "correction:",
374    "state_meta:",
375    "idempotency:",
376    "compensated:",
377    "verbatim:",
378    "turn:",
379    "bg:",
380];
381
382fn runtime_owned(key: &str) -> bool {
383    RUNTIME_PREFIXES.iter().any(|p| key.starts_with(p))
384}
385
386impl Scenario {
387    /// Turn a recorded session into a regression scenario: the incident
388    /// becomes a test.
389    ///
390    /// `journal` is the session's mutation journal (see
391    /// [`FileJournalSink`](gemini_adk_rs::state::FileJournalSink) and
392    /// [`read_journal`](gemini_adk_rs::state::read_journal)), which a
393    /// governed session writes as one ordered timeline. The scenario replays
394    /// what the application and user contributed and checks what governance
395    /// decided:
396    ///
397    /// - a slot or flag write becomes `Set`;
398    /// - a tool the flow admitted becomes `ExpectAllowed`, one it refused
399    ///   `ExpectDenied`, and its outcome `ToolResult`;
400    /// - every turn boundary where the flow was evaluated (a `flow:active`
401    ///   write) becomes a `Turn` followed by `ExpectActive` with the steps
402    ///   the session had then.
403    ///
404    /// Keys the runtime owns (`session:`, `flow:`, `derived:`, repair and
405    /// correction signals, …) are not replayed: the simulator must reach them
406    /// itself.
407    ///
408    /// Run the result against the conversation spec in CI. If a change to
409    /// the spec alters what that session would have done, the scenario fails
410    /// at the step where the two diverge.
411    pub fn from_journal(
412        name: impl Into<String>,
413        journal: &[gemini_adk_rs::state::StateMutation],
414    ) -> Self {
415        use gemini_adk_rs::flow::{TOOL_CALL_KEY, TOOL_DENIED_KEY, TOOL_RESULT_KEY};
416
417        let tool_of = |v: &Option<Value>| {
418            v.as_ref()
419                .and_then(|v| v["tool"].as_str())
420                .map(str::to_string)
421        };
422        let mut steps = Vec::new();
423        // `None` is a removal (`State::remove`, `clear_prefix`), which must
424        // replay as one: a key set to `null` is still present.
425        let mut pending: Vec<(String, Option<Value>)> = Vec::new();
426        let flush = |pending: &mut Vec<(String, Option<Value>)>, steps: &mut Vec<SimStep>| {
427            for (key, value) in pending.drain(..) {
428                steps.push(match value {
429                    Some(value) => SimStep::Set { key, value },
430                    None => SimStep::Remove { key },
431                });
432            }
433        };
434        let mut ordered: Vec<&gemini_adk_rs::state::StateMutation> = journal.iter().collect();
435        ordered.sort_by_key(|m| m.sequence);
436        for m in ordered {
437            match m.key.as_str() {
438                "flow:active" => {
439                    flush(&mut pending, &mut steps);
440                    steps.push(SimStep::Turn);
441                    let active: Vec<String> = m
442                        .new
443                        .clone()
444                        .and_then(|v| serde_json::from_value(v).ok())
445                        .unwrap_or_default();
446                    steps.push(SimStep::ExpectActive(active));
447                }
448                TOOL_CALL_KEY => {
449                    if let Some(tool) = tool_of(&m.new) {
450                        flush(&mut pending, &mut steps);
451                        steps.push(SimStep::ExpectAllowed(tool));
452                    }
453                }
454                TOOL_DENIED_KEY => {
455                    if let Some(tool) = tool_of(&m.new) {
456                        flush(&mut pending, &mut steps);
457                        steps.push(SimStep::ExpectDenied(tool));
458                    }
459                }
460                TOOL_RESULT_KEY => {
461                    if let Some(tool) = tool_of(&m.new) {
462                        flush(&mut pending, &mut steps);
463                        let ok = m.new.as_ref().is_some_and(|v| v["ok"] == true);
464                        steps.push(SimStep::ToolResult { tool, ok });
465                    }
466                }
467                key if runtime_owned(key) => {}
468                key => {
469                    let value = m.new.clone();
470                    // Keep the latest value per key, in first-written order.
471                    match pending.iter_mut().find(|(k, _)| k == key) {
472                        Some(slot) => slot.1 = value,
473                        None => pending.push((key.to_string(), value)),
474                    }
475                }
476            }
477        }
478        flush(&mut pending, &mut steps);
479        Self {
480            name: name.into(),
481            steps,
482        }
483    }
484
485    /// Run the scenario against `convo`. Returns `Ok(())` if every `Expect*` step
486    /// holds, else `Err` with the failing step index and a diagnostic.
487    pub async fn run(&self, convo: &CompiledConversation, mode: Enforcement) -> Result<(), String> {
488        let mut sim = Sim::new(convo, mode);
489        for (i, step) in self.steps.iter().enumerate() {
490            sim.apply(step)
491                .await
492                .map_err(|msg| format!("[{}] step {i} ({step:?}): {msg}", self.name))?;
493        }
494        Ok(())
495    }
496}
497
498#[cfg(test)]
499mod tests {
500    use super::*;
501    use crate::conversation::Conversation;
502    use gemini_adk_rs::flow::Guard;
503    use gemini_adk_rs::frame::{Frame, FrameSpec, SlotRecognizer, SlotSpec};
504
505    struct Booking;
506    impl Frame for Booking {
507        fn frame() -> FrameSpec {
508            FrameSpec {
509                name: "booking".into(),
510                slots: vec![SlotSpec {
511                    recognizer: Some(SlotRecognizer::IntegerNear(vec!["people".into()])),
512                    ..SlotSpec::new("party_size")
513                }],
514            }
515        }
516    }
517
518    fn booking() -> CompiledConversation {
519        Conversation::new("booking")
520            .stage("collect")
521            .collect_frame::<Booking>()
522            .next("confirm", Guard::captured(["party_size"]))
523            .stage("confirm")
524            .commit("book", Guard::is_true("user_confirmed"))
525            .next("done", Guard::called_ok("book"))
526            .stage("done")
527            .terminal()
528            .require(["done"])
529            .compile()
530            .expect("compiles")
531    }
532
533    #[tokio::test]
534    async fn fake_user_fills_slots_and_gates_commit() {
535        let convo = booking();
536        let mut sim = Sim::new(&convo, Enforcement::Enforce);
537
538        assert!(sim.active().contains(&"collect".to_string()));
539        assert!(!sim.allowed("book"));
540
541        // The fake user speaks; the recognizer fills party_size.
542        sim.user("a table for 4 people").await;
543        assert_eq!(sim.slot::<u32>("party_size"), Some(4));
544        assert!(sim.active().contains(&"confirm".to_string()));
545
546        // book is gated until confirmation.
547        assert!(!sim.allowed("book"));
548        sim.set("user_confirmed", true);
549        sim.turn();
550        assert!(sim.allowed("book"));
551
552        sim.tool_ok("book");
553        assert!(sim.is_complete());
554    }
555
556    #[tokio::test]
557    async fn scenario_runs_and_round_trips() {
558        let scenario = Scenario {
559            name: "happy_path".into(),
560            steps: vec![
561                SimStep::ExpectActive(vec!["collect".into()]),
562                SimStep::ExpectDenied("book".into()),
563                SimStep::User("party of 4 people".into()),
564                SimStep::ExpectSlot {
565                    key: "party_size".into(),
566                    value: serde_json::json!(4),
567                },
568                SimStep::ExpectActive(vec!["confirm".into()]),
569                SimStep::Set {
570                    key: "user_confirmed".into(),
571                    value: serde_json::json!(true),
572                },
573                SimStep::Turn,
574                SimStep::ExpectAllowed("book".into()),
575                SimStep::ToolOk("book".into()),
576                SimStep::ExpectComplete,
577            ],
578        };
579
580        scenario
581            .run(&booking(), Enforcement::Enforce)
582            .await
583            .expect("scenario passes");
584
585        // Scenarios are serializable (authorable as YAML/JSON).
586        let json = serde_json::to_string(&scenario).unwrap();
587        let back: Scenario = serde_json::from_str(&json).unwrap();
588        back.run(&booking(), Enforcement::Enforce)
589            .await
590            .expect("round-tripped scenario passes");
591    }
592
593    #[tokio::test]
594    async fn scenario_reports_failed_expectation() {
595        let scenario = Scenario {
596            name: "bad".into(),
597            steps: vec![SimStep::ExpectComplete], // not complete at the start
598        };
599        let err = scenario
600            .run(&booking(), Enforcement::Enforce)
601            .await
602            .unwrap_err();
603        assert!(err.contains("expected conversation complete"));
604    }
605
606    #[tokio::test]
607    async fn tool_latency_resolves_after_delay() {
608        let convo = booking();
609        let mut sim = Sim::new(&convo, Enforcement::Enforce);
610        sim.user("4 people").await;
611        sim.set("user_confirmed", true);
612        sim.turn();
613        // book completes after 2 turns of latency rather than immediately.
614        sim.schedule_tool("book", 2);
615        assert!(!sim.is_complete());
616        sim.turn();
617        assert!(!sim.is_complete());
618        sim.turn();
619        assert!(sim.is_complete());
620    }
621}
gemini_adk_fluent_rs/simulation.rs

gemini_adk_fluent_rs/
simulation.rs
