awaken_runtime_contract/contract/
message.rs

1//! Core types for agent messages and tool calls.
2
3use serde::{Deserialize, Serialize};
4use serde_json::Value;
5
6use super::content::ContentBlock;
7
8mod delivery;
9
10pub use delivery::{
11    DeliveryBoundary, DeliveryGranularity, DeliveryMode, PendingMessageRecord,
12    pending_queue_revision, select_pending_for_freeze, select_pending_for_freeze_for_run,
13};
14
15/// Message role.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(rename_all = "lowercase")]
18pub enum Role {
19    System,
20    User,
21    Assistant,
22    Tool,
23}
24
25/// Message visibility — controls whether a message is exposed to external API consumers.
26#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
27#[serde(rename_all = "lowercase")]
28pub enum Visibility {
29    /// Visible to both the user and the LLM.
30    #[default]
31    All,
32    /// Only visible to the LLM, hidden from external API consumers.
33    Internal,
34}
35
36impl Visibility {
37    /// Returns `true` if this is the default visibility (`All`).
38    pub fn is_default(&self) -> bool {
39        *self == Visibility::All
40    }
41}
42
43/// Optional metadata associating a message with a run and step.
44#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
45pub struct MessageMetadata {
46    /// The run that produced this message.
47    #[serde(skip_serializing_if = "Option::is_none")]
48    pub run_id: Option<String>,
49    /// Step (round) index within the run (0-based).
50    #[serde(skip_serializing_if = "Option::is_none")]
51    pub step_index: Option<u32>,
52    /// Originating agent for a cross-agent durable message (`send_message`'s
53    /// `parent`/`agent` routes). Lets the recipient attribute, audit, or apply
54    /// sender-based logic instead of seeing an anonymous user message. `None`
55    /// for ordinary end-user/system messages.
56    #[serde(default, skip_serializing_if = "Option::is_none")]
57    pub sender_agent_id: Option<String>,
58    /// When set, this message is a compaction summary that logically replaces
59    /// the committed interval `[from_seq, to_seq]` (ADR-0038 D11/C7). The mark
60    /// rides in the message body so it persists through every store without a
61    /// schema change; [`MessageRecord::from_message`] projects it onto
62    /// [`MessageRecord::compaction`] for the [`effective_messages`] fold.
63    #[serde(default, skip_serializing_if = "Option::is_none")]
64    pub compaction: Option<CompactionMark>,
65}
66
67/// A message persisted as part of a thread's append-only log.
68///
69/// `Message` remains the protocol payload. `MessageRecord` is the durable
70/// thread-owned view that assigns a sequence number and exposes producer
71/// relationships without making runs own message bodies.
72#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct MessageRecord {
74    /// Stable message identifier.
75    pub message_id: String,
76    /// Thread that owns the message.
77    pub thread_id: String,
78    /// 1-based sequence number within the thread log.
79    pub seq: u64,
80    /// Message payload.
81    pub message: Message,
82    /// Run that produced this message, if any.
83    #[serde(default, skip_serializing_if = "Option::is_none")]
84    pub produced_by_run_id: Option<String>,
85    /// Step that produced this message, if known.
86    #[serde(default, skip_serializing_if = "Option::is_none")]
87    pub step_index: Option<u32>,
88    /// Tool call this message responds to, if this is a tool result.
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub tool_call_id: Option<String>,
91    /// Unix timestamp (seconds) when the message was recorded.
92    #[serde(default, skip_serializing_if = "Option::is_none")]
93    pub created_at: Option<u64>,
94    /// When set, this record is a compaction summary that logically replaces the
95    /// committed message interval `[from_seq, to_seq]` in the effective view.
96    /// Originals are retained (append-only); see [`effective_messages`].
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub compaction: Option<CompactionMark>,
99}
100
101/// Interval a compaction summary covers, on the summary record.
102///
103/// Compaction never deletes or overwrites: it appends a summary message whose
104/// `CompactionMark` says which committed interval it stands in for. The raw
105/// messages in `[from_seq, to_seq]` stay in the log; the effective (resume /
106/// context) view folds them away via [`effective_messages`].
107#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
108pub struct CompactionMark {
109    /// First committed seq covered by this summary (inclusive).
110    pub from_seq: u64,
111    /// Last committed seq covered by this summary (inclusive).
112    pub to_seq: u64,
113}
114
115impl MessageRecord {
116    /// Build a record from a thread-owned message payload.
117    pub fn from_message(thread_id: impl Into<String>, seq: u64, mut message: Message) -> Self {
118        let message_id = message.id.clone().unwrap_or_else(gen_message_id);
119        if message.id.is_none() {
120            message.id = Some(message_id.clone());
121        }
122        let produced_by_run_id = message
123            .metadata
124            .as_ref()
125            .and_then(|metadata| metadata.run_id.clone());
126        let step_index = message
127            .metadata
128            .as_ref()
129            .and_then(|metadata| metadata.step_index);
130        let tool_call_id = message.tool_call_id.clone();
131        let compaction = message
132            .metadata
133            .as_ref()
134            .and_then(|metadata| metadata.compaction);
135        Self {
136            message_id,
137            thread_id: thread_id.into(),
138            seq,
139            message,
140            produced_by_run_id,
141            step_index,
142            tool_call_id,
143            created_at: None,
144            compaction,
145        }
146    }
147}
148
149/// Fold committed message records into the effective view, applying compaction.
150///
151/// Compaction is append-only and interval-based: a summary record carries a
152/// [`CompactionMark`] for the interval `[from_seq, to_seq]` it replaces. This
153/// produces the view the runtime resumes/builds context from:
154///
155/// - each compacted interval is replaced by its summary message, positioned at
156///   the interval's start (the summary's own append position is skipped);
157/// - records outside every interval pass through in seq order;
158/// - records inside an interval (other than the summary itself) are dropped.
159///
160/// Records must be sorted by `seq` ascending (the committed log order).
161///
162/// Compaction summaries are cumulative (each incremental summary subsumes the
163/// previous one), so two summaries can share the same prefix — e.g. `[1, a]`
164/// and `[1, b]` with `b > a`. The fold resolves this **cumulative-prefix-wins**
165/// (ADR-0038 D11, scheme A): an interval fully contained in another is dropped,
166/// so the largest covering summary wins and the superseded summary is folded
167/// away with the raw records it covered. Genuinely disjoint intervals are all
168/// kept and ordered by start.
169#[must_use]
170pub fn effective_messages(records: &[MessageRecord]) -> Vec<Message> {
171    // Every summary record is folded out of the raw stream, whether its
172    // interval wins or is superseded by a larger covering interval.
173    let summary_seqs: std::collections::HashSet<u64> = records
174        .iter()
175        .filter(|r| r.compaction.is_some())
176        .map(|r| r.seq)
177        .collect();
178
179    // Candidate intervals, then drop any contained in another (cumulative
180    // prefix): sort by start asc, end desc so the widest at each start is seen
181    // first, and keep an interval only when no already-kept one covers it.
182    let mut candidates: Vec<(u64, u64, &Message)> = records
183        .iter()
184        .filter_map(|r| r.compaction.map(|c| (c.from_seq, c.to_seq, &r.message)))
185        .collect();
186    candidates.sort_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)));
187    let mut intervals: Vec<(u64, u64, &Message)> = Vec::new();
188    for cand in candidates {
189        let dominated = intervals
190            .iter()
191            .any(|kept| kept.0 <= cand.0 && cand.1 <= kept.1);
192        if !dominated {
193            intervals.push(cand);
194        }
195    }
196    intervals.sort_by_key(|(from, _, _)| *from);
197
198    let covered = |seq: u64| {
199        intervals
200            .iter()
201            .any(|(from, to, _)| seq >= *from && seq <= *to)
202    };
203
204    let mut out = Vec::new();
205    let mut next_interval = intervals.iter().peekable();
206    for record in records.iter().filter(|r| !summary_seqs.contains(&r.seq)) {
207        // Emit any interval summaries whose interval starts at/before this seq.
208        while next_interval
209            .peek()
210            .is_some_and(|(from, _, _)| *from <= record.seq)
211        {
212            let (_, _, summary) = next_interval.next().unwrap();
213            out.push((*summary).clone());
214        }
215        if covered(record.seq) {
216            continue; // raw record folded away by its interval's summary
217        }
218        out.push(record.message.clone());
219    }
220    // Trailing intervals whose start is beyond the last raw record.
221    for (_, _, summary) in next_interval {
222        out.push((*summary).clone());
223    }
224    out
225}
226
227/// Build the read-time message view used for committed thread history.
228///
229/// A cancelled or superseded run can leave an assistant message with tool calls
230/// that never received a `tool` result. Committed history is append-oriented, so
231/// stores expose those removals as a view filter rather than rewriting message
232/// bodies in place.
233pub fn strip_unpaired_tool_calls_from_view(messages: &mut Vec<Message>) {
234    use std::collections::HashSet;
235
236    // A tool call is answered only by a real (`All`-visibility) tool result.
237    // An `Internal` tool result is a retraction marker (ADR-0042 D6): the
238    // runtime appends one when a suspended call is superseded/abandoned, so the
239    // call and its pending tool result are removed from the view here at read time
240    // rather than by rewriting committed history. No tool result is committed
241    // `Internal` for any other reason.
242    let mut answered: HashSet<String> = HashSet::new();
243    let mut retracted: HashSet<String> = HashSet::new();
244    for message in messages.iter() {
245        if message.role != Role::Tool {
246            continue;
247        }
248        if let Some(call_id) = message.tool_call_id.clone() {
249            match message.visibility {
250                Visibility::All => {
251                    answered.insert(call_id);
252                }
253                Visibility::Internal => {
254                    retracted.insert(call_id);
255                }
256            }
257        }
258    }
259
260    for message in messages.iter_mut() {
261        if message.role != Role::Assistant {
262            continue;
263        }
264        if let Some(ref mut calls) = message.tool_calls {
265            calls.retain(|call| answered.contains(&call.id) && !retracted.contains(&call.id));
266            if calls.is_empty() {
267                message.tool_calls = None;
268            }
269        }
270    }
271
272    // Keep only real tool results for calls that survived: drop `Internal`
273    // retraction markers, retracted pending results, and orphan tool results.
274    messages.retain(|message| {
275        if message.role != Role::Tool {
276            return true;
277        }
278        match message.tool_call_id.as_ref() {
279            Some(call_id) => message.visibility == Visibility::All && !retracted.contains(call_id),
280            None => true,
281        }
282    });
283}
284
285/// Return `messages` after applying the committed-history read view filter.
286pub fn strip_unpaired_tool_calls_from_owned_view(mut messages: Vec<Message>) -> Vec<Message> {
287    strip_unpaired_tool_calls_from_view(&mut messages);
288    messages
289}
290
291/// Build the effective resume/context view from a raw committed message log
292/// (ADR-0038 D11/C7): project each message to a record (reading its compaction
293/// mark from metadata), fold compaction intervals via [`effective_messages`],
294/// then apply the unpaired-tool-call view filter. Messages without a mark pass
295/// through unchanged, so a thread that has never been compacted yields the same
296/// view as the plain strip. The committed `message_version` (the optimistic
297/// guard) is the raw length and is computed by the caller, not from this view.
298#[must_use]
299pub fn effective_committed_view(committed: Vec<Message>, thread_id: &str) -> Vec<Message> {
300    let records: Vec<MessageRecord> = committed
301        .into_iter()
302        .enumerate()
303        .map(|(index, message)| MessageRecord::from_message(thread_id, index as u64 + 1, message))
304        .collect();
305    strip_unpaired_tool_calls_from_owned_view(effective_messages(&records))
306}
307
308/// Generate a time-ordered UUID v7 message identifier.
309pub fn gen_message_id() -> String {
310    uuid::Uuid::now_v7().to_string()
311}
312
313/// A message in the conversation.
314#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
315pub struct Message {
316    /// Stable message identifier (UUID v7, auto-generated).
317    #[serde(skip_serializing_if = "Option::is_none")]
318    pub id: Option<String>,
319    pub role: Role,
320    /// Multimodal content blocks.
321    pub content: Vec<ContentBlock>,
322    /// Tool calls made by the assistant.
323    #[serde(skip_serializing_if = "Option::is_none")]
324    pub tool_calls: Option<Vec<ToolCall>>,
325    /// Tool call ID this message responds to (for tool role).
326    #[serde(skip_serializing_if = "Option::is_none")]
327    pub tool_call_id: Option<String>,
328    /// Message visibility. Defaults to `All` (visible everywhere).
329    /// Internal messages (e.g. system reminders) are only sent to the LLM.
330    #[serde(default, skip_serializing_if = "Visibility::is_default")]
331    pub visibility: Visibility,
332    /// Optional run/step association metadata.
333    #[serde(default, skip_serializing_if = "Option::is_none")]
334    pub metadata: Option<MessageMetadata>,
335}
336
337impl Message {
338    /// Create a system message.
339    ///
340    /// # Examples
341    ///
342    /// ```
343    /// use awaken_runtime_contract::contract::message::Message;
344    ///
345    /// let msg = Message::system("You are helpful");
346    /// assert_eq!(msg.text(), "You are helpful");
347    /// ```
348    pub fn system(text: impl Into<String>) -> Self {
349        Self {
350            id: Some(gen_message_id()),
351            role: Role::System,
352            content: vec![ContentBlock::text(text)],
353            tool_calls: None,
354            tool_call_id: None,
355            visibility: Visibility::All,
356            metadata: None,
357        }
358    }
359
360    /// Create an internal system message (visible only to LLM, hidden from API consumers).
361    pub fn internal_system(text: impl Into<String>) -> Self {
362        Self {
363            id: Some(gen_message_id()),
364            role: Role::System,
365            content: vec![ContentBlock::text(text)],
366            tool_calls: None,
367            tool_call_id: None,
368            visibility: Visibility::Internal,
369            metadata: None,
370        }
371    }
372
373    /// Create an internal user message (visible only to the LLM).
374    pub fn internal_user(text: impl Into<String>) -> Self {
375        Self {
376            id: Some(gen_message_id()),
377            role: Role::User,
378            content: vec![ContentBlock::text(text)],
379            tool_calls: None,
380            tool_call_id: None,
381            visibility: Visibility::Internal,
382            metadata: None,
383        }
384    }
385
386    /// Create a user message with text.
387    ///
388    /// # Examples
389    ///
390    /// ```
391    /// use awaken_runtime_contract::contract::message::Message;
392    ///
393    /// let msg = Message::user("Hello");
394    /// assert_eq!(msg.text(), "Hello");
395    /// ```
396    pub fn user(text: impl Into<String>) -> Self {
397        Self {
398            id: Some(gen_message_id()),
399            role: Role::User,
400            content: vec![ContentBlock::text(text)],
401            tool_calls: None,
402            tool_call_id: None,
403            visibility: Visibility::All,
404            metadata: None,
405        }
406    }
407
408    /// Create a user message with multimodal content blocks.
409    pub fn user_with_content(content: Vec<ContentBlock>) -> Self {
410        Self {
411            id: Some(gen_message_id()),
412            role: Role::User,
413            content,
414            tool_calls: None,
415            tool_call_id: None,
416            visibility: Visibility::All,
417            metadata: None,
418        }
419    }
420
421    /// Create an assistant message.
422    pub fn assistant(text: impl Into<String>) -> Self {
423        Self {
424            id: Some(gen_message_id()),
425            role: Role::Assistant,
426            content: vec![ContentBlock::text(text)],
427            tool_calls: None,
428            tool_call_id: None,
429            visibility: Visibility::All,
430            metadata: None,
431        }
432    }
433
434    /// Create an assistant message with tool calls.
435    pub fn assistant_with_tool_calls(text: impl Into<String>, calls: Vec<ToolCall>) -> Self {
436        Self {
437            id: Some(gen_message_id()),
438            role: Role::Assistant,
439            content: vec![ContentBlock::text(text)],
440            tool_calls: if calls.is_empty() { None } else { Some(calls) },
441            tool_call_id: None,
442            visibility: Visibility::All,
443            metadata: None,
444        }
445    }
446
447    /// Create a tool response message with text.
448    pub fn tool(call_id: impl Into<String>, text: impl Into<String>) -> Self {
449        Self {
450            id: Some(gen_message_id()),
451            role: Role::Tool,
452            content: vec![ContentBlock::text(text)],
453            tool_calls: None,
454            tool_call_id: Some(call_id.into()),
455            visibility: Visibility::All,
456            metadata: None,
457        }
458    }
459
460    /// Create a tool response message with multimodal content.
461    pub fn tool_with_content(call_id: impl Into<String>, content: Vec<ContentBlock>) -> Self {
462        Self {
463            id: Some(gen_message_id()),
464            role: Role::Tool,
465            content,
466            tool_calls: None,
467            tool_call_id: Some(call_id.into()),
468            visibility: Visibility::All,
469            metadata: None,
470        }
471    }
472
473    /// Extract concatenated text from content blocks.
474    ///
475    /// # Examples
476    ///
477    /// ```
478    /// use awaken_runtime_contract::contract::message::Message;
479    ///
480    /// let msg = Message::user("Hello world");
481    /// assert_eq!(msg.text(), "Hello world");
482    /// ```
483    pub fn text(&self) -> String {
484        super::content::extract_text(&self.content)
485    }
486
487    pub fn is_internal_tool_result(&self) -> bool {
488        self.role == Role::Tool && self.visibility == Visibility::Internal
489    }
490
491    /// Override the auto-generated message ID.
492    #[must_use]
493    pub fn with_id(mut self, id: String) -> Self {
494        self.id = Some(id);
495        self
496    }
497
498    /// Attach run/step metadata to this message.
499    #[must_use]
500    pub fn with_metadata(mut self, metadata: MessageMetadata) -> Self {
501        self.metadata = Some(metadata);
502        self
503    }
504
505    /// Return the run that produced this message, if recorded.
506    #[must_use]
507    pub fn produced_by_run_id(&self) -> Option<&str> {
508        self.metadata
509            .as_ref()
510            .and_then(|metadata| metadata.run_id.as_deref())
511    }
512
513    /// Mark this message as produced by a run without overwriting existing
514    /// producer metadata.
515    pub fn mark_produced_by(&mut self, run_id: &str, step_index: Option<u32>) {
516        let metadata = self.metadata.get_or_insert_with(MessageMetadata::default);
517        if metadata.run_id.is_none() {
518            metadata.run_id = Some(run_id.to_string());
519        }
520        if metadata.step_index.is_none() {
521            metadata.step_index = step_index;
522        }
523    }
524}
525
526/// A tool call requested by the LLM.
527#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
528pub struct ToolCall {
529    /// Unique identifier for this tool call.
530    pub id: String,
531    /// Name of the tool to call.
532    pub name: String,
533    /// Arguments for the tool as JSON.
534    pub arguments: Value,
535}
536
537impl ToolCall {
538    /// Create a new tool call.
539    pub fn new(id: impl Into<String>, name: impl Into<String>, arguments: Value) -> Self {
540        Self {
541            id: id.into(),
542            name: name.into(),
543            arguments,
544        }
545    }
546}
547
548#[cfg(test)]
549#[path = "message/tests.rs"]
550mod tests;
awaken_runtime_contract/contract/message.rs

awaken_runtime_contract/contract/
message.rs
