mobius/middleware/
messages.rs

1//! Durable conversation-message delivery.
2
3use super::{
4    ActiveCommandContext, MessageRouteContext, MessageSubmitContext, Middleware,
5    SessionStartContext, SessionStartSource, SubmissionResult,
6};
7use crate::backend::checkpoint::{CheckpointStore, EventPageRequest, QueuedMessageBoundary};
8use crate::backend::model::internal_user_message;
9use crate::protocol::{
10    ActiveMessageDelivery, EventMsg, FrontendBlock, FrontendContribution, FrontendEvent,
11    FrontendSettingValue, FrontendSlot, FrontendSymbol, FrontendTone, FrontendWidget,
12    MAX_CAPABILITY_INPUT_BYTES, MessageAuthor, MessageDelivery, MessageEvent, MessageReply, Op,
13};
14use crate::{BoxFuture, Result};
15
16mod text {
17    #[derive(serde::Deserialize)]
18    #[serde(deny_unknown_fields)]
19    pub(super) struct Definition {
20        #[serde(deserialize_with = "crate::middleware::manifest::deserialize_settings")]
21        pub(super) settings: Vec<crate::middleware::manifest::MiddlewareSettingManifest>,
22        pub(super) permission_notice: String,
23        pub(super) reply_target_invalid: String,
24        pub(super) reply_text_changed: String,
25        pub(super) stale_turn: String,
26        pub(super) queue_full: String,
27        pub(super) queue_failed: String,
28        pub(super) edit_stale: String,
29        pub(super) edit_empty: String,
30        pub(super) edit_too_large: String,
31        pub(super) edit_peer: String,
32        pub(super) received_title: String,
33        pub(super) default_enabled: bool,
34        pub(super) manifest_description: String,
35        pub(super) manifest_label: String,
36    }
37    crate::embedded_config! { pub(super) static DEFINITION: Definition = include_str!("messages.toml"); }
38}
39const REPLY_EVENT_PAGE_SIZE: usize = 256;
40const MAX_PENDING: &str = "max_pending";
41const DELIVERY: &str = "delivery";
42
43super::manifest::middleware_manifest! {
44/// Configuration and presentation metadata for message delivery.
45    "messages", text::DEFINITION, required: true, settings: &text::DEFINITION.settings
46}
47
48const EDIT_COMMAND: &str = "edit";
49
50fn rejected(message: &str) -> SubmissionResult {
51    SubmissionResult::Rejected(message.into())
52}
53
54/// Prepares every conversation message and owns its durable delivery lifecycle.
55pub struct Messages {
56    max_pending: usize,
57    delivery: ActiveMessageDelivery,
58}
59
60impl Default for Messages {
61    fn default() -> Self {
62        let settings = &text::DEFINITION.settings;
63        Self {
64            max_pending: super::manifest::integer_default(settings, MAX_PENDING) as usize,
65            delivery: super::manifest::string_default(settings, DELIVERY)
66                .parse()
67                .expect("valid embedded delivery"),
68        }
69    }
70}
71
72impl Messages {
73    /// Creates message delivery with a bounded queue and active-turn default.
74    /// # Errors
75    ///
76    /// Returns an error if configuration is invalid or a required resource cannot be initialized.
77    pub fn new(max_pending: usize, delivery: ActiveMessageDelivery) -> Result<Self> {
78        text::DEFINITION
79            .settings
80            .iter()
81            .find(|setting| setting.id() == MAX_PENDING)
82            .expect("embedded max_pending setting")
83            .validate(
84                MANIFEST.id,
85                Some(&FrontendSettingValue::Integer(
86                    i64::try_from(max_pending).unwrap_or(i64::MAX),
87                )),
88            )?;
89        Ok(Self {
90            max_pending,
91            delivery,
92        })
93    }
94
95    fn remove_widget(&self, id: &str) -> FrontendEvent {
96        FrontendEvent::RemoveWidget {
97            capability: self.name().into(),
98            id: id.into(),
99        }
100    }
101
102    fn queued_widget(
103        &self,
104        id: &str,
105        author: &MessageAuthor,
106        delivery: MessageDelivery,
107        text: &str,
108    ) -> FrontendEvent {
109        let action = matches!(author, MessageAuthor::User).then(|| Op::CapabilityCommand {
110            capability: self.name().into(),
111            command: EDIT_COMMAND.into(),
112            arguments: id.into(),
113            input: Some(text.into()),
114            target: None,
115        });
116        FrontendEvent::Widget {
117            capability: self.name().into(),
118            item: FrontendWidget {
119                id: id.into(),
120                slot: FrontendSlot::TranscriptTail,
121                text: text.into(),
122                tone: FrontendTone::Neutral,
123                symbol: Some(match delivery {
124                    MessageDelivery::Turn => FrontendSymbol::Chat,
125                    MessageDelivery::Steer => FrontendSymbol::Custom("steer".into()),
126                    MessageDelivery::Queue => FrontendSymbol::Custom("queue".into()),
127                }),
128                icon_only: false,
129                progress: None,
130                content: None,
131                action,
132            },
133        }
134    }
135
136    fn prepare(
137        &self,
138        context: &MessageRouteContext<'_>,
139    ) -> std::result::Result<QueuedMessageBoundary, String> {
140        let Some(turn_id) = context.active_turn_id else {
141            return if context.message.target_turn_id.is_some() {
142                Err(text::DEFINITION.stale_turn.as_str().into())
143            } else {
144                Ok(QueuedMessageBoundary::Turn)
145            };
146        };
147        if let Some(target) = &context.message.target_turn_id
148            && target != turn_id
149        {
150            return Err(text::DEFINITION.stale_turn.as_str().into());
151        }
152        match context.message.requested_delivery.unwrap_or(self.delivery) {
153            ActiveMessageDelivery::Steer => Ok(QueuedMessageBoundary::Steer {
154                turn_id: turn_id.into(),
155            }),
156            ActiveMessageDelivery::Queue => Ok(QueuedMessageBoundary::Queue),
157        }
158    }
159
160    fn enqueue(&self, context: &mut MessageRouteContext<'_>) -> Result<SubmissionResult> {
161        let boundary = match self.prepare(context) {
162            Ok(boundary) => boundary,
163            Err(message) => return Ok(SubmissionResult::Rejected(message)),
164        };
165        if context.queued_messages.count() >= self.max_pending {
166            return Ok(rejected(&text::DEFINITION.queue_full));
167        }
168        let message = &mut *context.message;
169        let event = MessageEvent {
170            author: std::mem::replace(&mut message.author, MessageAuthor::User),
171            delivery: boundary.delivery(),
172            text: std::mem::take(&mut message.text),
173            attachments: std::mem::take(&mut message.attachments),
174            reply: message.reply.take(),
175            message_target: None,
176        };
177        let widget = (!matches!(boundary, QueuedMessageBoundary::Turn)).then(|| {
178            self.queued_widget(
179                context.submission_id,
180                &event.author,
181                event.delivery,
182                &event.text,
183            )
184        });
185        let input_changed = matches!(boundary, QueuedMessageBoundary::Steer { .. });
186        if !context
187            .queued_messages
188            .enqueue(context.submission_id, boundary, event)?
189        {
190            return Ok(rejected(&text::DEFINITION.queue_failed));
191        }
192        if let Some(widget) = widget {
193            context.events.push(EventMsg::Frontend(widget));
194        }
195        Ok(SubmissionResult::Accepted { input_changed })
196    }
197
198    fn edit(&self, context: &mut ActiveCommandContext<'_>) -> Result<SubmissionResult> {
199        let Some(input) = context.input.filter(|input| !input.trim().is_empty()) else {
200            return Ok(rejected(&text::DEFINITION.edit_empty));
201        };
202        if input.len() > MAX_CAPABILITY_INPUT_BYTES {
203            return Ok(rejected(&text::DEFINITION.edit_too_large));
204        }
205        let Some(queued) = context.queued_messages.find(context.arguments) else {
206            return Ok(rejected(&text::DEFINITION.edit_stale));
207        };
208        if !matches!(queued.author(), MessageAuthor::User) {
209            return Ok(rejected(&text::DEFINITION.edit_peer));
210        }
211        let event = MessageEvent {
212            author: MessageAuthor::User,
213            delivery: queued.delivery(),
214            text: input.into(),
215            attachments: queued.attachments().to_vec(),
216            reply: queued.reply().cloned(),
217            message_target: None,
218        };
219        let input_changed = event.delivery == MessageDelivery::Steer;
220        let widget = self.queued_widget(
221            context.submission_id,
222            &event.author,
223            event.delivery,
224            &event.text,
225        );
226        if !context
227            .queued_messages
228            .replace(context.arguments, context.submission_id, event)?
229        {
230            return Ok(rejected(&text::DEFINITION.edit_stale));
231        }
232        context
233            .events
234            .push(EventMsg::Frontend(self.remove_widget(context.arguments)));
235        context.events.push(EventMsg::Frontend(widget));
236        Ok(SubmissionResult::Accepted { input_changed })
237    }
238}
239
240impl Middleware for Messages {
241    fn name(&self) -> &'static str {
242        MANIFEST.id
243    }
244
245    fn frontend(&self, _session_id: &str) -> FrontendContribution {
246        FrontendContribution {
247            capability: self.name().into(),
248            ..FrontendContribution::default()
249        }
250    }
251
252    fn render(&self, event: &EventMsg, _session_id: &str) -> Option<FrontendBlock> {
253        let EventMsg::Message(message) = event else {
254            return None;
255        };
256        let MessageAuthor::Source {
257            message_id,
258            source,
259            handle,
260            symbol,
261            ..
262        } = &message.author
263        else {
264            return None;
265        };
266        if matches!(source, crate::protocol::MessageSource::Session { .. }) {
267            return None;
268        }
269        Some(FrontendBlock {
270            id: Some(format!(
271                "message_received:{}:{session_id}:{message_id}",
272                source.id().len(),
273                session_id = source.id()
274            )),
275            title: text::DEFINITION.received_title.replace("{handle}", handle),
276            text: message.text.clone(),
277            symbol: Some(symbol.clone().unwrap_or(FrontendSymbol::Chat)),
278            ..Default::default()
279        })
280    }
281
282    fn handles_messages(&self) -> bool {
283        true
284    }
285
286    fn route_message<'a>(
287        &'a self,
288        context: &'a mut MessageRouteContext<'_>,
289    ) -> BoxFuture<'a, Result<SubmissionResult>> {
290        Box::pin(async move {
291            if let Some(reply) = &context.message.reply
292                && let Some(rejection) =
293                    validate_reply(context.checkpoints, context.session_id, reply).await?
294            {
295                return Ok(SubmissionResult::Rejected(rejection.into()));
296            }
297            self.enqueue(context)
298        })
299    }
300    fn message_boundary_events(&self, submission_id: &str) -> Vec<EventMsg> {
301        vec![EventMsg::Frontend(self.remove_widget(submission_id))]
302    }
303
304    fn message_submit<'a>(
305        &'a self,
306        context: &'a mut MessageSubmitContext<'_>,
307    ) -> BoxFuture<'a, Result<()>> {
308        Box::pin(async move {
309            if matches!(context.author, MessageAuthor::Source { .. }) {
310                context.deliver_once("permissions", || {
311                    internal_user_message(
312                        "message_permissions",
313                        text::DEFINITION.permission_notice.as_str(),
314                    )
315                })?;
316            }
317            Ok(())
318        })
319    }
320
321    fn active_command<'a>(
322        &'a self,
323        context: &'a mut ActiveCommandContext<'_>,
324    ) -> BoxFuture<'a, Result<Option<SubmissionResult>>> {
325        Box::pin(async move {
326            if context.command == EDIT_COMMAND {
327                self.edit(context).map(Some)
328            } else {
329                Ok(None)
330            }
331        })
332    }
333
334    fn session_start<'a>(
335        &'a self,
336        context: &'a mut SessionStartContext<'_>,
337    ) -> BoxFuture<'a, Result<()>> {
338        Box::pin(async move {
339            if context.source() == SessionStartSource::Compact {
340                return Ok(());
341            }
342            for queued in context.queued_messages().views() {
343                (context.runtime.frontend)(self.queued_widget(
344                    queued.id(),
345                    queued.author(),
346                    queued.delivery(),
347                    queued.text(),
348                ))?;
349            }
350            Ok(())
351        })
352    }
353}
354
355/// Accepts a reply only when it quotes a durable message of this chat exactly.
356async fn validate_reply(
357    checkpoints: &dyn CheckpointStore,
358    session_id: &str,
359    reply: &MessageReply,
360) -> Result<Option<&'static str>> {
361    let mut before_sequence = None;
362    loop {
363        let page = checkpoints
364            .event_page(
365                session_id,
366                EventPageRequest {
367                    before_sequence,
368                    limit: REPLY_EVENT_PAGE_SIZE,
369                },
370            )
371            .await?;
372        for record in page.events {
373            let original = match &record.event.msg {
374                EventMsg::Message(message)
375                    if message.message_target.as_ref() == Some(&reply.target) =>
376                {
377                    Some(message.reply_text())
378                }
379                EventMsg::AssistantMessage(message)
380                    if message.message_target.as_ref() == Some(&reply.target) =>
381                {
382                    Some(message.reply_text())
383                }
384                _ => None,
385            };
386            if let Some(original) = original {
387                return Ok(match original {
388                    Some(original) if original.as_ref() == reply.text => None,
389                    Some(_) => Some(text::DEFINITION.reply_text_changed.as_str()),
390                    None => Some(text::DEFINITION.reply_target_invalid.as_str()),
391                });
392            }
393        }
394        let Some(next) = page.next_before_sequence else {
395            return Ok(Some(text::DEFINITION.reply_target_invalid.as_str()));
396        };
397        before_sequence = Some(next);
398    }
399}
400
401#[cfg(test)]
402mod tests {
403    use crate::protocol::FrontendBlockRole;
404    #[test]
405    fn declared_deliveries_parse_at_the_protocol_boundary() {
406        for choice in
407            super::super::manifest::static_choices(&super::text::DEFINITION.settings, "delivery")
408                .iter()
409        {
410            assert_eq!(
411                choice
412                    .value
413                    .parse::<super::ActiveMessageDelivery>()
414                    .expect("declared delivery")
415                    .id(),
416                choice.value
417            );
418        }
419        assert!("other".parse::<super::ActiveMessageDelivery>().is_err());
420    }
421
422    use std::collections::BTreeMap;
423    use std::sync::Arc;
424
425    use super::*;
426    use crate::backend::checkpoint::sqlite::SqliteCheckpoint;
427    use crate::backend::checkpoint::{Checkpoint, QueuedMessage};
428    use crate::middleware::{
429        ActiveCommandContext, MessageQueue, MessageRouteContext, MiddlewareStack,
430    };
431    use crate::protocol::{MessageReply, MessageSubmission, MessageTarget, SessionFileReference};
432
433    #[test]
434    fn queue_limit_follows_the_declared_setting_bounds() {
435        for (limit, valid) in [(0, false), (1, true), (1024, true), (1025, false)] {
436            assert_eq!(
437                Messages::new(limit, ActiveMessageDelivery::Steer).is_ok(),
438                valid
439            );
440        }
441    }
442
443    #[test]
444    fn manifest_advertises_delivery_symbols() {
445        assert!(
446            MANIFEST
447                .feature(Default::default())
448                .settings
449                .iter()
450                .all(|setting| !setting.composer)
451        );
452        assert_eq!(
453            super::super::manifest::static_choices(&text::DEFINITION.settings, "delivery")
454                .iter()
455                .map(|choice| (choice.value.as_str(), choice.symbol.as_deref()))
456                .collect::<Vec<_>>(),
457            [("steer", Some("steer")), ("queue", Some("queue"))]
458        );
459    }
460
461    #[test]
462    fn queued_widgets_name_their_delivery() {
463        let messages = Messages::default();
464        let symbol = |delivery| {
465            let FrontendEvent::Widget { item, .. } =
466                messages.queued_widget("message-1", &MessageAuthor::User, delivery, "hello")
467            else {
468                panic!("queued message widget");
469            };
470            assert_eq!(item.text, "hello");
471            assert!(matches!(
472                item.action,
473                Some(Op::CapabilityCommand { arguments, input: Some(input), .. })
474                    if arguments == "message-1" && input == "hello"
475            ));
476            item.symbol
477        };
478
479        assert_eq!(
480            (
481                symbol(MessageDelivery::Steer),
482                symbol(MessageDelivery::Queue),
483            ),
484            (
485                Some(FrontendSymbol::Custom("steer".into())),
486                Some(FrontendSymbol::Custom("queue".into())),
487            )
488        );
489    }
490
491    fn user(delivery: Option<ActiveMessageDelivery>) -> MessageSubmission {
492        MessageSubmission {
493            author: MessageAuthor::User,
494            text: "hello".into(),
495            attachments: Vec::<SessionFileReference>::new(),
496            reply: None,
497            requested_delivery: delivery,
498            target_turn_id: None,
499        }
500    }
501
502    fn event(id: &str, msg: EventMsg) -> crate::protocol::Event {
503        crate::protocol::Event {
504            submission_id: Some(id.into()),
505            msg,
506        }
507    }
508
509    fn checkpoints() -> (tempfile::TempDir, SqliteCheckpoint) {
510        let directory = tempfile::tempdir().expect("checkpoint directory");
511        let store = SqliteCheckpoint::new(directory.path().join("checkpoints.sqlite3"))
512            .expect("checkpoint store");
513        (directory, store)
514    }
515
516    async fn route(
517        stack: &MiddlewareStack,
518        queued: &mut Vec<QueuedMessage>,
519        message: &MessageSubmission,
520        active_turn_id: Option<&str>,
521    ) -> SubmissionResult {
522        let (_directory, store) = checkpoints();
523        route_in(&store, stack, queued, message, active_turn_id).await
524    }
525
526    async fn route_in(
527        checkpoints: &dyn CheckpointStore,
528        stack: &MiddlewareStack,
529        queued: &mut Vec<QueuedMessage>,
530        message: &MessageSubmission,
531        active_turn_id: Option<&str>,
532    ) -> SubmissionResult {
533        let mut message = message.clone();
534        stack
535            .route_message(&mut MessageRouteContext {
536                checkpoints,
537                session_id: "session-1",
538                submission_id: "message-1",
539                message: &mut message,
540                active_turn_id,
541                queued_messages: MessageQueue::new(queued),
542                events: &mut Vec::new(),
543            })
544            .await
545            .expect("route message")
546    }
547
548    #[tokio::test]
549    async fn active_user_uses_the_configured_queue_boundary() {
550        let stack = MiddlewareStack::new(vec![Arc::new(
551            Messages::new(4, ActiveMessageDelivery::Queue).expect("messages"),
552        )])
553        .expect("stack");
554        let mut queued = Vec::new();
555
556        let result = route(&stack, &mut queued, &user(None), Some("turn-1")).await;
557
558        assert_eq!(
559            result,
560            SubmissionResult::Accepted {
561                input_changed: false
562            }
563        );
564        assert!(
565            !stack
566                .messages_ready(&queued, "turn-1")
567                .expect("message input readiness")
568        );
569        assert_eq!(
570            stack
571                .next_turn(&mut queued)
572                .expect("next turn")
573                .expect("queued message")
574                .event
575                .message()
576                .map(|message| message.delivery),
577            Some(MessageDelivery::Queue)
578        );
579    }
580
581    #[tokio::test]
582    async fn immediate_turn_does_not_publish_a_queued_widget() {
583        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
584        let mut queued = Vec::new();
585        let mut events = Vec::new();
586
587        let (_directory, store) = checkpoints();
588        let result = stack
589            .route_message(&mut MessageRouteContext {
590                checkpoints: &store,
591                session_id: "session-1",
592                submission_id: "message-1",
593                message: &mut user(None),
594                active_turn_id: None,
595                queued_messages: MessageQueue::new(&mut queued),
596                events: &mut events,
597            })
598            .await
599            .expect("route message");
600
601        assert_eq!(
602            result,
603            SubmissionResult::Accepted {
604                input_changed: false
605            }
606        );
607        assert!(events.is_empty());
608        assert_eq!(queued.len(), 1);
609    }
610
611    #[tokio::test]
612    async fn queued_message_edit_preserves_retained_content_after_validation_failure() {
613        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
614        let mut queued = Vec::new();
615        let mut message = user(Some(ActiveMessageDelivery::Queue));
616        message.attachments.push(SessionFileReference {
617            id: uuid::Uuid::new_v4().to_string(),
618            name: "notes.txt".into(),
619            size: 8,
620            media_type: "text/plain".into(),
621        });
622        let target = MessageTarget {
623            checkpoint_sequence: 5,
624            batch_item_count: 2,
625        };
626        message.reply = Some(MessageReply {
627            target,
628            text: "Earlier".into(),
629        });
630        let (_directory, checkpoints) = checkpoints();
631        let mut session = crate::backend::checkpoint::Checkpoint::empty("session-1");
632        session.session_context.owner_id = "test-bot".into();
633        checkpoints
634            .save(&session, &[], None)
635            .await
636            .expect("save session");
637        checkpoints
638            .append_event(
639                "session-1",
640                1,
641                &event(
642                    "earlier",
643                    EventMsg::Message(MessageEvent {
644                        author: MessageAuthor::User,
645                        delivery: MessageDelivery::Turn,
646                        text: "Earlier".into(),
647                        attachments: Vec::new(),
648                        reply: None,
649                        message_target: Some(target),
650                    }),
651                ),
652            )
653            .await
654            .expect("append quoted message");
655        route_in(&checkpoints, &stack, &mut queued, &message, Some("turn-1")).await;
656        let mut events = Vec::new();
657        let metadata = BTreeMap::new();
658
659        let original = queued.clone();
660        for submission_id in [" ", "message-2"] {
661            let result = stack
662                .active_command(
663                    MANIFEST.id,
664                    &mut ActiveCommandContext {
665                        checkpoints: &checkpoints,
666                        submission_id,
667                        session_id: "session-1",
668                        metadata: &metadata,
669                        active_turn_id: "turn-1",
670                        command: EDIT_COMMAND,
671                        arguments: "message-1",
672                        input: Some("Updated"),
673                        target: None,
674                        queued_messages: MessageQueue::new(&mut queued),
675                        events: &mut events,
676                    },
677                )
678                .await;
679            if submission_id.trim().is_empty() {
680                assert!(result.is_err());
681                assert_eq!(queued, original);
682                assert!(events.is_empty());
683            } else {
684                assert_eq!(
685                    result.expect("edit queued message"),
686                    Some(SubmissionResult::Accepted {
687                        input_changed: false
688                    })
689                );
690            }
691        }
692
693        let edited = queued[0].event();
694        assert_eq!(queued[0].id(), "message-2");
695        assert_eq!(
696            (
697                edited.text.as_str(),
698                edited.delivery,
699                edited.attachments,
700                edited.reply
701            ),
702            (
703                "Updated",
704                MessageDelivery::Queue,
705                message.attachments,
706                message.reply
707            )
708        );
709    }
710
711    #[tokio::test]
712    async fn explicitly_steered_source_remains_non_authoritative_input() {
713        let stack = MiddlewareStack::new(vec![Arc::new(
714            Messages::new(4, ActiveMessageDelivery::Queue).expect("messages"),
715        )])
716        .expect("stack");
717        let peer = MessageSubmission {
718            author: MessageAuthor::Source {
719                message_id: "board-1".into(),
720                source: crate::protocol::MessageSource::Session {
721                    session_id: "peer-1".into(),
722                },
723                cause_id: None,
724                ancestry: Vec::new(),
725                handle: "worker".into(),
726                symbol: None,
727            },
728            text: "Review this.\n\nKeep the validation.\n".into(),
729            attachments: Vec::new(),
730            reply: None,
731            requested_delivery: Some(ActiveMessageDelivery::Steer),
732            target_turn_id: None,
733        };
734        let mut queued = Vec::new();
735
736        let result = route(&stack, &mut queued, &peer, Some("turn-1")).await;
737        let staged = stack
738            .stage_model_messages(&mut queued, "turn-1")
739            .expect("stage message");
740
741        assert_eq!(
742            result,
743            SubmissionResult::Accepted {
744                input_changed: true
745            }
746        );
747        assert!(staged[0].input.get("_mobius_internal").is_some());
748        assert_eq!(
749            staged[0].event.message().map(|message| message.delivery),
750            Some(MessageDelivery::Steer)
751        );
752        assert!(
753            Messages::default()
754                .render(&staged[0].event, "turn-1")
755                .is_none(),
756            "peer messages use their typed message event, not an activity block"
757        );
758        assert_eq!(staged[0].event.message().unwrap().text, peer.text);
759    }
760
761    #[tokio::test]
762    async fn queued_external_event_waits_for_its_own_turn() {
763        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
764        let source = MessageSubmission {
765            author: MessageAuthor::Source {
766                message_id: "report".into(),
767                source: crate::protocol::MessageSource::External {
768                    source_id: "routine".into(),
769                    event_id: "completed".into(),
770                },
771                cause_id: None,
772                ancestry: Vec::new(),
773                handle: "monitor".into(),
774                symbol: None,
775            },
776            text: "Monitoring completed.".into(),
777            attachments: Vec::new(),
778            reply: None,
779            requested_delivery: Some(ActiveMessageDelivery::Queue),
780            target_turn_id: None,
781        };
782        let mut queued = Vec::new();
783        assert_eq!(
784            route(&stack, &mut queued, &source, Some("user-turn")).await,
785            SubmissionResult::Accepted {
786                input_changed: false
787            }
788        );
789        assert!(
790            stack
791                .stage_model_messages(&mut queued, "user-turn")
792                .expect("stage")
793                .is_empty()
794        );
795        let report = stack.next_turn(&mut queued).expect("next").expect("report");
796        let block = Messages::default()
797            .render(&report.event, "report-turn")
798            .expect("external activity");
799        assert_eq!(block.role, FrontendBlockRole::Activity);
800        assert_eq!(block.title, "Message received from @monitor");
801        assert_eq!(block.text, source.text);
802        assert_eq!(block.symbol, Some(FrontendSymbol::Chat));
803        assert!(matches!(report.event, EventMsg::Message(event) if event.author == source.author));
804    }
805
806    #[tokio::test]
807    async fn failed_turn_promotes_unstaged_steering_to_a_queued_turn() {
808        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
809        let mut queued = Vec::new();
810        route(&stack, &mut queued, &user(None), Some("turn-1")).await;
811
812        stack
813            .finish_message_turn(
814                &mut queued,
815                "turn-1",
816                crate::backend::checkpoint::ExecutionOutcome::Failed,
817            )
818            .expect("promote failed turn");
819        let next = stack
820            .next_turn(&mut queued)
821            .expect("next turn")
822            .expect("promoted message");
823
824        assert_eq!(
825            next.event.message().map(|message| message.delivery),
826            Some(MessageDelivery::Queue)
827        );
828    }
829
830    #[tokio::test]
831    async fn reply_matches_a_durable_message_in_its_own_session() {
832        let (_directory, store) = checkpoints();
833        for session_id in ["current", "other"] {
834            let mut checkpoint = Checkpoint::empty(session_id);
835            checkpoint.session_context.owner_id = "test-bot".into();
836            store
837                .save(&checkpoint, &[], None)
838                .await
839                .expect("save session");
840        }
841        let attachment_target = crate::protocol::MessageTarget {
842            checkpoint_sequence: 1,
843            batch_item_count: 1,
844        };
845        store
846            .append_event(
847                "current",
848                1,
849                &event(
850                    "attachment",
851                    EventMsg::Message(crate::protocol::MessageEvent {
852                        author: crate::protocol::MessageAuthor::User,
853                        delivery: crate::protocol::MessageDelivery::Turn,
854                        text: String::new(),
855                        attachments: vec![crate::protocol::SessionFileReference {
856                            id: "00000000-0000-0000-0000-000000000001".into(),
857                            name: "clip.mov".into(),
858                            size: 1,
859                            media_type: "video/quicktime".into(),
860                        }],
861                        reply: None,
862                        message_target: Some(attachment_target),
863                    }),
864                ),
865            )
866            .await
867            .expect("append attachment message");
868        let assistant_target = crate::protocol::MessageTarget {
869            checkpoint_sequence: 1,
870            batch_item_count: 2,
871        };
872        store
873            .append_event(
874                "current",
875                2,
876                &event(
877                    "assistant",
878                    EventMsg::AssistantMessage(crate::protocol::AssistantMessageEvent {
879                        session_id: "current".into(),
880                        turn_id: "turn".into(),
881                        model_step_id: "step".into(),
882                        content: vec![
883                            crate::protocol::ModelStepContent {
884                                output_index: 0,
885                                part_index: 0,
886                                phase: crate::protocol::ModelStepContentPhase::FinalAnswer,
887                                text: "first part".into(),
888                                annotations: Vec::new(),
889                            },
890                            crate::protocol::ModelStepContent {
891                                output_index: 1,
892                                part_index: 0,
893                                phase: crate::protocol::ModelStepContentPhase::FinalAnswer,
894                                text: "last part".into(),
895                                annotations: Vec::new(),
896                            },
897                        ],
898                        message_target: Some(assistant_target),
899                    }),
900                ),
901            )
902            .await
903            .expect("append assistant message");
904        let other_target = crate::protocol::MessageTarget {
905            checkpoint_sequence: 2,
906            batch_item_count: 1,
907        };
908        store
909            .append_event(
910                "other",
911                1,
912                &event(
913                    "other",
914                    EventMsg::Message(crate::protocol::MessageEvent {
915                        author: crate::protocol::MessageAuthor::User,
916                        delivery: crate::protocol::MessageDelivery::Turn,
917                        text: "other chat".into(),
918                        attachments: Vec::new(),
919                        reply: None,
920                        message_target: Some(other_target),
921                    }),
922                ),
923            )
924            .await
925            .expect("append other message");
926
927        let exact = MessageReply {
928            target: attachment_target,
929            text: "clip.mov".into(),
930        };
931        let changed = MessageReply {
932            text: "forged quote".into(),
933            ..exact.clone()
934        };
935        let other = MessageReply {
936            target: other_target,
937            text: "other chat".into(),
938        };
939        let assistant = MessageReply {
940            target: assistant_target,
941            text: "last part".into(),
942        };
943        let combined_assistant = MessageReply {
944            target: assistant_target,
945            text: "first partlast part".into(),
946        };
947
948        assert_eq!(
949            (
950                validate_reply(&store, "current", &exact)
951                    .await
952                    .expect("validate exact reply"),
953                validate_reply(&store, "current", &changed)
954                    .await
955                    .expect("validate changed reply"),
956                validate_reply(&store, "current", &other)
957                    .await
958                    .expect("validate cross-session reply"),
959                validate_reply(&store, "current", &assistant)
960                    .await
961                    .expect("validate assistant reply"),
962                validate_reply(&store, "current", &combined_assistant)
963                    .await
964                    .expect("validate combined assistant reply"),
965            ),
966            (
967                None,
968                Some(text::DEFINITION.reply_text_changed.as_str()),
969                Some(text::DEFINITION.reply_target_invalid.as_str()),
970                None,
971                Some(text::DEFINITION.reply_text_changed.as_str()),
972            )
973        );
974        let stack = MiddlewareStack::new(vec![Arc::new(Messages::default())]).expect("stack");
975        let mut queued = Vec::new();
976        let mut forged = user(None);
977        forged.reply = Some(changed);
978        let mut events = Vec::new();
979        let result = stack
980            .route_message(&mut MessageRouteContext {
981                checkpoints: &store,
982                session_id: "current",
983                submission_id: "message-1",
984                message: &mut forged,
985                active_turn_id: None,
986                queued_messages: MessageQueue::new(&mut queued),
987                events: &mut events,
988            })
989            .await
990            .expect("route forged reply");
991        assert_eq!(
992            result,
993            SubmissionResult::Rejected(text::DEFINITION.reply_text_changed.clone())
994        );
995        assert!(queued.is_empty() && events.is_empty());
996    }
997}
mobius/middleware/messages.rs

mobius/middleware/
messages.rs
