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

leviath_runtime/pipeline/
mod.rs
