leviath_runtime/pipeline/
messaging.rs

1//! Inbound message routing and inbox delivery.
2
3use super::*;
4
5/// The receiving end of the world's inbound-message channel. Clients (the
6/// control API) send `AgentMessage`s here; the delivery system routes and
7/// delivers them.
8#[derive(Resource)]
9pub struct MessageIntake(pub UnboundedReceiver<AgentMessage>);
10
11/// Message-delivery system: route inbound messages to their target agents'
12/// inboxes (by agent id), then deliver each inbox into the agent's context
13/// window - but only for agents whose current stage accepts messages; otherwise
14/// the messages wait in the inbox for a stage that does. Ported from
15/// `AgentEngine::process_messages` / `deliver_inbox_messages`.
16pub fn deliver_messages(
17    mut intake: ResMut<MessageIntake>,
18    mut agents: Query<(Entity, &AgentState, &mut MessageInbox, &mut ContextWindow)>,
19) {
20    crate::tick_scope::clear();
21    // Route inbound channel messages to their target agent's inbox.
22    let mut incoming = Vec::new();
23    while let Ok(msg) = intake.0.try_recv() {
24        incoming.push(msg);
25    }
26    for msg in incoming {
27        for (entity, state, mut inbox, _) in agents.iter_mut() {
28            crate::tick_scope::enter(entity);
29            if state.agent_id == msg.agent_id {
30                inbox.push(msg.clone());
31                break;
32            }
33        }
34        // Unmatched target ⇒ dropped (agent no longer exists).
35    }
36
37    // Deliver inboxes into context windows for agents that accept messages.
38    for (entity, state, mut inbox, mut window) in agents.iter_mut() {
39        crate::tick_scope::enter(entity);
40        if !state.accepts_messages {
41            continue; // hold until a stage that accepts messages
42        }
43        for msg in inbox.drain_all() {
44            let region = msg.target_region.as_deref().unwrap_or("conversation");
45            let tokens = leviath_core::estimate_tokens(&msg.content);
46            let _ = window.add_typed_entry(
47                region,
48                leviath_core::EntryKind::UserMessage,
49                msg.content.clone(),
50                tokens,
51            );
52        }
53    }
54}
leviath_runtime/pipeline/messaging.rs

leviath_runtime/pipeline/
messaging.rs
