1use std::collections::{HashMap, HashSet};
29use std::fmt;
30use std::ops::{Bound, Range, RangeBounds};
31use std::str::FromStr;
32use std::sync::Arc;
33
34use serde::{Deserialize, Serialize, Serializer};
35
36use crate::completion::Message;
37use crate::message::{StopReason, ToolCall, UserContent};
38
39mod aside;
40mod author;
41mod lineage;
42mod note;
43pub mod pairing;
44mod turn;
45
46pub use aside::{Aside, AsideId};
47pub use author::{Author, AuthorKind, RequestId};
48pub use lineage::Lineage;
49pub use note::Note;
50pub use turn::{TURN, TurnEnd, TurnRecord};
51
52use pairing::{Normalized, PairingError};
53
54#[derive(Clone, Debug, Default, Serialize, Deserialize)]
70#[serde(try_from = "State")]
71pub struct Conversation {
72 messages: Vec<Arc<Message>>,
75 meta: Vec<Meta>,
77 #[serde(serialize_with = "saved_asides")]
79 asides: Vec<(AsideId, Aside)>,
80 #[serde(skip_serializing_if = "Option::is_none")]
81 halted: Option<Halt>,
82 next_id: u64,
84 next_aside: u64,
86 #[serde(skip)]
89 revision: u64,
90 #[serde(skip)]
92 lineage: Lineage,
93 #[serde(skip)]
94 unsaved: Vec<ChangeRecord>,
95 #[serde(skip)]
98 clock: Option<u64>,
99}
100
101#[derive(Clone, Debug, Serialize, Deserialize)]
105struct Meta {
106 id: MessageId,
107 #[serde(default, skip_serializing_if = "Option::is_none")]
108 author: Option<Author>,
109 #[serde(default)]
111 made: u64,
112}
113
114#[derive(Deserialize)]
116struct State {
117 #[serde(default)]
118 messages: Vec<Arc<Message>>,
119 #[serde(default)]
120 meta: Vec<Meta>,
121 #[serde(default)]
122 asides: Vec<(AsideId, Aside)>,
123 #[serde(default)]
124 halted: Option<Halt>,
125 #[serde(default)]
126 next_id: u64,
127 #[serde(default)]
128 next_aside: u64,
129}
130
131impl TryFrom<State> for Conversation {
132 type Error = ApplyError;
133
134 fn try_from(state: State) -> Result<Self, ApplyError> {
135 if state.meta.len() != state.messages.len() {
136 return Err(mismatch(format!(
137 "{} messages but {} ids",
138 state.messages.len(),
139 state.meta.len()
140 )));
141 }
142 let mut ids = HashSet::new();
143 if let Some(meta) = state
144 .meta
145 .iter()
146 .find(|meta| meta.id.0 >= state.next_id || !ids.insert(meta.id))
147 {
148 return Err(mismatch(format!("message id {} reused", meta.id.0)));
149 }
150 let mut asides = HashSet::new();
151 if let Some((id, _)) = state
152 .asides
153 .iter()
154 .find(|(id, _)| id.0 >= state.next_aside || !asides.insert(*id))
155 {
156 return Err(mismatch(format!("aside id {} reused", id.0)));
157 }
158 Ok(Self {
159 messages: state.messages,
160 meta: state.meta,
161 asides: state.asides,
162 halted: state.halted,
163 next_id: state.next_id,
164 next_aside: state.next_aside,
165 revision: 0,
166 lineage: Lineage::DETACHED,
167 unsaved: Vec::new(),
168 clock: None,
169 })
170 }
171}
172
173fn saved_asides<S: Serializer>(
174 asides: &[(AsideId, Aside)],
175 serializer: S,
176) -> Result<S::Ok, S::Error> {
177 serializer.collect_seq(asides.iter().filter(|(_, aside)| aside.saved))
178}
179
180impl PartialEq for Conversation {
181 fn eq(&self, other: &Self) -> bool {
182 let saved = |(_, aside): &&(AsideId, Aside)| aside.saved;
183 let same = |a: &Meta, b: &Meta| a.id == b.id && a.author == b.author;
184 self.messages == other.messages
185 && self.meta.len() == other.meta.len()
186 && self.meta.iter().zip(&other.meta).all(|(a, b)| same(a, b))
187 && self.halted == other.halted
188 && self
189 .asides
190 .iter()
191 .filter(saved)
192 .eq(other.asides.iter().filter(saved))
193 }
194}
195
196#[derive(
200 Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
201)]
202#[serde(transparent)]
203pub struct MessageId(u64);
204
205impl MessageId {
206 pub const fn get(self) -> u64 {
208 self.0
209 }
210}
211
212impl fmt::Display for MessageId {
214 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
215 write!(f, "m{}", self.0)
216 }
217}
218
219impl FromStr for MessageId {
221 type Err = ParseIdError;
222
223 fn from_str(text: &str) -> Result<Self, ParseIdError> {
224 parse_id(text, 'm').map(Self)
225 }
226}
227
228#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
230#[error("{0:?} is not an id")]
231pub struct ParseIdError(pub String);
232
233fn parse_id(text: &str, prefix: char) -> Result<u64, ParseIdError> {
235 let number = text.strip_prefix(prefix).unwrap_or(text);
236 number.parse().map_err(|_| ParseIdError(text.to_owned()))
237}
238
239#[derive(
244 Clone, Copy, Debug, Default, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize,
245)]
246pub struct MessageKey {
247 pub made: u64,
249 pub id: MessageId,
251}
252
253#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
258pub struct Entry {
259 pub message: Arc<Message>,
261 #[serde(default, skip_serializing_if = "Option::is_none")]
264 pub author: Option<Author>,
265 #[serde(default, skip_serializing_if = "Option::is_none")]
267 pub id: Option<MessageId>,
268 #[serde(default, skip_serializing_if = "Option::is_none")]
272 pub made: Option<u64>,
273}
274
275impl Entry {
276 pub fn new(message: impl Into<Arc<Message>>) -> Self {
278 Self {
279 message: message.into(),
280 author: None,
281 id: None,
282 made: None,
283 }
284 }
285
286 #[must_use]
288 pub fn by(self, author: Author) -> Self {
289 Self {
290 author: Some(author),
291 ..self
292 }
293 }
294}
295
296impl From<Message> for Entry {
297 fn from(message: Message) -> Self {
298 Self::new(message)
299 }
300}
301
302impl From<Arc<Message>> for Entry {
303 fn from(message: Arc<Message>) -> Self {
304 Self::new(message)
305 }
306}
307
308#[derive(Clone, Copy, Debug)]
310pub struct EntryRef<'c> {
311 pub index: usize,
313 pub id: MessageId,
315 pub message: &'c Message,
317 pub author: Option<&'c Author>,
319 pub made: u64,
321}
322
323impl EntryRef<'_> {
324 pub fn key(&self) -> MessageKey {
326 MessageKey {
327 made: self.made,
328 id: self.id,
329 }
330 }
331
332 pub fn is_users_own(&self) -> bool {
336 let typed = match self.message {
337 Message::User { content } => content
338 .iter()
339 .any(|item| !matches!(item, UserContent::ToolResult(_))),
340 _ => false,
341 };
342 let users = self
343 .author
344 .is_none_or(|author| author.kind == AuthorKind::User && author.title.is_none());
345 typed && users
346 }
347
348 pub fn is_request(&self) -> bool {
352 let asks = match self.message {
353 Message::User { content } => content
354 .iter()
355 .any(|item| !matches!(item, UserContent::ToolResult(_))),
356 _ => false,
357 };
358 asks && self
359 .author
360 .is_none_or(|author| author.kind != AuthorKind::Rig)
361 }
362
363 pub fn is_stop_marker(&self) -> bool {
366 let by_rig = self
367 .author
368 .is_some_and(|author| author.kind == AuthorKind::Rig);
369 by_rig
370 && matches!(self.message, Message::User { content }
371 if matches!(content.as_slice(), [UserContent::Text(text)] if text.text == STOPPED))
372 }
373}
374
375pub const STOPPED: &str = "[The request above was stopped by the user before it was answered. \
379 Do not carry it out unless asked again.]";
380
381#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
384#[serde(rename_all = "snake_case")]
385pub enum Halt {
386 Stopped,
389 #[default]
392 Kept,
393}
394
395#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
398#[serde(tag = "type", rename_all = "snake_case")]
399pub enum Change {
400 Push {
402 entries: Vec<Entry>,
404 },
405 Splice {
409 from: usize,
411 to: usize,
413 removed: Vec<MessageId>,
415 with: Vec<Entry>,
417 },
418 Halt {
420 halt: Option<Halt>,
422 },
423 Aside {
425 id: AsideId,
427 aside: Option<Aside>,
429 },
430 Reset {
434 state: Box<Conversation>,
436 },
437}
438
439#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
441pub struct ChangeRecord {
442 #[serde(flatten)]
444 pub change: Change,
445 #[serde(default, skip_serializing_if = "Option::is_none")]
448 pub by: Option<Author>,
449 #[serde(default, skip_serializing_if = "Option::is_none")]
451 pub at: Option<u64>,
452}
453
454#[derive(Clone, Debug, Default, PartialEq, Eq)]
456pub struct Spliced {
457 pub removed: Vec<MessageId>,
459 pub changed: Vec<MessageId>,
461 pub restored: Vec<MessageId>,
463}
464
465#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
468pub enum ApplyError {
469 #[error("{0}")]
471 Mismatch(String),
472}
473
474fn mismatch(why: String) -> ApplyError {
475 ApplyError::Mismatch(why)
476}
477
478pub fn short_id(id: &str) -> &str {
480 id.get(..8).unwrap_or(id)
481}
482
483impl Conversation {
484 pub fn new() -> Self {
486 Self::default()
487 }
488
489 pub fn from_messages(messages: impl IntoIterator<Item = impl Into<Arc<Message>>>) -> Self {
492 let mut conversation = Self::new();
493 conversation.extend(messages);
494 conversation
495 }
496
497 pub fn messages(&self) -> &[Arc<Message>] {
501 &self.messages
502 }
503
504 pub fn into_messages(self) -> Vec<Arc<Message>> {
507 self.messages
508 }
509
510 pub fn entries(&self) -> impl DoubleEndedIterator<Item = EntryRef<'_>> + ExactSizeIterator {
512 let entries = self.messages.iter().zip(&self.meta).enumerate();
513 entries.map(|(index, (message, meta))| EntryRef {
514 index,
515 id: meta.id,
516 message,
517 author: meta.author.as_ref(),
518 made: meta.made,
519 })
520 }
521
522 pub fn entry(&self, at: usize) -> Option<Entry> {
525 let meta = self.meta.get(at)?;
526 Some(Entry {
527 message: self.messages.get(at)?.clone(),
528 author: meta.author.clone(),
529 id: Some(meta.id),
530 made: Some(meta.made),
531 })
532 }
533
534 pub fn id(&self, at: usize) -> Option<MessageId> {
536 self.meta.get(at).map(|meta| meta.id)
537 }
538
539 pub fn key(&self, at: usize) -> Option<MessageKey> {
541 let meta = self.meta.get(at)?;
542 Some(MessageKey {
543 made: meta.made,
544 id: meta.id,
545 })
546 }
547
548 pub fn index_of_key(&self, key: MessageKey) -> Option<usize> {
551 let at = self.index_of(key.id)?;
552 (self.meta.get(at)?.made == key.made).then_some(at)
553 }
554
555 pub fn author(&self, at: usize) -> Option<&Author> {
558 self.meta.get(at)?.author.as_ref()
559 }
560
561 pub fn index_of(&self, id: MessageId) -> Option<usize> {
563 self.meta.iter().position(|meta| meta.id == id)
564 }
565
566 pub fn start_of(&self, id: MessageId) -> usize {
571 self.index_of(id).unwrap_or(if id.0 >= self.next_id {
572 self.messages.len()
573 } else {
574 0
575 })
576 }
577
578 pub fn turn_start(&self) -> MessageId {
581 let trailing = self.messages.iter().rev();
582 let trailing = trailing
583 .take_while(|message| matches!(&***message, Message::User { .. }))
584 .count();
585 let first = self.messages.len().saturating_sub(trailing);
586 let answered = (trailing > 0).then(|| self.id(first)).flatten();
587 answered.unwrap_or(MessageId(self.next_id))
588 }
589
590 pub fn asked(&self, start: MessageId) -> impl Iterator<Item = EntryRef<'_>> {
596 let entries = self.entries().skip(self.start_of(start));
597 entries.take_while(|entry| matches!(entry.message, Message::User { .. }))
598 }
599
600 pub fn request(&self, start: MessageId) -> Option<EntryRef<'_>> {
604 self.asked(start).find(EntryRef::is_request)
605 }
606
607 pub fn halted(&self) -> Option<Halt> {
610 self.halted
611 }
612
613 pub fn awaits_model(&self) -> bool {
618 let read = self.messages.iter().rev().map(Arc::as_ref).find(|message| {
619 !matches!(message, Message::Assistant(turn)
620 if turn.stop.as_ref().is_some_and(StopReason::is_failure))
621 });
622 self.halted.is_none() && matches!(read, Some(Message::User { .. }))
623 }
624
625 pub fn pending_calls(&self) -> Vec<ToolCall> {
628 pairing::pending_calls(&self.messages)
629 }
630
631 pub fn validate(&self) -> Result<(), PairingError> {
634 pairing::validate(&self.messages)
635 }
636
637 pub fn asides(&self) -> impl DoubleEndedIterator<Item = (AsideId, &Aside)> {
639 self.asides.iter().map(|(id, aside)| (*id, aside))
640 }
641
642 pub fn aside(&self, id: AsideId) -> Option<&Aside> {
644 let mut asides = self.asides.iter();
645 asides
646 .find(|(known, _)| *known == id)
647 .map(|(_, aside)| aside)
648 }
649
650 pub fn aside_position(&self, aside: &Aside) -> usize {
654 let Some(after) = aside.after else {
655 return 0;
656 };
657 match self.index_of(after) {
658 Some(index) => index.saturating_add(1),
659 None => self
660 .messages
661 .len()
662 .min(self.start_of(after).saturating_add(1)),
663 }
664 }
665
666 pub fn push(&mut self, entry: impl Into<Entry>) -> MessageId {
673 let entry = entry.into();
674 let by = entry.author.clone();
675 let mut ids = self.push_entries(vec![entry], by);
676 ids.pop().unwrap_or(MessageId(self.next_id))
677 }
678
679 pub fn extend(&mut self, messages: impl IntoIterator<Item = impl Into<Arc<Message>>>) {
682 let entries = messages.into_iter().map(Entry::new).collect();
683 self.push_entries(entries, None);
684 }
685
686 fn push_entries(&mut self, entries: Vec<Entry>, by: Option<Author>) -> Vec<MessageId> {
687 if entries.is_empty() {
688 return Vec::new();
689 }
690 let stop = (self.halted == Some(Halt::Stopped))
691 .then(|| Entry::new(Message::user(STOPPED)).by(Author::rig()));
692 let mut next = self.next_id;
693 let mut ids = Vec::with_capacity(entries.len());
694 let mut fill = |mut entry: Entry| {
695 let id = MessageId(next);
696 next = next.saturating_add(1);
697 entry.id = Some(id);
698 entry.made = None;
699 (entry, id)
700 };
701 let stop = stop.map(|entry| fill(entry).0);
702 let entries = entries.into_iter().map(|entry| {
703 let (entry, id) = fill(entry);
704 ids.push(id);
705 entry
706 });
707 let entries = stop.into_iter().chain(entries).collect();
708 self.record(Change::Push { entries }, by);
709 self.clear_halt();
710 ids
711 }
712
713 pub fn splice(
720 &mut self,
721 range: impl RangeBounds<usize>,
722 with: impl IntoIterator<Item = Entry>,
723 by: Author,
724 ) -> Spliced {
725 self.replace(range, with.into_iter().collect(), by, false)
726 }
727
728 pub fn restore(
753 &mut self,
754 range: impl RangeBounds<usize>,
755 with: impl IntoIterator<Item = Entry>,
756 by: Author,
757 ) -> Spliced {
758 self.replace(range, with.into_iter().collect(), by, true)
759 }
760
761 pub fn set_entries(&mut self, entries: Vec<Entry>, by: Author) -> Spliced {
766 let same = |entry: &Entry, at: usize| {
767 entry.id.is_some()
768 && entry.id == self.id(at)
769 && entry.author.as_ref() == self.author(at)
770 && Some(&entry.message) == self.messages.get(at)
771 };
772 let (len, count) = (self.messages.len(), entries.len());
773 let start = entries
774 .iter()
775 .enumerate()
776 .take_while(|(at, entry)| same(entry, *at))
777 .count();
778 let end = entries
779 .iter()
780 .rev()
781 .zip((0..len).rev())
782 .take(len.min(count).saturating_sub(start))
783 .take_while(|(entry, at)| same(entry, *at))
784 .count();
785 if start == len && start == count {
786 return Spliced::default();
787 }
788 let with = entries
789 .into_iter()
790 .skip(start)
791 .take(count.saturating_sub(start).saturating_sub(end));
792 self.restore(start..len.saturating_sub(end), with, by)
793 }
794
795 fn replace(
798 &mut self,
799 range: impl RangeBounds<usize>,
800 mut with: Vec<Entry>,
801 by: Author,
802 restores: bool,
803 ) -> Spliced {
804 let len = self.messages.len();
805 let to = match range.end_bound() {
806 Bound::Included(&end) => end.saturating_add(1),
807 Bound::Excluded(&end) => end,
808 Bound::Unbounded => len,
809 }
810 .min(len);
811 let from = match range.start_bound() {
812 Bound::Included(&start) => start,
813 Bound::Excluded(&start) => start.saturating_add(1),
814 Bound::Unbounded => 0,
815 }
816 .min(to);
817 if from == to && with.is_empty() {
818 return Spliced::default();
819 }
820 let replaced = self.meta.get(from..to).unwrap_or_default();
821 let removed: Vec<MessageId> = replaced.iter().map(|meta| meta.id).collect();
822 let in_range: HashSet<MessageId> = removed.iter().copied().collect();
823 let mut outside = None;
824 let mut taken = HashSet::new();
825 let mut restored = Vec::new();
826 let mut next = self.next_id;
827 for entry in &mut with {
828 let id = entry.id.filter(|id| !taken.contains(id));
829 if let Some(id) = id.filter(|id| in_range.contains(id)) {
830 taken.insert(id);
831 entry.made = None;
832 continue;
833 }
834 let back = id.filter(|id| {
835 restores
836 && id.0 < self.next_id
837 && !self.outside(from..to, &mut outside).contains(id)
838 });
839 if let Some(id) = back {
840 taken.insert(id);
841 restored.push(id);
842 continue;
843 }
844 entry.id = Some(MessageId(next));
845 entry.made = None;
846 next = next.saturating_add(1);
847 }
848 let changed = with.iter().filter_map(|entry| entry.id).collect();
849 let gone = removed.iter().filter(|id| !taken.contains(id)).copied();
850 let spliced = Spliced {
851 removed: gone.collect(),
852 changed,
853 restored,
854 };
855 let change = Change::Splice {
856 from,
857 to,
858 removed,
859 with,
860 };
861 self.record(change, Some(by));
862 if to == len {
863 self.clear_halt();
864 }
865 spliced
866 }
867
868 fn outside<'s>(
871 &self,
872 range: Range<usize>,
873 outside: &'s mut Option<HashSet<MessageId>>,
874 ) -> &'s HashSet<MessageId> {
875 outside.get_or_insert_with(|| {
876 let before = self.meta.get(..range.start).unwrap_or_default();
877 let after = self.meta.get(range.end..).unwrap_or_default();
878 before.iter().chain(after).map(|meta| meta.id).collect()
879 })
880 }
881
882 pub fn edit_each(
907 &mut self,
908 by: Author,
909 mut edit: impl FnMut(EntryRef<'_>) -> Option<Message>,
910 ) -> Vec<MessageId> {
911 let rewritten: Vec<(usize, MessageId, Message)> = self
912 .entries()
913 .filter_map(|entry| {
914 let (at, id) = (entry.index, entry.id);
915 edit(entry).map(|message| (at, id, message))
916 })
917 .collect();
918 let (Some(&(first, ..)), Some(&(last, ..))) = (rewritten.first(), rewritten.last()) else {
919 return Vec::new();
920 };
921 let ids = rewritten.iter().map(|(_, id, _)| *id).collect();
922 let mut rewritten = rewritten.into_iter().peekable();
923 let entries: Vec<Entry> = (first..=last)
924 .filter_map(|at| match rewritten.next_if(|(next, ..)| *next == at) {
925 Some((_, id, message)) => Some(Entry {
926 message: Arc::new(message),
927 author: self.author(at).cloned(),
928 id: Some(id),
929 made: None,
930 }),
931 None => self.entry(at),
932 })
933 .collect();
934 self.splice(first..=last, entries, by);
935 ids
936 }
937
938 pub fn normalize(&mut self, by: Author) -> Normalized {
945 let entries = (0..self.messages.len()).filter_map(|at| self.entry(at));
946 let normalized = pairing::normalize(entries.collect());
947 let into: HashMap<MessageId, MessageId> = normalized
948 .merged
949 .iter()
950 .map(|(kept, gone)| (*gone, *kept))
951 .collect();
952 let dropped: HashSet<MessageId> = normalized.dropped.iter().copied().collect();
953 let anchor = |id: MessageId| -> Option<MessageId> {
956 let before = self.meta.iter().rev().skip_while(|meta| meta.id != id);
957 before
958 .map(|meta| meta.id)
959 .find_map(|id| match into.get(&id) {
960 Some(kept) => Some(*kept),
961 None => (!dropped.contains(&id)).then_some(id),
962 })
963 };
964 let moves: Vec<(AsideId, Option<MessageId>)> = self
965 .asides
966 .iter()
967 .filter_map(|(at, aside)| {
968 let after = aside.after?;
969 let moves = into.contains_key(&after) || dropped.contains(&after);
970 moves.then(|| (*at, anchor(after)))
971 })
972 .collect();
973 self.set_entries(normalized.entries.clone(), by.clone());
974 for (id, after) in moves {
975 self.move_aside(id, after, &by);
976 }
977 normalized
978 }
979
980 pub fn reread(&mut self, from_log: Conversation) -> Result<(), ApplyError> {
991 if !self.unsaved.is_empty() || !from_log.unsaved.is_empty() {
992 return Err(mismatch(
993 "a value with unsaved changes is not reread".to_owned(),
994 ));
995 }
996 let mut asides = from_log.asides;
997 asides.retain(|(_, aside)| aside.saved);
998 let unsaved = self.asides.iter().filter(|(_, aside)| !aside.saved);
999 if let Some((id, _)) = unsaved
1000 .clone()
1001 .find(|(id, _)| asides.iter().any(|(saved, _)| saved == id))
1002 {
1003 return Err(mismatch(format!("aside id {} is saved in the log", id.0)));
1004 }
1005 let next_aside = unsaved
1006 .clone()
1007 .map(|(id, _)| id.0.saturating_add(1))
1008 .fold(from_log.next_aside, u64::max);
1009 asides.extend(unsaved.cloned());
1010 asides.sort_by_key(|(id, _)| *id);
1011 *self = Self {
1012 messages: from_log.messages,
1013 meta: from_log.meta,
1014 asides,
1015 halted: from_log.halted,
1016 next_id: from_log.next_id,
1017 next_aside,
1018 revision: from_log.revision,
1019 lineage: from_log.lineage,
1020 unsaved: Vec::new(),
1021 clock: self.clock,
1022 };
1023 Ok(())
1024 }
1025
1026 pub fn halt(&mut self, halt: Halt) -> bool {
1029 let halts = self.awaits_model();
1030 if halts {
1031 self.record(Change::Halt { halt: Some(halt) }, None);
1032 }
1033 halts
1034 }
1035
1036 pub fn resume(&mut self) -> bool {
1039 self.clear_halt();
1040 self.awaits_model()
1041 }
1042
1043 fn clear_halt(&mut self) {
1044 if self.halted.is_some() {
1045 self.record(Change::Halt { halt: None }, None);
1046 }
1047 }
1048
1049 pub fn add_aside(&mut self, aside: Aside) -> AsideId {
1052 let id = AsideId(self.next_aside);
1053 if aside.saved {
1054 let by = Some(aside.author.clone());
1055 let aside = Some(aside);
1056 self.record(Change::Aside { id, aside }, by);
1057 } else {
1058 self.place_aside(id, aside);
1059 }
1060 id
1061 }
1062
1063 pub fn replace_aside(&mut self, id: AsideId, aside: Aside) -> bool {
1067 let Some(was_saved) = self.aside(id).map(|old| old.saved) else {
1068 return false;
1069 };
1070 if aside.saved {
1071 let by = Some(aside.author.clone());
1072 let aside = Some(aside);
1073 self.record(Change::Aside { id, aside }, by);
1074 } else {
1075 self.place_aside(id, aside);
1077 if was_saved {
1078 self.log(Change::Aside { id, aside: None }, None);
1079 }
1080 }
1081 true
1082 }
1083
1084 pub fn remove_aside(&mut self, id: AsideId) -> Option<Aside> {
1086 let at = self.asides.iter().position(|(known, _)| *known == id)?;
1087 let (_, aside) = self.asides.remove(at);
1088 if aside.saved {
1089 self.log(Change::Aside { id, aside: None }, None);
1090 }
1091 Some(aside)
1092 }
1093
1094 fn move_aside(&mut self, id: AsideId, after: Option<MessageId>, by: &Author) {
1097 let Some(aside) = self.aside(id).cloned() else {
1098 return;
1099 };
1100 let aside = aside.after(after);
1101 if aside.saved {
1102 let aside = Some(aside);
1103 self.record(Change::Aside { id, aside }, Some(by.clone()));
1104 } else {
1105 self.place_aside(id, aside);
1106 }
1107 }
1108
1109 fn place_aside(&mut self, id: AsideId, aside: Aside) {
1112 self.next_aside = self.next_aside.max(id.0.saturating_add(1));
1113 match self.asides.iter_mut().find(|(known, _)| *known == id) {
1114 Some((_, slot)) => *slot = aside,
1115 None => self.asides.push((id, aside)),
1116 }
1117 }
1118
1119 pub fn apply(&mut self, change: &ChangeRecord) -> Result<(), ApplyError> {
1125 let at = change.at.unwrap_or_else(|| self.now());
1126 self.change(&change.change, at)?;
1127 self.revision = self.revision.saturating_add(1);
1128 Ok(())
1129 }
1130
1131 pub fn dated<R>(&mut self, at: u64, edit: impl FnOnce(&mut Self) -> R) -> R {
1149 let outer = self.clock.replace(at);
1150 let made = edit(self);
1151 self.clock = outer;
1152 made
1153 }
1154
1155 fn now(&self) -> u64 {
1158 self.clock.unwrap_or_else(crate::now_ms)
1159 }
1160
1161 pub fn unsaved(&self) -> &[ChangeRecord] {
1163 &self.unsaved
1164 }
1165
1166 pub fn take_unsaved(&mut self) -> Vec<ChangeRecord> {
1168 std::mem::take(&mut self.unsaved)
1169 }
1170
1171 pub fn revision(&self) -> u64 {
1176 self.revision
1177 }
1178
1179 pub fn lineage(&self) -> Lineage {
1183 self.lineage
1184 }
1185
1186 pub fn continues(&self, lineage: Lineage, changes: u64) -> bool {
1192 let base = self.revision.saturating_sub(self.unsaved.len() as u64);
1193 base == changes
1194 && (self.lineage == lineage || (changes == 0 && self.lineage == Lineage::EMPTY))
1195 }
1196
1197 pub fn continue_log(&mut self, lineage: Lineage, changes: u64) {
1201 self.lineage = lineage;
1202 self.revision = changes.saturating_add(self.unsaved.len() as u64);
1203 }
1204
1205 pub fn view(&self) -> Self {
1208 Self {
1209 asides: self.asides.clone(),
1210 ..self.request_view()
1211 }
1212 }
1213
1214 pub fn request_view(&self) -> Self {
1217 Self {
1218 messages: self.messages.clone(),
1219 meta: self.meta.clone(),
1220 asides: Vec::new(),
1221 halted: self.halted,
1222 next_id: self.next_id,
1223 next_aside: self.next_aside,
1224 revision: self.revision,
1225 lineage: self.lineage,
1226 unsaved: Vec::new(),
1227 clock: None,
1228 }
1229 }
1230
1231 fn record(&mut self, change: Change, by: Option<Author>) {
1233 let at = self.now();
1234 match self.change(&change, at) {
1235 Ok(()) => self.log_at(change, by, at),
1236 Err(error) => tracing::error!(%error, "a conversation edit did not fit; dropped"),
1237 }
1238 }
1239
1240 fn log(&mut self, change: Change, by: Option<Author>) {
1242 let at = self.now();
1243 self.log_at(change, by, at);
1244 }
1245
1246 fn log_at(&mut self, change: Change, by: Option<Author>, at: u64) {
1247 self.revision = self.revision.saturating_add(1);
1248 self.unsaved.push(ChangeRecord {
1249 change,
1250 by,
1251 at: Some(at),
1252 });
1253 }
1254
1255 fn change(&mut self, change: &Change, at: u64) -> Result<(), ApplyError> {
1259 let meta = |entry: &Entry, made: u64| Meta {
1260 id: entry.id.unwrap_or_default(),
1261 author: entry.author.clone(),
1262 made,
1263 };
1264 match change {
1265 Change::Push { entries } => {
1266 let end = self.meta.len();
1267 let next = self.new_ids(entries, &HashSet::new(), end..end)?;
1268 for entry in entries {
1269 self.messages.push(entry.message.clone());
1270 self.meta.push(meta(entry, entry.made.unwrap_or(at)));
1271 }
1272 self.next_id = next;
1273 }
1274 Change::Splice {
1275 from,
1276 to,
1277 removed,
1278 with,
1279 } => {
1280 let (from, to) = (*from, *to);
1281 let replaced = self.meta.get(from..to).ok_or_else(|| {
1282 mismatch(format!(
1283 "splice of {from}..{to} in {} entries",
1284 self.meta.len()
1285 ))
1286 })?;
1287 if !replaced.iter().map(|meta| &meta.id).eq(removed) {
1288 return Err(mismatch(format!("splice of {from}..{to} names other ids")));
1289 }
1290 let kept: HashMap<MessageId, u64> =
1291 replaced.iter().map(|meta| (meta.id, meta.made)).collect();
1292 let next = self.new_ids(with, &removed.iter().copied().collect(), from..to)?;
1293 let metas: Vec<Meta> = with
1294 .iter()
1295 .map(|entry| {
1296 let made = entry.id.and_then(|id| kept.get(&id).copied());
1297 meta(entry, made.or(entry.made).unwrap_or(at))
1298 })
1299 .collect();
1300 self.messages
1301 .splice(from..to, with.iter().map(|entry| entry.message.clone()));
1302 self.meta.splice(from..to, metas);
1303 self.next_id = next;
1304 }
1305 Change::Halt { halt } => self.halted = *halt,
1306 Change::Aside {
1307 id,
1308 aside: Some(aside),
1309 } => self.place_aside(*id, aside.clone()),
1310 Change::Aside { id, aside: None } => {
1311 let at = self.asides.iter().position(|(known, _)| known == id);
1312 let at = at.ok_or_else(|| mismatch(format!("no aside {}", id.0)))?;
1313 self.asides.remove(at);
1314 }
1315 Change::Reset { state } => {
1318 let (unsaved, revision) = (std::mem::take(&mut self.unsaved), self.revision);
1319 let lineage = self.lineage;
1320 *self = state.view();
1321 self.unsaved = unsaved;
1322 self.revision = revision;
1323 self.lineage = lineage;
1324 }
1325 }
1326 Ok(())
1327 }
1328
1329 fn new_ids(
1334 &self,
1335 entries: &[Entry],
1336 reused: &HashSet<MessageId>,
1337 range: Range<usize>,
1338 ) -> Result<u64, ApplyError> {
1339 let mut next = self.next_id;
1340 let mut seen = HashSet::new();
1341 let mut outside = None;
1342 for entry in entries {
1343 let id = entry
1344 .id
1345 .ok_or_else(|| mismatch("a recorded entry without an id".to_owned()))?;
1346 if id.0 >= next {
1347 next = id.0.saturating_add(1);
1348 continue;
1349 }
1350 let fits = reused.contains(&id)
1351 || (id.0 < self.next_id
1352 && !self.outside(range.clone(), &mut outside).contains(&id));
1353 if !fits || !seen.insert(id) {
1354 return Err(mismatch(format!("message id {} reused", id.0)));
1355 }
1356 }
1357 Ok(next)
1358 }
1359}
1360
1361#[cfg(test)]
1362mod tests;