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, PersistMsg};
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::*;
53mod resolve;
54pub use resolve::*;
55mod stall;
56pub use stall::*;
57mod wedge;
58pub use wedge::*;
59mod circuit;
60pub use circuit::*;
61
62// ─── Phase marker components (an agent is in exactly one) ────────────────────
63//
64// A marker's presence is a claim that some system has this agent queued, which
65// is what keeps it reachable. Anything new here must also be added to
66// [`Unreachable`], or the wedge watchdog will read an agent resting on it as one
67// nothing can drive.
68
69/// The agent is active and ready to build a request and (permits allowing)
70/// dispatch inference.
71#[derive(Component, Debug, Clone, Copy, PartialEq, Eq)]
72pub struct ReadyToInfer;
73
74/// Inference has been dispatched to the pool; the agent is waiting for its
75/// result (which the inference-collect system will apply).
76#[derive(Component, Debug, Clone, Copy, PartialEq, Eq)]
77pub struct AwaitingInference;
78
79/// Transient tag: the agent just entered a stage (index + name). The
80/// [`sync_tool_stages`] system reads it to notify the [`ToolService`] of the
81/// stage change, then removes it. Carries the data so the tool service need not
82/// query the world.
83#[derive(Component, Debug, Clone)]
84pub struct StageJustEntered {
85    /// The new stage's index.
86    pub index: usize,
87    /// The new stage's name.
88    pub name: String,
89}
90
91// ─── Per-agent stage data the dispatch system reads ──────────────────────────
92
93/// Resolved inference parameters for the agent's current stage, set when it
94/// enters that stage. Pure data - the dispatch system reads it to build the
95/// request.
96#[derive(Component, Debug, Clone)]
97pub struct StageInference {
98    /// Registered provider to call.
99    pub provider_name: String,
100    /// Model id (also the key into the per-model inference pools).
101    pub model: String,
102    /// Tools advertised at this stage.
103    pub tools: Vec<Tool>,
104    /// Optional allow-list of tool names (`None`/empty = all `tools`).
105    pub tool_filter: Option<Vec<String>>,
106    /// Providers to fail over to, best first, when the current one turns out
107    /// to be unusable. Consumed from the front by `collect_inference`, so an
108    /// exhausted list means "nowhere left to go" (issue #201).
109    pub fallbacks: Vec<leviath_core::blueprint::ModelEntry>,
110}
111
112// ─── World resources for the inference stage ─────────────────────────────────
113
114/// The registered providers, as a world resource.
115#[derive(Resource)]
116pub struct Providers(pub ProviderRegistry);
117
118/// The plumbing the inference-dispatch system needs: the per-model pools, the
119/// channel to report outcomes on, the tick wake handle, and a runtime handle to
120/// spawn the (bounded, per-request) worker tasks onto.
121#[derive(Resource, Clone)]
122pub struct InferenceStage {
123    /// Per-model concurrency pools.
124    pub pools: Arc<InferencePools>,
125    /// Where completed inferences are reported.
126    pub outcomes: UnboundedSender<InferenceOutcome>,
127    /// Where completed *transition-choice* inferences are reported (a separate
128    /// lane so the collect systems don't confuse a routing decision with a normal
129    /// agent turn).
130    pub transition_outcomes: UnboundedSender<InferenceOutcome>,
131    /// Where completed *compaction* jobs (LLM context summarization) are
132    /// reported - again a separate lane so a summary isn't mistaken for a turn.
133    pub compaction_outcomes: UnboundedSender<crate::compaction_bridge::CompactionOutcome>,
134    /// Where completed *content-summary transform* jobs are reported (the
135    /// Summarize context-transform lane - see `context_transform`).
136    pub content_summary_outcomes: UnboundedSender<crate::compaction_bridge::CompactionOutcome>,
137    /// Signalled when an inference completes, to wake the tick loop.
138    pub wake: Arc<Notify>,
139    /// Runtime the worker tasks are spawned onto.
140    pub runtime: Handle,
141    /// Opt-in: perform an exact pre-inference token count and reject requests
142    /// that would overflow the model's context window (see
143    /// `InferenceJob::exact_token_counting`). Off by default.
144    pub exact_token_counting: bool,
145}
146
147/// Truncate `text` to at most `max_chars` characters, never splitting a
148/// multi-byte UTF-8 char. `max_chars` is an approximate char budget the caller
149/// derives from a token estimate.
150fn truncate_on_char_boundary(text: &str, max_chars: usize) -> String {
151    text.chars().take(max_chars).collect()
152}
153
154#[cfg(test)]
155mod tests;
leviath_runtime/pipeline/mod.rs

leviath_runtime/pipeline/
mod.rs
