lxy/DevFlow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增: 无人值守P0任务边界全链

Browse Source 
- 任务级授权边界判定与审批旁路单点插桩
- boundary_json 随会话落库(V45 可逆迁移)
- 六路非正常终局统一安全停靠与托盘通知
- 每五轮进度事件与会话头部进度条
- 前端边界配置卡侧栏徽标及 UX 校验补齐
This commit is contained in:
lxy
2026-10-04 16:27:30 +08:00
parent 8396849d04
commit b2d44f85e9
29 changed files with 2773 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src-tauri/src/commands/ai/conversation.rs
+74 -12
View File
@@ -171,6 +171,18 @@ pub(crate) fn truncate_parts_for_persist(parts: &[df_ai::provider::ContentPart])
}
}
/// save 元数据快照(P3-5):子代理出口落库显式携带父会话的 provider_id/created_at。
///
/// save 内部默认读 `session.active_*` 决定首插的 provider_id/created_at——这对主会话
/// 正确,对子代理错误:子代理后台长跑收尾时用户可能已切走会话,active 值已是别的会话的,
/// 子轨迹会继承错元数据。spawn 时刻快照、出口显式传入即可钉住。
/// 主会话路径不经过本类型(save_conversation 传 None),读法与历史版本一致。
#[derive(Debug, Clone, Default)]
pub(crate) struct SaveMetaOverride {
pub(crate) provider_id: Option<String>,
pub(crate) created_at: Option<String>,
}
/// 保存对话到数据库（按 conv_id 写库，不受 active_conversation_id 切换影响）
///
/// 写 ai_messages(每条消息一行,F-260619-03 拆分存储)+ updated_at + 累加 token 用量 +
@@ -190,13 +202,40 @@ pub(crate) async fn save_conversation(
model: Option<&str>,
touch_updated_at: bool,
) {
// 外层 timeout(SAVE_CONVERSATION_TIMEOUT_SECS):超时只 warn 不阻断。
// save 是非关键路径(失败已 warn,内存真相源不变),防 DB/fs 卡死拖垮 agentic loop / IPC。
// timeout 仅作用于 DB 写入段(此处包住整个 inner,因 session 锁/序列化也属 save 周期,
// 超时即放弃本次 save,下次 loop 轮次会重新落库,幂等覆盖无副作用)。
save_with_meta(session_arc, db, conv_id, usage, model, touch_updated_at, None).await;
}
/// 子代理出口落库入口(P3-5):显式携带父会话 save 元数据快照的 save 变体。
/// 与 save_conversation 共用同一落库实现,仅首插的 provider_id/created_at 来源不同。
pub(crate) async fn save_conversation_meta(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
conv_id: &str,
usage: Option<&df_ai::provider::TokenUsage>,
model: Option<&str>,
touch_updated_at: bool,
meta: SaveMetaOverride,
) {
save_with_meta(session_arc, db, conv_id, usage, model, touch_updated_at, Some(meta)).await;
}
/// save 统一实现(meta=None 即主会话既有行为,Some 即显式元数据覆盖)。
/// 外层 timeout(SAVE_CONVERSATION_TIMEOUT_SECS):超时只 warn 不阻断。
/// save 是非关键路径(失败已 warn,内存真相源不变),防 DB/fs 卡死拖垮 agentic loop / IPC。
/// timeout 仅作用于 DB 写入段(此处包住整个 inner,因 session 锁/序列化也属 save 周期,
/// 超时即放弃本次 save,下次 loop 轮次会重新落库,幂等覆盖无副作用)。
async fn save_with_meta(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
conv_id: &str,
usage: Option<&df_ai::provider::TokenUsage>,
model: Option<&str>,
touch_updated_at: bool,
meta: Option<SaveMetaOverride>,
) {
if let Err(_elapsed) = tokio::time::timeout(
Duration::from_secs(SAVE_CONVERSATION_TIMEOUT_SECS),
save_conversation_serialized(session_arc, db, conv_id, usage, model, touch_updated_at),
save_conversation_serialized(session_arc, db, conv_id, usage, model, touch_updated_at, meta),
)
.await
{
@@ -216,16 +255,17 @@ async fn save_conversation_serialized(
usage: Option<&df_ai::provider::TokenUsage>,
model: Option<&str>,
touch_updated_at: bool,
meta: Option<SaveMetaOverride>,
) {
let save_lock = {
let mut session = session_arc.lock().await;
session.save_lock(conv_id)
};
let _save_guard = save_lock.lock().await;
save_conversation_inner(session_arc, db, conv_id, usage, model, touch_updated_at).await;
save_conversation_inner(session_arc, db, conv_id, usage, model, touch_updated_at, meta).await;
}
/// save_conversation 内部实现(无超时,由外层 save_conversation 包 timeout 调用)。
/// save_conversation 内部实现(无超时,由外层 save_with_meta 包 timeout 调用)。
async fn save_conversation_inner(
session_arc: &Arc<Mutex<AiSession>>,
db: &Arc<Database>,
@@ -233,6 +273,7 @@ async fn save_conversation_inner(
usage: Option<&df_ai::provider::TokenUsage>,
model: Option<&str>,
touch_updated_at: bool,
meta: Option<SaveMetaOverride>,
) {
// 取 messages + 懒创建首次落库所需的 provider_id/created_at
// 工具结果(content)超 8KB50KB→8KB,治 GLM 1214)时截断头尾各 ~3KB + 中段
@@ -253,7 +294,7 @@ async fn save_conversation_inner(
// 避免同时持有 session 的 mut 和 immut 借用(E0502)。
// BE1无变化捷径提前到克隆之前——同一 lock 段内先判「消息数未增且非 dirty」,
// 无变化则不 clone(每轮无变化 save 省一次全量 clone + 后续 truncate/records 构建)。
let (provider_id, created_at, pinned_goals, msg_len, persisted_count, needs_full_rewrite, mut persist_msgs, conv_origin) = {
let (provider_id, created_at, pinned_goals, msg_len, persisted_count, needs_full_rewrite, mut persist_msgs, conv_origin, conv_boundary) = {
let mut session = session_arc.lock().await;
// 已删除对话直接跳过写入(loop 在「push 后、save 前」被删除的幽灵复活根治):
// 此处不惰性重建 per_conv、不触发 INSERT 空壳 + insert_batch 复活。
@@ -268,8 +309,15 @@ async fn save_conversation_inner(
let __wait = __lock_start.elapsed();
// 先取 session 顶层 owned 字段(immutable borrow 即刻结束),再取 conv mutable borrow,
// 避免同时持有 session 的 mut 和 immut 借用(E0502)。
let provider_id = session.active_provider_id.clone();
let created_at = session.active_conv_created_at.clone();
// P3-5:显式快照优先(SubAgent 出口落库用),否则读 active_*(主会话历史读法不变)。
let provider_id = match &meta {
Some(m) => m.provider_id.clone(),
None => session.active_provider_id.clone(),
};
let created_at = match &meta {
Some(m) => m.created_at.clone(),
None => session.active_conv_created_at.clone(),
};
// 会话来源标记(V44 子轨迹落库):子代理 conv 首插时随 record 落 origin 列,主会话 None
let conv_origin = session.conv_origins.get(conv_id).cloned();
let conv = session.conv(conv_id);
@@ -284,6 +332,8 @@ async fn save_conversation_inner(
conv.messages.all_messages_clone()
};
let pinned_goals = conv.pinned_goals.clone();
// 无人值守任务边界随行落库(V45 boundary_json):每轮 save 同步内存真相源快照
let conv_boundary = conv.boundary.clone();
let _ = __wait; // 诊断:lock 等待时长(下行 eprintln 输出 lock 段总时长)
(
provider_id,
@@ -294,6 +344,7 @@ async fn save_conversation_inner(
needs_full_rewrite,
cloned,
conv_origin,
conv_boundary,
)
};
let __lock_total = __lock_start.elapsed();
@@ -403,6 +454,10 @@ async fn save_conversation_inner(
rec.pinned_goals = Some(serde_json::to_string(&pinned_goals).unwrap_or_else(|_| "[]".to_string()));
// 挂起审批快照持久化:每轮 save 同步当前 pending_approvals 快照
rec.pending_approvals = Some(pending_approvals_json.clone());
// 无人值守任务边界随行持久化:None(未声明)保持 NULL,不写 "null" 脏串
rec.boundary_json = conv_boundary.as_ref()
.map(|b| serde_json::to_string(b).unwrap_or_default())
.filter(|s| !s.is_empty());
// update_full 仍写 messages 列(保留旧值,本批不改 messages 字段),写元数据 + updated_at
if let Err(e) = conv_repo.update_full(&rec).await {
tracing::warn!("更新对话元数据失败 {conv_id}: {e}");
@@ -473,6 +528,9 @@ async fn save_conversation_inner(
created_at: conv_created.clone(),
updated_at: now,
origin: conv_origin,
boundary_json: conv_boundary.as_ref()
.map(|b| serde_json::to_string(b).unwrap_or_default())
.filter(|s| !s.is_empty()),
};
if let Err(e) = conv_repo.insert(rec).await {
tracing::warn!("落库对话失败 {conv_id}: {e}");
@@ -492,10 +550,14 @@ async fn save_conversation_inner(
// 失败保基线不变:下次 save 会重新尝试写未落库的消息(增量路径重试同 slice,INSERT OR IGNORE
// 幂等不冲突;脏路径全量重写收敛)。基线推进必须在锁内与读时一致(本进程内单线程 save 串行)。
// 注意:写库期间若有其他操作 push/置 dirty,基线推进可能与最新内存错位——但 advance 用
// max 防回退,且 dirty 标志由那次操作自己设(下次 save 仍走全量重写收敛),最终一致。
// max 防回退;dirty 只在全量重写时清(增量/无变化路径写库期间新置的 dirty 如压缩 mark,
// 与本次 append 写库无关,误清会致改动滞留内存永不落库)。
if msgs_written_ok {
let mut session = session_arc.lock().await;
session.conv(conv_id).messages.advance_persisted_count(total_len);
session
.conv(conv_id)
.messages
.advance_persisted_count(total_len, needs_full_rewrite);
}
// 自动 checkpoint（每 20 轮或总 token > 150k 时创建）