How to handle the Parent Actor?

[jBernavaPrah]

Hi, me again :)

I'm trying to figure out how to pass to the child actor, the parent actor/reference.

The complexity is that I could have multiple implementations of the child for the same message. In the example posted, I have crate::agents::agent1 and crate::agents::agent2 (both implement Message<agents::Commit>) that could be used on the crate::assistant::Assistant actor. At the same time, I also have the recognizer actors that use the same concept.

The currently posted example is not working, but I hope to give more insight into what I'm trying to achieve here.

Do you have any suggestions on how I could manage this situation?

Example Code

```rust use kameo::request::MessageSend; use std::time::Duration; use tracing::info; use tracing_subscriber::EnvFilter;

pub(crate) mod manager {
use futures::future::BoxFuture;
use kameo::actor::ActorRef;
use kameo::mailbox::bounded::BoundedMailbox;
use kameo::message::{Context, Message};
use kameo::Actor;
use tracing::info;

enum Factory<P: Actor, A: Actor> {
    Sync(Box<dyn FnMut(ActorRef<P>) -> ActorRef<A> + Send + Sync + 'static>),
    Async(
        Box<dyn FnMut(ActorRef<P>) -> BoxFuture<'static, ActorRef<A>> + Send + Sync + 'static>,
    ),
}

pub struct ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>, 
    B: Actor + Message<super::recognizers::Recognize>,
{
    messages: Vec<ManagerMessage>,
    sender: tokio::sync::broadcast::Sender<ManagerMessage>,
    thinker: Factory<Self, A>,
    thinker_actor: Option<ActorRef<A>>,

    recognizer: Factory<Self, B>,
    recognizer_actor: Option<ActorRef<B>>,
}

impl<A, B> ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    pub fn new(
        messages: Vec<ManagerMessage>,
        factory1: impl FnMut(ActorRef<Self>) -> ActorRef<A> + Send + Sync + 'static,
        factory2: impl FnMut(ActorRef<Self>) -> ActorRef<B> + Send + Sync + 'static,
    ) -> Self {
        let (tx, _) = tokio::sync::broadcast::channel(10);

        Self {
            messages,
            sender: tx,
            thinker: Factory::Sync(Box::new(factory1)),
            thinker_actor: None,
            recognizer: Factory::Sync(Box::new(factory2)),
            recognizer_actor: None,
        }
    }
}

impl<A, B> Actor for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Mailbox = BoundedMailbox<Self>;
}

pub struct GetStream;

impl<A, B> Message<GetStream> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = tokio::sync::broadcast::Receiver<ManagerMessage>;
    async fn handle(
        &mut self,
        _msg: GetStream,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        self.sender.subscribe()
    }
}

pub struct Commit(pub(crate) usize);

impl<A, B> Message<Commit> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: Commit,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        // if previous agent was set, kill it
        if let Some(thinker) = &self.thinker_actor {
            // kill it
            thinker.kill();
            thinker.wait_for_stop().await;
        }

        // replace with a new agent actor
        self.thinker_actor.replace(match &mut self.thinker {
            Factory::Sync(f) => f(_ctx.actor_ref()),
            Factory::Async(f) => f(_ctx.actor_ref()).await,
        });

        // send the messages to the new agent actor
        let _ = self
            .thinker_actor
            .as_ref()
            .unwrap()
            .tell(super::thinkers::Commit {
                messages: self.messages.clone(),
                id: msg.0,
            })
            .await;
    }
}

#[derive(Clone, Debug)]
pub enum ManagerMessage {
    User(String),
    Thinker(String)
}
impl<A, B> Message<ManagerMessage> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Commit>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: ManagerMessage,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        info!("Assistant received message: {:?}", msg);
        self.messages.push(msg.clone());
        let _ = self.sender.send(msg); // if nobody is readying the messages, kill the actor!
    }
}

}

pub(crate) mod recognizers {
pub struct Recognize(pub(crate) Vec);

pub mod recognizer1 {
    use crate::recognizers::Recognize;
    use kameo::actor::ActorRef;
    use kameo::message::Message;
    use kameo::Actor;

    #[derive(Actor)]
    pub struct RecognizerActor<A>
    where
        A: Actor + Message<super::super::thinkers::Commit>
    {
        pub parent: ActorRef<super::super::manager::ManagerActor<A, Self>>,
    }

    impl<A> Message<Recognize> for RecognizerActor<A>
    where
        A: Actor + Message<super::super::thinkers::Commit>
    {
        type Reply = ();
        async fn handle(
            &mut self,
            msg: Recognize,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            println!("Recognizer1 received message: {:?}", msg.0);
        }
    }
}

}

pub(crate) mod thinkers {
use crate::manager;
pub struct Commit {
pub messages: Vecmanager::ManagerMessage,
pub id: usize,
}

pub mod agent1 {
    use crate::thinkers::Commit;
    use crate::manager;
    use crate::manager::ManagerActor;
    use kameo::actor::{ActorRef, WeakActorRef};
    use kameo::error::{ActorStopReason, BoxError, PanicError};
    use kameo::mailbox::bounded::BoundedMailbox;
    use kameo::message::Message;
    use kameo::Actor;
    use std::time::Duration;
    use tracing::info;

    pub struct AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        pub parent: ActorRef<ManagerActor<Self, B>>,
    }

    impl<B> Actor for AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        type Mailbox = BoundedMailbox<Self>;

        async fn on_panic(
            &mut self,
            actor_ref: WeakActorRef<Self>,
            err: PanicError,
        ) -> Result<Option<ActorStopReason>, BoxError> {
            info!(
                "Actor {:?} err: {:?}. Restarting...",
                actor_ref.id(),
                err.to_string()
            );
            Ok(None)
        }
    }

    impl<B> Message<Commit> for AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        type Reply = ();
        async fn handle(
            &mut self,
            _msg: Commit,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            loop {
                let _ = self
                    .parent
                    .tell(manager::ManagerMessage::Thinker(format!(
                        "Agent1 {:?} is processing messages",
                        _msg.id
                    )))
                    .await;
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

pub mod agent2 {
    use crate::thinkers::Commit;
    use crate::manager;
    use crate::manager::ManagerActor;
    use kameo::actor::{ActorRef, WeakActorRef};
    use kameo::error::{ActorStopReason, BoxError, PanicError};
    use kameo::mailbox::bounded::BoundedMailbox;
    use kameo::message::Message;
    use kameo::Actor;
    use std::time::Duration;
    use tracing::info;

    pub struct AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        pub parent: ActorRef<ManagerActor<Self, B>>,
    }

    impl<B> Actor for AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        type Mailbox = BoundedMailbox<Self>;
        async fn on_panic(
            &mut self,
            actor_ref: WeakActorRef<Self>,
            err: PanicError,
        ) -> Result<Option<ActorStopReason>, BoxError> {
            info!(
                "Actor {:?} err: {:?}. Restarting...",
                actor_ref.id(),
                err.to_string()
            );
            Ok(None)
        }
    }

    impl<B> Message<Commit> for AgentActor<B>
    where
        B: Actor + Message<super::super::recognizers::Recognize>,
    {
        type Reply = ();
        async fn handle(
            &mut self,
            _msg: Commit,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            loop {
                let _ = self
                    .parent
                    .tell(manager::ManagerMessage::Thinker(format!(
                        "Agent2 {:?} is processing messages",
                        _msg.id
                    )))
                    .await;
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

}

#[tokio::main]
async fn main() -> Result<(), Box> {
tracing_subscriber::fmt()
.with_env_filter("trace".parse::().unwrap())
.without_time()
.with_target(false)
.init();

let my_actor_ref = kameo::actor::spawn(manager::ManagerActor::new(
    vec![],
    |actor_ref| {
        kameo::spawn(thinkers::agent1::AgentActor {
            parent: actor_ref.clone(),
        })
    },
    |actor_ref| {
        kameo::spawn(recognizers::recognizer1::RecognizerActor {
            parent: actor_ref.clone(),
        })
    },
));

let mut stream = my_actor_ref.ask(manager::GetStream).send().await?;

tokio::spawn(async move {
    while let Ok(s) = stream.recv().await {
        info!("Token {s:?}");
    }
});

my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;
my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;
my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;

let _ = my_actor_ref.ask(manager::Commit(0)).send().await?;

tokio::time::sleep(Duration::from_secs(4)).await;

let _ = my_actor_ref.ask(manager::Commit(1)).send().await?;

tokio::time::sleep(Duration::from_secs(5)).await;

Ok(())

}

</details>

[moofone]

I may be wrong but I think you have a cyclical to the parent type. Try instead to replace:
pub parent: ActorRef<manager::ManagerActor<Self, B>>,

with

pub parent: ActorRef<manager::ManagerActor<A, B>>,

[jBernavaPrah]

Thanks for the reply!

Your solution, unfortunately, doesn't work.

I also tried directly using the generic ActorRef<A>, with where A: Actor, but it gave me the same issue.

[tqwewe]

I had a brief look into this, and received the error about cyclic type of infinite size. My initial thought was to try to Box the parent field in the RecognizerActor but that didn't solve it. I'm not entirely sure how this can be solved currently, or whether its something kameo can fix on its end or not, but I can try spending some more time looking into this soon

[jBernavaPrah]

Thank you for investigating this!

In the meantime, perhaps I could change my perspective on the problem.

Could you share your method for communicating with the "parent/siblings" actors?
Are you using Pub/Sub, or is there another approach?

Thanks a lot!

[moofone]

FWIW I now use a different pattern to look up the actor ref by name in the DHT rather then passing references around. For latency sensitive stuff, I do the lookup and store the reference if it's unlikely to change. Also note that the linking is backwards to what you may intuitively think it is (child to parent vs parent to child). I mostly used link_together() to create sibling vs parent child when I did it that way, so can't add much more on the parent/child stuff

pub async fn get_actor_ref<A: Actor + RemoteActor>(name: &str) -> Result<ActorRef<A>> {
    const TIMEOUT_DURATION: Duration = Duration::from_secs(10);
    const POLL_INTERVAL: Duration = Duration::from_millis(100);

    timeout(TIMEOUT_DURATION, async {
        loop {
            match ActorRef::<A>::lookup(name).await {
                Ok(Some(actor_ref)) => return Ok(actor_ref),
                Ok(None) => tokio::time::sleep(POLL_INTERVAL).await,
                Err(e) => {
                    error!("Failed to lookup actor {}: {}", name, e);
                    tokio::time::sleep(POLL_INTERVAL).await;
                }
            }
        }
    })
    .await
    .map_err(|_| {
        anyhow!(
            "Timeout after {}s waiting for actor {}",
            TIMEOUT_DURATION.as_secs(),
            name
        )
    })?
}

[tqwewe]

Yeah the old linking methods were a little confusing, but I've deprecated them in 0.13 in favor of simple link/unlink methods, which only create bi-directional links, and removes parent/child like linkings.

Could you share your method for communicating with the "parent/siblings" actors?

I haven't actually come across the error you experienced @jBernavaPrah. It seems like it's more of an issue around the type system being an infinite cycle. Perhaps if a new type was introduced into kameo such as actix's Recipient type, where instead of specifying the Actor as the generic in ActorRef, instead Recipient would be just a single message type that can be handled by an actor.

[jBernavaPrah]

@tqwewe I think the idea of Recipient could work.

For reference, I did something similar, removing the types from the "parent" and instead using a channel.

The downside is that requires more code and a specific tokio::spawn for each channel, which simply sends the message to the manager.

Please don't hesitate to suggest any improvements to this approach.

Code

```rust use kameo::request::MessageSend; use std::time::Duration; use tracing::info; use tracing_subscriber::EnvFilter;

pub(crate) mod manager {
use futures::future::BoxFuture;
use kameo::actor::ActorRef;
use kameo::error::BoxError;
use kameo::mailbox::bounded::BoundedMailbox;
use kameo::message::{Context, Message};
use kameo::Actor;
use tracing::info;
use crate::{recognizers, thinkers};

#[derive(Debug,Clone)]
pub enum Event {
    Text(String),
}

pub enum Factory<M, A: Actor> {
    Sync(Box<dyn FnMut(tokio::sync::mpsc::Sender<M>) -> ActorRef<A> + Send + Sync + 'static>),
    Async(
        Box<
            dyn FnMut(tokio::sync::mpsc::Sender<M>) -> BoxFuture<'static, ActorRef<A>>
                + Send
                + Sync
                + 'static,
        >,
    ),
}

pub struct ManagerActor<A, B>
where
    A: Actor,
    B: Actor,
{
    messages: Vec<ManagerMessage>,
    sender: tokio::sync::broadcast::Sender<Event>,
    thinker: Factory<thinkers::Event, A>,
    thinker_sender: Option<tokio::sync::mpsc::Sender<thinkers::Event>>,
    thinker_actor: Option<ActorRef<A>>,

    recognizer: Factory<recognizers::Event, B>,
    recognizer_sender: Option<tokio::sync::mpsc::Sender<recognizers::Event>>,
    recognizer_actor: Option<ActorRef<B>>,
}

impl<A, B> ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    pub fn new(
        messages: Vec<ManagerMessage>,
        factory1: Factory<thinkers::Event, A>,
        factory2: Factory<recognizers::Event, B>,
    ) -> Self {
        let (tx, _) = tokio::sync::broadcast::channel(10);

        Self {
            messages,
            sender: tx,
            thinker: factory1,
            thinker_actor: None,
            thinker_sender: None,
            recognizer: factory2,
            recognizer_actor: None,
            recognizer_sender: None,
        }
    }
}

impl<A, B> Actor for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Mailbox = BoundedMailbox<Self>;

    /// In this method, I create the channels that are then passed to the actors.
    /// I also spawn a function that will receive the messages from the actors and forward them to the parent.
    async fn on_start(&mut self, actor_ref: ActorRef<Self>) -> Result<(), BoxError> {
        
        // ********** Manage Events from attached **********
        
        let (tx, mut rx) = tokio::sync::mpsc::channel(10);
        self.thinker_sender.replace(tx);
        let _actor_ref = actor_ref.clone();
        // simulate the parent that receives the messages
        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                info!("Manager received message from thinker: {:?}", msg);
                let _ = _actor_ref.tell(msg).await;
            }
        });

        
        // simulate the parent that receives the messages
        let (tx, mut rx) = tokio::sync::mpsc::channel(10);
        self.recognizer_sender.replace(tx);
        let _actor_ref = actor_ref.clone();
        tokio::spawn(async move {
            while let Some(msg) = rx.recv().await {
                info!("Manager received message from recognizer: {:?}", msg);
                let _ = _actor_ref.tell(msg).await;
            }
        });


        self.recognizer_actor = Some(match &mut self.recognizer {
            Factory::Sync(f) => f(self.recognizer_sender.as_ref().unwrap().clone()),
            Factory::Async(f) => f(self.recognizer_sender.as_ref().unwrap().clone()).await,
        });

        Ok(())
    }
}

pub struct GetStream;

impl<A, B> Message<GetStream> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = tokio::sync::broadcast::Receiver<self::Event>;
    async fn handle(
        &mut self,
        _msg: GetStream,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        self.sender.subscribe()
    }
}

pub struct Commit(pub(crate) usize);

impl<A, B> Message<Commit> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: Commit,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        // simulate a thinker that could be stopped when a new commit is triggered
        
        // if previous thinker was set, kill it
        if let Some(thinker) = &self.thinker_actor {
            // kill it
            thinker.kill();
            thinker.wait_for_stop().await;
        }

        // replace with a new thinker actor
        self.thinker_actor.replace(match &mut self.thinker {
            Factory::Sync(f) => f(self.thinker_sender.as_ref().unwrap().clone()),
            Factory::Async(f) => f(self.thinker_sender.as_ref().unwrap().clone()).await,
        });

        // send the messages to the new thinker actor
        let _ = self
            .thinker_actor
            .as_ref()
            .unwrap()
            .tell(super::thinkers::Think {
                id: msg.0,
            })
            .await;
    }
}

#[derive(Clone, Debug)]
pub enum ManagerMessage {
    User(String),
    Thinker(String),
}
impl<A, B> Message<ManagerMessage> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: ManagerMessage,
        _ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        self.messages.push(msg.clone());
        info!("Assistant received message: {:?}", msg);
        
    }
}

/// Implement the message Event for the recognizer.
impl<A, B> Message<recognizers::Event> for ManagerActor<A, B>
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: recognizers::Event,
        ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        info!("Manager received message from thinker: {:?}", msg);

        match msg {
            recognizers::Event::Recognized(msg) => {
                let _ = ctx.actor_ref().tell(ManagerMessage::Thinker(msg.clone()));
                let _ = ctx.actor_ref().tell(Commit(1));
            }
        }
    }
}

/// Implement the message Event for the thinker.
impl<A, B> Message<thinkers::Event> for ManagerActor<A, B> 
where
    A: Actor + Message<super::thinkers::Think>,
    B: Actor + Message<super::recognizers::Recognize>,
{
    type Reply = ();
    async fn handle(
        &mut self,
        msg: thinkers::Event,
        ctx: Context<'_, Self, Self::Reply>,
    ) -> Self::Reply {
        info!("Manager received message from thinker: {:?}", msg);
        
        match msg {
            thinkers::Event::Thought(msg) => {
                let _ = ctx.actor_ref().tell(ManagerMessage::Thinker(msg.clone()));
                let _ = self.sender.send(self::Event::Text(msg));
            }
        }
    }
}

}

pub(crate) mod recognizers {

// Define the events that the recognizer can send
#[derive(Debug, Clone)]
pub enum Event {
    Recognized(String),
}

pub struct Recognize(pub(crate) Vec<u8>);

pub mod recognizer1 {

    use crate::recognizers::{Event, Recognize};
    use kameo::message::Message;
    use kameo::Actor;

    #[derive(Actor)]
    pub struct RecognizerActor {
        pub sender: tokio::sync::mpsc::Sender<Event>,
    }
    
    impl Message<Recognize> for RecognizerActor {
        type Reply = ();
        async fn handle(
            &mut self,
            msg: Recognize,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            println!("Recognizer1 received message: {:?}", msg.0);
        }
    }
}

}

pub(crate) mod thinkers {

#[derive(Debug, Clone)]
pub enum Event {
    Thought(String),
}

pub struct Think {
    pub id: usize,
}

pub mod agent1 {
    use crate::{thinkers};
    use kameo::actor::WeakActorRef;
    use kameo::error::{ActorStopReason, BoxError, PanicError};
    use kameo::mailbox::bounded::BoundedMailbox;
    use kameo::message::Message;
    use kameo::Actor;
    use std::time::Duration;
    use tracing::info;

    pub struct AgentActor {
        pub sender: tokio::sync::mpsc::Sender<thinkers::Event>,
    }

    impl Actor for AgentActor {
        type Mailbox = BoundedMailbox<Self>;

        async fn on_panic(
            &mut self,
            actor_ref: WeakActorRef<Self>,
            err: PanicError,
        ) -> Result<Option<ActorStopReason>, BoxError> {
            info!(
                "Actor {:?} err: {:?}. Restarting...",
                actor_ref.id(),
                err.to_string()
            );
            Ok(None)
        }
    }

    impl Message<thinkers::Think> for AgentActor {
        type Reply = ();
        async fn handle(
            &mut self,
            _msg: thinkers::Think,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            loop {
                let _ = self
                    .sender
                    .send(thinkers::Event::Thought(format!(
                        "Agent1 {:?} is processing messages",
                        _msg.id
                    )))
                    .await;
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

pub mod agent2 {
    use crate::thinkers::Think;
    use kameo::actor::WeakActorRef;
    use kameo::error::{ActorStopReason, BoxError, PanicError};
    use kameo::mailbox::bounded::BoundedMailbox;
    use kameo::message::Message;
    use kameo::Actor;
    use std::time::Duration;
    use tracing::info;
    use crate::thinkers;

    pub struct AgentActor {
        pub sender: tokio::sync::mpsc::Sender<thinkers::Event>,
    }

    impl Actor for AgentActor {
        type Mailbox = BoundedMailbox<Self>;

        async fn on_panic(
            &mut self,
            actor_ref: WeakActorRef<Self>,
            err: PanicError,
        ) -> Result<Option<ActorStopReason>, BoxError> {
            info!(
                "Actor {:?} err: {:?}. Restarting...",
                actor_ref.id(),
                err.to_string()
            );
            Ok(None)
        }
    }

    impl Message<Think> for AgentActor {
        type Reply = ();
        async fn handle(
            &mut self,
            _msg: Think,
            _ctx: kameo::message::Context<'_, Self, Self::Reply>,
        ) -> Self::Reply {
            loop {
                let _ = self
                    .sender
                    .send(thinkers::Event::Thought(format!(
                        "Agent2 {:?} is processing messages",
                        _msg.id
                    )))
                    .await;
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }
}

}

#[tokio::main]
async fn main() -> Result<(), Box> {
tracing_subscriber::fmt()
.with_env_filter("trace".parse::().unwrap())
.without_time()
.with_target(false)
.init();

let my_actor_ref = kameo::actor::spawn(manager::ManagerActor::new(
    vec![],
    // thinker factory
    manager::Factory::Sync(Box::new(|sender| {
        // start the second agent
        kameo::spawn(thinkers::agent2::AgentActor { sender })
    })),
    // recognizer factory
    manager::Factory::Sync(Box::new(|sender| {
        // start the first recognizer
        kameo::spawn(recognizers::recognizer1::RecognizerActor { sender })
    })),
));

let mut stream = my_actor_ref.ask(manager::GetStream).send().await?;

tokio::spawn(async move {
    while let Ok(s) = stream.recv().await {
        info!("Token {s:?}");
    }
});

my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;
my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;
my_actor_ref
    .tell(manager::ManagerMessage::User("ABC".to_string()))
    .send()
    .await?;

let _ = my_actor_ref.ask(manager::Commit(0)).send().await?;

tokio::time::sleep(Duration::from_secs(4)).await;

let _ = my_actor_ref.ask(manager::Commit(1)).send().await?;

tokio::time::sleep(Duration::from_secs(5)).await;

Ok(())

}

</details>