DevFlow/src-tauri/src/commands/ai/wake.rs

//! 异步唤醒原语(W-b 核心,设计文档《并行化与异步唤醒-2026-10-04》§二)
//!
//! 治「挂起即等人」:挂起任务登记唤醒条件,条件满足自动唤醒注入续跑。首期进程内存
//! registry(随 app 存亡,不落库——唤醒条件本就依附 app 存续,接力棒场景由进程外脚本
//! 兜底)。条件三型:delay(到点)/ file-appear(路径出现)/ process-exit(进程退出)。
//!
//! ## 不新增机制(设计 §2.2)
//! 唤醒动作复用既有注入通路:向目标会话注入一条 user 角色消息并续跑。既有无人在场
//! 领单走正典 `spawn_agentic_session`(会话生成态守卫 + epoch 语义俱在),本模块沿用
//! 同一路径,不造第二套注入。守卫语义照设计——目标会话正在生成时注入被拒,条目保留
//! 待下轮重试(对齐「轮界排队,不打断流」)。
//!
//! ## 防重(设计 §2.1)
//! once 语义(触发成功即移除)+ expires 过期清理(默认 24h)+ registry 上限 64。
//!
//! ## 判定纯函数化
//! 条件判定([`condition_met`]/[`decide`])为纯函数,IO(时间/文件存在/进程存活)由
//! [`observe`] 采集后喂入,便于单测三型条件与过期/触发优先级。
//!
//! ## 开关
//! 默认关(KV `df-wake-enabled`,关时调度器每轮只读配置空转,零行为变更)。
//!
//! ## 接线
//! 调度器入口 [`start`] 需在 lib.rs setup 挂载一次(与 unattended::start 同处),本批未越界。
//! 挂载前本模块整体不可达,`allow(dead_code)` 即为此待接线态;接线后应移除该属性恢复死码检查。
#![allow(dead_code)]

use std::sync::{LazyLock, RwLock};
use std::time::Duration;

use tauri::{AppHandle, Manager};

use crate::commands::ai::commands::chat::spawn_agentic_session;
use crate::state::AppState;

/// 开关 KV:默认 false。关时调度器空转,且无条目注册 → 全链路零行为变更。
const KV_ENABLED: &str = "df-wake-enabled";
/// 调度器轮询周期(秒)。delay 型精度上界 + file/process 型发现延迟上界。
const POLL_SECS: u64 = 30;
/// registry 条数上限(设计 §2.1「宁少勿预留」)。
const MAX_ENTRIES: usize = 64;
/// 默认条目存活期(24h,设计 §2.1)。到点未触发即过期清理。
const DEFAULT_TTL_MILLIS: i64 = 24 * 60 * 60 * 1000;

/// 唤醒条件(首期三型,设计 §2.1)。
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WakeCondition {
    /// 延时到点(绝对毫秒时间戳,对齐 [`df_types::types::now_millis`] 口径)。
    Delay { due_at_millis: i64 },
    /// 文件出现(路径存在即满足;接力棒 result.json 场景)。
    FileAppear { path: String },
    /// 进程退出(pid 不再存活即满足;自换装场景)。
    ProcessExit { pid: u32 },
}

/// registry 条目。
#[derive(Debug, Clone)]
pub struct WakeEntry {
    pub id: String,
    /// 唤醒注入的目标会话。
    pub conv_id: String,
    pub condition: WakeCondition,
    /// 唤醒上下文人话(注入消息正文;条件判据由消息尾缀追加,不混入 payload)。
    pub payload: String,
    pub created_at_millis: i64,
    pub expires_at_millis: i64,
}

/// 单条目条件判定的观测快照。IO 由 [`observe`] 采集,判定纯函数化便于单测。
/// 非对应型条件的字段无意义(判定只读所属型字段),采集侧可填占位值。
#[derive(Debug, Clone, Copy)]
pub struct WakeObservation {
    pub now_millis: i64,
    /// file-appear 目标当前是否存在。
    pub file_exists: bool,
    /// process-exit 目标进程当前是否仍存活。
    pub process_alive: bool,
}

/// 本轮对单条目的处置:触发 / 过期 / 继续等待。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WakeDecision {
    Fire,
    Expired,
    Wait,
}

/// 条件判定(纯函数,设计三型核心):条件是否已满足。
pub fn condition_met(cond: &WakeCondition, obs: &WakeObservation) -> bool {
    match cond {
        WakeCondition::Delay { due_at_millis } => obs.now_millis >= *due_at_millis,
        WakeCondition::FileAppear { .. } => obs.file_exists,
        WakeCondition::ProcessExit { .. } => !obs.process_alive,
    }
}

/// 单条目本轮终判(纯函数):过期优先于条件满足——过期即弃,不注入陈旧唤醒。
pub fn decide(cond: &WakeCondition, expires_at_millis: i64, obs: &WakeObservation) -> WakeDecision {
    if obs.now_millis >= expires_at_millis {
        return WakeDecision::Expired;
    }
    if condition_met(cond, obs) {
        WakeDecision::Fire
    } else {
        WakeDecision::Wait
    }
}

/// 条件人话(注入消息尾缀,让 LLM 知道因何被唤醒)。
fn condition_desc(cond: &WakeCondition) -> String {
    match cond {
        WakeCondition::Delay { due_at_millis } => format!("延时到点({due_at_millis})"),
        WakeCondition::FileAppear { path } => format!("文件出现:{path}"),
        WakeCondition::ProcessExit { pid } => format!("进程退出:pid {pid}"),
    }
}

/// 唤醒注入消息(纯函数):payload 为唤醒上下文,尾缀标明触发条件。
pub fn wake_message(entry: &WakeEntry) -> String {
    format!(
        "[异步唤醒] {}\n\n唤醒条件已满足({}),请基于当前会话状态续跑后续工作。",
        entry.payload,
        condition_desc(&entry.condition)
    )
}

/// 进程级 registry(条目少,Vec 有序 + retain 清理足够,不强求 HashMap)。
static REGISTRY: LazyLock<RwLock<Vec<WakeEntry>>> = LazyLock::new(|| RwLock::new(Vec::new()));

/// 登记一条唤醒条件。
/// - `ttl_millis` None → 默认 24h。
/// - 返回 Err = registry 已满(上限 64),登记方据此放弃或改期。
pub fn register(
    conv_id: &str,
    condition: WakeCondition,
    payload: &str,
    ttl_millis: Option<i64>,
) -> Result<String, String> {
    let now = df_types::types::now_millis();
    let mut reg = REGISTRY.write().expect("唤醒 registry 锁");
    reg.retain(|e| e.expires_at_millis > now); // 顺手清过期,腾槽位
    if reg.len() >= MAX_ENTRIES {
        return Err(format!("唤醒 registry 已满(上限 {MAX_ENTRIES})"));
    }
    let id = df_types::types::new_id();
    reg.push(WakeEntry {
        id: id.clone(),
        conv_id: conv_id.to_string(),
        condition,
        payload: payload.to_string(),
        created_at_millis: now,
        expires_at_millis: now + ttl_millis.unwrap_or(DEFAULT_TTL_MILLIS),
    });
    Ok(id)
}

/// 当前活跃条目快照(只读;顺手清过期)。
pub fn list() -> Vec<WakeEntry> {
    let now = df_types::types::now_millis();
    let mut reg = REGISTRY.write().expect("唤醒 registry 锁");
    reg.retain(|e| e.expires_at_millis > now);
    reg.clone()
}

/// 取消一条(按 id)。返回值 = 是否命中(供调用方区分「已触发/已过期」与「取消了在等条目」)。
pub fn cancel(id: &str) -> bool {
    let mut reg = REGISTRY.write().expect("唤醒 registry 锁");
    let before = reg.len();
    reg.retain(|e| e.id != id);
    reg.len() != before
}

fn remove_entry(id: &str) {
    let mut reg = REGISTRY.write().expect("唤醒 registry 锁");
    reg.retain(|e| e.id != id);
}

/// 采集单条目条件观测(IO 层:时间/文件存在/进程存活)。
fn observe(cond: &WakeCondition, now_millis: i64) -> WakeObservation {
    WakeObservation {
        now_millis,
        file_exists: match cond {
            WakeCondition::FileAppear { path } => std::path::Path::new(path).exists(),
            _ => false,
        },
        process_alive: match cond {
            WakeCondition::ProcessExit { pid } => process_alive(*pid),
            _ => true,
        },
    }
}

/// 进程存活查询。std 无跨平台接口,按平台最小实现:
/// - Unix: /proc/<pid> 存在。
/// - Windows: tasklist 按 PID 过滤(低频轮询 + registry ≤64,开销可接受)。
/// 查询失败保守判「存活」(不触发)——避免不确定时抛出误唤醒注入。
#[cfg(unix)]
fn process_alive(pid: u32) -> bool {
    std::path::Path::new(&format!("/proc/{pid}")).exists()
}

#[cfg(windows)]
fn process_alive(pid: u32) -> bool {
    let out = std::process::Command::new("tasklist")
        .args(["/FI", &format!("PID eq {pid}"), "/NH", "/FO", "CSV"])
        .output();
    match out {
        Ok(o) => String::from_utf8_lossy(&o.stdout).contains(&format!("\"{pid}\"")),
        Err(_) => true, // 查询失败:保守判存活,不误触发
    }
}

#[cfg(not(any(unix, windows)))]
fn process_alive(_pid: u32) -> bool {
    true
}

/// 唤醒注入:复用正典注入路径向目标会话投递 user 消息并续跑。
/// 目标会话生成中 → spawn 被 generating 守卫拒绝,返回 false(条目保留,下轮重试)。
async fn fire(app: &AppHandle, entry: &WakeEntry) -> bool {
    let state = app.state::<AppState>();
    let message = wake_message(entry);
    match spawn_agentic_session(
        app,
        &state,
        message,
        None, // language 沿用默认
        None, // skill 不挂
        None, // model_override 沿用路由
        Some(entry.conv_id.clone()),
        None,
        None,
        false, // pin_active=false:唤醒不劫持用户活跃面板
        None, // #958:唤醒路径无外部来源
    )
    .await
    {
        Ok(_) => {
            tracing::info!(conv_id = %entry.conv_id, "[wake] 唤醒条件满足,已注入续跑");
            true
        }
        Err(err) => {
            // 生成中/暂不可接受:保留条目下轮重试(对齐「轮界排队,不打断流」)
            tracing::warn!(conv_id = %entry.conv_id, "[wake] 注入暂不可用,条目保留下轮重试: {}", err);
            false
        }
    }
}

/// 单轮:扫描 registry,过期的清、条件满足的注入。开关关时调用方已短路。
async fn run_tick(app: &AppHandle) {
    let now = df_types::types::now_millis();
    // 先快照后处理:注入是 async,不得持 registry 锁跨 await。
    let entries = list();
    for entry in entries {
        let obs = observe(&entry.condition, now);
        match decide(&entry.condition, entry.expires_at_millis, &obs) {
            WakeDecision::Wait => {}
            WakeDecision::Expired => {
                remove_entry(&entry.id);
                tracing::info!(conv_id = %entry.conv_id, "[wake] 条目过期清理");
            }
            WakeDecision::Fire => {
                // once:注入成功即移除;失败保留(生成态/瞬时错误下轮重试)。
                if fire(app, &entry).await {
                    remove_entry(&entry.id);
                }
            }
        }
    }
}

/// 调度器挂载入口(仿 `unattended::start`,lib.rs setup 调用一次)。
/// 开关关时每轮只读配置空转,零副作用。
pub fn start(app: AppHandle) {
    tauri::async_runtime::spawn(async move {
        let mut interval = tokio::time::interval(Duration::from_secs(POLL_SECS));
        interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            interval.tick().await;
            let state = app.state::<AppState>();
            let enabled = matches!(
                state.settings.get(KV_ENABLED).await.ok().flatten(),
                Some(v) if {
                    let v = v.trim().trim_matches('"').to_lowercase();
                    v == "true" || v == "1"
                }
            );
            if !enabled {
                continue;
            }
            run_tick(&app).await;
        }
    });
}

#[cfg(test)]
mod tests {
    use super::*;

    const CONV: &str = "conv-wake";

    /// 单测注册表复用进程级 static,串行化避免相互清理干扰。
    static TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

    fn obs(now: i64, file: bool, alive: bool) -> WakeObservation {
        WakeObservation { now_millis: now, file_exists: file, process_alive: alive }
    }

    #[test]
    fn delay_fires_at_or_after_due() {
        let cond = WakeCondition::Delay { due_at_millis: 1000 };
        assert_eq!(decide(&cond, 9999, &obs(999, false, true)), WakeDecision::Wait);
        assert_eq!(decide(&cond, 9999, &obs(1000, false, true)), WakeDecision::Fire);
        assert_eq!(decide(&cond, 9999, &obs(1500, false, true)), WakeDecision::Fire);
    }

    #[test]
    fn file_appear_fires_only_when_exists() {
        let cond = WakeCondition::FileAppear { path: "E:/x/result.json".into() };
        assert_eq!(decide(&cond, 9999, &obs(0, false, true)), WakeDecision::Wait);
        assert_eq!(decide(&cond, 9999, &obs(0, true, true)), WakeDecision::Fire);
    }

    #[test]
    fn process_exit_fires_only_when_not_alive() {
        let cond = WakeCondition::ProcessExit { pid: 4242 };
        assert_eq!(decide(&cond, 9999, &obs(0, false, true)), WakeDecision::Wait);
        assert_eq!(decide(&cond, 9999, &obs(0, false, false)), WakeDecision::Fire);
    }

    #[test]
    fn expired_wins_over_met_condition() {
        // 过期优先:即便条件已满足(延时已到),过期即弃不注入陈旧唤醒。
        let cond = WakeCondition::Delay { due_at_millis: 100 };
        assert_eq!(decide(&cond, 500, &obs(1000, false, true)), WakeDecision::Expired);
    }

    #[test]
    fn wake_message_carries_payload_and_condition() {
        let entry = WakeEntry {
            id: "w1".into(),
            conv_id: CONV.into(),
            condition: WakeCondition::FileAppear { path: "E:/x/result.json".into() },
            payload: "接力棒已完成,请读取结果并继续".into(),
            created_at_millis: 0,
            expires_at_millis: 1,
        };
        let msg = wake_message(&entry);
        assert!(msg.contains("接力棒已完成,请读取结果并继续"), "应含 payload: {msg}");
        assert!(msg.contains("E:/x/result.json"), "应含触发条件人话: {msg}");
    }

    #[test]
    fn registry_register_cancel_and_cap() {
        let _guard = TEST_LOCK.lock().unwrap_or_else(|p| p.into_inner());
        // 清场:上轮遗留条目
        for e in list() {
            remove_entry(&e.id);
        }
        let id = register(CONV, WakeCondition::Delay { due_at_millis: 10 }, "p", None)
            .expect("登记应成功");
        assert_eq!(list().len(), 1);
        assert!(cancel(&id), "取消在等条目应命中");
        assert!(!cancel(&id), "重复取消不应再命中");
        assert!(list().is_empty(), "取消后 registry 应空");
    }
}