1use bevy_ecs::prelude::*;
14use bevy_reflect::Reflect;
15use rig_core::message::{self, AssistantContent, AssistantMessage, ToolResultContent, UserContent};
16use serde::{Deserialize, Serialize};
17use std::collections::BTreeMap;
18
19use super::binary::{BinaryAssets, BinaryError, PartSource};
20use crate::agent::{MessageParts, Role, Utterance};
21
22#[derive(Component, Debug, Clone, PartialEq, Serialize, Deserialize, Reflect)]
25#[reflect(Component)]
26pub enum ContentPart {
27 Text(#[reflect(remote = super::reflect::TextPartReflect)] message::Text),
29 Image(ImagePart),
31 Audio(AudioPart),
33 Video(VideoPart),
35 Document(DocumentPart),
37 ToolCall(#[reflect(remote = super::reflect::ToolCallPartReflect)] message::ToolCall),
39 Reasoning(#[reflect(remote = super::reflect::ReasoningPartReflect)] message::Reasoning),
41 Opaque(#[reflect(remote = super::reflect::OpaquePartReflect)] message::Opaque),
43 ToolResult {
45 #[reflect(remote = crate::agent::reflect::ToolCallIdReflect)]
47 call: message::CallId,
48 #[reflect(remote = crate::agent::reflect::ToolNameReflect)]
50 name: message::ToolName,
51 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
53 is_error: bool,
54 },
55 Json(#[reflect(remote = super::reflect::JsonPartReflect)] serde_json::Value),
57}
58
59#[derive(Component, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Reflect)]
63#[reflect(Component)]
64pub enum RequestPartEdit {
65 Text(String),
67 Remove,
69}
70
71#[derive(Component, Debug, Clone, Copy, PartialEq, Eq, Reflect)]
73#[relationship(relationship_target = EditedBy)]
74#[reflect(Component)]
75pub struct EditTarget(pub Entity);
76
77#[derive(Component, Debug, Default, Reflect)]
79#[relationship_target(relationship = EditTarget)]
80#[reflect(Component)]
81pub struct EditedBy(Vec<Entity>);
82
83#[derive(Component, Debug, Clone, Default, PartialEq, Serialize, Deserialize, Reflect)]
86#[reflect(Component)]
87pub struct AssistantHead {
88 #[reflect(remote = super::reflect::OriginReflect)]
90 pub origin: Option<message::Origin>,
91 #[reflect(remote = super::reflect::StopReflect)]
93 pub stop: Option<message::StopReason>,
94}
95
96impl AssistantHead {
97 pub fn of(message: &AssistantMessage) -> Self {
99 Self {
100 origin: message.origin.clone(),
101 stop: message.stop.clone(),
102 }
103 }
104
105 pub fn message(&self, content: Vec<AssistantContent>) -> AssistantMessage {
107 AssistantMessage::default()
108 .with_origin(self.origin.clone())
109 .with_stop(self.stop.clone())
110 .with_content(content)
111 }
112}
113
114#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Reflect)]
116pub struct ImagePart {
117 pub source: PartSource,
119 #[reflect(remote = super::reflect::ImageMediaReflect)]
121 pub media_type: Option<message::ImageMediaType>,
122 #[reflect(remote = super::reflect::NativeReflect)]
124 pub native: Option<message::Native>,
125 #[reflect(remote = super::reflect::ImageDetailReflect)]
127 pub detail: Option<message::ImageDetail>,
128}
129
130#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Reflect)]
132pub struct AudioPart {
133 pub source: PartSource,
135 #[reflect(remote = super::reflect::AudioMediaReflect)]
137 pub media_type: Option<message::AudioMediaType>,
138}
139
140#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Reflect)]
142pub struct VideoPart {
143 pub source: PartSource,
145 #[reflect(remote = super::reflect::VideoMediaReflect)]
147 pub media_type: Option<message::VideoMediaType>,
148 #[reflect(remote = super::reflect::PartParamsReflect)]
150 pub additional_params: Option<serde_json::Value>,
151}
152
153#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Reflect)]
155pub struct DocumentPart {
156 pub source: PartSource,
158 #[reflect(remote = super::reflect::DocumentMediaReflect)]
160 pub media_type: Option<message::DocumentMediaType>,
161 #[reflect(remote = super::reflect::PartParamsReflect)]
163 pub additional_params: Option<serde_json::Value>,
164}
165
166#[derive(Component, Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Reflect)]
169#[reflect(Component)]
170#[serde(rename_all = "snake_case")]
171pub enum ToolResultStatus {
172 Ok,
174 Error,
178 Refused,
180 Skipped,
182 Denied,
184 WrongFamily,
186}
187
188pub const TOOL_RESULT_LIMIT_MARKER: &str = "\n[… {omitted} bytes omitted …]\n";
191
192#[derive(Component, Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Reflect)]
198#[reflect(Component)]
199pub struct ToolResultLimit {
200 pub max_bytes: usize,
202 pub marker: String,
204}
205
206impl ToolResultLimit {
207 #[must_use]
209 pub fn new(max_bytes: usize) -> Self {
210 Self {
211 max_bytes,
212 marker: TOOL_RESULT_LIMIT_MARKER.to_owned(),
213 }
214 }
215}
216
217#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
219pub enum ContentError {
220 #[error(transparent)]
222 Binary(#[from] BinaryError),
223 #[error("content graph is missing an entity or required component")]
225 Missing,
226 #[error("content part has an invalid type or role")]
228 Shape,
229}
230
231type PreparedParts = Vec<(ContentPart, Vec<ContentPart>)>;
234
235fn image(assets: &mut BinaryAssets, value: message::Image) -> Result<ImagePart, ContentError> {
236 Ok(ImagePart {
237 source: assets.intern(value.data)?,
238 media_type: value.media_type,
239 detail: value.detail,
240 native: value.native,
241 })
242}
243
244fn user(
245 assets: &mut BinaryAssets,
246 value: UserContent,
247) -> Result<(ContentPart, Vec<ContentPart>), ContentError> {
248 let mut children = Vec::new();
249 let part = match value {
250 UserContent::Text(value) => ContentPart::Text(value),
251 UserContent::Image(value) => ContentPart::Image(image(assets, value)?),
252 UserContent::Audio(value) => ContentPart::Audio(AudioPart {
253 source: assets.intern(value.data)?,
254 media_type: value.media_type,
255 }),
256 UserContent::Video(value) => ContentPart::Video(VideoPart {
257 source: assets.intern(value.data)?,
258 media_type: value.media_type,
259 additional_params: value.additional_params,
260 }),
261 UserContent::Document(value) => ContentPart::Document(DocumentPart {
262 source: assets.intern(value.data)?,
263 media_type: value.media_type,
264 additional_params: value.additional_params,
265 }),
266 UserContent::ToolResult(value) => {
267 children = value
268 .content
269 .into_iter()
270 .map(|item| {
271 Ok(match item {
272 ToolResultContent::Text(text) => ContentPart::Text(text),
273 ToolResultContent::Image(value) => {
274 ContentPart::Image(image(assets, value)?)
275 }
276 ToolResultContent::Json { value } => ContentPart::Json(value),
277 })
278 })
279 .collect::<Result<_, ContentError>>()?;
280 ContentPart::ToolResult {
281 call: value.call,
282 name: value.name,
283 is_error: value.is_error,
284 }
285 }
286 };
287 Ok((part, children))
288}
289
290fn prepare(
291 assets: &mut BinaryAssets,
292 parts: MessageParts,
293) -> Result<(Role, Option<AssistantHead>, PreparedParts), ContentError> {
294 Ok(match parts {
295 MessageParts::User { content } => (
296 Role::User,
297 None,
298 content
299 .into_iter()
300 .map(|part| user(assets, part))
301 .collect::<Result<_, _>>()?,
302 ),
303 MessageParts::Assistant(message) => (
304 Role::Assistant,
305 Some(AssistantHead::of(&message)),
306 message
307 .content
308 .into_iter()
309 .map(|part| {
310 let part = match part {
311 AssistantContent::Text(value) => ContentPart::Text(value),
312 AssistantContent::Image(value) => ContentPart::Image(image(assets, value)?),
313 AssistantContent::ToolCall(value) => ContentPart::ToolCall(value),
314 AssistantContent::Reasoning(value) => ContentPart::Reasoning(value),
315 AssistantContent::Opaque(value) => ContentPart::Opaque(value),
316 _ => return Err(ContentError::Shape),
317 };
318 Ok((part, Vec::new()))
319 })
320 .collect::<Result<_, ContentError>>()?,
321 ),
322 })
323}
324
325fn spawn_parts(world: &mut World, parent: Entity, parts: PreparedParts) {
326 for (part, children) in parts {
327 let entity = world.spawn(ChildOf(parent)).insert(part).id();
329 for child in children {
330 world.spawn(ChildOf(entity)).insert(child);
331 }
332 }
333}
334
335pub fn write_message(
342 world: &mut World,
343 utterance: Entity,
344 parts: MessageParts,
345) -> Result<(), ContentError> {
346 if world.get::<Utterance>(utterance).is_none() {
347 return Err(ContentError::Missing);
348 }
349 world.init_resource::<BinaryAssets>();
350 let (role, head, parts) = prepare(&mut world.resource_mut::<BinaryAssets>(), parts)?;
351 replace_parts(world, utterance, role, head, parts);
352 Ok(())
353}
354
355fn replace_parts(
356 world: &mut World,
357 utterance: Entity,
358 role: Role,
359 head: Option<AssistantHead>,
360 parts: PreparedParts,
361) {
362 let old: Vec<_> = world
363 .get::<Children>(utterance)
364 .map(|children| children.iter().collect())
365 .unwrap_or_default();
366 for child in old {
367 world.despawn(child);
368 }
369 let mut entity = world.entity_mut(utterance);
370 entity.insert(role).remove::<AssistantHead>();
371 if let Some(head) = head {
372 entity.insert(head);
373 }
374 spawn_parts(world, utterance, parts);
375}
376
377fn ordered<'a>(
378 get: &impl Fn(Entity) -> Option<EntityRef<'a>>,
379 parent: Entity,
380) -> Result<Vec<Entity>, ContentError> {
381 let parent = get(parent).ok_or(ContentError::Missing)?;
382 let children = parent.get::<Children>();
383 let mut ordered = Vec::new();
384 for child in children.into_iter().flat_map(|c| c.iter()) {
385 let entity = get(child).ok_or(ContentError::Missing)?;
386 if entity.get::<ContentPart>().is_none() {
387 return Err(ContentError::Shape);
388 }
389 ordered.push(child);
390 }
391 Ok(ordered)
392}
393
394fn read_image(assets: &BinaryAssets, value: &ImagePart) -> Result<message::Image, ContentError> {
395 Ok(message::Image {
396 data: assets.resolve(&value.source)?,
397 media_type: value.media_type.clone(),
398 detail: value.detail.clone(),
399 native: value.native.clone(),
400 })
401}
402
403fn read_edited_part<'a>(
404 get: &impl Fn(Entity) -> Option<EntityRef<'a>>,
405 entity: Entity,
406 nested: bool,
407 edits: &BTreeMap<Entity, RequestPartEdit>,
408) -> Result<Option<ContentPart>, ContentError> {
409 let entity_ref = get(entity).ok_or(ContentError::Missing)?;
410 let mut value = entity_ref
411 .get::<ContentPart>()
412 .ok_or(ContentError::Shape)?
413 .clone();
414 if matches!(value, ContentPart::ToolResult { .. }) {
415 if nested {
416 return Err(ContentError::Shape);
417 }
418 } else if entity_ref
419 .get::<Children>()
420 .is_some_and(|children| !children.is_empty())
421 {
422 return Err(ContentError::Shape);
423 }
424 match edits.get(&entity) {
425 Some(RequestPartEdit::Remove) => return Ok(None),
426 Some(RequestPartEdit::Text(text)) => match &mut value {
427 ContentPart::Text(part) => part.text.clone_from(text),
428 _ => return Err(ContentError::Shape),
429 },
430 None => {}
431 }
432 Ok(Some(value))
433}
434
435fn to_user<'a>(
436 get: &impl Fn(Entity) -> Option<EntityRef<'a>>,
437 assets: &BinaryAssets,
438 entity: Entity,
439 value: ContentPart,
440 edits: &BTreeMap<Entity, RequestPartEdit>,
441) -> Result<UserContent, ContentError> {
442 Ok(match value {
443 ContentPart::Text(value) => UserContent::Text(value),
444 ContentPart::Image(value) => UserContent::Image(read_image(assets, &value)?),
445 ContentPart::Audio(value) => UserContent::Audio(message::Audio {
446 data: assets.resolve(&value.source)?,
447 media_type: value.media_type,
448 }),
449 ContentPart::Video(value) => UserContent::Video(message::Video {
450 data: assets.resolve(&value.source)?,
451 media_type: value.media_type,
452 additional_params: value.additional_params,
453 }),
454 ContentPart::Document(value) => UserContent::Document(message::Document {
455 data: assets.resolve(&value.source)?,
456 media_type: value.media_type,
457 additional_params: value.additional_params,
458 }),
459 ContentPart::ToolResult {
460 call,
461 name,
462 is_error,
463 } => UserContent::ToolResult(message::ToolResult {
464 call,
465 name,
466 is_error,
467 content: ordered(get, entity)?
468 .into_iter()
469 .map(|child| read_edited_part(get, child, true, edits))
470 .collect::<Result<Vec<_>, _>>()?
471 .into_iter()
472 .flatten()
473 .map(|part| {
474 Ok(match part {
475 ContentPart::Text(value) => ToolResultContent::Text(value),
476 ContentPart::Image(value) => {
477 ToolResultContent::Image(read_image(assets, &value)?)
478 }
479 ContentPart::Json(value) => ToolResultContent::Json { value },
480 _ => return Err(ContentError::Shape),
481 })
482 })
483 .collect::<Result<Vec<_>, ContentError>>()
484 .and_then(non_empty)?,
485 }),
486 _ => return Err(ContentError::Shape),
487 })
488}
489
490pub fn read_message(world: &World, utterance: Entity) -> Result<MessageParts, ContentError> {
493 let empty = BinaryAssets::default();
494 let assets = world.get_resource::<BinaryAssets>().unwrap_or(&empty);
495 read_message_from(
496 &|entity| world.get_entity(entity).ok(),
497 assets,
498 utterance,
499 &BTreeMap::new(),
500 )
501}
502
503#[derive(bevy_ecs::system::SystemParam)]
506pub struct ContentGraph<'w, 's> {
507 entities: Query<'w, 's, EntityRef<'static>, Without<bevy_ecs::resource::IsResource>>,
508 assets: Option<Res<'w, BinaryAssets>>,
509}
510
511impl ContentGraph<'_, '_> {
512 pub fn message_with(
515 &self,
516 utterance: Entity,
517 edits: &BTreeMap<Entity, RequestPartEdit>,
518 ) -> Result<MessageParts, ContentError> {
519 if edits.is_empty() {
520 return self.message(utterance);
521 }
522 self.message(utterance)?;
524 let empty = BinaryAssets::default();
525 read_message_from(
526 &|entity| self.entities.get(entity).ok(),
527 self.assets.as_deref().unwrap_or(&empty),
528 utterance,
529 edits,
530 )
531 }
532
533 pub fn target_utterance(&self, target: Entity) -> Result<Entity, ContentError> {
536 let part = self
537 .entities
538 .get(target)
539 .map_err(|_| ContentError::Missing)?;
540 if !part.contains::<ContentPart>() {
541 return Err(ContentError::Shape);
542 }
543 let parent = part.get::<ChildOf>().ok_or(ContentError::Missing)?.parent();
544 let owner = self
545 .entities
546 .get(parent)
547 .map_err(|_| ContentError::Missing)?;
548 if owner.contains::<Utterance>() {
549 return Ok(parent);
550 }
551 if !matches!(
552 owner.get::<ContentPart>(),
553 Some(ContentPart::ToolResult { .. })
554 ) {
555 return Err(ContentError::Shape);
556 }
557 let utterance = owner
558 .get::<ChildOf>()
559 .ok_or(ContentError::Missing)?
560 .parent();
561 if !self
562 .entities
563 .get(utterance)
564 .map_err(|_| ContentError::Missing)?
565 .contains::<Utterance>()
566 {
567 return Err(ContentError::Shape);
568 }
569 Ok(utterance)
570 }
571
572 pub fn message(&self, utterance: Entity) -> Result<MessageParts, ContentError> {
575 let empty = BinaryAssets::default();
576 let assets = self.assets.as_deref().unwrap_or(&empty);
577 read_message_from(
578 &|entity| self.entities.get(entity).ok(),
579 assets,
580 utterance,
581 &BTreeMap::new(),
582 )
583 }
584}
585
586fn read_message_from<'a>(
587 get: &impl Fn(Entity) -> Option<EntityRef<'a>>,
588 assets: &BinaryAssets,
589 utterance: Entity,
590 edits: &BTreeMap<Entity, RequestPartEdit>,
591) -> Result<MessageParts, ContentError> {
592 let entity = get(utterance).ok_or(ContentError::Missing)?;
593 if entity.get::<Utterance>().is_none() {
594 return Err(ContentError::Missing);
595 }
596 let role = entity.get::<Role>().ok_or(ContentError::Missing)?;
597 let values = ordered(get, utterance)?
598 .into_iter()
599 .map(|entity| {
600 read_edited_part(get, entity, false, edits).map(|part| part.map(|part| (entity, part)))
601 })
602 .collect::<Result<Vec<_>, _>>()?
603 .into_iter()
604 .flatten()
605 .collect::<Vec<_>>();
606 match role {
607 Role::User => {
608 if entity.get::<AssistantHead>().is_some() {
609 return Err(ContentError::Shape);
610 }
611 MessageParts::user(
612 values
613 .into_iter()
614 .map(|(entity, value)| to_user(get, assets, entity, value, edits))
615 .collect::<Result<_, _>>()?,
616 )
617 }
618 Role::Assistant => MessageParts::assistant(
619 entity
620 .get::<AssistantHead>()
621 .ok_or(ContentError::Missing)?
622 .message(
623 values
624 .into_iter()
625 .map(|(_, value)| {
626 Ok(match value {
627 ContentPart::Text(value) => AssistantContent::Text(value),
628 ContentPart::Image(value) => {
629 AssistantContent::Image(read_image(assets, &value)?)
630 }
631 ContentPart::ToolCall(value) => AssistantContent::ToolCall(value),
632 ContentPart::Reasoning(value) => AssistantContent::Reasoning(value),
633 ContentPart::Opaque(value) => AssistantContent::Opaque(value),
634 _ => return Err(ContentError::Shape),
635 })
636 })
637 .collect::<Result<_, ContentError>>()?,
638 ),
639 ),
640 }
641}
642
643fn non_empty<T>(items: Vec<T>) -> Result<Vec<T>, ContentError> {
645 if items.is_empty() {
646 return Err(ContentError::Shape);
647 }
648 Ok(items)
649}
650
651pub fn collect_binary_assets(
655 world: &mut World,
656 pins: impl IntoIterator<Item = super::binary::BinaryId>,
657) -> Result<(), BinaryError> {
658 let mut roots: Vec<_> = pins.into_iter().collect();
659 for part in world.query::<&ContentPart>().iter(world) {
660 let source = match part {
661 ContentPart::Image(part) => &part.source,
662 ContentPart::Audio(part) => &part.source,
663 ContentPart::Video(part) => &part.source,
664 ContentPart::Document(part) => &part.source,
665 _ => continue,
666 };
667 if let PartSource::Binary { id, .. } = source {
668 roots.push(*id);
669 }
670 }
671 match world.get_resource_mut::<BinaryAssets>() {
672 Some(mut assets) => assets.retain(roots),
673 None if roots.is_empty() => Ok(()),
674 None => Err(BinaryError::Missing),
675 }
676}
677
678pub(crate) fn spawn_deferred(
681 commands: &mut Commands,
682 assets: &mut BinaryAssets,
683 parent: Entity,
684 message: MessageParts,
685) -> Result<Entity, ContentError> {
686 spawn_deferred_with(commands, assets, parent, message, Vec::new())
687}
688
689pub(crate) fn spawn_deferred_with(
692 commands: &mut Commands,
693 assets: &mut BinaryAssets,
694 parent: Entity,
695 message: MessageParts,
696 statuses: Vec<ToolResultStatus>,
697) -> Result<Entity, ContentError> {
698 let (role, head, parts) = prepare(assets, message)?;
699 let mut utterance = commands.spawn((Utterance, role, ChildOf(parent)));
700 if let Some(head) = head {
701 utterance.insert(head);
702 }
703 let entity = utterance.id();
704 commands.queue(move |world: &mut World| {
705 spawn_parts(world, entity, parts);
706 stamp_statuses(world, entity, &statuses);
707 });
708 Ok(entity)
709}
710
711fn stamp_statuses(world: &mut World, utterance: Entity, statuses: &[ToolResultStatus]) {
712 if statuses.is_empty() {
713 return;
714 }
715 let results: Vec<Entity> = world
716 .get::<Children>(utterance)
717 .into_iter()
718 .flat_map(|children| children.iter())
719 .filter(|child| {
720 matches!(
721 world.get::<ContentPart>(*child),
722 Some(ContentPart::ToolResult { .. })
723 )
724 })
725 .collect();
726 for (entity, status) in results.into_iter().zip(statuses) {
727 world.entity_mut(entity).insert(*status);
728 }
729}
730
731pub(crate) fn replace_deferred(
732 commands: &mut Commands,
733 assets: &mut BinaryAssets,
734 utterance: Entity,
735 message: MessageParts,
736) -> Result<(), ContentError> {
737 let (role, head, parts) = prepare(assets, message)?;
738 commands.queue(move |world: &mut World| {
739 if world.get::<Utterance>(utterance).is_none() {
740 return;
741 }
742 replace_parts(world, utterance, role, head, parts);
743 });
744 Ok(())
745}