From 6391d1c75a7d47e1320968103ef54cc65d7326e3 Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 15 Jul 2026 23:17:11 +0200 Subject: [PATCH] feat(server): agents CLI via le PTY WebSocket sur idea --serve (#13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- crates/app-tauri/src/server.rs | 465 ++++++++++++++++++++++++++++++++- 1 file changed, 460 insertions(+), 5 deletions(-) diff --git a/crates/app-tauri/src/server.rs b/crates/app-tauri/src/server.rs index 0e78821..8a4c62a 100644 --- a/crates/app-tauri/src/server.rs +++ b/crates/app-tauri/src/server.rs @@ -28,15 +28,16 @@ use tokio::sync::mpsc; use uuid::Uuid; use application::{ - CloseTerminalInput, GetProjectWorkStateInput, OpenProjectInput, ResizeTerminalInput, - WriteToTerminalInput, + CloseTerminalInput, GetProjectWorkStateInput, LaunchAgentInput, McpRuntime, OpenProjectInput, + ResizeTerminalInput, RotateConversationLogInput, WriteToTerminalInput, }; use domain::ports::PtyHandle; use domain::{Project, SessionId}; use crate::dto::{ - parse_project_id, parse_session_id, ErrorDto, HealthRequestDto, HealthResponseDto, - OpenTerminalRequestDto, ProjectDto, ProjectListDto, ProjectWorkStateDto, TerminalSessionDto, + parse_agent_id, parse_node_id, parse_project_id, parse_session_id, ErrorDto, HealthRequestDto, + HealthResponseDto, LaunchAgentRequestDto, OpenTerminalRequestDto, ProjectDto, ProjectListDto, + ProjectWorkStateDto, TerminalSessionDto, }; use crate::pty::PtyChunk; use crate::state::AppState; @@ -580,6 +581,7 @@ async fn handle_client_frame( ) { let result = match frame.kind.as_str() { "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.input" | "input" => ws_input(&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) } +async fn ws_launch_agent( + frame: &ClientFrame, + state: &Arc, + tx: &mpsc::Sender, + owned: &Arc>>, +) -> 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, + tx: &mpsc::Sender, + owned: &Arc>>, + 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, + request: LaunchAgentRequestDto, +) -> Result { + 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( frame: &ClientFrame, state: &Arc, @@ -1649,7 +1771,14 @@ fn payload_session_id(payload: &Value) -> Option { #[cfg(test)] mod tests { 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 std::time::Duration; @@ -1743,6 +1872,54 @@ mod tests { output.project.id.to_string() } + async fn create_raw_cli_agent_for_test( + state: &Arc, + 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 { let mut headers = HeaderMap::new(); headers.insert(ORIGIN, HeaderValue::from_str(origin).unwrap()); @@ -1777,6 +1954,19 @@ mod tests { .expect("server frame channel is open") } + async fn recv_server_frame_where( + rx: &mut mpsc::Receiver, + 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) { 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, session_id: &str) { let (tx, mut rx) = mpsc::channel(8); let owned = Arc::new(Mutex::new(Vec::new())); @@ -2108,6 +2314,255 @@ mod tests { 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] fn websocket_client_unmasked_frame_is_rejected() { let raw = [0x81, 0x02, b'h', b'i'];