feat(server): agents CLI via le PTY WebSocket sur idea --serve (#13)
Lot B6 du chantier server/client mode : le serveur --serve permet de lancer et piloter un agent CLI à travers le PTY WebSocket B5. - Frame WS agent.launch → ack terminal.attached avec assignedConversationId. - Réutilise LaunchAgent + TerminalSessions + sink WS B5, réattache sans respawn, singleton AGENT_ALREADY_RUNNING, structured → UNSUPPORTED. - 42 tests server (launch/reattach/singleton/structured), tests flaky stabilisés. Clippy server.rs propre. Validé : app-tauri 286 tests verts, backend 28, contrat B6↔F4 aligné (aucun écart de frame), desktop non régressé. Réserve connue : le round-trip socket réel relève d'une validation live hors sandbox. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -28,15 +28,16 @@ use tokio::sync::mpsc;
|
|||||||
use uuid::Uuid;
|
use uuid::Uuid;
|
||||||
|
|
||||||
use application::{
|
use application::{
|
||||||
CloseTerminalInput, GetProjectWorkStateInput, OpenProjectInput, ResizeTerminalInput,
|
CloseTerminalInput, GetProjectWorkStateInput, LaunchAgentInput, McpRuntime, OpenProjectInput,
|
||||||
WriteToTerminalInput,
|
ResizeTerminalInput, RotateConversationLogInput, WriteToTerminalInput,
|
||||||
};
|
};
|
||||||
use domain::ports::PtyHandle;
|
use domain::ports::PtyHandle;
|
||||||
use domain::{Project, SessionId};
|
use domain::{Project, SessionId};
|
||||||
|
|
||||||
use crate::dto::{
|
use crate::dto::{
|
||||||
parse_project_id, parse_session_id, ErrorDto, HealthRequestDto, HealthResponseDto,
|
parse_agent_id, parse_node_id, parse_project_id, parse_session_id, ErrorDto, HealthRequestDto,
|
||||||
OpenTerminalRequestDto, ProjectDto, ProjectListDto, ProjectWorkStateDto, TerminalSessionDto,
|
HealthResponseDto, LaunchAgentRequestDto, OpenTerminalRequestDto, ProjectDto, ProjectListDto,
|
||||||
|
ProjectWorkStateDto, TerminalSessionDto,
|
||||||
};
|
};
|
||||||
use crate::pty::PtyChunk;
|
use crate::pty::PtyChunk;
|
||||||
use crate::state::AppState;
|
use crate::state::AppState;
|
||||||
@ -580,6 +581,7 @@ async fn handle_client_frame(
|
|||||||
) {
|
) {
|
||||||
let result = match frame.kind.as_str() {
|
let result = match frame.kind.as_str() {
|
||||||
"terminal.open" | "open_terminal" => ws_open_terminal(&frame, state, tx, owned).await,
|
"terminal.open" | "open_terminal" => ws_open_terminal(&frame, state, tx, owned).await,
|
||||||
|
"agent.launch" | "launch_agent" => ws_launch_agent(&frame, state, tx, owned).await,
|
||||||
"terminal.attach" | "attach_terminal" => ws_attach_terminal(&frame, state, tx, owned).await,
|
"terminal.attach" | "attach_terminal" => ws_attach_terminal(&frame, state, tx, owned).await,
|
||||||
"terminal.input" | "input" => ws_input(&frame, state),
|
"terminal.input" | "input" => ws_input(&frame, state),
|
||||||
"terminal.resize" | "resize" => ws_resize(&frame, state),
|
"terminal.resize" | "resize" => ws_resize(&frame, state),
|
||||||
@ -646,6 +648,126 @@ async fn ws_open_terminal(
|
|||||||
attach_sink_and_pump(state, sid, sink, owned)
|
attach_sink_and_pump(state, sid, sink, owned)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn ws_launch_agent(
|
||||||
|
frame: &ClientFrame,
|
||||||
|
state: &Arc<ServerState>,
|
||||||
|
tx: &mpsc::Sender<ServerFrame>,
|
||||||
|
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
|
||||||
|
) -> Result<(), ErrorDto> {
|
||||||
|
let request_value = frame
|
||||||
|
.payload
|
||||||
|
.get("request")
|
||||||
|
.cloned()
|
||||||
|
.unwrap_or_else(|| frame.payload.clone());
|
||||||
|
let request: LaunchAgentRequestDto =
|
||||||
|
serde_json::from_value(request_value).map_err(invalid_args_error)?;
|
||||||
|
let output = execute_launch_agent_for_ws(state, request).await?;
|
||||||
|
send_launch_agent_attached(frame, state, tx, owned, output).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn send_launch_agent_attached(
|
||||||
|
frame: &ClientFrame,
|
||||||
|
state: &Arc<ServerState>,
|
||||||
|
tx: &mpsc::Sender<ServerFrame>,
|
||||||
|
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
|
||||||
|
output: application::LaunchAgentOutput,
|
||||||
|
) -> Result<(), ErrorDto> {
|
||||||
|
if output.structured.is_some() {
|
||||||
|
return Err(ErrorDto {
|
||||||
|
code: "UNSUPPORTED".to_owned(),
|
||||||
|
message: "structured agent sessions do not stream over the PTY websocket".to_owned(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
let session_id = output.session.id;
|
||||||
|
let dto = TerminalSessionDto::from(output);
|
||||||
|
let sid = parse_session_id(&dto.session_id)?;
|
||||||
|
let sink = WsPtySink::new(sid, tx.clone(), 0);
|
||||||
|
tx.send(ServerFrame::attached(&frame.id, &dto, Vec::new(), 0, false))
|
||||||
|
.await
|
||||||
|
.map_err(|_| ErrorDto {
|
||||||
|
code: "WS_CLOSED".to_owned(),
|
||||||
|
message: "websocket output closed".to_owned(),
|
||||||
|
})?;
|
||||||
|
attach_sink_and_pump(state, session_id, sink, owned)
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn execute_launch_agent_for_ws(
|
||||||
|
state: &Arc<ServerState>,
|
||||||
|
request: LaunchAgentRequestDto,
|
||||||
|
) -> Result<application::LaunchAgentOutput, ErrorDto> {
|
||||||
|
let project = resolve_project_readonly(&request.project_id, &state.app).await?;
|
||||||
|
let agent_id = parse_agent_id(&request.agent_id)?;
|
||||||
|
let node_id = request.node_id.as_deref().map(parse_node_id).transpose()?;
|
||||||
|
let mcp_runtime = crate::mcp_endpoint::idea_exe_path().map(|exe| McpRuntime {
|
||||||
|
exe,
|
||||||
|
endpoint: crate::mcp_endpoint::mcp_endpoint(&project.id)
|
||||||
|
.as_cli_arg()
|
||||||
|
.to_owned(),
|
||||||
|
project_id: project.id.as_uuid().simple().to_string(),
|
||||||
|
requester: agent_id.to_string(),
|
||||||
|
});
|
||||||
|
|
||||||
|
state.app.reconcile_claude_run_dirs(&project).await;
|
||||||
|
|
||||||
|
let resume_project = project.clone();
|
||||||
|
let rotation_root = project.root.clone();
|
||||||
|
let watch_root = project.root.clone();
|
||||||
|
let output = state
|
||||||
|
.app
|
||||||
|
.launch_agent
|
||||||
|
.execute(LaunchAgentInput {
|
||||||
|
project,
|
||||||
|
agent_id,
|
||||||
|
rows: request.rows,
|
||||||
|
cols: request.cols,
|
||||||
|
node_id,
|
||||||
|
conversation_id: request.conversation_id.clone(),
|
||||||
|
mcp_runtime,
|
||||||
|
allow_structured_alongside_pty: false,
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.map_err(ErrorDto::from)?;
|
||||||
|
|
||||||
|
if let Ok(mut contexts) = state.app.resume_contexts.lock() {
|
||||||
|
contexts.insert(
|
||||||
|
agent_id,
|
||||||
|
crate::state::ResumeContext {
|
||||||
|
project: resume_project,
|
||||||
|
rows: request.rows,
|
||||||
|
cols: request.cols,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(profile) = output.profile.as_ref() {
|
||||||
|
state.app.arm_turn_watch(
|
||||||
|
&watch_root,
|
||||||
|
agent_id,
|
||||||
|
profile,
|
||||||
|
output.assigned_conversation_id.clone(),
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Some(conversation) = request
|
||||||
|
.conversation_id
|
||||||
|
.as_deref()
|
||||||
|
.and_then(|raw| uuid::Uuid::parse_str(raw).ok())
|
||||||
|
.map(domain::ConversationId::from_uuid)
|
||||||
|
{
|
||||||
|
let rotate = Arc::clone(&state.app.rotate_conversation_log);
|
||||||
|
tokio::spawn(async move {
|
||||||
|
let _ = rotate
|
||||||
|
.execute(RotateConversationLogInput {
|
||||||
|
project_root: rotation_root,
|
||||||
|
conversation,
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(output)
|
||||||
|
}
|
||||||
|
|
||||||
async fn ws_attach_terminal(
|
async fn ws_attach_terminal(
|
||||||
frame: &ClientFrame,
|
frame: &ClientFrame,
|
||||||
state: &Arc<ServerState>,
|
state: &Arc<ServerState>,
|
||||||
@ -1649,7 +1771,14 @@ fn payload_session_id(payload: &Value) -> Option<String> {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
use application::CreateProjectInput;
|
use application::{
|
||||||
|
CreateAgentInput, CreateProjectInput, LaunchAgentOutput, SaveProfileInput,
|
||||||
|
StructuredSessionDescriptor,
|
||||||
|
};
|
||||||
|
use domain::{
|
||||||
|
AgentId, AgentProfile, ContextInjection, NodeId, ProfileId, ProjectPath, PtySize,
|
||||||
|
SessionKind, SessionStatus, SessionStrategy, TerminalSession,
|
||||||
|
};
|
||||||
use http::header::HeaderName;
|
use http::header::HeaderName;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
@ -1743,6 +1872,54 @@ mod tests {
|
|||||||
output.project.id.to_string()
|
output.project.id.to_string()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn create_raw_cli_agent_for_test(
|
||||||
|
state: &Arc<ServerState>,
|
||||||
|
name: &str,
|
||||||
|
) -> (String, String) {
|
||||||
|
let project_id = create_project_for_test(state, name).await;
|
||||||
|
let project = resolve_project_readonly(&project_id, &state.app)
|
||||||
|
.await
|
||||||
|
.expect("test project resolves");
|
||||||
|
let profile_id = ProfileId::from_uuid(Uuid::new_v4());
|
||||||
|
let profile = AgentProfile::new(
|
||||||
|
profile_id,
|
||||||
|
format!("{name} shell profile"),
|
||||||
|
"/bin/sh",
|
||||||
|
vec![
|
||||||
|
"-c".to_owned(),
|
||||||
|
"printf agent-ready; sleep 30".to_owned(),
|
||||||
|
"idea-test-sh".to_owned(),
|
||||||
|
],
|
||||||
|
ContextInjection::env("IDEA_CONTEXT").expect("valid env injection"),
|
||||||
|
None,
|
||||||
|
"{agentRunDir}",
|
||||||
|
Some(
|
||||||
|
SessionStrategy::new(Some("--session-id".to_owned()), "--resume")
|
||||||
|
.expect("valid session strategy"),
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.expect("valid raw CLI profile");
|
||||||
|
state
|
||||||
|
.app
|
||||||
|
.save_profile
|
||||||
|
.execute(SaveProfileInput { profile })
|
||||||
|
.await
|
||||||
|
.expect("test profile saved");
|
||||||
|
let agent = state
|
||||||
|
.app
|
||||||
|
.create_agent
|
||||||
|
.execute(CreateAgentInput {
|
||||||
|
project,
|
||||||
|
name: format!("{name} agent"),
|
||||||
|
profile_id,
|
||||||
|
initial_content: Some("Test agent context".to_owned()),
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.expect("test agent created")
|
||||||
|
.agent;
|
||||||
|
(project_id, agent.id.to_string())
|
||||||
|
}
|
||||||
|
|
||||||
fn ws_headers(origin: &str, cookie: Option<&str>) -> HeaderMap {
|
fn ws_headers(origin: &str, cookie: Option<&str>) -> HeaderMap {
|
||||||
let mut headers = HeaderMap::new();
|
let mut headers = HeaderMap::new();
|
||||||
headers.insert(ORIGIN, HeaderValue::from_str(origin).unwrap());
|
headers.insert(ORIGIN, HeaderValue::from_str(origin).unwrap());
|
||||||
@ -1777,6 +1954,19 @@ mod tests {
|
|||||||
.expect("server frame channel is open")
|
.expect("server frame channel is open")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn recv_server_frame_where(
|
||||||
|
rx: &mut mpsc::Receiver<ServerFrame>,
|
||||||
|
mut predicate: impl FnMut(&ServerFrame) -> bool,
|
||||||
|
) -> ServerFrame {
|
||||||
|
for _ in 0..16 {
|
||||||
|
let frame = recv_server_frame(rx).await;
|
||||||
|
if predicate(&frame) {
|
||||||
|
return frame;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
panic!("matching server frame was not received");
|
||||||
|
}
|
||||||
|
|
||||||
fn drain_server_frames(rx: &mut mpsc::Receiver<ServerFrame>) {
|
fn drain_server_frames(rx: &mut mpsc::Receiver<ServerFrame>) {
|
||||||
while rx.try_recv().is_ok() {}
|
while rx.try_recv().is_ok() {}
|
||||||
}
|
}
|
||||||
@ -1809,6 +1999,22 @@ mod tests {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn agent_launch_payload(
|
||||||
|
project_id: &str,
|
||||||
|
agent_id: &str,
|
||||||
|
node_id: &str,
|
||||||
|
conversation_id: Option<&str>,
|
||||||
|
) -> Value {
|
||||||
|
json!({
|
||||||
|
"projectId": project_id,
|
||||||
|
"agentId": agent_id,
|
||||||
|
"nodeId": node_id,
|
||||||
|
"rows": 24,
|
||||||
|
"cols": 80,
|
||||||
|
"conversationId": conversation_id,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
async fn close_ws_session(state: &Arc<ServerState>, session_id: &str) {
|
async fn close_ws_session(state: &Arc<ServerState>, session_id: &str) {
|
||||||
let (tx, mut rx) = mpsc::channel(8);
|
let (tx, mut rx) = mpsc::channel(8);
|
||||||
let owned = Arc::new(Mutex::new(Vec::new()));
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
@ -2108,6 +2314,255 @@ mod tests {
|
|||||||
assert!(state.app.terminal_sessions.is_empty());
|
assert!(state.app.terminal_sessions.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn websocket_launch_agent_emits_attached_with_assigned_conversation_id() {
|
||||||
|
let state = state();
|
||||||
|
let (project_id, agent_id) = create_raw_cli_agent_for_test(&state, "ws-launch").await;
|
||||||
|
let node_id = Uuid::new_v4().to_string();
|
||||||
|
let (tx, mut rx) = mpsc::channel(32);
|
||||||
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-1",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(&project_id, &agent_id, &node_id, None),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let attached = recv_server_frame(&mut rx).await;
|
||||||
|
let session_id = attached_session_id(&attached);
|
||||||
|
|
||||||
|
assert_eq!(attached.kind, "terminal.attached");
|
||||||
|
assert_eq!(attached.reply_to.as_deref(), Some("agent-launch-1"));
|
||||||
|
assert_eq!(attached.payload["scrollback"].as_array().unwrap().len(), 0);
|
||||||
|
assert!(attached.payload["assignedConversationId"]
|
||||||
|
.as_str()
|
||||||
|
.and_then(|raw| Uuid::parse_str(raw).ok())
|
||||||
|
.is_some());
|
||||||
|
|
||||||
|
close_ws_session(&state, &session_id).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn websocket_launch_agent_structured_is_unsupported() {
|
||||||
|
let state = state();
|
||||||
|
let (tx, mut rx) = mpsc::channel(4);
|
||||||
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
let session_id = SessionId::new_random();
|
||||||
|
let agent_id = AgentId::from_uuid(Uuid::new_v4());
|
||||||
|
let node_id = NodeId::from_uuid(Uuid::new_v4());
|
||||||
|
let size = PtySize::new(24, 80).unwrap();
|
||||||
|
let mut session = TerminalSession::starting(
|
||||||
|
session_id,
|
||||||
|
node_id,
|
||||||
|
ProjectPath::new("/".to_owned()).unwrap(),
|
||||||
|
SessionKind::Agent { agent_id },
|
||||||
|
size,
|
||||||
|
);
|
||||||
|
session.status = SessionStatus::Running;
|
||||||
|
let frame = ws_frame("agent-structured", "agent.launch", json!({}));
|
||||||
|
let output = LaunchAgentOutput {
|
||||||
|
session,
|
||||||
|
assigned_conversation_id: Some("pair-conversation".to_owned()),
|
||||||
|
engine_session_id: Some("engine-session".to_owned()),
|
||||||
|
structured: Some(StructuredSessionDescriptor {
|
||||||
|
session_id,
|
||||||
|
agent_id,
|
||||||
|
node_id,
|
||||||
|
conversation_id: Some("engine-session".to_owned()),
|
||||||
|
}),
|
||||||
|
profile: None,
|
||||||
|
};
|
||||||
|
|
||||||
|
if let Err(err) = send_launch_agent_attached(&frame, &state, &tx, &owned, output).await {
|
||||||
|
tx.send(ServerFrame::error(
|
||||||
|
Some(frame.id),
|
||||||
|
payload_session_id(&frame.payload),
|
||||||
|
err.code,
|
||||||
|
err.message,
|
||||||
|
))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
}
|
||||||
|
let error = recv_server_frame(&mut rx).await;
|
||||||
|
|
||||||
|
assert_eq!(error.kind, "error");
|
||||||
|
assert_eq!(error.reply_to.as_deref(), Some("agent-structured"));
|
||||||
|
assert_eq!(error.payload["code"], "UNSUPPORTED");
|
||||||
|
assert_eq!(
|
||||||
|
error.payload["message"],
|
||||||
|
"structured agent sessions do not stream over the PTY websocket"
|
||||||
|
);
|
||||||
|
assert_eq!(state.ws_pty_bridge.active_sessions(), 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn websocket_launch_agent_reattach_replays_scrollback_without_respawn() {
|
||||||
|
let state = state();
|
||||||
|
let (project_id, agent_id) = create_raw_cli_agent_for_test(&state, "ws-reattach").await;
|
||||||
|
let node_id = Uuid::new_v4().to_string();
|
||||||
|
let (tx, mut rx) = mpsc::channel(32);
|
||||||
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-reattach",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(&project_id, &agent_id, &node_id, None),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let launched = recv_server_frame(&mut rx).await;
|
||||||
|
let session_id = attached_session_id(&launched);
|
||||||
|
let scrollback = wait_for_scrollback(&state, &session_id, b"agent-ready").await;
|
||||||
|
drain_server_frames(&mut rx);
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-attach",
|
||||||
|
"terminal.attach",
|
||||||
|
json!({ "sessionId": session_id, "lastSeq": 0, "rows": 24, "cols": 80 }),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let attached = recv_server_frame_where(&mut rx, |frame| {
|
||||||
|
frame.kind == "terminal.attached" && frame.reply_to.as_deref() == Some("agent-attach")
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
let replay = attached.payload["scrollback"].as_array().unwrap();
|
||||||
|
|
||||||
|
assert_eq!(attached.kind, "terminal.attached");
|
||||||
|
assert_eq!(attached.payload["session"]["sessionId"], session_id);
|
||||||
|
assert_eq!(
|
||||||
|
base64_decode(replay[0]["bytesBase64"].as_str().unwrap()).unwrap(),
|
||||||
|
scrollback
|
||||||
|
);
|
||||||
|
let aid = parse_agent_id(&agent_id).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
state.app.terminal_sessions.sessions_for_agent(&aid).len(),
|
||||||
|
1
|
||||||
|
);
|
||||||
|
|
||||||
|
close_ws_session(&state, &session_id).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn websocket_launch_agent_same_cell_is_idempotent_singleton() {
|
||||||
|
let state = state();
|
||||||
|
let (project_id, agent_id) = create_raw_cli_agent_for_test(&state, "ws-singleton").await;
|
||||||
|
let node_id = Uuid::new_v4().to_string();
|
||||||
|
let (tx, mut rx) = mpsc::channel(32);
|
||||||
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-first",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(&project_id, &agent_id, &node_id, None),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let first = recv_server_frame(&mut rx).await;
|
||||||
|
let first_session = attached_session_id(&first);
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-second",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(
|
||||||
|
&project_id,
|
||||||
|
&agent_id,
|
||||||
|
&node_id,
|
||||||
|
first.payload["assignedConversationId"].as_str(),
|
||||||
|
),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let second = recv_server_frame_where(&mut rx, |frame| {
|
||||||
|
frame.kind == "terminal.attached"
|
||||||
|
&& frame.reply_to.as_deref() == Some("agent-launch-second")
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert_eq!(second.kind, "terminal.attached");
|
||||||
|
assert_eq!(attached_session_id(&second), first_session);
|
||||||
|
let aid = parse_agent_id(&agent_id).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
state.app.terminal_sessions.sessions_for_agent(&aid).len(),
|
||||||
|
1
|
||||||
|
);
|
||||||
|
|
||||||
|
close_ws_session(&state, &first_session).await;
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn websocket_launch_agent_different_cell_is_refused_by_singleton_guard() {
|
||||||
|
let state = state();
|
||||||
|
let (project_id, agent_id) =
|
||||||
|
create_raw_cli_agent_for_test(&state, "ws-singleton-refuse").await;
|
||||||
|
let first_node = Uuid::new_v4().to_string();
|
||||||
|
let second_node = Uuid::new_v4().to_string();
|
||||||
|
let (tx, mut rx) = mpsc::channel(32);
|
||||||
|
let owned = Arc::new(Mutex::new(Vec::new()));
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-first-cell",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(&project_id, &agent_id, &first_node, None),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let first = recv_server_frame(&mut rx).await;
|
||||||
|
let session_id = attached_session_id(&first);
|
||||||
|
|
||||||
|
handle_client_frame(
|
||||||
|
ws_frame(
|
||||||
|
"agent-launch-other-cell",
|
||||||
|
"agent.launch",
|
||||||
|
agent_launch_payload(&project_id, &agent_id, &second_node, None),
|
||||||
|
),
|
||||||
|
&state,
|
||||||
|
&tx,
|
||||||
|
&owned,
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let error = recv_server_frame_where(&mut rx, |frame| {
|
||||||
|
frame.kind == "error" && frame.reply_to.as_deref() == Some("agent-launch-other-cell")
|
||||||
|
})
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert_eq!(error.kind, "error");
|
||||||
|
assert_eq!(error.reply_to.as_deref(), Some("agent-launch-other-cell"));
|
||||||
|
assert_eq!(error.payload["code"], "AGENT_ALREADY_RUNNING");
|
||||||
|
let aid = parse_agent_id(&agent_id).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
state.app.terminal_sessions.sessions_for_agent(&aid).len(),
|
||||||
|
1
|
||||||
|
);
|
||||||
|
|
||||||
|
close_ws_session(&state, &session_id).await;
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn websocket_client_unmasked_frame_is_rejected() {
|
fn websocket_client_unmasked_frame_is_rejected() {
|
||||||
let raw = [0x81, 0x02, b'h', b'i'];
|
let raw = [0x81, 0x02, b'h', b'i'];
|
||||||
|
|||||||
Reference in New Issue
Block a user