feat(server): relais live-state + actions background via web (event.domain, allowlist) (#13)

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 <noreply@anthropic.com>

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
+//! The shared backend core stays unaware of HTTP, cookies, origins and
-//! live WebSocket streams are left to later lots. The shared backend core stays
+//! WebSocket framing; this module owns the secure web driving adapter.
-//! unaware of HTTP, cookies and origins.
 
 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,
+    parse_agent_id, parse_node_id, parse_project_id, parse_session_id, parse_task_id,
-    HealthResponseDto, LaunchAgentRequestDto, OpenTerminalRequestDto, ProjectDto, ProjectListDto,
+    BackgroundTaskDto, ErrorDto, HealthRequestDto, HealthResponseDto, LaunchAgentRequestDto,
-    ProjectWorkStateDto, TerminalSessionDto,
+    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::<ServerFrame>(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<DomainEvent>,
+    tx: mpsc::Sender<ServerFrame>,
+) -> 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<ServerState>,
@@ -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<Value, ErrorDto> {
+    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::<domain::TaskId, domain::BackgroundTask>::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::<Vec<_>>();
+    tasks.sort_by_key(|task| (task.created_at_ms, task.id));
+    let dto = tasks
+        .into_iter()
+        .map(BackgroundTaskDto::from)
+        .collect::<Vec<_>>();
+    serde_json::to_value(dto).map_err(serialization_error)
+}
+
+async fn invoke_cancel_background_task(args: &Value, state: &AppState) -> Result<Value, ErrorDto> {
+    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<Value, ErrorDto> {
+    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<domain::TaskId, ErrorDto> {
+    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<Project, ErrorDto> {
     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<String>,
         session_id: Option<String>,
@@ -1775,9 +1939,11 @@ mod tests {
         CreateAgentInput, CreateProjectInput, LaunchAgentOutput, SaveProfileInput,
         StructuredSessionDescriptor,
     };
+    use domain::ports::EventBus;
     use domain::{
-        AgentId, AgentProfile, ContextInjection, NodeId, ProfileId, ProjectPath, PtySize,
+        AgentId, AgentProfile, BackgroundTask, BackgroundTaskKind, BackgroundTaskWakePolicy,
-        SessionKind, SessionStatus, SessionStrategy, TerminalSession,
+        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<ServerState>,
+        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<ServerState>,
         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();