fix(llm/flow/replay): review wf_6783a4b3 — CRITICAL history poisoning + 9 majors

CRITICAL: max_rounds bail in BOTH tool loops (B-G4 run_tool_loop, B-G3
run_flow_loop) left the already-recorded assistant{tool_calls} unpaired —
every later request 400s on strict providers, permanently bricking the
session (the backstop fired exactly when models loop). Both loops now write
synthetic terminal results ("aborted: max rounds") for the final batch
before stopping; pinned with a history-pairing audit in the capped-loop test.

- ConversationHistory trim is pair-aware: evictions never leave a leading
  orphan tool result (the same 400-forever shape) — pinned.
- Tool-result batches append under ONE history-lock acquisition
  (add_tool_results_batch): a concurrent turn's user message can no longer
  interleave inside a batch (persisted pairing corruption).
- B-G5 double-commit closed: once the initial inference completes (recorded
  with its preamble), the streamed accumulator is cleared before the tool
  loop — a cancel inside the loop no longer commits the preamble twice.
- spoke-flag fix: a turn that streamed a round-0 preamble then ran the tool
  loop now SPEAKS the loop's final answer (was: bot permanently silent after
  "let me check..."); pinned end-to-end through the orchestrator with a
  preamble+tool_calls SSE mock.
- Flows: stale pending_transition from an aborted turn is discarded at the
  next turn's start (state machine no longer jumps nodes uncommanded);
  initialize() routes entry-inference tool calls through the flow loop
  (lookup-then-greet nodes work; no dangling tool_calls from turn one);
  flow MAX_ROUNDS bail pairs terminal results (same critical shape);
  persona is tracked EXPLICITLY (seeded from the client's configured
  system_prompt, updated by role_message) instead of inferred from
  history[0] — embedders wiring flows from the standard config keep their
  system instruction.
- D-G1: clear() now keeps the most recent ~250ms onset window (a final
  never covers audio still in flight — clear-all lost the next utterance's
  first syllables when the socket died before the next final); clear_all()
  added for teardown; chaos choreography re-pinned (pre-window audio never
  replays, onset survivor may, tail byte-identical in order).
- Cheap hardening from the unverified tail: orchestrator tool-loop guard
  requires a NON-EMPTY registry; continue_from_history injects the system
  prompt on an empty session instead of sending an empty messages array;
  LiveKit forwarder warn properly rate-limited (1 per 100 sheds) and its
  doc now matches the newest-drop behavior it implements.

Floor: 6094/0 lib (all features) + chaos 6/6 + conversation_loop 10/10 +
flow_manager 7/7 + clippy clean.
Live: Sarvam tool loop, 3-node intake flow, Deepgram nova-3 STT roundtrip
all green post-fix.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: jithinAB

gateway/src/core/conversation/mod.rs

@@ -451,10 +451,23 @@ impl ConversationOrchestrator {
         // content lands on the normal speak path below (the streaming pump
         // saw no text for a tool-call response, so the !spoke fallback
         // speaks it).
+        let mut ran_tool_loop = false;
         let result = match result {
             Ok(response)
-                if !response.tool_calls.is_empty() && self.llm.functions().is_some() =>
+                if !response.tool_calls.is_empty()
+                    && self.llm.functions().is_some_and(|r| !r.is_empty()) =>
             {
+                // The initial inference COMPLETED (recorded with its
+                // preamble): the streamed accumulator is spent — without
+                // this, a cancel inside the tool loop would commit the
+                // preamble a SECOND time after the tool results (review
+                // wf_6783a4b3: duplicate assistant text).
+                streamed_text.lock().clear();
+                // The pump spoke (at most) round-0 preamble; the loop's
+                // FINAL answer must still reach TTS through the fallback
+                // speak below (review wf_6783a4b3: a true `spoke` here left
+                // the bot permanently silent after "let me check...").
+                ran_tool_loop = true;
                 let registry =
                     Arc::clone(self.llm.functions().expect("guarded by condition"));
                 crate::core::llm::run_tool_loop(
@@ -485,7 +498,9 @@ impl ConversationOrchestrator {
                     // deltas and deliver the answer (or nothing) at the end. If
                     // the pump spoke nothing, fall back to the final content; if
                     // that is empty too, say so loudly instead of going silent.
-                    !spoke.load(std::sync::atomic::Ordering::Relaxed)
+                    // A completed TOOL LOOP always speaks its final answer —
+                    // the pump only ever saw the round-0 preamble.
+                    ran_tool_loop || !spoke.load(std::sync::atomic::Ordering::Relaxed)
                 } else {
                     true
                 };

gateway/src/core/flow/mod.rs

@@ -147,6 +147,11 @@ pub struct FlowManager {
     /// Available at every node, alongside the node's own functions.
     global_functions: Vec<FlowFunctionSchema>,
     action_sink: Option<ActionSink>,
+    /// The session PERSONA — tracked explicitly (review wf_6783a4b3: the
+    /// "system is history[0]" inference broke when embedders seeded flows
+    /// from the standard config). Seeded from the LlmClient's configured
+    /// system prompt; updated by every node `role_message`.
+    persona: Mutex<Option<String>>,
 }
 
 impl std::fmt::Debug for FlowManager {
@@ -165,6 +170,7 @@ impl FlowManager {
         session_id: impl Into<String>,
         api_key: Option<String>,
     ) -> Self {
+        let persona = llm.config().system_prompt.clone();
         Self {
             llm,
             registry,
@@ -174,6 +180,7 @@ impl FlowManager {
             pending_transition: Arc::new(Mutex::new(None)),
             global_functions: Vec::new(),
             action_sink: None,
+            persona: Mutex::new(persona),
         }
     }
 
@@ -200,7 +207,14 @@ impl FlowManager {
         first: NodeConfig,
         cancel: &CancellationToken,
     ) -> LlmResult<Option<LlmResponse>> {
-        self.set_node(first, cancel).await
+        match self.set_node(first, cancel).await? {
+            // The entry inference may itself call tools (lookup-then-greet
+            // nodes are core pipecat-flows usage): run the flow loop, or
+            // the calls are silently dropped AND the dangling
+            // assistant{tool_calls} poisons history (review wf_6783a4b3).
+            Some(response) => self.run_flow_loop(response, cancel).await,
+            None => Ok(None),
+        }
     }
 
     /// Drive one USER turn through the flow: inference on the current
@@ -214,6 +228,15 @@ impl FlowManager {
         transcript: &str,
         cancel: &CancellationToken,
     ) -> LlmResult<Option<LlmResponse>> {
+        // A transition staged by an aborted/dropped earlier turn must never
+        // fire on THIS unrelated turn (review wf_6783a4b3: pending leak).
+        if let Some(stale) = self.pending_transition.lock().take() {
+            warn!(
+                session = %self.session_id,
+                node = ?stale.name,
+                "discarding stale pending transition from an aborted turn"
+            );
+        }
         let response = self
             .llm
             .complete(&self.session_id, transcript, self.api_key.as_deref(), cancel, None)
@@ -235,7 +258,24 @@ impl FlowManager {
                 return Ok(Some(response));
             }
             if rounds >= MAX_ROUNDS {
-                warn!(session = %self.session_id, "flow loop hit max rounds");
+                // Same CRITICAL shape as B-G4's loop: bailing with the
+                // assistant{tool_calls} already recorded but unanswered
+                // bricks the session on strict providers. Pair them with
+                // terminal results first (review wf_6783a4b3).
+                warn!(session = %self.session_id, "flow loop hit max rounds; writing terminal results");
+                let results: Vec<(String, String)> = response
+                    .tool_calls
+                    .iter()
+                    .map(|c| {
+                        (
+                            c.id.clone(),
+                            serde_json::json!({"error": "aborted: flow loop reached max rounds"})
+                                .to_string(),
+                        )
+                    })
+                    .collect();
+                self.llm.add_tool_results_batch(&self.session_id, results).await;
+                response.tool_calls.clear();
                 return Ok(Some(response));
             }
             rounds += 1;
@@ -246,11 +286,12 @@ impl FlowManager {
             // are captured into pending_transition by the wrappers.
             let results =
                 execute_batch(&self.registry, &self.session_id, &batch, cancel, true).await;
-            for (call, value) in batch.iter().zip(results) {
-                self.llm
-                    .add_tool_result(&self.session_id, &call.id, &value.to_string())
-                    .await;
-            }
+            let rendered: Vec<(String, String)> = batch
+                .iter()
+                .zip(results)
+                .map(|(call, value)| (call.id.clone(), value.to_string()))
+                .collect();
+            self.llm.add_tool_results_batch(&self.session_id, rendered).await;
 
             // TWO-PHASE: results are in context; NOW the transition fires.
             let next = self.pending_transition.lock().take();
@@ -314,9 +355,14 @@ impl FlowManager {
         }
         self.registry.swap_tools(items);
 
-        // Persona persists until another node changes it.
+        // Persona persists until another node changes it (tracked
+        // explicitly — never inferred from history position).
         if let Some(role) = &node.role_message {
-            self.llm.upsert_system(&self.session_id, role).await;
+            *self.persona.lock() = Some(role.clone());
+        }
+        let persona = self.persona.lock().clone();
+        if let Some(p) = &persona {
+            self.llm.upsert_system(&self.session_id, p).await;
         }
 
         // Context strategy.
@@ -334,20 +380,9 @@ impl FlowManager {
                         "ResetWithSummary: native summarizer (B-G6) not built; behaving as Reset"
                     );
                 }
-                // Rebuild: persona system message (the node's, else the one
-                // already in context) + the node's task messages.
-                let system = match &node.role_message {
-                    Some(r) => Some(r.clone()),
-                    None => self
-                        .llm
-                        .history_snapshot(&self.session_id)
-                        .await
-                        .first()
-                        .filter(|m| m.role == crate::core::llm::MessageRole::System)
-                        .and_then(|m| m.content.clone()),
-                };
+                // Rebuild: the tracked persona + the node's task messages.
                 let mut messages = Vec::new();
-                if let Some(system) = system {
+                if let Some(system) = &persona {
                     messages.push(ChatMessage::system(system));
                 }
                 messages.extend(node.task_messages.clone());

gateway/src/core/llm/functions.rs

@@ -281,10 +281,27 @@ pub async fn run_tool_loop(
     let mut rounds = 0usize;
     while !response.tool_calls.is_empty() {
         if rounds >= opts.max_rounds {
+            // CRITICAL (review wf_6783a4b3): the assistant{tool_calls}
+            // message is ALREADY in history — bailing without results leaves
+            // it unpaired and every later request 400s on strict providers
+            // (session permanently bricked). Write synthetic terminal
+            // results so pairing holds, then stop.
             warn!(
                 session = session_id,
-                rounds, "tool loop hit max_rounds; returning last response"
+                rounds, "tool loop hit max_rounds; writing terminal results and stopping"
             );
+            let results: Vec<(String, String)> = response
+                .tool_calls
+                .iter()
+                .map(|c| {
+                    (
+                        c.id.clone(),
+                        json!({"error": "aborted: tool loop reached max_rounds"}).to_string(),
+                    )
+                })
+                .collect();
+            llm.add_tool_results_batch(session_id, results).await;
+            response.tool_calls.clear();
             break;
         }
         rounds += 1;
@@ -299,11 +316,16 @@ pub async fn run_tool_loop(
 
         let results = execute_batch(registry, session_id, &batch, cancel, opts.parallel).await;
 
-        // Pairing invariant: exactly one result per tool_call_id, batch order.
-        for (call, value) in batch.iter().zip(results) {
-            let rendered = value.to_string();
-            llm.add_tool_result(session_id, &call.id, &rendered).await;
-        }
+        // Pairing invariant: exactly one result per tool_call_id, batch
+        // order — appended under ONE history-lock acquisition so a
+        // concurrent turn's user message can never interleave between a
+        // batch's results (review wf_6783a4b3).
+        let rendered: Vec<(String, String)> = batch
+            .iter()
+            .zip(results)
+            .map(|(call, value)| (call.id.clone(), value.to_string()))
+            .collect();
+        llm.add_tool_results_batch(session_id, rendered).await;
 
         // ONE re-inference for the whole batch (group_id semantics).
         response = llm
@@ -838,9 +860,81 @@ mod tests {
         let opts = ToolLoopOptions { parallel: true, max_rounds: 3 };
         let resp = run_tool_loop(&llm, &reg, "s1", first, None, &token, opts).await.unwrap();
         assert!(
-            !resp.tool_calls.is_empty(),
-            "loop returns the last tool-call response when capped"
+            resp.tool_calls.is_empty(),
+            "capped loop must not hand back live tool calls (they were terminally answered)"
         );
         assert_eq!(mock.requests.lock().len(), 1 + 3, "initial + max_rounds inferences");
+
+        // CRITICAL pairing pin (review wf_6783a4b3): the bail must leave NO
+        // unpaired assistant{tool_calls} — every call id in history has a
+        // tool result, so the session is NOT bricked on strict providers.
+        let history = llm.history_snapshot("s1").await;
+        let result_ids: std::collections::HashSet<&str> = history
+            .iter()
+            .filter(|m| matches!(m.role, MessageRole::Tool))
+            .filter_map(|m| m.tool_call_id.as_deref())
+            .collect();
+        for m in &history {
+            if let Some(calls) = &m.tool_calls {
+                for c in calls {
+                    assert!(
+                        result_ids.contains(c.id.as_str()),
+                        "unpaired tool call {} after max_rounds bail: {history:?}",
+                        c.id
+                    );
+                }
+            }
+        }
+        let last_results: Vec<_> = history
+            .iter()
+            .rev()
+            .take_while(|m| matches!(m.role, MessageRole::Tool))
+            .collect();
+        assert_eq!(last_results.len(), 2, "the final batch got terminal results");
+        assert!(
+            last_results[0]
+                .content
+                .as_deref()
+                .unwrap()
+                .contains("max_rounds"),
+            "terminal results say why"
+        );
+    }
+
+    #[test]
+    fn history_trim_never_orphans_tool_messages() {
+        // review wf_6783a4b3: count-based eviction crossing a tool exchange
+        // must take the whole exchange, never leaving a leading orphan tool
+        // result (strict providers 400 on it forever).
+        use crate::core::llm::{ChatMessage, ConversationHistory, MessageRole};
+        let mut h = ConversationHistory::new(4);
+        h.add(ChatMessage::system("p"));
+        h.add(ChatMessage::user("q1"));
+        let mut call = ChatMessage::assistant("");
+        call.tool_calls = Some(vec![crate::core::llm::ToolCall {
+            id: "c1".into(),
+            call_type: "function".into(),
+            function: crate::core::llm::FunctionCall { name: "f".into(), arguments: "{}".into() },
+        }]);
+        h.add(call);
+        h.add(ChatMessage::tool("c1", "result"));
+        // Over capacity: evictions start. q1 goes, then the assistant{call}
+        // — its orphaned result MUST go with it.
+        h.add(ChatMessage::user("q2"));
+        h.add(ChatMessage::assistant("a2"));
+        let msgs = h.messages();
+        assert!(
+            !msgs
+                .iter()
+                .enumerate()
+                .any(|(i, m)| m.role == MessageRole::Tool
+                    && !msgs[..i].iter().any(|p| p
+                        .tool_calls
+                        .as_ref()
+                        .is_some_and(|cs| cs.iter().any(|c| Some(c.id.as_str())
+                            == m.tool_call_id.as_deref())))),
+            "orphan tool message survived the trim: {msgs:?}"
+        );
+        assert_eq!(msgs[0].role, MessageRole::System, "system always survives");
     }
 }

gateway/src/core/llm/mod.rs

@@ -524,6 +524,16 @@ impl ConversationHistory {
                 .position(|m| m.role != MessageRole::System)
             {
                 self.messages.remove(idx);
+                // Pair-aware eviction (review wf_6783a4b3): the oldest
+                // non-system slot must never be left on a tool RESULT whose
+                // call was just (or previously) evicted — strict providers
+                // 400 on orphan tool messages. Any tool message now at the
+                // front is by definition an orphan: drop it with its call.
+                while idx < self.messages.len()
+                    && self.messages[idx].role == MessageRole::Tool
+                {
+                    self.messages.remove(idx);
+                }
             } else {
                 break;
             }
@@ -755,8 +765,16 @@ impl LlmClient {
             .or_insert_with(|| ConversationHistory::new(self.config.max_history));
 
         // CONTINUE mode (B-G4 re-inference after tool results): the request
-        // is the stored history exactly as-is — no new user message.
+        // is the stored history exactly as-is — no new user message. An
+        // EMPTY history would render an empty messages array (provider
+        // 400): inject the configured system prompt so the request is at
+        // least well-formed (review wf_6783a4b3, unverified #25).
         let Some(input) = input else {
+            if history.messages().is_empty()
+                && let Some(system_prompt) = &self.config.system_prompt
+            {
+                history.add(ChatMessage::system(system_prompt));
+            }
             return history.messages().to_vec();
         };
 
@@ -1191,6 +1209,22 @@ impl LlmClient {
         }
     }
 
+    /// Add a BATCH of tool results under one lock acquisition: a concurrent
+    /// turn's user message must never interleave between a batch's results
+    /// (pairing violation → strict providers 400 forever).
+    pub async fn add_tool_results_batch(
+        &self,
+        session_id: &str,
+        results: Vec<(String, String)>,
+    ) {
+        let mut histories = self.histories.write().await;
+        if let Some(history) = histories.get_mut(session_id) {
+            for (id, result) in results {
+                history.add(ChatMessage::tool(id, result));
+            }
+        }
+    }
+
     /// Append messages to a session's context (B-G3 flows: APPEND strategy /
     /// task messages). Creates the session entry if missing.
     pub async fn append_context(&self, session_id: &str, messages: Vec<ChatMessage>) {