mobius/middleware/
messages.rs

1//! Durable conversation-message delivery.
2
3pub mod voice;
4
5use super::manifest::{
6    MiddlewareManifest, MiddlewareSettingChoice, MiddlewareSettingChoices,
7    MiddlewareSettingManifest,
8};
9use super::{
10    ActiveCommandContext, MessageRouteContext, Middleware, MiddlewareCommandContext,
11    MiddlewareCommandOutput, SessionStartContext, SessionStartSource, SubmissionResult,
12};
13use crate::backend::checkpoint::QueuedMessageBoundary;
14use crate::protocol::{
15    ActiveMessageDelivery, EventMsg, FrontendBlock, FrontendBlockFormat, FrontendBlockRole,
16    FrontendBlockState, FrontendBlockUpdate, FrontendCommand, FrontendContribution, FrontendEvent,
17    FrontendSlot, FrontendSymbol, FrontendTone, FrontendWidget, MAX_CAPABILITY_INPUT_BYTES,
18    MessageAuthor, MessageDelivery, MessageEvent, Op,
19};
20use crate::{BoxFuture, Error, Result};
21
22mod text {
23    pub const DEFAULTS_MAX_PENDING: i64 = 64;
24    pub const MANIFEST_DESCRIPTION: &str =
25        "Deliver conversation messages at the correct turn boundary";
26    pub const MANIFEST_LABEL: &str = "Messages";
27    pub const SETTING_DELIVERY_DESCRIPTION: &str =
28        "Default behavior for user messages sent while the agent is working";
29    pub const SETTING_DELIVERY_LABEL: &str = "Active message delivery";
30    pub const SETTING_DELIVERY_QUEUE_DESCRIPTION: &str =
31        "Start a new turn after the active turn finishes";
32    pub const SETTING_DELIVERY_QUEUE_LABEL: &str = "Queue";
33    pub const SETTING_DELIVERY_STEER_DESCRIPTION: &str =
34        "Inject at the next model boundary of the active turn";
35    pub const SETTING_DELIVERY_STEER_LABEL: &str = "Steer";
36    pub const SETTING_MAX_PENDING_DESCRIPTION: &str = "Maximum messages waiting for delivery";
37    pub const SETTING_MAX_PENDING_LABEL: &str = "Maximum pending messages";
38    pub const SETTING_MAX_PENDING_STEP: i64 = 1;
39}
40const MAX_PENDING_MESSAGES: usize = 1_024;
41const _: () = {
42    assert!(text::DEFAULTS_MAX_PENDING >= 1);
43    assert!(text::DEFAULTS_MAX_PENDING <= MAX_PENDING_MESSAGES as i64);
44    assert!(text::SETTING_MAX_PENDING_STEP > 0);
45};
46
47/// Default number of pending messages retained by the delivery queue.
48pub const DEFAULT_MAX_PENDING: usize = text::DEFAULTS_MAX_PENDING as usize;
49/// Default delivery for user messages submitted during an active turn.
50pub const DEFAULT_DELIVERY: ActiveMessageDelivery = ActiveMessageDelivery::Steer;
51
52const DELIVERIES: &[MiddlewareSettingChoice] = &[
53    MiddlewareSettingChoice {
54        disables: &[],
55        value: "steer",
56        label: text::SETTING_DELIVERY_STEER_LABEL,
57        description: text::SETTING_DELIVERY_STEER_DESCRIPTION,
58        symbol: Some("steer"),
59        tone: FrontendTone::Neutral,
60    },
61    MiddlewareSettingChoice {
62        disables: &[],
63        value: "queue",
64        label: text::SETTING_DELIVERY_QUEUE_LABEL,
65        description: text::SETTING_DELIVERY_QUEUE_DESCRIPTION,
66        symbol: Some("queue"),
67        tone: FrontendTone::Neutral,
68    },
69];
70
71const SETTINGS: &[MiddlewareSettingManifest] = &[
72    MiddlewareSettingManifest::Select {
73        id: "delivery",
74        label: text::SETTING_DELIVERY_LABEL,
75        description: text::SETTING_DELIVERY_DESCRIPTION,
76        choices: MiddlewareSettingChoices::Static(DELIVERIES),
77        unset_label: None,
78        default: Some("steer"),
79        max_bytes: 8,
80        composer: false,
81    },
82    MiddlewareSettingManifest::Integer {
83        id: "max_pending",
84        label: text::SETTING_MAX_PENDING_LABEL,
85        description: text::SETTING_MAX_PENDING_DESCRIPTION,
86        min: 1,
87        max: Some(MAX_PENDING_MESSAGES as i64),
88        step: text::SETTING_MAX_PENDING_STEP,
89        default: DEFAULT_MAX_PENDING as i64,
90    },
91];
92
93/// Configuration and presentation metadata for message delivery.
94pub const MANIFEST: MiddlewareManifest = MiddlewareManifest {
95    id: "messages",
96    label: text::MANIFEST_LABEL,
97    description: text::MANIFEST_DESCRIPTION,
98    required: true,
99    default_enabled: true,
100    settings: SETTINGS,
101};
102
103const EDIT_COMMAND: &str = "edit";
104const STALE_EDIT: &str = "message is no longer queued";
105const INVALID_EDIT: &str = "message edit requires non-empty text";
106
107/// Prepares every conversation message and owns its durable delivery lifecycle.
108pub struct Messages {
109    max_pending: usize,
110    delivery: ActiveMessageDelivery,
111}
112
113impl Default for Messages {
114    fn default() -> Self {
115        Self {
116            max_pending: DEFAULT_MAX_PENDING,
117            delivery: DEFAULT_DELIVERY,
118        }
119    }
120}
121
122impl Messages {
123    /// Creates message delivery with a bounded queue and active-turn default.
124    pub fn new(max_pending: usize, delivery: ActiveMessageDelivery) -> Result<Self> {
125        if max_pending == 0 || max_pending > MAX_PENDING_MESSAGES {
126            return Err(Error::Config(format!(
127                "message queue limit must be between 1 and {MAX_PENDING_MESSAGES}"
128            )));
129        }
130        Ok(Self {
131            max_pending,
132            delivery,
133        })
134    }
135
136    fn remove_widget(&self, id: &str) -> FrontendEvent {
137        FrontendEvent::RemoveWidget {
138            capability: self.name().into(),
139            id: id.into(),
140        }
141    }
142
143    fn queued_widget(&self, id: &str, message: &MessageEvent) -> FrontendEvent {
144        let action = matches!(message.author, MessageAuthor::User).then(|| Op::CapabilityCommand {
145            capability: self.name().into(),
146            command: EDIT_COMMAND.into(),
147            arguments: id.into(),
148            input: Some(message.text.clone()),
149            target: None,
150        });
151        FrontendEvent::Widget {
152            capability: self.name().into(),
153            item: FrontendWidget {
154                id: id.into(),
155                slot: FrontendSlot::TranscriptTail,
156                text: message.text.clone(),
157                tone: FrontendTone::Neutral,
158                symbol: Some(match message.delivery {
159                    MessageDelivery::Turn => FrontendSymbol::Chat,
160                    MessageDelivery::Steer => FrontendSymbol::Custom("steer".into()),
161                    MessageDelivery::Queue => FrontendSymbol::Custom("queue".into()),
162                }),
163                icon_only: false,
164                progress: None,
165                content: None,
166                action,
167            },
168        }
169    }
170
171    fn prepare(
172        &self,
173        context: &MessageRouteContext<'_>,
174    ) -> std::result::Result<QueuedMessageBoundary, String> {
175        let Some(turn_id) = context.active_turn_id else {
176            return if context.message.target_turn_id.is_some() {
177                Err("message targeted a stale turn".into())
178            } else {
179                Ok(QueuedMessageBoundary::Turn)
180            };
181        };
182        if matches!(context.message.author, MessageAuthor::Peer { .. }) {
183            return Ok(QueuedMessageBoundary::Steer {
184                turn_id: turn_id.into(),
185            });
186        }
187        if let Some(target) = &context.message.target_turn_id
188            && target != turn_id
189        {
190            return Err("message targeted a stale turn".into());
191        }
192        match context.message.requested_delivery.unwrap_or(self.delivery) {
193            ActiveMessageDelivery::Steer => Ok(QueuedMessageBoundary::Steer {
194                turn_id: turn_id.into(),
195            }),
196            ActiveMessageDelivery::Queue => Ok(QueuedMessageBoundary::Queue),
197        }
198    }
199}
200
201impl Middleware for Messages {
202    fn name(&self) -> &'static str {
203        MANIFEST.id
204    }
205
206    fn frontend(&self) -> FrontendContribution {
207        FrontendContribution {
208            capability: self.name().into(),
209            commands: vec![FrontendCommand {
210                name: voice::transcript::COMMAND.into(),
211                arguments: String::new(),
212                description: "Open the voice transcript".into(),
213                requires_idle: false,
214            }],
215            ..FrontendContribution::default()
216        }
217    }
218
219    fn render(&self, event: &EventMsg, _session_id: &str) -> Option<FrontendBlock> {
220        let EventMsg::Message(message) = event else {
221            return None;
222        };
223        let MessageAuthor::Peer {
224            message_id,
225            session_id,
226            handle,
227            symbol,
228        } = &message.author
229        else {
230            return None;
231        };
232        Some(FrontendBlock {
233            id: Some(format!(
234                "message_received:{}:{session_id}:{message_id}",
235                session_id.len()
236            )),
237            group: None,
238            update: FrontendBlockUpdate::Replace,
239            state: FrontendBlockState::Complete,
240            role: FrontendBlockRole::Activity,
241            title: format!("Message received from @{handle}"),
242            text: message.text.clone(),
243            symbol: Some(symbol.clone().unwrap_or(FrontendSymbol::Chat)),
244            files: Vec::new(),
245            format: FrontendBlockFormat::PlainText,
246            tone: FrontendTone::Neutral,
247        })
248    }
249
250    fn command<'a>(
251        &'a self,
252        context: MiddlewareCommandContext<'a>,
253    ) -> BoxFuture<'a, Result<MiddlewareCommandOutput>> {
254        Box::pin(async move {
255            if context.command != voice::transcript::COMMAND {
256                return Err(Error::Unknown(format!(
257                    "messages command `{}`",
258                    context.command
259                )));
260            }
261            Ok(MiddlewareCommandOutput::events(vec![
262                voice::transcript::read_preview(
263                    context.checkpoints.as_ref(),
264                    context.session_id,
265                    context.arguments,
266                )
267                .await?,
268            ]))
269        })
270    }
271
272    fn handles_messages(&self) -> bool {
273        true
274    }
275
276    fn route_message(&self, context: &mut MessageRouteContext<'_>) -> Result<SubmissionResult> {
277        let boundary = match self.prepare(context) {
278            Ok(boundary) => boundary,
279            Err(message) => return Ok(SubmissionResult::Rejected(message)),
280        };
281        if context.queued_messages.count() >= self.max_pending {
282            return Ok(SubmissionResult::Rejected("message queue is full".into()));
283        }
284        let event = MessageEvent {
285            author: context.message.author.clone(),
286            delivery: boundary.delivery(),
287            text: context.message.text.clone(),
288            attachments: context.message.attachments.clone(),
289            reply: context.message.reply.clone(),
290            message_target: None,
291        };
292        if !context.queued_messages.enqueue(
293            context.submission_id,
294            boundary.clone(),
295            event.clone(),
296        )? {
297            return Ok(SubmissionResult::Rejected(
298                "message could not be queued".into(),
299            ));
300        }
301        context.events.push(EventMsg::Frontend(
302            self.queued_widget(context.submission_id, &event),
303        ));
304        Ok(SubmissionResult::Accepted {
305            input_changed: matches!(boundary, QueuedMessageBoundary::Steer { .. }),
306        })
307    }
308
309    fn message_boundary_events(&self, submission_id: &str) -> Vec<EventMsg> {
310        vec![EventMsg::Frontend(self.remove_widget(submission_id))]
311    }
312
313    fn active_command<'a>(
314        &'a self,
315        context: &'a mut ActiveCommandContext<'_>,
316    ) -> BoxFuture<'a, Result<Option<SubmissionResult>>> {
317        Box::pin(async move {
318            if context.command == voice::transcript::COMMAND {
319                let result = voice::transcript::read_preview(
320                    context.checkpoints,
321                    context.session_id,
322                    context.arguments,
323                )
324                .await;
325                return Ok(Some(match result {
326                    Ok(event) => {
327                        context.events.push(EventMsg::Frontend(event));
328                        SubmissionResult::Handled
329                    }
330                    Err(error) => SubmissionResult::Rejected(error.to_string()),
331                }));
332            }
333            if context.command != EDIT_COMMAND {
334                return Ok(None);
335            }
336            let Some(input) = context.input.filter(|input| !input.trim().is_empty()) else {
337                return Ok(Some(SubmissionResult::Rejected(INVALID_EDIT.into())));
338            };
339            if input.len() > MAX_CAPABILITY_INPUT_BYTES {
340                return Ok(Some(SubmissionResult::Rejected(
341                    "message exceeds editable size limit".into(),
342                )));
343            }
344            let Some(queued) = context.queued_messages.find(context.arguments) else {
345                return Ok(Some(SubmissionResult::Rejected(STALE_EDIT.into())));
346            };
347            let mut event = queued.event();
348            if !matches!(event.author, MessageAuthor::User) {
349                return Ok(Some(SubmissionResult::Rejected(
350                    "peer messages cannot be edited".into(),
351                )));
352            }
353            event.text = input.into();
354            let input_changed = event.delivery == MessageDelivery::Steer;
355            if !context.queued_messages.replace(
356                context.arguments,
357                context.submission_id,
358                event.clone(),
359            )? {
360                return Ok(Some(SubmissionResult::Rejected(STALE_EDIT.into())));
361            }
362            context
363                .events
364                .push(EventMsg::Frontend(self.remove_widget(context.arguments)));
365            context.events.push(EventMsg::Frontend(
366                self.queued_widget(context.submission_id, &event),
367            ));
368            Ok(Some(SubmissionResult::Accepted { input_changed }))
369        })
370    }
371
372    fn session_start<'a>(
373        &'a self,
374        context: &'a mut SessionStartContext<'_>,
375    ) -> BoxFuture<'a, Result<()>> {
376        Box::pin(async move {
377            if context.source() == SessionStartSource::Compact {
378                return Ok(());
379            }
380            for queued in context.queued_messages().views() {
381                (context.runtime.frontend)(self.queued_widget(queued.id(), &queued.event()))?;
382            }
383            if let Some(widget) = voice::transcript::restore_widget(
384                context.runtime.checkpoints.as_ref(),
385                &context.runtime.session_id,
386            )
387            .await?
388            {
389                (context.runtime.frontend)(widget)?;
390            }
391            Ok(())
392        })
393    }
394}
395
396#[cfg(test)]
397mod tests {
398    use std::collections::BTreeMap;
399    use std::sync::Arc;
400
401    use super::*;
402    use crate::backend::checkpoint::QueuedMessage;
403    use crate::middleware::{
404        ActiveCommandContext, MessageQueue, MessageRouteContext, MiddlewareStack,
405    };
406    use crate::protocol::{MessageReply, MessageSubmission, MessageTarget, SessionFileReference};
407
408    #[test]
409    fn manifest_advertises_delivery_symbols() {
410        assert!(
411            MANIFEST
412                .feature(&[])
413                .settings
414                .iter()
415                .all(|setting| !setting.composer)
416        );
417        assert_eq!(
418            DELIVERIES
419                .iter()
420                .map(|choice| (choice.value, choice.symbol))
421                .collect::<Vec<_>>(),
422            [("steer", Some("steer")), ("queue", Some("queue"))]
423        );
424    }
425
426    #[test]
427    fn queued_widgets_name_their_delivery() {
428        let messages = Messages::default();
429        let symbol = |delivery| {
430            let FrontendEvent::Widget { item, .. } = messages.queued_widget(
431                "message-1",
432                &MessageEvent {
433                    author: MessageAuthor::User,
434                    delivery,
435                    text: "hello".into(),
436                    attachments: Vec::new(),
437                    reply: None,
438                    message_target: None,
439                },
440            ) else {
441                panic!("queued message widget");
442            };
443            item.symbol
444        };
445
446        assert_eq!(
447            (
448                symbol(MessageDelivery::Steer),
449                symbol(MessageDelivery::Queue),
450            ),
451            (
452                Some(FrontendSymbol::Custom("steer".into())),
453                Some(FrontendSymbol::Custom("queue".into())),
454            )
455        );
456    }
457
458    fn user(delivery: Option<ActiveMessageDelivery>) -> MessageSubmission {
459        MessageSubmission {
460            author: MessageAuthor::User,
461            text: "hello".into(),
462            attachments: Vec::<SessionFileReference>::new(),
463            reply: None,
464            requested_delivery: delivery,
465            target_turn_id: None,
466        }
467    }
468
469    fn route(
470        stack: &MiddlewareStack,
471        queued: &mut Vec<QueuedMessage>,
472        message: &MessageSubmission,
473        active_turn_id: Option<&str>,
474    ) -> SubmissionResult {
475        stack
476            .route_message(&mut MessageRouteContext {
477                submission_id: "message-1",
478                message,
479                active_turn_id,
480                queued_messages: MessageQueue::new(queued),
481                events: &mut Vec::new(),
482            })
483            .expect("route message")
484    }
485
486    #[test]
487    fn active_user_uses_the_configured_queue_boundary() {
488        let stack = MiddlewareStack::new(vec![Arc::new(
489            Messages::new(4, ActiveMessageDelivery::Queue).expect("messages"),
490        )])
491        .expect("stack");
492        let mut queued = Vec::new();
493
494        let result = route(&stack, &mut queued, &user(None), Some("turn-1"));
495
496        assert_eq!(
497            result,
498            SubmissionResult::Accepted {
499                input_changed: false
500            }
501        );
502        assert!(
503            !stack
504                .messages_ready(&queued, "turn-1")
505                .expect("message input readiness")
506        );
507        assert_eq!(
508            stack
509                .next_turn(&mut queued)
510                .expect("next turn")
511                .expect("queued message")
512                .event
513                .message()
514                .map(|message| message.delivery),
515            Some(MessageDelivery::Queue)
516        );
517    }
518
519    #[tokio::test]
520    async fn queued_message_edit_preserves_its_reply_snapshot() {
521        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
522        let mut queued = Vec::new();
523        let mut message = user(Some(ActiveMessageDelivery::Queue));
524        message.reply = Some(MessageReply {
525            target: MessageTarget {
526                checkpoint_sequence: 5,
527                batch_item_count: 2,
528            },
529            text: "Earlier".into(),
530        });
531        route(&stack, &mut queued, &message, Some("turn-1"));
532        let mut events = Vec::new();
533        let metadata = BTreeMap::new();
534        let directory = tempfile::tempdir().expect("checkpoint directory");
535        let checkpoints = crate::backend::checkpoint::sqlite::SqliteCheckpoint::new(
536            directory.path().join("checkpoints.sqlite3"),
537        )
538        .expect("checkpoint store");
539
540        stack
541            .active_command(
542                MANIFEST.id,
543                &mut ActiveCommandContext {
544                    checkpoints: &checkpoints,
545                    submission_id: "message-2",
546                    session_id: "session-1",
547                    metadata: &metadata,
548                    active_turn_id: "turn-1",
549                    command: EDIT_COMMAND,
550                    arguments: "message-1",
551                    input: Some("Updated"),
552                    target: None,
553                    queued_messages: MessageQueue::new(&mut queued),
554                    events: &mut events,
555                },
556            )
557            .await
558            .expect("edit queued message")
559            .expect("message command");
560
561        let edited = queued[0].event();
562        assert_eq!(
563            (edited.text.as_str(), edited.reply),
564            ("Updated", message.reply)
565        );
566    }
567
568    #[test]
569    fn active_peer_always_steers_as_non_authoritative_input() {
570        let stack = MiddlewareStack::new(vec![Arc::new(
571            Messages::new(4, ActiveMessageDelivery::Queue).expect("messages"),
572        )])
573        .expect("stack");
574        let peer = MessageSubmission {
575            author: MessageAuthor::Peer {
576                message_id: "board-1".into(),
577                session_id: "peer-1".into(),
578                handle: "worker".into(),
579                symbol: None,
580            },
581            text: "Review this.\n\nKeep the validation.\n".into(),
582            attachments: Vec::new(),
583            reply: None,
584            requested_delivery: Some(ActiveMessageDelivery::Queue),
585            target_turn_id: None,
586        };
587        let mut queued = Vec::new();
588
589        let result = route(&stack, &mut queued, &peer, Some("turn-1"));
590        let staged = stack
591            .stage_model_messages(&mut queued, "turn-1")
592            .expect("stage message");
593
594        assert_eq!(
595            result,
596            SubmissionResult::Accepted {
597                input_changed: true
598            }
599        );
600        assert!(staged[0].input.get("_mobius_internal").is_some());
601        assert_eq!(
602            staged[0].event.message().map(|message| message.delivery),
603            Some(MessageDelivery::Steer)
604        );
605        let block = Messages::default()
606            .render(&staged[0].event, "turn-1")
607            .expect("received activity");
608        assert_eq!(block.role, FrontendBlockRole::Activity);
609        assert_eq!(block.title, "Message received from @worker");
610        assert_eq!(block.text, peer.text);
611        assert_eq!(block.symbol, Some(FrontendSymbol::Chat));
612    }
613
614    #[test]
615    fn failed_turn_promotes_unstaged_steering_to_a_queued_turn() {
616        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
617        let mut queued = Vec::new();
618        route(&stack, &mut queued, &user(None), Some("turn-1"));
619
620        stack
621            .finish_message_turn(
622                &mut queued,
623                "turn-1",
624                crate::backend::checkpoint::ExecutionOutcome::Failed,
625            )
626            .expect("promote failed turn");
627        let next = stack
628            .next_turn(&mut queued)
629            .expect("next turn")
630            .expect("promoted message");
631
632        assert_eq!(
633            next.event.message().map(|message| message.delivery),
634            Some(MessageDelivery::Queue)
635        );
636    }
637}
mobius/middleware/messages.rs

mobius/middleware/
messages.rs
