funera_core/chat/
session.rs

1use std::sync::Arc;
2
3use serde_json::Value as JsonValue;
4use tokio::sync::{broadcast, mpsc, oneshot};
5use uuid::Uuid;
6
7use crate::{
8    chat::message::FuneraMessage,
9    event_bus::env_state_bus::EnvStateEvent,
10    middleware::{ErrorsEnabled, EventSenderFn, MiddlewareChain, MiddlewareEvent},
11    re_act::{ReActLoop, ReActLoopConfig},
12};
13
14// ═══════════════════════════════════════════════════════════════
15// SessionCmd — 通过 channel 与 SessionActor 通信的协议
16// ═══════════════════════════════════════════════════════════════
17
18pub enum SessionCmd {
19    /// 向 session 末尾追加一批消息
20    PushMessages { msgs: Vec<FuneraMessage> },
21    /// 获取当前上下文（JSON 格式，用于构建 LLM 请求）
22    FetchContext {
23        respond: oneshot::Sender<Vec<JsonValue>>,
24    },
25    /// 获取当前消息列表（核心版本，用于检查/显示）
26    GetMessages {
27        respond: oneshot::Sender<Vec<FuneraMessage>>,
28    },
29    /// 清空 session
30    Clear,
31}
32
33/// 启动一个长期运行的 SessionActor，返回其控制通道。
34///
35/// actor 内部维护 `Vec<FuneraMessage>`，通过 `SessionCmd` 处理
36/// 推送、查询、清空操作。`ReActLoop` 通过克隆的 `session_tx`
37/// 在 reactor 循环中读写 session。
38///
39/// # 示例
40///
41/// ```rust,no_run
42/// # use funera_core::chat::session::spawn_session_actor;
43/// let session_tx = spawn_session_actor();
44/// ```
45pub fn spawn_session_actor() -> mpsc::UnboundedSender<SessionCmd> {
46    let (tx, mut rx) = mpsc::unbounded_channel();
47    // Actor runs until all sender handles are dropped; rx.recv() then
48    // returns None and the task exits cleanly — no leak.
49    tokio::spawn(async move {
50        let mut msgs: Vec<FuneraMessage> = Vec::new();
51        while let Some(cmd) = rx.recv().await {
52            match cmd {
53                SessionCmd::PushMessages { msgs: new } => msgs.extend(new),
54                SessionCmd::FetchContext { respond } => {
55                    let ctx: Vec<JsonValue> = msgs.iter().map(|m| m.format_json()).collect();
56                    let _ = respond.send(ctx);
57                }
58                SessionCmd::GetMessages { respond } => {
59                    let _ = respond.send(msgs.clone());
60                }
61                SessionCmd::Clear => msgs.clear(),
62            }
63        }
64    });
65    tx
66}
67
68/// Session 薄壳——通过 `session_tx` 与后端 Actor 通信。
69///
70/// 所有权始终由 `AgentRuntime` 持有（通过 `spawn_session_actor` 创建）。
71/// `FuneraSession` 仅提供便利方法，不直接持有消息数据。
72pub struct FuneraSession {
73    id: Uuid,
74    session_tx: mpsc::UnboundedSender<SessionCmd>,
75}
76
77impl FuneraSession {
78    pub fn new(session_tx: mpsc::UnboundedSender<SessionCmd>) -> Self {
79        Self {
80            id: Uuid::new_v4(),
81            session_tx,
82        }
83    }
84
85    pub fn id(&self) -> Uuid {
86        self.id
87    }
88
89    pub fn session_tx(&self) -> mpsc::UnboundedSender<SessionCmd> {
90        self.session_tx.clone()
91    }
92
93    /// 向 session 追加一条消息。
94    pub fn push_message(&self, msg: FuneraMessage) {
95        let _ = self
96            .session_tx
97            .send(SessionCmd::PushMessages { msgs: vec![msg] });
98    }
99
100    /// 异步获取 session 上下文（JSON 格式）。
101    pub async fn session_context(&self) -> Vec<JsonValue> {
102        let (respond, rx) = oneshot::channel();
103        let _ = self.session_tx.send(SessionCmd::FetchContext { respond });
104        rx.await.unwrap_or_default()
105    }
106
107    /// 异步获取消息列表（不可变引用）。
108    pub async fn get_messages(&self) -> Vec<FuneraMessage> {
109        let (respond, rx) = oneshot::channel();
110        let _ = self.session_tx.send(SessionCmd::GetMessages { respond });
111        rx.await.unwrap_or_default()
112    }
113
114    /// 运行 ReAct 循环。
115    ///
116    /// `init_msg` 会通过 `session_tx` 推送到 session actor。
117    /// 通过 `session_tx.clone()` 传递给 `ReActLoop`，使其可在循环期间
118    /// 读写 session 而不移出所有权。
119    ///
120    /// 完成后返回 `()`。session 数据保留在 actor 中（外部通过 `session_context()` 查询）。
121    pub async fn react_loop<P: crate::provider::ChatProvider, E: MiddlewareEvent>(
122        &self,
123        init_msg: FuneraMessage,
124        mut config: ReActLoopConfig,
125        env_state_tx: broadcast::Sender<EnvStateEvent>,
126        middleware: Option<Arc<MiddlewareChain<E, ErrorsEnabled>>>,
127        event_sender: Option<EventSenderFn<E>>,
128    ) -> anyhow::Result<()> {
129        let _ = env_state_tx.send(EnvStateEvent::SessionStart);
130
131        // Push init message to session actor
132        self.push_message(init_msg);
133
134        config.session_tx = Some(self.session_tx.clone());
135        let react_loop = ReActLoop::<P>::from_config(config);
136        let loop_handle = react_loop.run::<E>(middleware, event_sender);
137
138        loop_handle.task.await??;
139
140        let _ = env_state_tx.send(EnvStateEvent::SessionClosed);
141        Ok(())
142    }
143}
funera_core/chat/session.rs

funera_core/chat/
session.rs
