blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
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>
This commit is contained in:
Blomios
2026-07-15 23:36:50 +02:00
parent 5254f16095
commit 1bc5217dbc
1 changed files with 378 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

386
crates/app-tauri/src/server.rs
View File

@ -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::<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,
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<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();