From 1bc5217dbc724fcfa212475e2e0f27146066964c Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 15 Jul 2026 23:36:50 +0200 Subject: [PATCH] feat(server): relais live-state + actions background via web (event.domain, allowlist) (#13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lot B7 du chantier server/client mode : le serveur --serve relaie l'état live et les tâches de fond au client web. - Relais du bus domaine en frames event.domain, queue bornée avec drop, exclusion de PtyOutput. - Allowlist étendue : list_background_tasks (read) + cancel_background_task / retry_background_task (actions), sous auth. - 49 tests server (fan-out, drop sur queue pleine, relais de complétion background, exclusion PtyOutput, actions allowlist + auth). Clippy propre. Validé : app-tauri 292+ tests verts, backend 28, cœur backend agnostique, desktop non régressé. Réserve connue : le fan-out socket réel relève d'une validation live hors sandbox. Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/server.rs | 386 ++++++++++++++++++++++++++++++++- 1 file changed, 378 insertions(+), 8 deletions(-) diff --git a/crates/app-tauri/src/server.rs b/crates/app-tauri/src/server.rs index 8a4c62a..7f71c5a 100644 --- a/crates/app-tauri/src/server.rs +++ b/crates/app-tauri/src/server.rs @@ -1,8 +1,7 @@ //! Secure HTTP driving adapter for `idea --serve`. //! -//! B3 deliberately exposes only request/response RPC over `/api/invoke`; PTY and -//! live WebSocket streams are left to later lots. The shared backend core stays -//! unaware of HTTP, cookies and origins. +//! The shared backend core stays unaware of HTTP, cookies, origins and +//! WebSocket framing; this module owns the secure web driving adapter. use std::collections::HashSet; use std::env; @@ -15,6 +14,8 @@ use std::sync::{Arc, Mutex}; use backend::stream::{OutputBridge, OutputSink, OutputSinkError}; use bytes::Bytes; use cookie::{Cookie, SameSite}; +use domain::events::DomainEvent; +use domain::ports::BackgroundTaskPortError; use http::header::{HeaderValue, CONTENT_TYPE, COOKIE, ORIGIN, SET_COOKIE}; #[cfg(test)] use http::Request; @@ -35,10 +36,11 @@ use domain::ports::PtyHandle; use domain::{Project, SessionId}; use crate::dto::{ - parse_agent_id, parse_node_id, parse_project_id, parse_session_id, ErrorDto, HealthRequestDto, - HealthResponseDto, LaunchAgentRequestDto, OpenTerminalRequestDto, ProjectDto, ProjectListDto, - ProjectWorkStateDto, TerminalSessionDto, + parse_agent_id, parse_node_id, parse_project_id, parse_session_id, parse_task_id, + BackgroundTaskDto, ErrorDto, HealthRequestDto, HealthResponseDto, LaunchAgentRequestDto, + OpenTerminalRequestDto, ProjectDto, ProjectListDto, ProjectWorkStateDto, TerminalSessionDto, }; +use crate::events::DomainEventDto; use crate::pty::PtyChunk; use crate::state::AppState; @@ -501,6 +503,8 @@ async fn run_ws_connection( let (mut reader, mut writer) = stream.into_split(); let (tx, mut rx) = mpsc::channel::(WS_OUTPUT_BUFFER); let owned = Arc::new(Mutex::new(Vec::<(SessionId, u64)>::new())); + let event_relay_task = + spawn_ws_domain_event_relay(state.app.event_bus.raw_receiver(), tx.clone()); let writer_task = tokio::spawn(async move { while let Some(frame) = rx.recv().await { let text = serde_json::to_vec(&frame) @@ -567,12 +571,38 @@ async fn run_ws_connection( state.ws_pty_bridge.unregister_if(session, *gen); } } + event_relay_task.abort(); drop(tx); writer_task .await .map_err(|err| format!("websocket writer task failed: {err}"))? } +fn spawn_ws_domain_event_relay( + mut rx: tokio::sync::broadcast::Receiver, + tx: mpsc::Sender, +) -> tokio::task::JoinHandle<()> { + use tokio::sync::broadcast::error::RecvError; + + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(event) => { + if matches!(event, DomainEvent::PtyOutput { .. }) { + continue; + } + match tx.try_send(ServerFrame::domain_event(&event)) { + Ok(()) | Err(mpsc::error::TrySendError::Full(_)) => {} + Err(mpsc::error::TrySendError::Closed(_)) => break, + } + } + Err(RecvError::Lagged(_)) => continue, + Err(RecvError::Closed) => break, + } + } + }) +} + async fn handle_client_frame( frame: ClientFrame, state: &Arc, @@ -1039,6 +1069,9 @@ async fn invoke( "list_projects" => invoke_list_projects(&state.app).await, "open_project" => invoke_open_project(&request.args, &state.app).await, "get_project_work_state" => invoke_get_project_work_state(&request.args, &state.app).await, + "list_background_tasks" => invoke_list_background_tasks(&request.args, &state.app).await, + "cancel_background_task" => invoke_cancel_background_task(&request.args, &state.app).await, + "retry_background_task" => invoke_retry_background_task(&request.args, &state.app).await, _ => Err(ErrorDto { code: "UNKNOWN_COMMAND".to_owned(), message: format!("unknown command: {}", request.command), @@ -1101,6 +1134,128 @@ async fn invoke_get_project_work_state(args: &Value, state: &AppState) -> Result serde_json::to_value(output).map_err(serialization_error) } +async fn invoke_list_background_tasks(args: &Value, state: &AppState) -> Result { + let project_id = args + .get("projectId") + .and_then(Value::as_str) + .ok_or_else(|| ErrorDto { + code: "INVALID".to_owned(), + message: "list_background_tasks requires args.projectId".to_owned(), + }) + .and_then(parse_project_id)?; + let agent_id = args + .get("agentId") + .and_then(Value::as_str) + .map(parse_agent_id) + .transpose()?; + + let mut by_id = std::collections::HashMap::::new(); + if let Some(agent_id) = agent_id { + for task in state + .background_task_store + .list_open_for_agent(agent_id) + .await + .map_err(background_error)? + { + by_id.insert(task.id, task); + } + } + for task in state + .background_task_store + .list_undelivered_completions() + .await + .map_err(background_error)? + { + by_id.entry(task.id).or_insert(task); + } + + let mut tasks = by_id + .into_values() + .filter(|task| task.project_id == project_id) + .filter(|task| agent_id.map_or(true, |agent_id| task.owner_agent_id == agent_id)) + .collect::>(); + tasks.sort_by_key(|task| (task.created_at_ms, task.id)); + let dto = tasks + .into_iter() + .map(BackgroundTaskDto::from) + .collect::>(); + serde_json::to_value(dto).map_err(serialization_error) +} + +async fn invoke_cancel_background_task(args: &Value, state: &AppState) -> Result { + let task_id = task_id_arg(args, "cancel_background_task")?; + verify_background_task_project(args, state, task_id).await?; + let output = state + .cancel_background_task + .execute(task_id) + .await + .map_err(ErrorDto::from)?; + serde_json::to_value(output.task.map(BackgroundTaskDto::from)).map_err(serialization_error) +} + +async fn invoke_retry_background_task(args: &Value, state: &AppState) -> Result { + let task_id = task_id_arg(args, "retry_background_task")?; + verify_background_task_project(args, state, task_id).await?; + let output = state + .retry_background_task + .execute(task_id) + .await + .map_err(ErrorDto::from)?; + serde_json::to_value(BackgroundTaskDto::from(output.task)).map_err(serialization_error) +} + +fn task_id_arg(args: &Value, command: &str) -> Result { + args.get("taskId") + .and_then(Value::as_str) + .ok_or_else(|| ErrorDto { + code: "INVALID".to_owned(), + message: format!("{command} requires args.taskId"), + }) + .and_then(parse_task_id) +} + +async fn verify_background_task_project( + args: &Value, + state: &AppState, + task_id: domain::TaskId, +) -> Result<(), ErrorDto> { + let Some(project_id) = args.get("projectId").and_then(Value::as_str) else { + return Ok(()); + }; + let project_id = parse_project_id(project_id)?; + let Some(task) = state + .background_task_store + .get(task_id) + .await + .map_err(background_error)? + else { + return Err(ErrorDto { + code: "NOT_FOUND".to_owned(), + message: format!("background task not found: {task_id}"), + }); + }; + if task.project_id != project_id { + return Err(ErrorDto { + code: "FORBIDDEN".to_owned(), + message: "background task does not belong to requested project".to_owned(), + }); + } + Ok(()) +} + +fn background_error(err: BackgroundTaskPortError) -> ErrorDto { + let code = match &err { + BackgroundTaskPortError::NotFound => "NOT_FOUND", + BackgroundTaskPortError::AlreadyExists | BackgroundTaskPortError::Invalid(_) => "INVALID", + BackgroundTaskPortError::Runner(_) => "PROCESS", + BackgroundTaskPortError::Store(_) => "STORE", + }; + ErrorDto { + code: code.to_owned(), + message: err.to_string(), + } +} + async fn resolve_project_readonly(project_id: &str, state: &AppState) -> Result { let id = parse_project_id(project_id)?; state @@ -1407,6 +1562,15 @@ impl ServerFrame { } } + fn domain_event(event: &DomainEvent) -> Self { + Self { + kind: "event.domain".to_owned(), + reply_to: None, + payload: serde_json::to_value(DomainEventDto::from(event)) + .expect("DomainEventDto serializes"), + } + } + fn error( request_id: Option, session_id: Option, @@ -1775,9 +1939,11 @@ mod tests { CreateAgentInput, CreateProjectInput, LaunchAgentOutput, SaveProfileInput, StructuredSessionDescriptor, }; + use domain::ports::EventBus; use domain::{ - AgentId, AgentProfile, ContextInjection, NodeId, ProfileId, ProjectPath, PtySize, - SessionKind, SessionStatus, SessionStrategy, TerminalSession, + AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskWakePolicy, + ContextInjection, NodeId, ProfileId, ProjectId, ProjectPath, PtySize, SessionKind, + SessionStatus, SessionStrategy, TaskId, TerminalSession, }; use http::header::HeaderName; use std::time::Duration; @@ -1872,6 +2038,35 @@ mod tests { output.project.id.to_string() } + async fn create_background_task_for_test( + state: &Arc, + project_id: &str, + label: &str, + ) -> (TaskId, AgentId) { + let task_id = TaskId::from_uuid(Uuid::new_v4()); + let owner = AgentId::from_uuid(Uuid::new_v4()); + let project_id = ProjectId::from_uuid(Uuid::parse_str(project_id).unwrap()); + let task = BackgroundTask::new( + task_id, + project_id, + owner, + BackgroundTaskKind::Command { + label: label.to_owned(), + }, + BackgroundTaskWakePolicy::WakeOwner, + 1_000, + None, + ) + .expect("test background task is valid"); + state + .app + .background_task_store + .create(&task) + .await + .expect("test background task is stored"); + (task_id, owner) + } + async fn create_raw_cli_agent_for_test( state: &Arc, name: &str, @@ -2186,6 +2381,125 @@ mod tests { assert!(frame.payload.as_object().unwrap().is_empty()); } + #[test] + fn websocket_domain_event_frame_uses_contract_shape() { + let project_id = ProjectId::from_uuid(Uuid::new_v4()); + let frame = ServerFrame::domain_event(&DomainEvent::ProjectCreated { project_id }); + + assert_eq!(frame.kind, "event.domain"); + assert!(frame.reply_to.is_none()); + assert_eq!(frame.payload["type"], "projectCreated"); + assert_eq!(frame.payload["projectId"], project_id.to_string()); + } + + #[tokio::test] + async fn websocket_domain_event_relay_forwards_domain_events() { + let state = state(); + let project_id = ProjectId::from_uuid(Uuid::new_v4()); + let event_rx = state.app.event_bus.raw_receiver(); + let (tx, mut rx) = mpsc::channel(4); + let relay = spawn_ws_domain_event_relay(event_rx, tx); + + state + .app + .event_bus + .publish(DomainEvent::ProjectCreated { project_id }); + let frame = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("event frame is forwarded") + .expect("event frame channel is open"); + relay.abort(); + + assert_eq!(frame.kind, "event.domain"); + assert_eq!(frame.payload["type"], "projectCreated"); + assert_eq!(frame.payload["projectId"], project_id.to_string()); + } + + #[tokio::test] + async fn websocket_domain_event_relay_forwards_background_completion() { + let state = state(); + let project_id = ProjectId::from_uuid(Uuid::new_v4()); + let task_id = TaskId::from_uuid(Uuid::new_v4()); + let owner = AgentId::from_uuid(Uuid::new_v4()); + let event_rx = state.app.event_bus.raw_receiver(); + let (tx, mut rx) = mpsc::channel(4); + let relay = spawn_ws_domain_event_relay(event_rx, tx); + + state + .app + .event_bus + .publish(DomainEvent::BackgroundTaskCompleted { + project_id, + task_id, + owner_agent_id: owner, + }); + let frame = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("background completion event is forwarded") + .expect("event frame channel is open"); + relay.abort(); + + assert_eq!(frame.kind, "event.domain"); + assert_eq!(frame.payload["type"], "backgroundTaskChanged"); + assert_eq!(frame.payload["projectId"], project_id.to_string()); + assert_eq!(frame.payload["taskId"], task_id.to_string()); + assert_eq!(frame.payload["agentId"], owner.to_string()); + assert_eq!(frame.payload["state"], "completed"); + } + + #[tokio::test] + async fn websocket_domain_event_relay_excludes_pty_output() { + let state = state(); + let project_id = ProjectId::from_uuid(Uuid::new_v4()); + let event_rx = state.app.event_bus.raw_receiver(); + let (tx, mut rx) = mpsc::channel(4); + let relay = spawn_ws_domain_event_relay(event_rx, tx); + + state.app.event_bus.publish(DomainEvent::PtyOutput { + session_id: SessionId::new_random(), + bytes: b"hidden from global live bus".to_vec(), + }); + state + .app + .event_bus + .publish(DomainEvent::ProjectCreated { project_id }); + let frame = tokio::time::timeout(Duration::from_secs(1), rx.recv()) + .await + .expect("sentinel event is forwarded") + .expect("event frame channel is open"); + relay.abort(); + + assert_eq!(frame.kind, "event.domain"); + assert_eq!(frame.payload["type"], "projectCreated"); + assert_eq!(frame.payload["projectId"], project_id.to_string()); + assert!( + rx.try_recv().is_err(), + "PtyOutput must not be emitted on the global event.domain stream" + ); + } + + #[tokio::test] + async fn websocket_domain_event_relay_drops_when_client_queue_is_full() { + let state = state(); + let event_rx = state.app.event_bus.raw_receiver(); + let (tx, mut rx) = mpsc::channel(1); + assert!(tx.try_send(ServerFrame::pong()).is_ok()); + let relay = spawn_ws_domain_event_relay(event_rx, tx); + + state.app.event_bus.publish(DomainEvent::ProjectCreated { + project_id: ProjectId::from_uuid(Uuid::new_v4()), + }); + tokio::time::sleep(Duration::from_millis(20)).await; + let first = rx.try_recv().expect("pre-filled frame remains queued"); + relay.abort(); + + assert_eq!(first.kind, "pong"); + assert!( + rx.try_recv().is_err(), + "full client queue should drop the live event instead of buffering" + ); + } + #[tokio::test] async fn websocket_open_terminal_emits_attached_with_empty_scrollback() { let state = state(); @@ -2932,6 +3246,62 @@ mod tests { assert!(body["conversations"].as_array().is_some()); } + #[tokio::test] + async fn authorized_list_background_tasks_returns_tauri_contract() { + let state = state(); + let project_id = create_project_for_test(&state, "Background Snapshot").await; + let (task_id, owner) = + create_background_task_for_test(&state, &project_id, "web background").await; + let cookie = pair_and_cookie(Arc::clone(&state)).await; + + let response = request( + state, + Method::POST, + "/api/invoke", + json!({ + "command": "list_background_tasks", + "args": { "projectId": project_id, "agentId": owner.to_string() } + }), + &[("cookie", &cookie)], + ) + .await; + let (status, body, _) = response_json(response).await; + + assert_eq!(status, StatusCode::OK); + let tasks = body.as_array().expect("background task list is an array"); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0]["taskId"], task_id.to_string()); + assert_eq!(tasks[0]["ownerAgentId"], owner.to_string()); + assert_eq!(tasks[0]["projectId"], project_id); + assert_eq!(tasks[0]["kind"], "command"); + assert_eq!(tasks[0]["state"], "queued"); + } + + #[tokio::test] + async fn background_actions_are_allowlisted_with_auth_gate() { + let state = state(); + let cookie = pair_and_cookie(Arc::clone(&state)).await; + let bad_task_id = "not-a-task-id"; + + for command in ["cancel_background_task", "retry_background_task"] { + let response = request( + Arc::clone(&state), + Method::POST, + "/api/invoke", + json!({ "command": command, "args": { "taskId": bad_task_id } }), + &[("cookie", &cookie)], + ) + .await; + let (status, body, _) = response_json(response).await; + + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!( + body["code"], "INVALID", + "{command} must be an explicit action, not UNKNOWN_COMMAND" + ); + } + } + #[tokio::test] async fn mutation_and_pty_commands_stay_out_of_readonly_allowlist() { let state = state();