xz_agent_core/

engine.rs

//! Core engine loop implementation.
//!
//! Provides the generic [`CoreEngine<Ctx, Out>`](CoreEngine) struct with a
//! [`run`](CoreEngine::run) method that executes the agent loop,
//! along with an [`EngineBuilder<Ctx, Out>`](EngineBuilder) for fluent construction.
//!
//! # Architecture
//!
//! The engine orchestrates the agent data-flow pipeline:
//!
//! ```text
//! Ctx → [ContextBuilder chain] → Thinker → [OutputProcessor chain] → Signal
//! ```
//!
//! Each iteration:
//! 1. Context flows through a chain of context builders (each can modify it)
//! 2. The thinker produces output from the context
//! 3. Output flows through a chain of processors (any can signal Stop)
//!
//! # Type Erasure
//!
//! Since Rust's native async functions in traits are not object-safe,
//! this module provides type-erased wrapper types (`DynThinker`, `DynContextBuilder`,
//! `DynOutputProcessor`) that allow heterogeneous chains of different concrete
//! types via boxing. Each wrapper struct implements the corresponding trait
//! by dispatching to an internal erased trait object.
//!
//! # Example
//!
//! ```rust
//! use xz_agent_core::engine::{CoreEngine, EngineBuilder};
//! use xz_agent_core::traits::thinker::Thinker;
//! use xz_agent_core::traits::context::ContextBuilder;
//! use xz_agent_core::traits::processor::OutputProcessor;
//! use xz_agent_core::error::EngineError;
//! use xz_agent_core::types::signal::Signal;
//!
//! // A simple thinker that echoes input.
//! struct EchoThinker;
//! impl Thinker for EchoThinker {
//!     type Context = String;
//!     type Output = String;
//!     async fn think(&self, ctx: &Self::Context) -> Result<Self::Output, EngineError> {
//!         Ok(ctx.clone())
//!     }
//! }
//!
//! // A processor that stops after the first turn.
//! struct OneShot;
//! impl OutputProcessor for OneShot {
//!     type Context = String;
//!     type Output = String;
//!     async fn process(
//!         &self,
//!         _output: &Self::Output,
//!         _ctx: &mut Self::Context,
//!     ) -> Result<Signal, EngineError> {
//!         Ok(Signal::Stop)
//!     }
//! }
//!
//! let engine: CoreEngine<String, String> = EngineBuilder::new()
//!     .thinker(EchoThinker)
//!     .processor(OneShot)
//!     .build()
//!     .unwrap();
//! ```

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use tokio_util::sync::CancellationToken;

use crate::error::EngineError;
use crate::traits::context::ContextBuilder;
use crate::traits::processor::OutputProcessor;
use crate::traits::thinker::Thinker;
use crate::types::signal::Signal;

// ─────────────────────────────────────────────────────────────────────────────
// Type-erased wrapper traits
// ─────────────────────────────────────────────────────────────────────────────

/// Erased form of [`Thinker`] — boxes the async output.
///
/// Generic over the context type `Ctx` and output type `Out` so that
/// different concrete `Thinker` implementations with the same associated
/// types can be stored heterogeneously.
trait ErasedThinker<Ctx, Out>: Send + Sync {
    /// Type-erased [`Thinker::think`].
    fn think_erased<'a>(
        &'a self,
        ctx: &'a Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Out, EngineError>> + Send + 'a>>;
}

/// Erased form of [`ContextBuilder`] — boxes the async output.
///
/// Generic over the context type `Ctx` so that different concrete
/// `ContextBuilder` implementations can be chained dynamically.
trait ErasedContextBuilder<Ctx>: Send + Sync {
    /// Type-erased [`ContextBuilder::build`].
    fn build_erased<'a>(
        &'a self,
        ctx: Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Ctx, EngineError>> + Send + 'a>>;
}

/// Erased form of [`OutputProcessor`] — boxes the async output.
///
/// Generic over the context type `Ctx` and output type `Out` so that
/// different concrete `OutputProcessor` implementations can be chained.
trait ErasedOutputProcessor<Ctx, Out>: Send + Sync {
    /// Type-erased [`OutputProcessor::process`].
    fn process_erased<'a>(
        &'a self,
        output: &'a Out,
        ctx: &'a mut Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Signal, EngineError>> + Send + 'a>>;
}

// ── Blanket implementations ──

impl<T, Ctx, Out> ErasedThinker<Ctx, Out> for T
where
    T: Thinker<Context = Ctx, Output = Out> + 'static,
    Ctx: Send + Sync,
    Out: Send,
{
    fn think_erased<'a>(
        &'a self,
        ctx: &'a Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Out, EngineError>> + Send + 'a>> {
        Box::pin(T::think(self, ctx))
    }
}

impl<T, Ctx> ErasedContextBuilder<Ctx> for T
where
    T: ContextBuilder<Context = Ctx> + 'static,
    Ctx: Send,
{
    fn build_erased<'a>(
        &'a self,
        ctx: Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Ctx, EngineError>> + Send + 'a>> {
        Box::pin(T::build(self, ctx))
    }
}

impl<T, Ctx, Out> ErasedOutputProcessor<Ctx, Out> for T
where
    T: OutputProcessor<Context = Ctx, Output = Out> + 'static,
    Ctx: Send,
    Out: Sync,
{
    fn process_erased<'a>(
        &'a self,
        output: &'a Out,
        ctx: &'a mut Ctx,
    ) -> Pin<Box<dyn Future<Output = Result<Signal, EngineError>> + Send + 'a>> {
        Box::pin(T::process(self, output, ctx))
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Wrapper structs
// ─────────────────────────────────────────────────────────────────────────────

/// A concrete wrapper that implements [`Thinker`] via type-erased dispatch.
///
/// Allows heterogeneous chains of different concrete `Thinker` types.
pub struct DynThinker<Ctx, Out>(Arc<dyn ErasedThinker<Ctx, Out>>);

/// A concrete wrapper that implements [`ContextBuilder`] via type-erased dispatch.
///
/// Allows heterogeneous chains of different concrete `ContextBuilder` types.
pub struct DynContextBuilder<Ctx>(Arc<dyn ErasedContextBuilder<Ctx>>);

/// A concrete wrapper that implements [`OutputProcessor`] via type-erased dispatch.
///
/// Allows heterogeneous chains of different concrete `OutputProcessor` types.
pub struct DynOutputProcessor<Ctx, Out>(Arc<dyn ErasedOutputProcessor<Ctx, Out>>);

// ── Constructors ──

impl<Ctx, Out> DynThinker<Ctx, Out> {
    /// Creates a [`DynThinker`] from any [`Thinker`] implementation.
    ///
    /// The thinker must have `Context = Ctx` and `Output = Out`.
    pub fn new<T>(thinker: T) -> Self
    where
        T: Thinker<Context = Ctx, Output = Out> + 'static,
        Ctx: Send + Sync,
        Out: Send,
    {
        DynThinker(Arc::new(thinker))
    }
}

impl<Ctx> DynContextBuilder<Ctx> {
    /// Creates a [`DynContextBuilder`] from any [`ContextBuilder`] implementation.
    ///
    /// The builder must have `Context = Ctx`.
    pub fn new<T>(builder: T) -> Self
    where
        T: ContextBuilder<Context = Ctx> + 'static,
        Ctx: Send,
    {
        DynContextBuilder(Arc::new(builder))
    }
}

impl<Ctx, Out> DynOutputProcessor<Ctx, Out> {
    /// Creates a [`DynOutputProcessor`] from any [`OutputProcessor`] implementation.
    ///
    /// The processor must have `Context = Ctx` and `Output = Out`.
    pub fn new<T>(processor: T) -> Self
    where
        T: OutputProcessor<Context = Ctx, Output = Out> + 'static,
        Ctx: Send,
        Out: Sync,
    {
        DynOutputProcessor(Arc::new(processor))
    }
}

// ── Trait implementations for wrappers ──

impl<Ctx, Out> Thinker for DynThinker<Ctx, Out>
where
    Ctx: Sync,
{
    type Context = Ctx;
    type Output = Out;

    async fn think(&self, ctx: &Self::Context) -> Result<Self::Output, EngineError> {
        self.0.think_erased(ctx).await
    }
}

impl<Ctx> ContextBuilder for DynContextBuilder<Ctx>
where
    Ctx: Send,
{
    type Context = Ctx;

    async fn build(&self, ctx: Self::Context) -> Result<Self::Context, EngineError> {
        self.0.build_erased(ctx).await
    }
}

impl<Ctx, Out> OutputProcessor for DynOutputProcessor<Ctx, Out>
where
    Ctx: Send,
    Out: Sync,
{
    type Context = Ctx;
    type Output = Out;

    async fn process(
        &self,
        output: &Self::Output,
        ctx: &mut Self::Context,
    ) -> Result<Signal, EngineError> {
        self.0.process_erased(output, ctx).await
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// CoreEngine
// ─────────────────────────────────────────────────────────────────────────────

/// Result of a single engine turn ([`CoreEngine::run_once`]).
///
/// Always carries the thinker's output. [`stopped`](Self::stopped) is `true`
/// when a processor returned [`Signal::Stop`] (or when no processors are
/// registered and the single turn is treated as complete).
#[derive(Debug, Clone)]
pub struct TurnResult<Out> {
    /// Output produced by the thinker for this turn.
    pub output: Out,
    /// Whether the engine should terminate after this turn.
    pub stopped: bool,
}

/// The core agent engine that orchestrates the agent loop.
///
/// Generic over the context type `Ctx` and output type `Out`.
/// Construct using [`EngineBuilder`]. The engine drives the main
/// think-process cycle with plugin-based extension points for
/// context building and output processing.
///
/// The loop continues until an output processor signals
/// [`Signal::Stop`](Signal::Stop) or the cancellation token is triggered.
pub struct CoreEngine<Ctx, Out> {
    context_builders: Vec<DynContextBuilder<Ctx>>,
    thinker: DynThinker<Ctx, Out>,
    output_processors: Vec<DynOutputProcessor<Ctx, Out>>,
    cancel: CancellationToken,
}

impl<Ctx, Out> CoreEngine<Ctx, Out>
where
    Ctx: Clone + Send + Sync,
    Out: Send + Sync,
{
    /// Runs the agent loop to completion and returns the final context.
    ///
    /// Each iteration:
    /// 1. Checks the cancellation token
    /// 2. Runs the context builder chain (each may modify the context)
    /// 3. Calls the thinker to produce output
    /// 4. Runs the output processor chain (first Stop wins)
    ///
    /// The loop continues until a processor returns [`Signal::Stop`] or
    /// cancellation is triggered.
    ///
    /// If **no** output processors are registered, the loop executes exactly
    /// one turn and returns. This prevents an infinite loop when the engine
    /// has nothing that can signal stop.
    ///
    /// # Parameters
    ///
    /// * `initial` — The starting context for the first iteration.
    ///
    /// # Returns
    ///
    /// The final context after the last processor chain (or after a single
    /// turn when no processors are configured). Processors may have mutated
    /// the context (for example, appended messages).
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Cancelled`] if the cancellation token is
    /// triggered. Propagates errors from thinkers, context builders,
    /// and output processors.
    pub async fn run(&self, initial: Ctx) -> Result<Ctx, EngineError> {
        let (ctx, _last) = self.run_with_output(initial).await?;
        Ok(ctx)
    }

    /// Runs the agent loop to completion, returning the final context and
    /// the **last** thinker output (if any turn completed).
    ///
    /// Same loop as [`run`](Self::run), but also yields the output from the
    /// final turn so outer paradigms (e.g. Ralph) can verify it without
    /// losing the result when processors signal [`Signal::Stop`].
    ///
    /// # Returns
    ///
    /// `(final_ctx, Some(last_output))` after at least one successful think.
    /// `(final_ctx, None)` only if the loop exits before thinking (should not
    /// occur under normal processor configurations).
    pub async fn run_with_output(&self, initial: Ctx) -> Result<(Ctx, Option<Out>), EngineError> {
        let mut ctx = initial;
        let mut last_output: Option<Out> = None;

        loop {
            if self.cancel.is_cancelled() {
                return Err(EngineError::Cancelled);
            }

            for builder in &self.context_builders {
                ctx = builder.build(ctx).await?;
            }

            let output = self.thinker.think(&ctx).await?;

            if self.output_processors.is_empty() {
                last_output = Some(output);
                break;
            }

            let mut stop = false;
            for processor in &self.output_processors {
                match processor.process(&output, &mut ctx).await? {
                    Signal::Stop => {
                        stop = true;
                        break;
                    }
                    Signal::Continue => {}
                }
            }

            last_output = Some(output);

            if stop {
                break;
            }
        }

        Ok((ctx, last_output))
    }

    /// Runs a single turn of the engine loop.
    ///
    /// Executes one complete iteration:
    ///
    /// 1. Checks the cancellation token
    /// 2. Runs the context builder chain on the mutable context
    /// 3. Calls the thinker to produce output
    /// 4. Runs the output processor chain (shared `&output`, first Stop wins)
    ///
    /// Always returns the thinker output. [`TurnResult::stopped`] is `true`
    /// when a processor signalled [`Signal::Stop`] (including a normal
    /// terminal text turn from tool orchestration).
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Cancelled`] if the cancellation token is
    /// triggered. Propagates errors from thinkers, context builders,
    /// and output processors.
    pub async fn run_once(&self, ctx: &mut Ctx) -> Result<TurnResult<Out>, EngineError> {
        if self.cancel.is_cancelled() {
            return Err(EngineError::Cancelled);
        }

        for builder in &self.context_builders {
            let next = builder.build(ctx.clone()).await?;
            *ctx = next;
        }

        let output = self.thinker.think(ctx).await?;

        // No processors → single-turn completion (stopped = true).
        let mut stopped = self.output_processors.is_empty();
        for processor in &self.output_processors {
            match processor.process(&output, ctx).await? {
                Signal::Stop => {
                    stopped = true;
                    break;
                }
                Signal::Continue => {}
            }
        }

        Ok(TurnResult { output, stopped })
    }

    /// Returns a cloneable handle to the engine's cancellation token.
    ///
    /// External code can clone this token and call `.cancel()` to
    /// trigger graceful shutdown of the engine loop at the start
    /// of the next iteration.
    pub fn cancel_handle(&self) -> CancellationToken {
        self.cancel.clone()
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// EngineBuilder
// ─────────────────────────────────────────────────────────────────────────────

/// Fluent builder for constructing a [`CoreEngine`].
///
/// # Example
///
/// ```rust
/// use xz_agent_core::engine::{CoreEngine, EngineBuilder};
/// use xz_agent_core::traits::thinker::Thinker;
/// use xz_agent_core::error::EngineError;
///
/// struct MyThinker;
/// impl Thinker for MyThinker {
///     type Context = String;
///     type Output = String;
///     async fn think(&self, ctx: &Self::Context) -> Result<Self::Output, EngineError> {
///         Ok(format!("processed: {}", ctx))
///     }
/// }
///
/// let engine: CoreEngine<String, String> = EngineBuilder::new()
///     .thinker(MyThinker)
///     .build()
///     .unwrap();
/// ```
pub struct EngineBuilder<Ctx, Out> {
    thinker: Option<DynThinker<Ctx, Out>>,
    context_builders: Vec<DynContextBuilder<Ctx>>,
    output_processors: Vec<DynOutputProcessor<Ctx, Out>>,
    cancel: CancellationToken,
}

impl<Ctx, Out> Default for EngineBuilder<Ctx, Out> {
    fn default() -> Self {
        Self::new()
    }
}

impl<Ctx, Out> EngineBuilder<Ctx, Out> {
    /// Creates a new builder with default settings.
    ///
    /// A thinker must be provided via [`thinker`](Self::thinker)
    /// before calling [`build`](Self::build).
    pub fn new() -> Self {
        EngineBuilder {
            thinker: None,
            context_builders: Vec::new(),
            output_processors: Vec::new(),
            cancel: CancellationToken::new(),
        }
    }
}

impl<Ctx, Out> EngineBuilder<Ctx, Out>
where
    Ctx: Send + Sync,
    Out: Send + Sync,
{
    /// Sets the thinker used to produce output from context.
    ///
    /// The thinker is the only required component. All other chains
    /// are optional.
    pub fn thinker(mut self, thinker: impl Thinker<Context = Ctx, Output = Out> + 'static) -> Self {
        self.thinker = Some(DynThinker::new(thinker));
        self
    }

    /// Appends a context builder to the builder chain.
    ///
    /// Builders are applied in the order they are added. Each builder
    /// receives the context produced by the previous one and may
    /// modify it.
    pub fn context(mut self, builder: impl ContextBuilder<Context = Ctx> + 'static) -> Self {
        self.context_builders.push(DynContextBuilder::new(builder));
        self
    }

    /// Appends an output processor to the processor chain.
    ///
    /// Processors are consulted in the order they are added. The first
    /// processor to return [`Signal::Stop`] terminates the loop.
    pub fn processor(
        mut self,
        processor: impl OutputProcessor<Context = Ctx, Output = Out> + 'static,
    ) -> Self {
        self.output_processors.push(DynOutputProcessor::new(processor));
        self
    }

    /// Sets the cancellation token for the engine.
    ///
    /// When the token is cancelled, the engine loop exits at the start
    /// of the next turn with [`EngineError::Cancelled`].
    pub fn cancel(mut self, token: CancellationToken) -> Self {
        self.cancel = token;
        self
    }

    /// Builds the [`CoreEngine`].
    ///
    /// # Errors
    ///
    /// Returns [`EngineError::Config`] if no thinker has been set. The
    /// thinker is the only required component; all other chains are optional.
    pub fn build(self) -> Result<CoreEngine<Ctx, Out>, EngineError> {
        let thinker = self
            .thinker
            .ok_or_else(|| EngineError::Config("thinker is required".into()))?;
        Ok(CoreEngine {
            context_builders: self.context_builders,
            thinker,
            output_processors: self.output_processors,
            cancel: self.cancel,
        })
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// Tests
// ─────────────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc as StdArc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    // ── Test mocks ──

    /// A thinker that echoes the context as output.
    struct EchoThinker;
    impl Thinker for EchoThinker {
        type Context = String;
        type Output = String;
        async fn think(&self, ctx: &Self::Context) -> Result<Self::Output, EngineError> {
            Ok(ctx.clone())
        }
    }

    /// A thinker that returns a fixed output (ignoring context).
    struct FixedThinker {
        output: String,
    }
    impl Thinker for FixedThinker {
        type Context = String;
        type Output = String;
        async fn think(&self, _ctx: &Self::Context) -> Result<Self::Output, EngineError> {
            Ok(self.output.clone())
        }
    }

    /// A context builder that appends a suffix.
    struct AppendSuffix {
        suffix: String,
    }
    impl ContextBuilder for AppendSuffix {
        type Context = String;
        async fn build(&self, mut ctx: Self::Context) -> Result<Self::Context, EngineError> {
            ctx.push_str(&self.suffix);
            Ok(ctx)
        }
    }

    /// A counting context builder — records how many times it was called.
    struct CountingBuilder {
        count: StdArc<AtomicUsize>,
    }
    impl CountingBuilder {
        fn new(count: StdArc<AtomicUsize>) -> Self {
            CountingBuilder { count }
        }
    }
    impl ContextBuilder for CountingBuilder {
        type Context = String;
        async fn build(&self, ctx: Self::Context) -> Result<Self::Context, EngineError> {
            self.count.fetch_add(1, Ordering::SeqCst);
            Ok(ctx)
        }
    }

    /// An output processor that always returns Continue.
    struct ContinueProcessor;
    impl OutputProcessor for ContinueProcessor {
        type Context = String;
        type Output = String;
        async fn process(
            &self,
            _output: &Self::Output,
            _ctx: &mut Self::Context,
        ) -> Result<Signal, EngineError> {
            Ok(Signal::Continue)
        }
    }

    /// An output processor that always returns Stop.
    struct StopProcessor;
    impl OutputProcessor for StopProcessor {
        type Context = String;
        type Output = String;
        async fn process(
            &self,
            _output: &Self::Output,
            _ctx: &mut Self::Context,
        ) -> Result<Signal, EngineError> {
            Ok(Signal::Stop)
        }
    }

    /// An output processor that modifies the context (appends output).
    struct AppendOutput;
    impl OutputProcessor for AppendOutput {
        type Context = String;
        type Output = String;
        async fn process(
            &self,
            output: &Self::Output,
            ctx: &mut Self::Context,
        ) -> Result<Signal, EngineError> {
            ctx.push_str(output);
            Ok(Signal::Continue)
        }
    }

    /// An output processor that stops when the output contains "stop".
    struct StopOnMatch {
        keyword: &'static str,
    }
    impl OutputProcessor for StopOnMatch {
        type Context = String;
        type Output = String;
        async fn process(
            &self,
            output: &Self::Output,
            _ctx: &mut Self::Context,
        ) -> Result<Signal, EngineError> {
            if output.contains(self.keyword) { Ok(Signal::Stop) } else { Ok(Signal::Continue) }
        }
    }

    /// A thinker that echoes the context and counts invocations.
    struct CountingEchoThinker {
        count: StdArc<AtomicUsize>,
    }
    impl Thinker for CountingEchoThinker {
        type Context = String;
        type Output = String;
        async fn think(&self, ctx: &Self::Context) -> Result<Self::Output, EngineError> {
            self.count.fetch_add(1, Ordering::SeqCst);
            Ok(ctx.clone())
        }
    }

    /// A context builder that records its identity in the context and counts calls.
    struct RecordingBuilder {
        id: &'static str,
        count: StdArc<AtomicUsize>,
    }
    impl ContextBuilder for RecordingBuilder {
        type Context = String;
        async fn build(&self, mut ctx: Self::Context) -> Result<Self::Context, EngineError> {
            self.count.fetch_add(1, Ordering::SeqCst);
            ctx.push_str(&format!("|{}", self.id));
            Ok(ctx)
        }
    }

    /// An output processor that counts calls and returns a configurable signal.
    struct CountingSignalProcessor {
        count: StdArc<AtomicUsize>,
        signal: Signal,
    }
    impl OutputProcessor for CountingSignalProcessor {
        type Context = String;
        type Output = String;
        async fn process(
            &self,
            _output: &Self::Output,
            _ctx: &mut Self::Context,
        ) -> Result<Signal, EngineError> {
            self.count.fetch_add(1, Ordering::SeqCst);
            Ok(self.signal)
        }
    }

    // ── Tests ──

    #[test]
    fn test_builder_missing_thinker_returns_err() {
        let result: Result<CoreEngine<String, String>, _> = EngineBuilder::new().build();
        assert!(result.is_err(), "build without thinker should fail");
    }

    #[test]
    fn test_builder_with_thinker_returns_ok() {
        let result = EngineBuilder::new().thinker(EchoThinker).build();
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_minimal_engine_single_iteration() {
        // Engine with a thinker that echoes and a Stop processor.
        // Should run exactly one iteration and succeed.
        let engine: CoreEngine<String, String> =
            EngineBuilder::new().thinker(EchoThinker).processor(StopProcessor).build().unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_context_builder_chain() {
        let count = StdArc::new(AtomicUsize::new(0));

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(AppendSuffix { suffix: " world".into() })
            .context(CountingBuilder::new(StdArc::clone(&count)))
            .context(AppendSuffix { suffix: "!".into() })
            .thinker(EchoThinker)
            .processor(StopProcessor)
            .build()
            .unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());

        // Verify the counting builder was called exactly once
        assert_eq!(count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_context_builder_modifies_context() {
        // Verify that context builders actually modify the context.
        // The thinker echoes the (modified) context.
        // We use a processor that appends output to context and runs
        // exactly once. After the run, we can't easily inspect the
        // final context — but we can verify the pipeline completes.
        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(AppendSuffix { suffix: " world".into() })
            .thinker(EchoThinker)
            .processor(StopProcessor)
            .build()
            .unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_processor_continues_loop() {
        // With ContinueProcessor, the engine would loop forever without
        // a stop mechanism. Use a StopAfterN processor approach instead:
        // a counting context builder that tracks iterations and a
        // processor that stops after N iterations.
        let iteration_count = StdArc::new(AtomicUsize::new(0));
        let count_clone = StdArc::clone(&iteration_count);

        struct StopAfterN {
            count: StdArc<AtomicUsize>,
            limit: usize,
        }
        impl OutputProcessor for StopAfterN {
            type Context = String;
            type Output = String;
            async fn process(
                &self,
                _output: &Self::Output,
                _ctx: &mut Self::Context,
            ) -> Result<Signal, EngineError> {
                let current = self.count.fetch_add(1, Ordering::SeqCst) + 1;
                if current >= self.limit { Ok(Signal::Stop) } else { Ok(Signal::Continue) }
            }
        }

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(EchoThinker)
            .processor(StopAfterN { count: StdArc::clone(&count_clone), limit: 3 })
            .build()
            .unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());

        // Should have run exactly 3 iterations
        assert_eq!(count_clone.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn test_stop_signal_breaks_loop() {
        let engine: CoreEngine<String, String> =
            EngineBuilder::new().thinker(EchoThinker).processor(StopProcessor).build().unwrap();

        let result = engine.run("test".into()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_cancellation_returns_cancelled_error() {
        let token = CancellationToken::new();
        token.cancel(); // Pre-cancel before engine starts

        let engine: CoreEngine<String, String> =
            EngineBuilder::new().thinker(EchoThinker).cancel(token).build().unwrap();

        let result = engine.run("test".into()).await;

        assert!(result.is_err());
        match result.unwrap_err() {
            EngineError::Cancelled => {} // expected
            other => panic!("expected Cancelled, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn test_cancel_handle() {
        let engine: CoreEngine<String, String> =
            EngineBuilder::new().thinker(EchoThinker).processor(ContinueProcessor).build().unwrap();

        let handle = engine.cancel_handle();
        assert!(!handle.is_cancelled());

        handle.cancel();
        assert!(handle.is_cancelled());

        let result = engine.run("test".into()).await;
        assert!(matches!(result.unwrap_err(), EngineError::Cancelled));
    }

    #[tokio::test]
    async fn test_processor_chain_order() {
        // Processors are consulted in order. The first Stop wins.
        // With [ContinueProcessor, StopProcessor], the first returns
        // Continue, the second returns Stop → loop stops after 1 iteration.
        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(EchoThinker)
            .processor(ContinueProcessor)
            .processor(StopProcessor)
            .build()
            .unwrap();

        let result = engine.run("test".into()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_processor_modifies_context() {
        // AppendOutput processor appends the thinker output to the context.
        // After the first iteration, context becomes "hellohello".
        // The StopOnMatch processor stops when output contains a keyword.
        // Since output = context (EchoThinker), on 2nd iter context is "hellohello",
        // thinker echoes it → output "hellohello", doesn't contain "stop" →
        // StopOnMatch won't stop... this loops forever.
        //
        // Better test: use a processor that stops after first iteration
        // combined with a processor that modifies context.
        let count = StdArc::new(AtomicUsize::new(0));
        let count2 = StdArc::clone(&count);

        struct StopAfterOne {
            count: StdArc<AtomicUsize>,
        }
        impl OutputProcessor for StopAfterOne {
            type Context = String;
            type Output = String;
            async fn process(
                &self,
                _output: &Self::Output,
                _ctx: &mut Self::Context,
            ) -> Result<Signal, EngineError> {
                let n = self.count.fetch_add(1, Ordering::SeqCst) + 1;
                if n >= 2 { Ok(Signal::Stop) } else { Ok(Signal::Continue) }
            }
        }

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(EchoThinker)
            .processor(AppendOutput)
            .processor(StopAfterOne { count: StdArc::clone(&count2) })
            .build()
            .unwrap();

        // Run with "hello" — first iter: echo → "hello", append → ctx="hellohello",
        // StopAfterOne count=1 → Continue. Second iter: echo → "hellohello",
        // append → ctx="hellohellohellohello", StopAfterOne count=2 → Stop.
        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());
        assert_eq!(count2.load(Ordering::SeqCst), 2);
        // Final context must be returned to the caller.
        assert_eq!(result.unwrap(), "hellohellohellohello");
    }

    #[tokio::test]
    async fn test_multiple_context_builders() {
        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(AppendSuffix { suffix: " world".into() })
            .context(AppendSuffix { suffix: "!".into() })
            .thinker(FixedThinker { output: "done".into() })
            .processor(StopOnMatch { keyword: "done" })
            .build()
            .unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_dyn_context_builder_clone() {
        // Verify DynContextBuilder can be constructed from any ContextBuilder
        let builder = DynContextBuilder::<String>::new(AppendSuffix { suffix: " test".into() });
        let ctx: String = builder.build("hello".into()).await.unwrap();
        assert_eq!(ctx, "hello test");
    }

    #[tokio::test]
    async fn test_dyn_thinker_clone() {
        let thinker = DynThinker::<String, String>::new(EchoThinker);
        let result = thinker.think(&"input".to_string()).await.unwrap();
        assert_eq!(result, "input");
    }

    #[tokio::test]
    async fn test_dyn_output_processor_clone() {
        let processor = DynOutputProcessor::<String, String>::new(StopProcessor);
        let mut ctx = String::from("test");
        let result = processor.process(&"output".into(), &mut ctx).await.unwrap();
        assert_eq!(result, Signal::Stop);
    }

    // ── Integration tests ──

    #[tokio::test]
    async fn test_complete_loop_with_mock() {
        // Verify all three components (ContextBuilder, Thinker, OutputProcessor)
        // are invoked during a complete engine loop.
        let cb_count = StdArc::new(AtomicUsize::new(0));
        let t_count = StdArc::new(AtomicUsize::new(0));
        let op_count = StdArc::new(AtomicUsize::new(0));

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(CountingBuilder::new(StdArc::clone(&cb_count)))
            .thinker(CountingEchoThinker { count: StdArc::clone(&t_count) })
            .processor(CountingSignalProcessor {
                count: StdArc::clone(&op_count),
                signal: Signal::Stop,
            })
            .build()
            .unwrap();

        let result = engine.run("hello".into()).await;
        assert!(result.is_ok());

        assert_eq!(cb_count.load(Ordering::SeqCst), 1);
        assert_eq!(t_count.load(Ordering::SeqCst), 1);
        assert_eq!(op_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_three_context_builders_executed() {
        // Register 3 ContextBuilders and verify they all execute.
        let count_a = StdArc::new(AtomicUsize::new(0));
        let count_b = StdArc::new(AtomicUsize::new(0));
        let count_c = StdArc::new(AtomicUsize::new(0));

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(RecordingBuilder { id: "A", count: StdArc::clone(&count_a) })
            .context(RecordingBuilder { id: "B", count: StdArc::clone(&count_b) })
            .context(RecordingBuilder { id: "C", count: StdArc::clone(&count_c) })
            .thinker(FixedThinker { output: "done".into() })
            .processor(StopOnMatch { keyword: "done" })
            .build()
            .unwrap();

        let result = engine.run("init".into()).await;
        assert!(result.is_ok());

        assert_eq!(count_a.load(Ordering::SeqCst), 1);
        assert_eq!(count_b.load(Ordering::SeqCst), 1);
        assert_eq!(count_c.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_output_processor_stop_signal() {
        // Stop signal exits the loop after exactly one iteration.
        let t_count = StdArc::new(AtomicUsize::new(0));
        let op_count = StdArc::new(AtomicUsize::new(0));

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(CountingEchoThinker { count: StdArc::clone(&t_count) })
            .processor(CountingSignalProcessor {
                count: StdArc::clone(&op_count),
                signal: Signal::Stop,
            })
            .build()
            .unwrap();

        let result = engine.run("test".into()).await;
        assert!(result.is_ok());

        assert_eq!(t_count.load(Ordering::SeqCst), 1);
        assert_eq!(op_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_continue_signal_continues() {
        // Continue signal does not stop the loop; chain proceeds to next
        // processor (Stop) which terminates the loop.
        let continue_count = StdArc::new(AtomicUsize::new(0));
        let stop_count = StdArc::new(AtomicUsize::new(0));

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(EchoThinker)
            .processor(CountingSignalProcessor {
                count: StdArc::clone(&continue_count),
                signal: Signal::Continue,
            })
            .processor(CountingSignalProcessor {
                count: StdArc::clone(&stop_count),
                signal: Signal::Stop,
            })
            .build()
            .unwrap();

        let result = engine.run("test".into()).await;
        assert!(result.is_ok());

        // Both processors ran: Continue let the chain proceed to Stop.
        assert_eq!(continue_count.load(Ordering::SeqCst), 1);
        assert_eq!(stop_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_empty_context_and_no_processors() {
        // Engine with no context builders and no output processors can be
        // built and enters the run loop. With a pre-cancelled token it
        // returns Cancelled immediately, proving it runs.
        let token = CancellationToken::new();
        token.cancel();

        let engine: CoreEngine<String, String> =
            EngineBuilder::new().thinker(EchoThinker).cancel(token).build().unwrap();

        let result = engine.run("data".into()).await;
        assert!(
            matches!(result, Err(EngineError::Cancelled)),
            "expected Cancelled, got {result:?}"
        );
    }

    #[tokio::test]
    async fn test_run_returns_final_context() {
        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .context(AppendSuffix { suffix: " world".into() })
            .thinker(EchoThinker)
            .processor(AppendOutput)
            .processor(StopProcessor)
            .build()
            .unwrap();

        // initial "hi" → context "hi world" → think echoes → append → "hi worldhi world"
        let final_ctx = engine.run("hi".into()).await.unwrap();
        assert_eq!(final_ctx, "hi worldhi world");
    }

    #[tokio::test]
    async fn test_no_processors_stops_after_one_turn() {
        let t_count = StdArc::new(AtomicUsize::new(0));
        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(CountingEchoThinker { count: StdArc::clone(&t_count) })
            .build()
            .unwrap();

        let final_ctx = engine.run("once".into()).await.unwrap();
        assert_eq!(final_ctx, "once");
        assert_eq!(t_count.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn test_iterator_counting() {
        // Verify the engine runs exactly N iterations before stopping.
        let iter_count = StdArc::new(AtomicUsize::new(0));
        let count_clone = StdArc::clone(&iter_count);

        struct StopAfterN {
            count: StdArc<AtomicUsize>,
            limit: usize,
        }
        impl OutputProcessor for StopAfterN {
            type Context = String;
            type Output = String;
            async fn process(
                &self,
                _output: &Self::Output,
                _ctx: &mut Self::Context,
            ) -> Result<Signal, EngineError> {
                let current = self.count.fetch_add(1, Ordering::SeqCst) + 1;
                if current >= self.limit { Ok(Signal::Stop) } else { Ok(Signal::Continue) }
            }
        }

        let engine: CoreEngine<String, String> = EngineBuilder::new()
            .thinker(EchoThinker)
            .processor(StopAfterN { count: StdArc::clone(&count_clone), limit: 5 })
            .build()
            .unwrap();

        let result = engine.run("start".into()).await;
        assert!(result.is_ok());
        assert_eq!(count_clone.load(Ordering::SeqCst), 5);
    }
}