opendev_context/
validated_list.rs

1//! ValidatedMessageList — write-time enforcement of message pair invariants.
2//!
3//! Wraps a `Vec<ApiMessage>` and enforces structural invariants on every
4//! mutation. All reads work identically to Vec. Mutations are intercepted
5//! and routed through validated methods.
6//!
7//! State machine:
8//!     EXPECT_ANY --add_assistant(tc)--> EXPECT_TOOL_RESULTS{pending_ids}
9//!          ^                                    |
10//!          |                          add_tool_result(id) removes from pending
11//!          |                                    |
12//!          +------ all pending satisfied -------+
13
14use std::collections::HashSet;
15use std::sync::Mutex;
16
17use tracing::warn;
18
19use crate::compaction::ApiMessage;
20
21/// Synthetic error message for incomplete tool results.
22pub const SYNTHETIC_TOOL_RESULT: &str =
23    "Error: Tool execution result was lost. The tool may have been interrupted or crashed.";
24
25/// Drop-in Vec replacement that enforces message pair invariants.
26///
27/// All reads (iteration, indexing, len) work through `messages()`.
28/// Mutations are intercepted via validated methods.
29pub struct ValidatedMessageList {
30    messages: Vec<ApiMessage>,
31    pending_tool_ids: Mutex<HashSet<String>>,
32    strict: bool,
33}
34
35impl ValidatedMessageList {
36    /// Create a new validated message list.
37    ///
38    /// If `initial` is provided, bulk-loads without per-message validation
39    /// (trusts existing data), then rebuilds pending state.
40    pub fn new(initial: Option<Vec<ApiMessage>>, strict: bool) -> Self {
41        let messages = initial.unwrap_or_default();
42        let mut list = Self {
43            messages,
44            pending_tool_ids: Mutex::new(HashSet::new()),
45            strict,
46        };
47        list.rebuild_pending_state();
48        list
49    }
50
51    /// Access the underlying messages (read-only).
52    pub fn messages(&self) -> &[ApiMessage] {
53        &self.messages
54    }
55
56    /// Consume self and return the inner Vec.
57    pub fn into_inner(self) -> Vec<ApiMessage> {
58        self.messages
59    }
60
61    pub fn len(&self) -> usize {
62        self.messages.len()
63    }
64
65    pub fn is_empty(&self) -> bool {
66        self.messages.is_empty()
67    }
68
69    /// Tool call IDs still awaiting results.
70    pub fn pending_tool_ids(&self) -> HashSet<String> {
71        self.pending_tool_ids.lock().unwrap().clone()
72    }
73
74    /// True if in EXPECT_TOOL_RESULTS state.
75    pub fn has_pending_tools(&self) -> bool {
76        !self.pending_tool_ids.lock().unwrap().is_empty()
77    }
78
79    /// Append a user message. Auto-completes pending tool results if any.
80    pub fn add_user(&mut self, content: &str) {
81        self.auto_complete_pending("add_user");
82        let mut msg = ApiMessage::new();
83        msg.insert(
84            "role".to_string(),
85            serde_json::Value::String("user".to_string()),
86        );
87        msg.insert(
88            "content".to_string(),
89            serde_json::Value::String(content.to_string()),
90        );
91        self.messages.push(msg);
92    }
93
94    /// Append assistant message. If tool_calls present, enters EXPECT_TOOL_RESULTS.
95    pub fn add_assistant(
96        &mut self,
97        content: Option<&str>,
98        tool_calls: Option<Vec<serde_json::Value>>,
99    ) {
100        self.auto_complete_pending("add_assistant");
101        let mut msg = ApiMessage::new();
102        msg.insert(
103            "role".to_string(),
104            serde_json::Value::String("assistant".to_string()),
105        );
106        msg.insert(
107            "content".to_string(),
108            serde_json::Value::String(content.unwrap_or("").to_string()),
109        );
110        if let Some(tcs) = tool_calls {
111            let mut pending = self.pending_tool_ids.lock().unwrap();
112            for tc in &tcs {
113                if let Some(id) = tc.get("id").and_then(|v| v.as_str())
114                    && !id.is_empty()
115                {
116                    pending.insert(id.to_string());
117                }
118            }
119            msg.insert("tool_calls".to_string(), serde_json::Value::Array(tcs));
120        }
121        self.messages.push(msg);
122    }
123
124    /// Append tool result. Rejects orphaned IDs not in pending set (in strict mode).
125    pub fn add_tool_result(&mut self, tool_call_id: &str, content: &str) -> Result<(), String> {
126        let mut pending = self.pending_tool_ids.lock().unwrap();
127        if !pending.contains(tool_call_id) {
128            let detail = format!("Orphaned tool result for id={tool_call_id}");
129            if self.strict {
130                return Err(detail);
131            }
132            warn!(
133                "ValidatedMessageList: {} (permissive mode, accepting)",
134                detail
135            );
136        } else {
137            pending.remove(tool_call_id);
138        }
139        drop(pending);
140
141        let mut msg = ApiMessage::new();
142        msg.insert(
143            "role".to_string(),
144            serde_json::Value::String("tool".to_string()),
145        );
146        msg.insert(
147            "tool_call_id".to_string(),
148            serde_json::Value::String(tool_call_id.to_string()),
149        );
150        msg.insert(
151            "content".to_string(),
152            serde_json::Value::String(content.to_string()),
153        );
154        self.messages.push(msg);
155        Ok(())
156    }
157
158    /// Batch-add tool results. Fills missing with synthetic errors.
159    pub fn add_tool_results_batch(
160        &mut self,
161        tool_calls: &[serde_json::Value],
162        results_by_id: &std::collections::HashMap<String, String>,
163    ) {
164        let mut pending = self.pending_tool_ids.lock().unwrap();
165        for tc in tool_calls {
166            let tc_id = tc.get("id").and_then(|v| v.as_str()).unwrap_or("");
167            if tc_id.is_empty() {
168                continue;
169            }
170            let content = if let Some(result) = results_by_id.get(tc_id) {
171                result.clone()
172            } else {
173                let tool_name = tc
174                    .get("function")
175                    .and_then(|f| f.get("name"))
176                    .and_then(|n| n.as_str())
177                    .unwrap_or("unknown");
178                warn!(
179                    "ValidatedMessageList: Missing result for {} (id={}), inserting synthetic error",
180                    tool_name, tc_id
181                );
182                SYNTHETIC_TOOL_RESULT.to_string()
183            };
184            pending.remove(tc_id);
185
186            let mut msg = ApiMessage::new();
187            msg.insert(
188                "role".to_string(),
189                serde_json::Value::String("tool".to_string()),
190            );
191            msg.insert(
192                "tool_call_id".to_string(),
193                serde_json::Value::String(tc_id.to_string()),
194            );
195            msg.insert("content".to_string(), serde_json::Value::String(content));
196            drop(pending);
197            self.messages.push(msg);
198            pending = self.pending_tool_ids.lock().unwrap();
199        }
200    }
201
202    /// Replace all messages (e.g., after compaction). Rebuilds pending state.
203    pub fn replace_all(&mut self, messages: Vec<ApiMessage>) {
204        self.messages = messages;
205        self.rebuild_pending_state();
206    }
207
208    /// Scan all messages to reconstruct pending tool_call IDs.
209    fn rebuild_pending_state(&mut self) {
210        let mut expected: HashSet<String> = HashSet::new();
211        for msg in &self.messages {
212            let role = msg.get("role").and_then(|v| v.as_str()).unwrap_or("");
213            if role == "assistant" {
214                if let Some(tcs) = msg.get("tool_calls").and_then(|v| v.as_array()) {
215                    for tc in tcs {
216                        if let Some(id) = tc.get("id").and_then(|v| v.as_str())
217                            && !id.is_empty()
218                        {
219                            expected.insert(id.to_string());
220                        }
221                    }
222                }
223            } else if role == "tool"
224                && let Some(id) = msg.get("tool_call_id").and_then(|v| v.as_str())
225            {
226                expected.remove(id);
227            }
228        }
229        *self.pending_tool_ids.lock().unwrap() = expected;
230    }
231
232    /// Insert synthetic error results for any pending tool calls.
233    fn auto_complete_pending(&mut self, source: &str) {
234        let mut pending = self.pending_tool_ids.lock().unwrap();
235        if pending.is_empty() {
236            return;
237        }
238        warn!(
239            "ValidatedMessageList: Auto-completing {} pending tool results before {}: {:?}",
240            pending.len(),
241            source,
242            pending,
243        );
244        let ids: Vec<String> = pending.drain().collect();
245        drop(pending);
246
247        for tc_id in ids {
248            let mut msg = ApiMessage::new();
249            msg.insert(
250                "role".to_string(),
251                serde_json::Value::String("tool".to_string()),
252            );
253            msg.insert("tool_call_id".to_string(), serde_json::Value::String(tc_id));
254            msg.insert(
255                "content".to_string(),
256                serde_json::Value::String(SYNTHETIC_TOOL_RESULT.to_string()),
257            );
258            self.messages.push(msg);
259        }
260    }
261}
262
263#[cfg(test)]
264#[path = "validated_list_tests.rs"]
265mod tests;
opendev_context/validated_list.rs

opendev_context/
validated_list.rs
