heartbit_core/agent/flow/
mod.rs

1//! Dynamic-workflow combinator core (P1).
2//!
3//! A pure-Rust async combinator layer that reproduces the *semantics* of
4//! Claude Code "dynamic workflows" — [`parallel`] (barrier, fail-soft),
5//! [`pipeline`] (no-barrier per-item streaming), [`phase`]/[`log`]
6//! observability, and a run-wide concurrency cap — on top of the existing
7//! [`AgentRunner`](crate::AgentRunner). A "workflow" is ordinary developer-written
8//! async Rust against these free functions; we do **not** embed a script
9//! engine, IR, or macro DSL.
10//!
11//! These primitives are the deliberate *duals* of the compile-time workflow
12//! agents in [`crate::agent::workflow`]: [`parallel`] is fail-SOFT where
13//! [`ParallelAgent`](crate::ParallelAgent) is fail-FAST, and [`pipeline`] has
14//! NO inter-item barrier where [`SequentialAgent`](crate::SequentialAgent)
15//! advances stage-by-stage. Both layers coexist; nothing existing is changed.
16//!
17//! # Load-bearing invariant
18//!
19//! The `Err -> None` collapse lives **only** in [`parallel`]/[`pipeline`] at the
20//! join/stage boundary, **never** inside [`agent`]. If `agent()` swallowed
21//! errors, the concurrency backstop (and, in later phases, the token budget)
22//! would stop propagating and silently break. Control errors must reach the
23//! caller; only agent-domain failures collapse to `None`.
24
25pub mod agent;
26pub mod budget;
27pub mod ctx;
28pub mod event;
29pub mod journal;
30pub mod parallel;
31pub mod pipeline;
32pub mod progress;
33pub mod worktree;
34
35#[cfg(test)]
36mod audit_gap_tests;
37
38// Re-export the free functions at the `flow` level so callers write
39// `flow::agent(..)`, `flow::parallel(..)`, etc. A module and a same-named
40// function coexist (type vs value namespace), so `flow::agent` is the module in
41// path position and the function in call position.
42pub use agent::agent;
43pub use ctx::{ProviderFactory, WorkflowCtx};
44pub use event::{log, phase};
45pub use parallel::{parallel, thunk};
46pub use pipeline::pipeline;
47
48/// Run `body` as a nested sub-workflow sharing this run's budget, journal,
49/// concurrency cap, runaway backstop, and cancellation. Exactly **one** level of
50/// nesting is allowed: calling `workflow()` from inside a `body` returns
51/// [`Error::Config`](crate::Error::Config).
52///
53/// The child receives a fresh [`WorkflowCtx`] whose default-phase is independent
54/// of the parent's, but whose limits and journal are the *same shared state*, so
55/// a child's spend counts against the parent budget and a child's breach halts
56/// the whole run.
57///
58/// ```ignore
59/// let summary = workflow(&ctx, "research", |child| async move {
60///     let findings = parallel(&child, finders).await;
61///     agent(&child, summarize(findings)).run().await
62/// }).await?;
63/// ```
64pub async fn workflow<R, F, Fut>(
65    ctx: &WorkflowCtx,
66    _name: impl Into<String>,
67    body: F,
68) -> Result<R, crate::error::Error>
69where
70    F: FnOnce(WorkflowCtx) -> Fut,
71    Fut: std::future::Future<Output = Result<R, crate::error::Error>>,
72{
73    let child = ctx.nested()?;
74    body(child).await
75}
76
77#[cfg(test)]
78mod smoke_tests {
79    //! End-to-end smoke test exercising the whole P1 surface through the
80    //! **public crate-root re-exports** (`crate::WorkflowCtx`, `crate::agent`,
81    //! `crate::parallel`, `crate::pipeline`, `crate::phase`), proving the wiring
82    //! in `lib.rs` resolves and the combinators compose.
83
84    use std::sync::Arc;
85
86    use serde_json::Value;
87
88    use crate::agent::test_helpers::MockProvider;
89    use crate::llm::BoxedProvider;
90
91    #[tokio::test]
92    async fn p1_surface_composes_through_public_paths() {
93        // Mock provider with plenty of responses (single + 2 parallel + 2 pipeline).
94        let responses: Vec<_> = (0..6)
95            .map(|i| MockProvider::text_response(&format!("r{i}"), 5, 3))
96            .collect();
97        let provider = Arc::new(BoxedProvider::new(MockProvider::new(responses)));
98        // Root-level type re-export.
99        let ctx = crate::WorkflowCtx::builder(provider)
100            .max_concurrency(4)
101            .build()
102            .expect("build ctx");
103
104        // Free functions via the `crate::flow::` namespace.
105        // phase() returns an RAII guard; agents issued under it adopt the phase.
106        let _phase = crate::flow::phase(&ctx, "smoke");
107
108        // Single agent leaf.
109        let single = crate::flow::agent(&ctx, "solo")
110            .label("solo")
111            .run()
112            .await
113            .expect("single ok");
114        assert!(single.is_some());
115
116        // Fail-soft barrier fan-out over a homogeneous work-list (no boxing).
117        let thunks: Vec<_> = (0..2)
118            .map(|i| {
119                let ctx = ctx.clone();
120                move || async move { crate::flow::agent(&ctx, format!("p{i}")).run().await }
121            })
122            .collect();
123        let fanned = crate::flow::parallel(&ctx, thunks).await;
124        assert_eq!(fanned.iter().filter(|o| o.is_some()).count(), 2);
125
126        // No-barrier pipeline whose stage calls an agent leaf.
127        let piped = crate::flow::pipeline(&ctx, vec![10usize, 20usize])
128            .stage(move |_prev, item, _idx| {
129                let ctx = ctx.clone();
130                async move {
131                    let text = crate::flow::agent(&ctx, format!("item {item}"))
132                        .run()
133                        .await?
134                        .unwrap_or_default();
135                    Ok(Value::String(text))
136                }
137            })
138            .run()
139            .await;
140        assert_eq!(piped.iter().filter(|o| o.is_some()).count(), 2);
141    }
142}
heartbit_core/agent/flow/mod.rs

heartbit_core/agent/flow/
mod.rs
