use std::sync::Arc;
use std::sync::atomic::{AtomicU8, AtomicU64, Ordering};
use std::time::Duration;
use sqlx::sqlite::SqlitePool;
use tokio::time::interval;
use crate::acp::chat_persistence;
use crate::acp::client::{AcpClient, SystemNotice, TurnEndEvent};
use crate::acp::permission::{
MAX_PERM_NOTICE_REQUESTS, PermissionRequestSummary, pick_auto_option,
summarize_permission_request,
};
use crate::acp::supervisor::AcpSupervisor;
pub const IDLE_RECYCLE_SECS: u64 = 300;
pub const REQUIRES_ACTION_RECYCLE_SECS: u64 = 1800;
pub const PERM_TIMEOUT_NEVER_SECS: u64 = 0;
pub const PERM_TIMEOUT_STEP_SECS: u64 = 30;
pub const MAX_PERM_TIMEOUT_SECS: u64 = 3600;
pub const DEFAULT_PERM_TIMEOUT_SECS: u64 = REQUIRES_ACTION_RECYCLE_SECS;
pub fn is_valid_perm_timeout_secs(secs: u64) -> bool {
secs <= MAX_PERM_TIMEOUT_SECS && secs.is_multiple_of(PERM_TIMEOUT_STEP_SECS)
}
pub fn perm_timeout_secs_from_setting(setting: Option<&str>) -> Option<u64> {
let secs = setting.map(str::trim)?.parse::<u64>().ok()?;
is_valid_perm_timeout_secs(secs).then_some(secs)
}
pub fn perm_timeout_secs_from_legacy_min(setting: Option<&str>) -> Option<u64> {
let secs = setting.map(str::trim)?.parse::<u64>().ok()?.checked_mul(60)?;
is_valid_perm_timeout_secs(secs).then_some(secs)
}
fn format_perm_duration(secs: u64) -> String {
if secs.is_multiple_of(60) { format!("{} 分钟", secs / 60) } else { format!("{secs} 秒") }
}
pub const PROMPT_STALE_SECS: u64 = 600;
const TICK_SECS: u64 = 30;
pub const SYSTEM_LABEL_PERM_TIMEOUT_ABORT: &str = "system.permTimeout.abort";
pub const SYSTEM_LABEL_PERM_TIMEOUT_AUTO: &str = "system.permTimeout.auto";
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum PermissionTimeoutMode {
#[default]
Abort,
Auto,
Wait,
}
impl PermissionTimeoutMode {
pub fn as_str(&self) -> &'static str {
match self {
PermissionTimeoutMode::Abort => "abort",
PermissionTimeoutMode::Auto => "auto",
PermissionTimeoutMode::Wait => "wait",
}
}
pub fn from_str_opt(s: &str) -> Option<Self> {
match s.trim() {
"abort" => Some(PermissionTimeoutMode::Abort),
"auto" => Some(PermissionTimeoutMode::Auto),
"wait" => Some(PermissionTimeoutMode::Wait),
_ => None,
}
}
fn as_u8(self) -> u8 {
match self {
PermissionTimeoutMode::Abort => 0,
PermissionTimeoutMode::Auto => 1,
PermissionTimeoutMode::Wait => 2,
}
}
fn from_u8(v: u8) -> Self {
match v {
1 => PermissionTimeoutMode::Auto,
2 => PermissionTimeoutMode::Wait,
_ => PermissionTimeoutMode::Abort,
}
}
}
#[derive(Debug)]
pub struct PermissionTimeoutConfig {
mode: AtomicU8,
secs: AtomicU64,
}
impl Default for PermissionTimeoutConfig {
fn default() -> Self {
Self::new(PermissionTimeoutMode::Abort, REQUIRES_ACTION_RECYCLE_SECS)
}
}
impl PermissionTimeoutConfig {
pub fn new(mode: PermissionTimeoutMode, secs: u64) -> Self {
Self { mode: AtomicU8::new(mode.as_u8()), secs: AtomicU64::new(secs) }
}
pub fn snapshot(&self) -> (PermissionTimeoutMode, u64) {
(
PermissionTimeoutMode::from_u8(self.mode.load(Ordering::Relaxed)),
self.secs.load(Ordering::Relaxed),
)
}
pub fn store(&self, mode: PermissionTimeoutMode, secs: u64) {
self.mode.store(mode.as_u8(), Ordering::Relaxed);
self.secs.store(secs, Ordering::Relaxed);
}
}
struct PermissionTimeoutNotice {
label: &'static str,
text: String,
detail: serde_json::Value,
}
fn format_request_line(summary: &PermissionRequestSummary) -> String {
let tool = match (&summary.tool, &summary.kind) {
(Some(t), Some(k)) if t != k => format!("{t}({k})"),
(Some(t), _) => t.clone(),
(None, Some(k)) => k.clone(),
(None, None) => "未知工具".to_string(),
};
let mut parts = vec![format!("请求:{tool}")];
if let Some(content) = &summary.content {
let mut line = format!("内容:{content}");
if summary.content_omitted > 0 {
line.push_str(&format!("…(已省略 {} 字符)", summary.content_omitted));
}
parts.push(line);
}
if !summary.options.is_empty() {
parts.push(format!("可选项:{}", summary.options.join(" / ")));
}
parts.join(";")
}
fn build_perm_notice_abort(
summaries: &[PermissionRequestSummary],
extra: usize,
secs: u64,
) -> PermissionTimeoutNotice {
let mut text = format!(
"权限请求 {}未获响应,系统已自动取消该请求并回收会话(agent 已终止)。可重新打开会话继续。",
format_perm_duration(secs)
);
for s in summaries {
text.push(' ');
text.push_str(&format_request_line(s));
}
if extra > 0 {
text.push_str(&format!("(另有 {extra} 项审批一并取消)"));
}
PermissionTimeoutNotice {
label: SYSTEM_LABEL_PERM_TIMEOUT_ABORT,
text,
detail: detail_json(summaries.first(), None, secs, extra),
}
}
fn build_perm_notice_auto(
summary: &PermissionRequestSummary,
selected: &str,
secs: u64,
) -> PermissionTimeoutNotice {
let waited = if secs == PERM_TIMEOUT_NEVER_SECS {
"权限请求未获响应".to_string()
} else {
format!("权限请求 {}未获响应", format_perm_duration(secs))
};
let mut text = format!("{waited},已按设置自动选择「{selected}」继续执行。");
text.push(' ');
text.push_str(&format_request_line(summary));
PermissionTimeoutNotice {
label: SYSTEM_LABEL_PERM_TIMEOUT_AUTO,
text,
detail: detail_json(Some(summary), Some(selected), secs, 0),
}
}
fn detail_json(
summary: Option<&PermissionRequestSummary>,
selected: Option<&str>,
secs: u64,
extra: usize,
) -> serde_json::Value {
serde_json::json!({
"seconds": secs,
"tool": summary.and_then(|s| s.tool.clone()),
"kind": summary.and_then(|s| s.kind.clone()),
"content": summary.and_then(|s| s.content.clone()),
"content_omitted": summary.map_or(0, |s| s.content_omitted),
"options": summary.map(|s| s.options.clone()).unwrap_or_default(),
"selected": selected,
"extra": extra,
})
}
async fn persist_and_broadcast(
db: &SqlitePool,
session_id: &str,
client: &AcpClient,
notice: &PermissionTimeoutNotice,
) {
let blocks = serde_json::json!([
{ "type": "system", "label": notice.label, "detail": notice.detail }
])
.to_string();
if let Err(e) =
chat_persistence::insert_message(db, session_id, "system", ¬ice.text, Some(&blocks))
.await
{
tracing::warn!(
session_id = %session_id,
error = %e,
"reaper: failed to persist permission-timeout system message"
);
}
client.notify_system_message(SystemNotice {
label: notice.label.to_string(),
detail: Some(notice.detail.clone()),
});
}
async fn auto_advance_permissions(
db: &SqlitePool,
session_id: &str,
client: &AcpClient,
secs: u64,
) -> bool {
let events = client.pending_permission_events().await;
let mut resolved = 0usize;
for event in &events {
let Some((option_id, selected)) = pick_auto_option(&event.request) else {
tracing::warn!(
session_id = %session_id,
"reaper: permission request has no selectable option; skipping auto-advance for it"
);
continue;
};
if !client.resolve_permission(&event.id, &option_id).await {
continue;
}
resolved += 1;
if resolved > MAX_PERM_NOTICE_REQUESTS {
continue;
}
let summary = summarize_permission_request(&event.request);
let notice = build_perm_notice_auto(&summary, &selected, secs);
persist_and_broadcast(db, session_id, client, ¬ice).await;
}
if resolved > MAX_PERM_NOTICE_REQUESTS {
tracing::warn!(
session_id = %session_id,
resolved,
cap = MAX_PERM_NOTICE_REQUESTS,
"reaper: auto-advance resolved more permissions than the notice cap; extra ones answered without a chat notice"
);
}
resolved > 0
}
pub async fn run_reaper(
supervisor: AcpSupervisor,
db: SqlitePool,
idle_recycle_secs: Arc<AtomicU64>,
perm_timeout: Arc<PermissionTimeoutConfig>,
) {
let mut ticker = interval(Duration::from_secs(TICK_SECS));
loop {
ticker.tick().await;
let idle_secs = idle_recycle_secs.load(Ordering::Relaxed);
let (perm_mode, perm_secs) = perm_timeout.snapshot();
let mut to_reap: Vec<(String, bool /*perm_stale*/)> = Vec::new();
for (sid, client) in supervisor.snapshot().await {
if client.is_idle_stale(idle_secs).await {
to_reap.push((sid, false));
continue;
}
let pending = client.pending_permissions().await;
if pending > 0 {
let fired = if perm_secs == PERM_TIMEOUT_NEVER_SECS {
perm_mode == PermissionTimeoutMode::Auto
} else {
client.is_permission_stale(perm_secs).await
};
if fired {
match perm_mode {
PermissionTimeoutMode::Wait => {
}
PermissionTimeoutMode::Auto => {
if !auto_advance_permissions(&db, &sid, &client, perm_secs).await {
tracing::warn!(
session_id = %sid,
"reaper: auto-advance found no selectable option; falling back to abort"
);
to_reap.push((sid, true));
}
}
PermissionTimeoutMode::Abort => to_reap.push((sid, true)),
}
continue;
}
if perm_secs == PERM_TIMEOUT_NEVER_SECS || perm_mode == PermissionTimeoutMode::Wait
{
continue;
}
}
if client.is_prompt_stale(PROMPT_STALE_SECS) {
tracing::warn!(
session_id = %sid,
"prompt active but no agent activity for {}s; force-finalizing turn",
PROMPT_STALE_SECS
);
client.mark_prompt_idle();
client.notify_turn_end(TurnEndEvent::Done {
stop_reason: "InactivityTimeout".into(),
row_id: client.turn_row_id(),
duration: client.turn_timing(),
abnormal: false,
});
}
}
for (sid, perm_stale) in to_reap {
if let Some(client) = supervisor.dispose(&sid).await {
if perm_stale {
let events = client.pending_permission_events().await;
let summaries: Vec<PermissionRequestSummary> = events
.iter()
.take(MAX_PERM_NOTICE_REQUESTS)
.map(|e| summarize_permission_request(&e.request))
.collect();
let notice = build_perm_notice_abort(
&summaries,
events.len().saturating_sub(MAX_PERM_NOTICE_REQUESTS),
perm_secs,
);
persist_and_broadcast(&db, &sid, &client, ¬ice).await;
let _ = client.cancel();
}
client.shutdown().await;
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn mode_wire_format_is_whitelisted() {
assert_eq!(PermissionTimeoutMode::default(), PermissionTimeoutMode::Abort);
for mode in
[PermissionTimeoutMode::Abort, PermissionTimeoutMode::Auto, PermissionTimeoutMode::Wait]
{
assert_eq!(PermissionTimeoutMode::from_str_opt(mode.as_str()), Some(mode));
}
for bad in ["", "Abort", "AUTO", "never", "0"] {
assert_eq!(PermissionTimeoutMode::from_str_opt(bad), None, "bad mode: {bad:?}");
}
assert_eq!(
PermissionTimeoutMode::from_str_opt(" auto "),
Some(PermissionTimeoutMode::Auto)
);
}
#[test]
fn config_defaults_match_legacy_behavior() {
let cfg = PermissionTimeoutConfig::default();
assert_eq!(cfg.snapshot(), (PermissionTimeoutMode::Abort, REQUIRES_ACTION_RECYCLE_SECS));
}
#[test]
fn config_snapshot_store_roundtrip() {
let cfg = PermissionTimeoutConfig::new(PermissionTimeoutMode::Wait, 600);
assert_eq!(cfg.snapshot(), (PermissionTimeoutMode::Wait, 600));
cfg.store(PermissionTimeoutMode::Auto, 120);
assert_eq!(cfg.snapshot(), (PermissionTimeoutMode::Auto, 120));
cfg.mode.store(9, Ordering::Relaxed);
assert_eq!(cfg.snapshot().0, PermissionTimeoutMode::Abort);
}
#[test]
fn perm_timeout_secs_accepts_only_slider_notches() {
for ok in [
PERM_TIMEOUT_NEVER_SECS,
PERM_TIMEOUT_STEP_SECS,
60,
DEFAULT_PERM_TIMEOUT_SECS,
MAX_PERM_TIMEOUT_SECS,
] {
assert!(is_valid_perm_timeout_secs(ok), "should accept {ok}");
assert_eq!(perm_timeout_secs_from_setting(Some(&ok.to_string())), Some(ok));
}
for bad in [1, 45, 59, MAX_PERM_TIMEOUT_SECS + PERM_TIMEOUT_STEP_SECS, u64::MAX] {
assert!(!is_valid_perm_timeout_secs(bad), "should reject {bad}");
}
for bad in [None, Some(""), Some("abc"), Some(" 45 "), Some("-30")] {
assert_eq!(perm_timeout_secs_from_setting(bad), None, "bad value: {bad:?}");
}
assert_eq!(perm_timeout_secs_from_setting(Some(" 1800 ")), Some(1800));
}
#[test]
fn perm_timeout_secs_legacy_minutes_fallback() {
assert_eq!(perm_timeout_secs_from_legacy_min(Some("45")), Some(2700));
assert_eq!(perm_timeout_secs_from_legacy_min(Some(" 1 ")), Some(60));
for bad in [None, Some(""), Some("abc"), Some("61"), Some("18446744073709551615")] {
assert_eq!(perm_timeout_secs_from_legacy_min(bad), None, "bad legacy: {bad:?}");
}
}
fn sample_summary() -> PermissionRequestSummary {
PermissionRequestSummary {
tool: Some("Bash".into()),
kind: Some("execute".into()),
content: Some("git push origin main".into()),
content_omitted: 0,
options: vec!["允许一次".into(), "总是允许".into(), "拒绝".into()],
}
}
#[test]
fn abort_notice_carries_request_details() {
let notice = build_perm_notice_abort(&[sample_summary()], 0, DEFAULT_PERM_TIMEOUT_SECS);
assert_eq!(notice.label, SYSTEM_LABEL_PERM_TIMEOUT_ABORT);
assert!(notice.text.contains("30 分钟"), "{}", notice.text);
assert!(notice.text.contains("git push origin main"), "{}", notice.text);
assert!(notice.text.contains("可选项:允许一次 / 总是允许 / 拒绝"), "{}", notice.text);
let detail = ¬ice.detail;
assert_eq!(detail["seconds"], DEFAULT_PERM_TIMEOUT_SECS);
assert_eq!(detail["tool"], "Bash");
assert_eq!(detail["options"][1], "总是允许");
assert_eq!(detail["extra"], 0);
assert!(detail["selected"].is_null());
}
#[test]
fn notice_duration_spells_out_sub_minute_notches() {
let notice = build_perm_notice_abort(&[sample_summary()], 0, 30);
assert!(notice.text.starts_with("权限请求 30 秒未获响应"), "{}", notice.text);
assert_eq!(notice.detail["seconds"], 30);
let notice = build_perm_notice_abort(&[], 0, 90);
assert!(notice.text.starts_with("权限请求 90 秒未获响应"), "{}", notice.text);
let notice = build_perm_notice_abort(&[], 0, 60);
assert!(notice.text.starts_with("权限请求 1 分钟未获响应"), "{}", notice.text);
}
#[test]
fn auto_notice_for_never_notch_omits_duration() {
let notice = build_perm_notice_auto(&sample_summary(), "总是允许", PERM_TIMEOUT_NEVER_SECS);
assert_eq!(notice.label, SYSTEM_LABEL_PERM_TIMEOUT_AUTO);
assert!(
notice.text.starts_with("权限请求未获响应,已按设置自动选择「总是允许」继续执行。"),
"{}",
notice.text
);
assert!(
!notice.text.contains("0 分钟") && !notice.text.contains("0 秒"),
"{}",
notice.text
);
assert_eq!(notice.detail["seconds"], PERM_TIMEOUT_NEVER_SECS);
}
#[test]
fn abort_notice_reports_extra_requests() {
let notice = build_perm_notice_abort(&[sample_summary()], 3, DEFAULT_PERM_TIMEOUT_SECS);
assert!(notice.text.contains("另有 3 项审批一并取消"), "{}", notice.text);
assert_eq!(notice.detail["extra"], 3);
}
#[test]
fn auto_notice_names_the_selected_option() {
let notice = build_perm_notice_auto(&sample_summary(), "总是允许", 600);
assert_eq!(notice.label, SYSTEM_LABEL_PERM_TIMEOUT_AUTO);
assert!(notice.text.contains("自动选择「总是允许」"), "{}", notice.text);
assert!(notice.text.contains("10 分钟"), "{}", notice.text);
assert_eq!(notice.detail["selected"], "总是允许");
assert_eq!(notice.detail["seconds"], 600);
}
#[test]
fn auto_notice_marks_omitted_content() {
let mut s = sample_summary();
s.content = Some("x".repeat(10));
s.content_omitted = 42;
let notice = build_perm_notice_auto(&s, "允许一次", 5);
assert!(notice.text.contains("已省略 42 字符"), "{}", notice.text);
assert_eq!(notice.detail["content_omitted"], 42);
}
#[test]
fn abort_notice_without_summary_still_readable() {
let notice = build_perm_notice_abort(&[], 0, DEFAULT_PERM_TIMEOUT_SECS);
assert!(notice.text.starts_with("权限请求 30 分钟未获响应"), "{}", notice.text);
assert!(notice.detail["tool"].is_null());
}
}