leviath_runtime/pipeline/
mod.rs

1//! The ECS pipeline (Phase 2): components + systems that drive every agent
2//! through check-input → infer → tools → apply → repeat, entirely as data.
3//!
4//! Agents are entities; their execution phase is a **marker component**
5//! (`ReadyToInfer`, `AwaitingInference`, …) so systems can query by phase. A
6//! system never blocks on I/O: the dispatch systems hand work to the async
7//! bridges (`inference_bridge`, [`crate::tool_bridge`]) and the collect
8//! systems apply the results on a later tick. This module is built alongside the
9//! existing imperative engine; the two are unified in a later phase.
10
11use std::sync::Arc;
12
13use bevy_ecs::prelude::*;
14use tokio::runtime::Handle;
15use tokio::sync::Notify;
16use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
17
18use leviath_providers::{InferenceRequest, Provider, Tool};
19
20use crate::compaction_bridge::{CompactionJob, CompactionOutcome, run_compaction_job};
21use crate::components::{
22    AgentMessage, AgentState, AgentStatus, AwaitingInteraction, ContextWindow, InferenceConfig,
23    MessageInbox,
24};
25use crate::fanout::FanOutWaiting;
26use crate::inference_bridge::{InferenceJob, InferenceOutcome, run_inference_job};
27use crate::inference_pool::InferencePools;
28use crate::interaction_hub::InteractionHub;
29use crate::persistence::{RunMetadata, TokenTotals, build_context_snapshot, build_run_meta};
30use crate::persistence_bridge::PersistJob;
31use crate::providers::ProviderRegistry;
32use crate::tool_bridge::{BoxedToolExec, ToolJob, ToolOutcome};
33
34// Sections of the former single-file pipeline, one per concern.
35mod transition;
36pub use transition::*;
37mod messaging;
38pub use messaging::*;
39mod persist;
40pub use persist::*;
41mod compaction;
42pub use compaction::*;
43mod tool_results;
44pub use tool_results::*;
45mod gate;
46pub use gate::*;
47mod tools;
48pub use tools::*;
49mod response;
50pub use response::*;
51mod inference;
52pub use inference::*;
53
54// ─── Phase marker components (an agent is in exactly one) ────────────────────
55
56/// The agent is active and ready to build a request and (permits allowing)
57/// dispatch inference.
58#[derive(Component, Debug, Clone, Copy, PartialEq, Eq)]
59pub struct ReadyToInfer;
60
61/// Inference has been dispatched to the pool; the agent is waiting for its
62/// result (which the inference-collect system will apply).
63#[derive(Component, Debug, Clone, Copy, PartialEq, Eq)]
64pub struct AwaitingInference;
65
66/// Transient tag: the agent just entered a stage (index + name). The
67/// [`sync_tool_stages`] system reads it to notify the [`ToolService`] of the
68/// stage change, then removes it. Carries the data so the tool service need not
69/// query the world.
70#[derive(Component, Debug, Clone)]
71pub struct StageJustEntered {
72    /// The new stage's index.
73    pub index: usize,
74    /// The new stage's name.
75    pub name: String,
76}
77
78// ─── Per-agent stage data the dispatch system reads ──────────────────────────
79
80/// Resolved inference parameters for the agent's current stage, set when it
81/// enters that stage. Pure data - the dispatch system reads it to build the
82/// request.
83#[derive(Component, Debug, Clone)]
84pub struct StageInference {
85    /// Registered provider to call.
86    pub provider_name: String,
87    /// Model id (also the key into the per-model inference pools).
88    pub model: String,
89    /// Tools advertised at this stage.
90    pub tools: Vec<Tool>,
91    /// Optional allow-list of tool names (`None`/empty = all `tools`).
92    pub tool_filter: Option<Vec<String>>,
93}
94
95// ─── World resources for the inference stage ─────────────────────────────────
96
97/// The registered providers, as a world resource.
98#[derive(Resource)]
99pub struct Providers(pub ProviderRegistry);
100
101/// The plumbing the inference-dispatch system needs: the per-model pools, the
102/// channel to report outcomes on, the tick wake handle, and a runtime handle to
103/// spawn the (bounded, per-request) worker tasks onto.
104#[derive(Resource, Clone)]
105pub struct InferenceStage {
106    /// Per-model concurrency pools.
107    pub pools: Arc<InferencePools>,
108    /// Where completed inferences are reported.
109    pub outcomes: UnboundedSender<InferenceOutcome>,
110    /// Where completed *transition-choice* inferences are reported (a separate
111    /// lane so the collect systems don't confuse a routing decision with a normal
112    /// agent turn).
113    pub transition_outcomes: UnboundedSender<InferenceOutcome>,
114    /// Where completed *compaction* jobs (LLM context summarization) are
115    /// reported - again a separate lane so a summary isn't mistaken for a turn.
116    pub compaction_outcomes: UnboundedSender<crate::compaction_bridge::CompactionOutcome>,
117    /// Where completed *content-summary transform* jobs are reported (the
118    /// Summarize context-transform lane - see `context_transform`).
119    pub content_summary_outcomes: UnboundedSender<crate::compaction_bridge::CompactionOutcome>,
120    /// Signalled when an inference completes, to wake the tick loop.
121    pub wake: Arc<Notify>,
122    /// Runtime the worker tasks are spawned onto.
123    pub runtime: Handle,
124    /// Opt-in: perform an exact pre-inference token count and reject requests
125    /// that would overflow the model's context window (see
126    /// `InferenceJob::exact_token_counting`). Off by default.
127    pub exact_token_counting: bool,
128}
129
130/// Truncate `text` to at most `max_chars` characters, never splitting a
131/// multi-byte UTF-8 char. `max_chars` is an approximate char budget the caller
132/// derives from a token estimate.
133fn truncate_on_char_boundary(text: &str, max_chars: usize) -> String {
134    text.chars().take(max_chars).collect()
135}
136
137#[cfg(test)]
138mod tests;
leviath_runtime/pipeline/mod.rs

leviath_runtime/pipeline/
mod.rs
