283 lines
13 KiB
Rust
283 lines
13 KiB
Rust
//! 应用全局状态 — 数据库、Repo、事件总线、节点注册表、AI 会话
|
||
|
||
use std::collections::HashMap;
|
||
use std::path::Path;
|
||
use std::sync::Arc;
|
||
use std::sync::atomic::AtomicUsize;
|
||
|
||
use anyhow::Result;
|
||
use serde::{Deserialize, Serialize};
|
||
use tokio::sync::{Mutex, Semaphore};
|
||
|
||
use df_ai::ai_tools::AiToolRegistry;
|
||
use df_storage::crud::{
|
||
AiConversationRepo, AiProviderRepo, AiToolExecutionRepo, IdeaRepo, KnowledgeEventsRepo,
|
||
KnowledgeRepo, NodeExecutionRepo, ProjectRepo, ReleaseRepo, SettingsRepo, TaskRepo,
|
||
WorkflowRepo,
|
||
};
|
||
use df_storage::db::Database;
|
||
use df_workflow::eventbus::EventBus;
|
||
use df_workflow::registry::NodeRegistry;
|
||
use df_workflow::state::StateMachine;
|
||
|
||
use crate::commands::ai::AiSession;
|
||
|
||
// ============================================================
|
||
// 知识库配置(提取 + 注入)
|
||
// ============================================================
|
||
|
||
/// AI 提炼触发方式
|
||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
|
||
#[serde(rename_all = "snake_case")]
|
||
pub enum ExtractTrigger {
|
||
/// 对话正常完成时(默认)
|
||
OnComplete,
|
||
/// 对话闲置 N 秒后
|
||
OnIdle,
|
||
/// 仅手动按钮触发
|
||
ManualOnly,
|
||
}
|
||
|
||
/// 知识库行为配置(存 AppState 内存,前后端通过 IPC 读写)
|
||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||
pub struct KnowledgeConfig {
|
||
/// 提炼总开关,默认 true
|
||
pub auto_extract: bool,
|
||
/// 提炼触发方式,默认 OnComplete
|
||
pub trigger_mode: ExtractTrigger,
|
||
/// 最少消息数守卫(防闲聊噪音),默认 4
|
||
pub min_messages: u32,
|
||
/// 闲置触发超时(ms),默认 30000
|
||
pub idle_timeout_ms: u64,
|
||
/// 聊天时自动注入相关知识开关,默认 true
|
||
pub auto_inject: bool,
|
||
/// 语义检索(向量)总开关,默认 false——关闭时纯 LIKE 零外部依赖
|
||
#[serde(default)]
|
||
pub vector_enabled: bool,
|
||
/// embedding 用的 provider id(仅 openai_compat 类型,Anthropic 无 embed API)
|
||
#[serde(default)]
|
||
pub embedding_provider_id: Option<String>,
|
||
/// embedding 模型名(如 embedding-3 / text-embedding-3-small)
|
||
#[serde(default)]
|
||
pub embedding_model: Option<String>,
|
||
}
|
||
|
||
impl Default for KnowledgeConfig {
|
||
fn default() -> Self {
|
||
Self {
|
||
auto_extract: true,
|
||
trigger_mode: ExtractTrigger::OnComplete,
|
||
min_messages: 4,
|
||
idle_timeout_ms: 30_000,
|
||
auto_inject: true,
|
||
vector_enabled: false,
|
||
embedding_provider_id: None,
|
||
embedding_model: None,
|
||
}
|
||
}
|
||
}
|
||
|
||
// ============================================================
|
||
// LLM 调用并发控制（双层 Semaphore）
|
||
// ============================================================
|
||
|
||
/// LLM 调用并发控制 — 全局 + 单对话双层 Semaphore
|
||
///
|
||
/// 限流对象：所有真实 LLM 调用（主循环 stream_llm / 标题生成 / 知识提炼）。
|
||
/// 不限流本地工具执行（tools.execute）——本地操作无外部成本、不受 RPM 约束。
|
||
///
|
||
/// 运行时调整：tokio Semaphore 的 permits 数构造时固定、不可增减，
|
||
/// 故用 `Arc<Mutex<Arc<Semaphore>>>` 双层包装——替换内层 Arc 即重建 Semaphore。
|
||
/// 已持有旧 permit 的任务不受影响（permit 绑定旧 Semaphore，软收敛），
|
||
/// 新请求 lock 后克隆到最新 Arc、自动走新限制。旧 Semaphore 随最后 permit 释放而 drop。
|
||
///
|
||
/// 注意：per_conv 当前是应用级单一信号量(非 per-conv map)。因 AiSession 为单例 +
|
||
/// generating 互斥,同一时刻仅一个对话的 loop 在跑,per_conv 退化为"单对话内并发"
|
||
/// (主循环 stream_llm + 标题生成 + 知识提炼)。未来若支持多对话并发,
|
||
/// 需改为 HashMap<conv_id, Semaphore>。
|
||
#[derive(Clone)]
|
||
pub struct LlmConcurrency {
|
||
global: Arc<Mutex<Arc<Semaphore>>>,
|
||
per_conv: Arc<Mutex<Arc<Semaphore>>>,
|
||
}
|
||
|
||
impl LlmConcurrency {
|
||
pub fn new(global: usize, per_conv: usize) -> Self {
|
||
Self {
|
||
global: Arc::new(Mutex::new(Arc::new(Semaphore::new(global)))),
|
||
per_conv: Arc::new(Mutex::new(Arc::new(Semaphore::new(per_conv)))),
|
||
}
|
||
}
|
||
|
||
/// 取全局并发 permit（重建后新请求自动走最新 Semaphore）
|
||
pub async fn acquire_global(&self) -> tokio::sync::OwnedSemaphorePermit {
|
||
let sema = self.global.lock().await.clone();
|
||
sema.acquire_owned().await.expect("llm global semaphore closed")
|
||
}
|
||
|
||
/// 取单对话并发 permit
|
||
pub async fn acquire_per_conv(&self) -> tokio::sync::OwnedSemaphorePermit {
|
||
let sema = self.per_conv.lock().await.clone();
|
||
sema.acquire_owned().await.expect("llm per_conv semaphore closed")
|
||
}
|
||
|
||
/// 重建全局 Semaphore（软收敛：旧 permit 不回收，待其释放后新限制完全生效）
|
||
pub async fn set_global(&self, permits: usize) {
|
||
*self.global.lock().await = Arc::new(Semaphore::new(permits));
|
||
}
|
||
|
||
/// 重建单对话 Semaphore
|
||
pub async fn set_per_conv(&self, permits: usize) {
|
||
*self.per_conv.lock().await = Arc::new(Semaphore::new(permits));
|
||
}
|
||
}
|
||
|
||
/// 应用全局状态 — 通过 `app.manage()` 注入，command 中以 `State<'_, AppState>` 取用
|
||
pub struct AppState {
|
||
/// 数据库句柄（Arc 包装，便于在异步任务中重建 Repo）
|
||
pub db: Arc<Database>,
|
||
/// 灵感表 Repo
|
||
pub ideas: IdeaRepo,
|
||
/// 项目表 Repo
|
||
pub projects: ProjectRepo,
|
||
/// 任务表 Repo
|
||
pub tasks: TaskRepo,
|
||
/// 发布表 Repo
|
||
pub releases: ReleaseRepo,
|
||
/// 工作流执行表 Repo
|
||
pub workflows: WorkflowRepo,
|
||
/// 节点执行表 Repo
|
||
pub node_executions: NodeExecutionRepo,
|
||
/// 工作流事件总线（tokio broadcast）
|
||
pub event_bus: EventBus,
|
||
/// 节点注册表（已注册内置节点）
|
||
pub registry: Arc<NodeRegistry>,
|
||
// ── AI ──
|
||
/// AI 提供商配置 Repo
|
||
pub ai_providers: AiProviderRepo,
|
||
/// AI 对话历史 Repo
|
||
pub ai_conversations: AiConversationRepo,
|
||
/// AI 工具执行审计 Repo
|
||
pub ai_tool_executions: AiToolExecutionRepo,
|
||
/// AI 工具注册表
|
||
pub ai_tools: Arc<AiToolRegistry>,
|
||
/// AI 会话状态
|
||
pub ai_session: Arc<Mutex<AiSession>>,
|
||
// ── 知识库 ──
|
||
/// 知识库 Repo
|
||
pub knowledge: KnowledgeRepo,
|
||
/// 知识生命线事件 Repo(产生/审核/引用/归档审计)
|
||
pub knowledge_events: KnowledgeEventsRepo,
|
||
/// 知识库行为配置(提取 + 注入)
|
||
pub knowledge_config: Arc<Mutex<KnowledgeConfig>>,
|
||
/// 通用应用设置 KV Repo(前端 localStorage 迁移目标)
|
||
pub settings: SettingsRepo,
|
||
// ── LLM 并发控制 ──
|
||
/// LLM 调用并发上限（全局 + 单对话双层 Semaphore，运行时可调）
|
||
pub llm_concurrency: LlmConcurrency,
|
||
// ── Agentic 循环轮次上限 ──
|
||
/// Agentic 循环最大轮次（前端 Settings 数字配置 → AppState 字段 → 热改 command →
|
||
/// loop 入口 load 快照透传形参；当前 loop 锁定边界，热改下次发消息生效）。
|
||
/// 默认 10，与 agentic.rs::DEFAULT_MAX_AGENT_ITERATIONS 对齐。
|
||
pub agent_max_iterations: Arc<AtomicUsize>,
|
||
// ── 流式对话失败自动重试 ──
|
||
/// 流式对话失败自动重试次数（F-260616-07 / 决策 a1：只重试流前失败 Init Err——未输出
|
||
/// 任何 token；流中途失败 MidStream Partial 保文不重试。退避复用 retry::backoff_delay +
|
||
/// is_status_retryable Fatal 分类 + 30s 总预算,详见 agentic.rs 重试循环。默认 3）。
|
||
pub agent_max_retries: Arc<AtomicUsize>,
|
||
// ── 工作流执行状态 ──
|
||
/// 工作流执行 → 节点状态机注册表
|
||
///
|
||
/// run_workflow 创建执行器后注册其 state_machine（StateMachine 内部 Arc<Mutex>，
|
||
/// clone 共享底层 HashMap，与下沉到 NodeContext.node_status 的是同一份）；
|
||
/// cancel_workflow_node IPC 经 execution_id 取出后 set_cancelled，
|
||
/// 直达运行中 HumanNode 的 is_cancelled 轮询。执行完成（成功/失败）后移除条目。
|
||
pub workflow_state_registry: Arc<Mutex<HashMap<String, StateMachine>>>,
|
||
}
|
||
|
||
impl AppState {
|
||
/// 初始化应用状态：打开（或创建）数据库并执行迁移，构建各 Repo 与节点注册表
|
||
pub async fn init(db_path: &Path) -> Result<Self> {
|
||
let db = Arc::new(Database::open(db_path).await?);
|
||
let ai_tools = Arc::new(crate::commands::ai::build_ai_tool_registry(&db));
|
||
// build_registry 需注入 Arc<Database>(TaskAdvanceNode 持 db)。
|
||
// 在 struct 字段 `db` move 前 clone,避免 E0382。
|
||
let registry = Arc::new(build_registry(db.clone()));
|
||
let state = Self {
|
||
ideas: IdeaRepo::new(&db),
|
||
projects: ProjectRepo::new(&db),
|
||
tasks: TaskRepo::new(&db),
|
||
releases: ReleaseRepo::new(&db),
|
||
workflows: WorkflowRepo::new(&db),
|
||
node_executions: NodeExecutionRepo::new(&db),
|
||
ai_providers: AiProviderRepo::new(&db),
|
||
ai_conversations: AiConversationRepo::new(&db),
|
||
ai_tool_executions: AiToolExecutionRepo::new(&db),
|
||
ai_session: Arc::new(Mutex::new(AiSession::new())),
|
||
knowledge: KnowledgeRepo::new(&db),
|
||
knowledge_events: KnowledgeEventsRepo::new(&db),
|
||
knowledge_config: Arc::new(Mutex::new(KnowledgeConfig::default())),
|
||
settings: SettingsRepo::new(&db),
|
||
llm_concurrency: LlmConcurrency::new(3, 2),
|
||
agent_max_iterations: Arc::new(AtomicUsize::new(
|
||
crate::commands::ai::agentic::DEFAULT_MAX_AGENT_ITERATIONS,
|
||
)),
|
||
agent_max_retries: Arc::new(AtomicUsize::new(
|
||
crate::commands::ai::agentic::DEFAULT_MAX_AGENT_RETRIES,
|
||
)),
|
||
workflow_state_registry: Arc::new(Mutex::new(HashMap::new())),
|
||
db,
|
||
event_bus: EventBus::new(),
|
||
registry,
|
||
ai_tools,
|
||
};
|
||
// 启动恢复:重启前卡 pending 的工具审批(内存 pending_approvals 已丢)从审计表重建,
|
||
// 使重启后待审批不丢。前端经 ai_pending_tool_calls + switchConversation 恢复 toolCard 态。
|
||
crate::commands::ai::restore_pending_approvals(&state).await;
|
||
Ok(state)
|
||
}
|
||
}
|
||
|
||
/// 构建节点注册表 — 注册内置节点
|
||
///
|
||
/// 注意：不使用 `NodeRegistry::default()`，其 script 工厂为占位实现（会 panic），
|
||
/// 这里注册 df-nodes 提供的真实 HumanNode / AiNode。
|
||
///
|
||
/// "script" 节点不注册：ScriptNode 走 cmd /C | sh -c 执行 config.command 原始串，
|
||
/// 前端可构造任意 DagDef 触达无审批 shell（R-PD-2）。DevFlow 工作流当前为纯演示
|
||
/// 功能（前端唯一构造点 ProjectDetail.vue demoDag 三步 echo），无真实构建/部署脚本
|
||
/// 需求。需要脚本执行能力时新建独立 BuildNode（白名单 + 项目目录锚定 + 复用 AI 工具
|
||
/// RiskLevel 审批链），而非回头启用 ScriptNode + 黑名单。
|
||
/// 详见 docs/02-架构设计/工作流脚本执行边界-2026-06-15.md。
|
||
fn build_registry(db: Arc<Database>) -> NodeRegistry {
|
||
let mut registry = NodeRegistry::new();
|
||
registry.register("human", |_config| {
|
||
Box::new(df_nodes::human_node::HumanNode)
|
||
});
|
||
// AiNode(df_nodes::ai_node, impl Node trait):
|
||
// 决策 a(AiNode 自审闭环)步骤②:AiNode 持 Arc<Database>,execute 完成后若有 task_id
|
||
// 则把产出落 task.output_json。工厂闭包 move 捕获 db 句柄(Arc clone 廉价)。
|
||
let ai_db = db.clone();
|
||
registry.register("ai", move |_config| {
|
||
Box::new(df_nodes::ai_node::AiNode::new(ai_db.clone()))
|
||
});
|
||
// AiSelfReviewNode(df_nodes::ai_node, impl Node trait):
|
||
// 决策 a 步骤③:四维度自审(prompt 固定+JSON 解析兜底),写回 output_json 加 review 子字段,
|
||
// review 摘要塞 NodeOutput.data 供下游 human_review 经 DAG inputs 透传。
|
||
// testing 模板 ai_self_review 节点类型对齐此注册 key。
|
||
let review_db = db.clone();
|
||
registry.register("ai_self_review", move |_config| {
|
||
Box::new(df_nodes::ai_node::AiSelfReviewNode::new(review_db.clone()))
|
||
});
|
||
// TaskAdvanceNode(df_nodes::task_advance_node, impl Node trait):
|
||
// 推进链阶段 2 工作流联动入口 — DAG 内触发 advance_task。
|
||
// 持有 Arc<Database> 在此构造时注入(NodeRegistry::register 工厂闭包 move 捕获 db,
|
||
// Arc clone 廉价),Node::execute 从 NodeContext.config 读 task_id/target_status。
|
||
// D-260616-03: 推进链/状态机/闸门走 df-nodes Node trait,非复活 df-task。
|
||
registry.register("task_advance", move |_config| {
|
||
Box::new(df_nodes::task_advance_node::TaskAdvanceNode::new(db.clone()))
|
||
});
|
||
registry
|
||
}
|
||
|