use serde_json::json;
use tokio::time::{Duration, timeout};
use crate::app_server::AppServerInbound;
use crate::storage::PRIMARY_RUNTIME_ID;
use super::super::events::handle_app_server_message;
use super::support::bootstrap_test_state;
#[tokio::test]
async fn command_execution_render_snapshot_preserves_exec_kind() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/completed".to_string(),
params: json!({
"threadId": "thread-render-1",
"turnId": "turn-1",
"item": {
"id": "item-explored",
"type": "commandExecution",
"status": "completed",
"command": "ls -la",
"aggregatedOutput": "file-a\nfile-b\n",
"exitCode": 0,
"commandActions": [
{
"type": "listFiles",
"command": "ls -la",
"path": "."
}
]
}
}),
},
),
)
.await
.expect("处理 commandExecution item/completed 超时")
.expect("处理 commandExecution item/completed 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-1")
.expect("应缓存 thread render snapshot");
let exec_node = snapshot
.nodes
.iter()
.find_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::ExecGroup {
kind, output_text, ..
} => Some((kind.clone(), output_text.clone())),
_ => None,
})
.expect("应存在 exec group 节点");
assert_eq!("explored", exec_node.0);
assert_eq!(Some("file-a\nfile-b\n".to_string()), exec_node.1);
}
#[tokio::test]
async fn file_change_render_snapshot_preserves_per_file_diff() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/completed".to_string(),
params: json!({
"threadId": "thread-render-2",
"turnId": "turn-2",
"item": {
"id": "item-file-change",
"type": "fileChange",
"status": "completed",
"changes": [
{
"path": "/srv/workspace/src/main.rs",
"kind": {
"update": {}
},
"diff": "diff --git a/src/main.rs b/src/main.rs\n--- a/src/main.rs\n+++ b/src/main.rs\n@@ -1 +1 @@\n-old\n+new\n"
}
]
}
}),
},
),
)
.await
.expect("处理 fileChange item/completed 超时")
.expect("处理 fileChange item/completed 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-2")
.expect("应缓存 thread render snapshot");
let change = snapshot
.nodes
.iter()
.find_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::FileChange { changes, .. } => {
changes.first().cloned()
}
_ => None,
})
.expect("应存在 file change 节点");
assert_eq!("/srv/workspace/src/main.rs", change.path);
assert_eq!(
Some(
"diff --git a/src/main.rs b/src/main.rs\n--- a/src/main.rs\n+++ b/src/main.rs\n@@ -1 +1 @@\n-old\n+new\n"
.to_string()
),
change.diff,
);
}
#[tokio::test]
async fn user_message_render_snapshot_collects_data_urls_without_polluting_text() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/completed".to_string(),
params: json!({
"threadId": "thread-render-user-message",
"turnId": "turn-user-1",
"item": {
"id": "item-user-1",
"type": "userMessage",
"content": [
{
"type": "text",
"text": "请分析这张图"
},
{
"type": "image",
"url": "data:image/png;base64,aGVsbG8="
}
]
}
}),
},
),
)
.await
.expect("处理 userMessage item/completed 超时")
.expect("处理 userMessage item/completed 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-user-message")
.expect("应缓存用户消息快照");
let user_node = snapshot
.nodes
.iter()
.find_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::UserMessage {
text, image_urls, ..
} => Some((text.clone(), image_urls.clone())),
_ => None,
})
.expect("应存在 user message 节点");
assert_eq!("请分析这张图", user_node.0);
assert_eq!(
vec!["data:image/png;base64,aGVsbG8=".to_string()],
user_node.1
);
}
#[tokio::test]
async fn hook_notifications_upsert_single_render_node() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "hook/started".to_string(),
params: json!({
"threadId": "thread-render-hook",
"turnId": "turn-hook",
"run": {
"id": "hook-run-1",
"eventName": "preToolUse",
"statusMessage": "checking tool arguments"
}
}),
},
),
)
.await
.expect("处理 hook/started 超时")
.expect("处理 hook/started 失败");
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "hook/completed".to_string(),
params: json!({
"threadId": "thread-render-hook",
"turnId": "turn-hook",
"run": {
"id": "hook-run-1",
"eventName": "preToolUse",
"status": "completed",
"statusMessage": "checking tool arguments",
"entries": [
{
"kind": "feedback",
"text": "validated"
}
]
}
}),
},
),
)
.await
.expect("处理 hook/completed 超时")
.expect("处理 hook/completed 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-hook")
.expect("应缓存 hook thread render snapshot");
let hook_nodes = snapshot
.nodes
.iter()
.filter_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::HookEvent {
title,
state,
detail_lines,
..
} => Some((title.clone(), state.clone(), detail_lines.clone())),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(1, hook_nodes.len());
assert_eq!("PreToolUse hook (completed)", hook_nodes[0].0);
assert_eq!("completed", hook_nodes[0].1);
assert_eq!(
vec![
"checking tool arguments".to_string(),
"feedback: validated".to_string(),
],
hook_nodes[0].2,
);
}
#[tokio::test]
async fn terminal_interaction_notification_preserves_command_and_input() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/completed".to_string(),
params: json!({
"threadId": "thread-render-terminal",
"turnId": "turn-terminal",
"item": {
"id": "item-exec-terminal",
"type": "commandExecution",
"status": "completed",
"command": "python manage.py migrate"
}
}),
},
),
)
.await
.expect("处理 terminal command item/completed 超时")
.expect("处理 terminal command item/completed 失败");
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/commandExecution/terminalInteraction".to_string(),
params: json!({
"threadId": "thread-render-terminal",
"turnId": "turn-terminal",
"itemId": "item-exec-terminal",
"stdin": "yes\ncontinue\n"
}),
},
),
)
.await
.expect("处理 terminalInteraction 超时")
.expect("处理 terminalInteraction 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-terminal")
.expect("应缓存 terminal thread render snapshot");
let terminal = snapshot
.nodes
.iter()
.find_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::TerminalInteraction {
title,
command,
stdin,
waited,
..
} => Some((title.clone(), command.clone(), stdin.clone(), *waited)),
_ => None,
})
.expect("应存在 terminal interaction 节点");
assert_eq!("Interacted with background terminal", terminal.0);
assert_eq!(Some("python manage.py migrate".to_string()), terminal.1);
assert_eq!("yes\ncontinue\n".to_string(), terminal.2);
assert!(!terminal.3);
}
#[tokio::test]
async fn approval_review_notifications_upsert_single_render_node() {
let state = bootstrap_test_state().await;
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/autoApprovalReview/started".to_string(),
params: json!({
"threadId": "thread-render-approval",
"turnId": "turn-approval",
"targetItemId": "item-command-1",
"reviewId": "review-1",
"action": {
"type": "command",
"command": "git status",
"cwd": "/workspace"
},
"review": {
"status": "inProgress",
"riskLevel": "medium"
}
}),
},
),
)
.await
.expect("处理 approval review started 超时")
.expect("处理 approval review started 失败");
timeout(
Duration::from_secs(2),
handle_app_server_message(
&state,
AppServerInbound::Notification {
runtime_id: PRIMARY_RUNTIME_ID.to_string(),
method: "item/autoApprovalReview/completed".to_string(),
params: json!({
"threadId": "thread-render-approval",
"turnId": "turn-approval",
"targetItemId": "item-command-1",
"reviewId": "review-1",
"decisionSource": "guardian",
"action": {
"type": "command",
"command": "git status",
"cwd": "/workspace"
},
"review": {
"status": "approved",
"riskLevel": "medium",
"rationale": "matches allowlist"
}
}),
},
),
)
.await
.expect("处理 approval review completed 超时")
.expect("处理 approval review completed 失败");
let snapshots = state
.thread_render_snapshots
.lock()
.expect("thread render snapshots poisoned");
let snapshot = snapshots
.get("thread-render-approval")
.expect("应缓存 approval thread render snapshot");
let approval_nodes = snapshot
.nodes
.iter()
.filter_map(|node| match node {
crate::bridge_protocol::ThreadRenderNode::ApprovalReview {
title,
state,
detail_lines,
..
} => Some((title.clone(), state.clone(), detail_lines.clone())),
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(1, approval_nodes.len());
assert_eq!("Reviewed command approval request", approval_nodes[0].0);
assert_eq!("approved", approval_nodes[0].1);
assert_eq!(
vec![
"command: git status @ /workspace".to_string(),
"status: approved".to_string(),
"risk: medium".to_string(),
"matches allowlist".to_string(),
"decision source: guardian".to_string(),
],
approval_nodes[0].2,
);
}