use std::collections::HashMap;
use std::time::Duration;
use agent_client_protocol::schema::v1::{ContentBlock, SessionUpdate, StopReason};
use crate::acp::agent_proc;
use crate::acp::agent_proc::spawn_test_lock_async;
use crate::acp::client::AcpClient;
use crate::acp::terminal::TerminalActivity;
use crate::acp::test_support::{
KILL_TIMEOUT, REAP_TIMEOUT, SPAWN_TIMEOUT, UNWIND_TIMEOUT, agent_for, agent_for_with_env,
proc_alive, proc_dead, proc_reaped, read_events, spawn_connect, unique_dir, wait_for_event,
wait_until, wait_until_some, write_fake_agent,
};
use crate::models::agent::AgentEnvVar;
#[tokio::test]
async fn agent_dying_mid_prompt_fails_prompt_and_shuts_down_clean() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("crash");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "crash", &dir), workspace).await;
let pid = client.agent_pid().expect("D1:crash 模式也必须捕获 pid");
let prompt_result =
tokio::time::timeout(UNWIND_TIMEOUT, client.send_prompt("hi", vec![], vec![], vec![]))
.await
.expect("send_prompt 未在限时内返回(挂死)");
assert!(
prompt_result.is_err(),
"agent 死于 prompt 在途时 send_prompt 应失败,实际: {prompt_result:?}"
);
assert!(wait_until(|| proc_reaped(pid), UNWIND_TIMEOUT).await, "agent 死亡后未被 crate 回收");
assert!(
client.is_alive(),
"is_alive() 对已崩溃 agent 的误报存活是 crate 既有行为(见模块文档),\
此处 pin 住;若本断言失败说明 crate 行为已变,须复查 is_alive 判定与 \
backend.md 的描述"
);
client.shutdown().await;
assert!(wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await, "shutdown 后 agent 仍残留");
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn spawn_fails_promptly_when_agent_dies_during_handshake() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("handshake");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let result = tokio::time::timeout(
SPAWN_TIMEOUT,
AcpClient::spawn_and_connect(
agent_for(&script, "handshake", &dir),
workspace,
&HashMap::new(),
),
)
.await
.expect("spawn 挂起:agent 死于 handshake 时 spawn_and_connect 未限时返回");
assert!(result.is_err(), "agent 死于 handshake 时 spawn_and_connect 应快速失败");
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn agent_exits_without_traffic_shutdown_stays_clean() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("exit");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let exit_gate = dir.join("exit.gate");
let client = spawn_connect(agent_for(&script, "exit", &dir), workspace).await;
let pid = client.agent_pid().expect("D1:exit 模式也必须捕获 pid");
assert!(proc_alive(pid), "放行前 agent 必须驻留");
std::fs::write(&exit_gate, b"go").expect("write exit gate");
assert!(wait_until(|| proc_reaped(pid), UNWIND_TIMEOUT).await, "agent 退出后未被 crate 回收");
client.shutdown().await;
assert!(wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await, "shutdown 后 agent 仍残留");
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn shutdown_kills_live_agent_process_group_promptly() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("live");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "live", &dir), workspace).await;
let pid = client.agent_pid().expect("D1:真实 spawn 路径必须捕获 agent pid");
assert!(proc_alive(pid), "shutdown 前 agent 必须存活");
client.shutdown().await;
assert!(
wait_until(|| proc_dead(pid), KILL_TIMEOUT).await,
"shutdown 后 agent 未在 {KILL_TIMEOUT:?} 内死亡(killpg 未生效或走了 1s 优雅宽限期)"
);
assert!(
wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await,
"agent 死亡后未被回收(僵尸残留)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn shutdown_kill_reaches_grandchild_in_same_process_group() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("group");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let grandchild_pid_file = dir.join("grandchild.pid");
let client = spawn_connect(agent_for(&script, "group", &dir), workspace).await;
let leader = client.agent_pid().expect("D1:group 模式也必须捕获 leader pid");
let grandchild = wait_until_some(
|| {
std::fs::read_to_string(&grandchild_pid_file)
.ok()
.and_then(|s| s.trim().parse::<u32>().ok())
},
UNWIND_TIMEOUT,
)
.await
.expect("孙进程未在限时内登记 pid");
assert_ne!(grandchild, leader);
assert!(proc_alive(leader), "shutdown 前 leader 必须存活");
assert!(proc_alive(grandchild), "shutdown 前孙进程必须存活");
client.shutdown().await;
assert!(
wait_until(|| proc_dead(leader), KILL_TIMEOUT).await,
"shutdown 后 leader 未在 {KILL_TIMEOUT:?} 内死亡"
);
assert!(
wait_until(|| proc_dead(grandchild), KILL_TIMEOUT).await,
"shutdown 后孙进程未随进程组被击杀(D2 回归:只杀了单进程?)"
);
assert!(
wait_until(|| proc_reaped(leader), REAP_TIMEOUT).await,
"leader 死亡后未被回收(僵尸残留)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn restore_path_captures_pid_and_shutdown_kills_agent() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("restore");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = tokio::time::timeout(
SPAWN_TIMEOUT,
AcpClient::spawn_and_load(
agent_for(&script, "live", &dir),
workspace,
"fake-acp-session".to_string(),
&HashMap::new(),
),
)
.await
.expect("spawn_and_load 超时:fake agent 未响应 initialize")
.expect("spawn_and_load 失败");
assert!(client.supports_load_session(), "fake agent 已声明 loadSession 能力");
let pid = client.agent_pid().expect("D1:restore 路径必须捕获 agent pid");
assert!(proc_alive(pid), "shutdown 前 agent 必须存活");
client.shutdown().await;
assert!(
wait_until(|| proc_dead(pid), KILL_TIMEOUT).await,
"restore 路径 shutdown 后 agent 未在 {KILL_TIMEOUT:?} 内死亡"
);
assert!(
wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await,
"restore 路径 agent 死亡后未被回收(僵尸残留)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn spawn_timeout_during_handshake_does_not_leak_agent_process() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("hang");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let children_before = agent_proc::snapshot_direct_children();
let result = tokio::time::timeout(
Duration::from_secs(1),
AcpClient::spawn_and_connect(
agent_for(&script, "hang", &dir),
workspace.clone(),
&HashMap::new(),
),
)
.await;
assert!(result.is_err(), "hang agent 必须让 spawn 限时失败而不是挂死");
let pid = agent_proc::resolve_child_pid(&children_before, &workspace)
.expect("应能 diff 出 hang agent 的 pid");
assert!(proc_alive(pid), "spawn 超时前 agent 必须还在跑(否则测了个空)");
assert!(
wait_until(|| proc_dead(pid), Duration::from_secs(4)).await,
"spawn 超时(调用方消失)后 agent 进程泄漏(P2-3 回归)"
);
assert!(
wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await,
"hang agent 死亡后未被回收(僵尸残留)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn prompt_roundtrip_streams_session_update_and_completes() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("prompt-ok");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "prompt_ok", &dir), workspace).await;
let mut updates = client.session_update_subscribe();
client.mark_prompt_active();
let resp =
tokio::time::timeout(UNWIND_TIMEOUT, client.send_prompt("hi", vec![], vec![], vec![]))
.await
.expect("send_prompt 未在限时内返回(挂死)")
.expect("prompt 正常链路应返回 Ok");
assert_eq!(resp.stop_reason, StopReason::EndTurn, "fake agent 以 end_turn 应答");
client.mark_prompt_idle();
let frame = updates.try_recv().expect("应收到 session/update 广播");
assert!(frame.seq.is_some(), "turn 内折叠的帧必须带 seq(重连对账依赖)");
match frame.notification.update {
SessionUpdate::AgentMessageChunk(chunk) => assert!(
matches!(&chunk.content, ContentBlock::Text(t) if t.text == "hello-from-agent"),
"AgentMessageChunk 文本不符"
),
other => panic!("期望 AgentMessageChunk,实际 {other:?}"),
}
let snap = client.turn_snapshot();
assert!(snap.text.contains("hello-from-agent"), "累积器应折叠正文,实际: {:?}", snap.text);
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn cancel_notification_reaches_agent_and_prompt_ends_cancelled() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("cancel");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "cancel", &dir), workspace).await;
client.mark_prompt_active();
let prompt_fut = client.send_prompt("hi", vec![], vec![], vec![]);
let cancel_when_delivered = async {
assert!(
wait_for_event(&dir, "prompt", UNWIND_TIMEOUT).await,
"agent 未在限时内收到 prompt:{}",
read_events(&dir)
);
client.cancel().expect("session/cancel 通知应发送成功");
};
let (resp, ()) = tokio::join!(prompt_fut, cancel_when_delivered);
let resp = resp.expect("cancel 后 prompt 应以 cancelled 正常返回");
assert_eq!(resp.stop_reason, StopReason::Cancelled);
assert!(
wait_for_event(&dir, "cancel", UNWIND_TIMEOUT).await,
"agent 未收到 session/cancel:{}",
read_events(&dir)
);
client.mark_prompt_idle();
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn permission_request_resolve_roundtrip() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("perm-resolve");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "perm", &dir), workspace).await;
let mut perm_rx = client.permission_subscribe();
let mut resolved_rx = client.permission_resolved_subscribe();
client.mark_prompt_active();
let prompt_fut = client.send_prompt("hi", vec![], vec![], vec![]);
let resolve = async {
let ev = tokio::time::timeout(UNWIND_TIMEOUT, perm_rx.recv())
.await
.expect("等待权限请求事件超时")
.expect("权限事件通道关闭");
assert_eq!(client.pending_permissions().await, 1, "请求登记后未决数应为 1");
assert!(
client.resolve_permission(&ev.id, "allow").await,
"首个 resolve 应消费未决项并返回 true"
);
let resolved = tokio::time::timeout(UNWIND_TIMEOUT, resolved_rx.recv())
.await
.expect("等待 resolved 广播超时")
.expect("resolved 通道关闭");
assert_eq!(resolved, ev.id, "resolved 广播载荷应为审批 id");
};
let (resp, ()) = tokio::join!(prompt_fut, resolve);
assert!(resp.is_ok(), "权限被应答后 prompt 应正常收尾: {resp:?}");
assert_eq!(client.pending_permissions().await, 0, "resolve 后未决数应归零");
assert!(
wait_for_event(&dir, "\"optionId\":\"allow\"", UNWIND_TIMEOUT).await,
"agent 未收到 Selected 应答:{}",
read_events(&dir)
);
client.mark_prompt_idle();
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn cancel_answers_pending_permission_with_cancelled_outcome() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("perm-cancel");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "perm", &dir), workspace).await;
let mut perm_rx = client.permission_subscribe();
client.mark_prompt_active();
let prompt_fut = client.send_prompt("hi", vec![], vec![], vec![]);
let cancel_when_pending = async {
let _ev = tokio::time::timeout(UNWIND_TIMEOUT, perm_rx.recv())
.await
.expect("等待权限请求事件超时")
.expect("权限事件通道关闭");
client.cancel().expect("session/cancel 通知应发送成功");
};
let (resp, ()) = tokio::join!(prompt_fut, cancel_when_pending);
let resp = resp.expect("cancel 后 prompt 应正常返回");
assert_eq!(resp.stop_reason, StopReason::Cancelled);
assert!(
wait_for_event(&dir, "\"outcome\":\"cancelled\"", UNWIND_TIMEOUT).await,
"agent 未收到 Cancelled 应答:{}",
read_events(&dir)
);
assert_eq!(client.pending_permissions().await, 0, "cancel_all 后未决数应归零");
client.mark_prompt_idle();
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn shutdown_flips_alive_false_and_prompt_fails_fast() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("shutdown-alive");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "live", &dir), workspace).await;
assert!(client.is_alive(), "shutdown 前应存活");
let pid = client.agent_pid().expect("D1:live 模式必须捕获 pid");
client.shutdown().await;
assert!(!client.is_alive(), "shutdown 后 is_alive() 必须为 false");
let result =
tokio::time::timeout(UNWIND_TIMEOUT, client.send_prompt("hi", vec![], vec![], vec![]))
.await
.expect("shutdown 后 send_prompt 应快速失败而不是挂死");
assert!(result.is_err(), "shutdown 后发送必须失败,实际: {result:?}");
assert!(
wait_until(|| proc_dead(pid), KILL_TIMEOUT).await,
"shutdown 后 agent 未在 {KILL_TIMEOUT:?} 内死亡"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn disconnect_consumes_client_and_kills_agent() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("disconnect");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "live", &dir), workspace).await;
let pid = client.agent_pid().expect("D1:live 模式必须捕获 pid");
client.disconnect().await;
assert!(
wait_until(|| proc_dead(pid), KILL_TIMEOUT).await,
"disconnect 后 agent 未在 {KILL_TIMEOUT:?} 内死亡"
);
assert!(
wait_until(|| proc_reaped(pid), REAP_TIMEOUT).await,
"disconnect 后 agent 未被回收(僵尸残留)"
);
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn terminal_roundtrip_create_wait_output_release() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("term");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "term", &dir), workspace).await;
let mut terminal_events = client.terminal_event_subscribe();
client.mark_prompt_active();
let resp =
tokio::time::timeout(UNWIND_TIMEOUT, client.send_prompt("hi", vec![], vec![], vec![]))
.await
.expect("terminal 链路的 prompt 未限时返回")
.expect("terminal 链路应正常收尾");
assert_eq!(resp.stop_reason, StopReason::EndTurn);
client.mark_prompt_idle();
assert!(
wait_for_event(&dir, "\"exitCode\":7", UNWIND_TIMEOUT).await,
"terminal/wait_for_exit 未返回 exitCode 7:{}",
read_events(&dir)
);
assert!(
wait_for_event(&dir, "term-hello", UNWIND_TIMEOUT).await,
"terminal/output 未返回命令输出:{}",
read_events(&dir)
);
assert!(
wait_for_event(&dir, "term-release ", UNWIND_TIMEOUT).await,
"terminal/release 未得到应答:{}",
read_events(&dir)
);
let (mut created, mut exited) = (false, false);
while !(created && exited) {
let ev = tokio::time::timeout(UNWIND_TIMEOUT, terminal_events.recv())
.await
.expect("等待终端生命周期事件超时")
.expect("终端事件通道关闭");
match ev {
TerminalActivity::Created { .. } => created = true,
TerminalActivity::Exited { .. } => exited = true,
}
}
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn terminal_kill_then_wait_reports_no_exit_code() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("termkill");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let client = spawn_connect(agent_for(&script, "termkill", &dir), workspace).await;
client.mark_prompt_active();
let resp =
tokio::time::timeout(UNWIND_TIMEOUT, client.send_prompt("hi", vec![], vec![], vec![]))
.await
.expect("terminal kill 链路的 prompt 未限时返回")
.expect("terminal kill 链路应正常收尾");
assert_eq!(resp.stop_reason, StopReason::EndTurn);
client.mark_prompt_idle();
assert!(
wait_for_event(&dir, "\"id\":\"term-killwait\",\"result\":{}", UNWIND_TIMEOUT).await,
"terminal/kill 后 wait_for_exit 应返回无 exit_code 的退出状态:{}",
read_events(&dir)
);
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}
#[tokio::test]
async fn duplicate_env_keys_last_value_wins() {
let _guard = spawn_test_lock_async().await;
let dir = unique_dir("dup-env");
let workspace = dir.join("ws");
std::fs::create_dir_all(&workspace).expect("create workspace");
let script = write_fake_agent(&dir);
let extra = vec![
AgentEnvVar { key: "FAKE_DUP".into(), value: "first".into() },
AgentEnvVar { key: "FAKE_DUP".into(), value: "second".into() },
];
let client = spawn_connect(agent_for_with_env(&script, "live", &dir, extra), workspace).await;
let dup_file = dir.join("dup.txt");
let value = wait_until_some(|| std::fs::read_to_string(&dup_file).ok(), UNWIND_TIMEOUT)
.await
.expect("fake agent 未落盘 FAKE_DUP(脚本未生效?)");
assert_eq!(value, "second", "重复 env 键必须最后一条胜出(计划 D4 pin)");
client.shutdown().await;
let _ = std::fs::remove_dir_all(&dir);
}