blomios/IdeaSDK
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6391d1c75a7d47e1320968103ef54cc65d7326e3
IdeaSDK/crates/app-tauri/src/server.rs
Blomios 6391d1c75a 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>
2026-07-15 23:17:11 +02:00

2983 lines
93 KiB
Rust
Raw Blame History

//! 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.
use std::collections::HashSet;
use std::env;
use std::net::SocketAddr;
use std::path::PathBuf;
use std::process::ExitCode;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};
use backend::stream::{OutputBridge, OutputSink, OutputSinkError};
use bytes::Bytes;
use cookie::{Cookie, SameSite};
use http::header::{HeaderValue, CONTENT_TYPE, COOKIE, ORIGIN, SET_COOKIE};
#[cfg(test)]
use http::Request;
use http::{HeaderMap, Method, Response, StatusCode, Uri};
use http_body_util::{BodyExt, Full};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::sync::mpsc;
use uuid::Uuid;
use application::{
CloseTerminalInput, GetProjectWorkStateInput, LaunchAgentInput, McpRuntime, OpenProjectInput,
ResizeTerminalInput, RotateConversationLogInput, WriteToTerminalInput,
};
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,
};
use crate::pty::PtyChunk;
use crate::state::AppState;
const DEFAULT_LISTEN: &str = "127.0.0.1:17373";
const SESSION_COOKIE: &str = "idea_session";
const WS_PATH: &str = "/api/ws";
const WS_MAGIC: &str = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
const WS_MAX_PAYLOAD: usize = 64 * 1024;
const WS_OUTPUT_BUFFER: usize = 512;
type ResponseBody = Full<Bytes>;
/// Runs the `idea --serve` subcommand from already-split CLI arguments.
pub fn run_from_args(args: Vec<String>) -> ExitCode {
let config = match ServerConfig::from_args(args).and_then(|config| {
config.validate()?;
Ok(config)
}) {
Ok(config) => config,
Err(err) => {
eprintln!("idea --serve: {err}");
return ExitCode::from(2);
}
};
let state = Arc::new(ServerState::new(config.clone()));
eprintln!("IdeA pairing code: {}", state.pairing_code());
eprintln!("IdeA server listening on {}", config.listen);
match tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
{
Ok(runtime) => match runtime.block_on(run_server(config, state)) {
Ok(()) => ExitCode::SUCCESS,
Err(err) => {
eprintln!("idea --serve: {err}");
ExitCode::from(1)
}
},
Err(err) => {
eprintln!("idea --serve: failed to build runtime: {err}");
ExitCode::from(1)
}
}
}
#[derive(Clone)]
struct ServerConfig {
listen: SocketAddr,
public_origin: Option<String>,
allow_remote: bool,
trust_reverse_proxy: bool,
app_data_dir: PathBuf,
}
impl ServerConfig {
fn from_args(args: Vec<String>) -> Result<Self, String> {
let mut listen = DEFAULT_LISTEN
.parse::<SocketAddr>()
.expect("default listen address is valid");
let mut public_origin = None;
let mut allow_remote = false;
let mut trust_reverse_proxy = false;
let mut app_data_dir = default_app_data_dir();
let mut it = args.into_iter();
while let Some(arg) = it.next() {
match arg.as_str() {
"--listen" => {
let value = it
.next()
.ok_or_else(|| "--listen requires an address".to_owned())?;
listen = value
.parse()
.map_err(|_| format!("invalid --listen address: {value}"))?;
}
"--public-origin" => {
let value = it
.next()
.ok_or_else(|| "--public-origin requires an origin".to_owned())?;
public_origin = Some(value);
}
"--allow-remote" => allow_remote = true,
"--trust-reverse-proxy" => trust_reverse_proxy = true,
"--app-data-dir" => {
let value = it
.next()
.ok_or_else(|| "--app-data-dir requires a path".to_owned())?;
app_data_dir = PathBuf::from(value);
}
"--help" | "-h" => return Err(Self::usage()),
other => return Err(format!("unknown --serve argument: {other}")),
}
}
Ok(Self {
listen,
public_origin,
allow_remote,
trust_reverse_proxy,
app_data_dir,
})
}
fn validate(&self) -> Result<(), String> {
if let Some(origin) = &self.public_origin {
validate_origin(origin)?;
}
if !self.listen.ip().is_loopback() && !self.allow_remote {
return Err(
"refusing non-loopback bind without --allow-remote and HTTPS proxy config"
.to_owned(),
);
}
if self.allow_remote {
let Some(origin) = &self.public_origin else {
return Err("--allow-remote requires --public-origin https://...".to_owned());
};
if !origin.starts_with("https://") {
return Err("--allow-remote requires an HTTPS public origin".to_owned());
}
if !self.trust_reverse_proxy {
return Err("--allow-remote requires --trust-reverse-proxy".to_owned());
}
}
Ok(())
}
fn secure_cookie(&self) -> bool {
self.allow_remote
|| self
.public_origin
.as_deref()
.is_some_and(|o| o.starts_with("https://"))
}
fn usage() -> String {
"usage: idea --serve [--listen IP:PORT] [--app-data-dir PATH] [--allow-remote --public-origin https://host --trust-reverse-proxy]".to_owned()
}
}
struct ServerState {
config: ServerConfig,
app: AppState,
pairing_code: String,
sessions: Mutex<HashSet<String>>,
ws_pty_bridge: Arc<OutputBridge<SessionId, PtyChunk>>,
}
impl ServerState {
fn new(config: ServerConfig) -> Self {
Self {
app: AppState::build(config.app_data_dir.clone()),
config,
pairing_code: new_pairing_code(),
sessions: Mutex::new(HashSet::new()),
ws_pty_bridge: Arc::new(OutputBridge::new()),
}
}
#[cfg(test)]
fn new_for_test(config: ServerConfig, pairing_code: impl Into<String>) -> Self {
Self {
app: AppState::build(config.app_data_dir.clone()),
config,
pairing_code: pairing_code.into(),
sessions: Mutex::new(HashSet::new()),
ws_pty_bridge: Arc::new(OutputBridge::new()),
}
}
fn pairing_code(&self) -> &str {
&self.pairing_code
}
fn create_session(&self) -> String {
let token = new_session_token();
if let Ok(mut sessions) = self.sessions.lock() {
sessions.insert(token.clone());
}
token
}
fn has_session(&self, token: &str) -> bool {
self.sessions
.lock()
.map(|sessions| sessions.contains(token))
.unwrap_or(false)
}
}
async fn run_server(config: ServerConfig, state: Arc<ServerState>) -> Result<(), String> {
let listener = TcpListener::bind(config.listen)
.await
.map_err(|err| format!("failed to bind {}: {err}", config.listen))?;
loop {
let (stream, _) = listener
.accept()
.await
.map_err(|err| format!("failed to accept connection: {err}"))?;
let state = Arc::clone(&state);
tokio::spawn(async move {
if let Err(err) = handle_tcp_connection(stream, state).await {
eprintln!("idea --serve: connection error: {err}");
}
});
}
}
async fn handle_tcp_connection(
mut stream: tokio::net::TcpStream,
state: Arc<ServerState>,
) -> Result<(), String> {
let mut buffer = Vec::with_capacity(8192);
let mut chunk = [0_u8; 2048];
let header_end = loop {
let read = stream
.read(&mut chunk)
.await
.map_err(|err| format!("failed to read request: {err}"))?;
if read == 0 {
return Ok(());
}
buffer.extend_from_slice(&chunk[..read]);
if buffer.len() > 1024 * 1024 {
return Err("request too large".to_owned());
}
if let Some(pos) = find_header_end(&buffer) {
break pos;
}
};
let (method, uri, headers, content_length) = parse_http_request_head(&buffer[..header_end])?;
if method == Method::GET && uri.path() == WS_PATH {
return handle_ws_upgrade(stream, headers, state).await;
}
let body_start = header_end + 4;
while buffer.len() < body_start + content_length {
let read = stream
.read(&mut chunk)
.await
.map_err(|err| format!("failed to read request body: {err}"))?;
if read == 0 {
break;
}
buffer.extend_from_slice(&chunk[..read]);
if buffer.len() > 1024 * 1024 {
return Err("request too large".to_owned());
}
}
if buffer.len() < body_start + content_length {
return Err("truncated request body".to_owned());
}
let response = dispatch_http(
method,
uri,
headers,
Bytes::copy_from_slice(&buffer[body_start..body_start + content_length]),
state,
)
.await;
write_http_response(&mut stream, response).await
}
#[cfg(test)]
async fn handle_request(
req: Request<ResponseBody>,
state: Arc<ServerState>,
) -> Response<ResponseBody> {
let (parts, body) = req.into_parts();
let body = match body.collect().await {
Ok(body) => body.to_bytes(),
Err(err) => {
return error_response(
StatusCode::BAD_REQUEST,
"INVALID",
format!("invalid request body: {err}"),
None,
);
}
};
dispatch_http(
parts.method,
parts.uri,
parts.headers,
body,
Arc::clone(&state),
)
.await
}
fn find_header_end(buffer: &[u8]) -> Option<usize> {
buffer.windows(4).position(|window| window == b"\r\n\r\n")
}
fn parse_http_request_head(head: &[u8]) -> Result<(Method, Uri, HeaderMap, usize), String> {
let text = std::str::from_utf8(head).map_err(|_| "request head is not UTF-8".to_owned())?;
let mut lines = text.split("\r\n");
let request_line = lines
.next()
.ok_or_else(|| "missing request line".to_owned())?;
let mut request_parts = request_line.split_whitespace();
let method = request_parts
.next()
.ok_or_else(|| "missing method".to_owned())?
.parse::<Method>()
.map_err(|_| "invalid method".to_owned())?;
let uri = request_parts
.next()
.ok_or_else(|| "missing uri".to_owned())?
.parse::<Uri>()
.map_err(|_| "invalid uri".to_owned())?;
let mut headers = HeaderMap::new();
let mut content_length = 0;
for line in lines {
if line.is_empty() {
continue;
}
let Some((name, value)) = line.split_once(':') else {
return Err("invalid header line".to_owned());
};
let name = http::header::HeaderName::from_bytes(name.trim().as_bytes())
.map_err(|_| "invalid header name".to_owned())?;
let value =
HeaderValue::from_str(value.trim()).map_err(|_| "invalid header value".to_owned())?;
if name == http::header::CONTENT_LENGTH {
content_length = value
.to_str()
.ok()
.and_then(|raw| raw.parse::<usize>().ok())
.ok_or_else(|| "invalid content-length".to_owned())?;
}
headers.insert(name, value);
}
Ok((method, uri, headers, content_length))
}
async fn write_http_response(
stream: &mut tokio::net::TcpStream,
response: Response<ResponseBody>,
) -> Result<(), String> {
let status = response.status();
let headers = response.headers().clone();
let body = response
.into_body()
.collect()
.await
.map_err(|err| format!("failed to collect response: {err}"))?
.to_bytes();
let reason = status.canonical_reason().unwrap_or("Unknown");
let mut bytes = format!("HTTP/1.1 {} {}\r\n", status.as_u16(), reason).into_bytes();
for (name, value) in &headers {
bytes.extend_from_slice(name.as_str().as_bytes());
bytes.extend_from_slice(b": ");
bytes.extend_from_slice(value.as_bytes());
bytes.extend_from_slice(b"\r\n");
}
bytes.extend_from_slice(format!("content-length: {}\r\n", body.len()).as_bytes());
bytes.extend_from_slice(b"connection: close\r\n\r\n");
bytes.extend_from_slice(&body);
stream
.write_all(&bytes)
.await
.map_err(|err| format!("failed to write response: {err}"))
}
async fn handle_ws_upgrade(
mut stream: tokio::net::TcpStream,
headers: HeaderMap,
state: Arc<ServerState>,
) -> Result<(), String> {
let accept = match validate_ws_upgrade(&headers, &state) {
Ok(accept) => accept,
Err(response) => return write_http_response(&mut stream, *response).await,
};
let response = format!(
"HTTP/1.1 101 Switching Protocols\r\n\
upgrade: websocket\r\n\
connection: Upgrade\r\n\
sec-websocket-accept: {accept}\r\n\r\n"
);
stream
.write_all(response.as_bytes())
.await
.map_err(|err| format!("failed to write websocket upgrade: {err}"))?;
run_ws_connection(stream, state).await
}
fn validate_ws_upgrade(
headers: &HeaderMap,
state: &ServerState,
) -> Result<String, Box<Response<ResponseBody>>> {
let origin = validate_request_origin(headers, &state.config)?;
let Some(token) = session_cookie(headers) else {
return Err(Box::new(error_response(
StatusCode::UNAUTHORIZED,
"UNAUTHORIZED",
"missing session cookie",
origin.as_deref(),
)));
};
if !state.has_session(&token) {
return Err(Box::new(error_response(
StatusCode::UNAUTHORIZED,
"UNAUTHORIZED",
"invalid session cookie",
origin.as_deref(),
)));
}
if !header_contains_token(headers, "upgrade", "websocket")
|| !header_contains_token(headers, "connection", "upgrade")
{
return Err(Box::new(error_response(
StatusCode::BAD_REQUEST,
"INVALID",
"missing websocket upgrade headers",
origin.as_deref(),
)));
}
let Some(key) = headers
.get("sec-websocket-key")
.and_then(|value| value.to_str().ok())
else {
return Err(Box::new(error_response(
StatusCode::BAD_REQUEST,
"INVALID",
"missing Sec-WebSocket-Key",
origin.as_deref(),
)));
};
Ok(websocket_accept(key))
}
fn header_contains_token(headers: &HeaderMap, name: &str, needle: &str) -> bool {
headers
.get(name)
.and_then(|value| value.to_str().ok())
.is_some_and(|value| {
value
.split(',')
.any(|token| token.trim().eq_ignore_ascii_case(needle))
})
}
async fn run_ws_connection(
stream: tokio::net::TcpStream,
state: Arc<ServerState>,
) -> Result<(), String> {
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 writer_task = tokio::spawn(async move {
while let Some(frame) = rx.recv().await {
let text = serde_json::to_vec(&frame)
.map_err(|err| format!("failed to encode server frame: {err}"))?;
let bytes = encode_ws_frame(WsOpcode::Text, &text);
writer
.write_all(&bytes)
.await
.map_err(|err| format!("failed to write websocket frame: {err}"))?;
}
Ok::<(), String>(())
});
loop {
let frame = match read_ws_frame(&mut reader).await {
Ok(frame) => frame,
Err(err) => {
let _ = tx
.send(ServerFrame::error(
None,
None,
"WS_PROTOCOL",
err.to_string(),
))
.await;
break;
}
};
match frame.opcode {
WsOpcode::Text => {
let parsed = serde_json::from_slice::<ClientFrame>(&frame.payload)
.map_err(|err| format!("invalid client frame JSON: {err}"));
match parsed {
Ok(frame) => {
handle_client_frame(frame, &state, &tx, &owned).await;
}
Err(err) => {
let _ = tx
.send(ServerFrame::error(None, None, "INVALID", err))
.await;
}
}
}
WsOpcode::Ping => {
let _ = tx.send(ServerFrame::pong()).await;
}
WsOpcode::Close => break,
WsOpcode::Pong => {}
WsOpcode::Binary => {
let _ = tx
.send(ServerFrame::error(
None,
None,
"UNSUPPORTED",
"binary websocket frames are not supported",
))
.await;
}
}
}
if let Ok(owned) = owned.lock() {
for (session, gen) in owned.iter() {
state.ws_pty_bridge.unregister_if(session, *gen);
}
}
drop(tx);
writer_task
.await
.map_err(|err| format!("websocket writer task failed: {err}"))?
}
async fn handle_client_frame(
frame: ClientFrame,
state: &Arc<ServerState>,
tx: &mpsc::Sender<ServerFrame>,
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
) {
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),
"terminal.detach" | "detach" => ws_detach(&frame, state, owned),
"terminal.close" | "close" => match ws_close(&frame, state, owned).await {
Ok((session_id, exit_code)) => {
let _ = tx
.send(ServerFrame::status(session_id, "exited", exit_code))
.await;
Ok(())
}
Err(err) => Err(err),
},
"ping" => {
let _ = tx.send(ServerFrame::pong()).await;
Ok(())
}
_ => Err(ErrorDto {
code: "UNKNOWN_FRAME".to_owned(),
message: format!("unknown websocket frame kind: {}", frame.kind),
}),
};
if let Err(err) = result {
let _ = tx
.send(ServerFrame::error(
Some(frame.id),
payload_session_id(&frame.payload),
err.code,
err.message,
))
.await;
}
}
async fn ws_open_terminal(
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: OpenTerminalRequestDto =
serde_json::from_value(request_value).map_err(invalid_args_error)?;
let output = state
.app
.open_terminal
.execute(request.into())
.await
.map_err(ErrorDto::from)?;
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, 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(
frame: &ClientFrame,
state: &Arc<ServerState>,
tx: &mpsc::Sender<ServerFrame>,
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
) -> Result<(), ErrorDto> {
let session_id = payload_string(&frame.payload, "sessionId")?;
let sid = parse_session_id(&session_id)?;
let handle = PtyHandle { session_id: sid };
let scrollback = state
.app
.pty_port
.scrollback(&handle)
.map_err(|err| ErrorDto::from(application::AppError::from(err)))?;
let rows = payload_u16(&frame.payload, "rows").unwrap_or(24);
let cols = payload_u16(&frame.payload, "cols").unwrap_or(80);
let dto = TerminalSessionDto {
session_id,
cwd: String::new(),
rows,
cols,
assigned_conversation_id: None,
engine_session_id: None,
cell_kind: crate::dto::CellKind::Pty,
};
let next_seq = u64::from(!scrollback.is_empty());
tx.send(ServerFrame::attached(
&frame.id,
&dto,
scrollback.clone(),
next_seq,
frame.payload.get("lastSeq").is_some(),
))
.await
.map_err(|_| ErrorDto {
code: "WS_CLOSED".to_owned(),
message: "websocket output closed".to_owned(),
})?;
let sink = WsPtySink::new(sid, tx.clone(), next_seq);
attach_sink_and_pump(state, sid, sink, owned)
}
fn attach_sink_and_pump(
state: &Arc<ServerState>,
sid: SessionId,
sink: WsPtySink,
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
) -> Result<(), ErrorDto> {
let gen = state.ws_pty_bridge.register(sid, Arc::new(sink));
if let Ok(mut owned) = owned.lock() {
owned.push((sid, gen));
}
let handle = PtyHandle { session_id: sid };
let stream = state
.app
.pty_port
.subscribe_output(&handle)
.map_err(|err| ErrorDto::from(application::AppError::from(err)))?;
let bridge = Arc::clone(&state.ws_pty_bridge);
std::thread::spawn(move || {
for chunk in stream {
if !bridge.send_output(&sid, chunk) {
break;
}
}
bridge.unregister_if(&sid, gen);
});
Ok(())
}
fn ws_input(frame: &ClientFrame, state: &Arc<ServerState>) -> Result<(), ErrorDto> {
let sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
let bytes = payload_bytes(&frame.payload)?;
state
.app
.write_terminal
.execute(WriteToTerminalInput {
session_id: sid,
data: bytes,
})
.map_err(ErrorDto::from)
}
fn ws_resize(frame: &ClientFrame, state: &Arc<ServerState>) -> Result<(), ErrorDto> {
let sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
let rows = payload_u16(&frame.payload, "rows").ok_or_else(|| ErrorDto {
code: "INVALID".to_owned(),
message: "resize requires rows".to_owned(),
})?;
let cols = payload_u16(&frame.payload, "cols").ok_or_else(|| ErrorDto {
code: "INVALID".to_owned(),
message: "resize requires cols".to_owned(),
})?;
state
.app
.resize_terminal
.execute(ResizeTerminalInput {
session_id: sid,
rows,
cols,
})
.map_err(ErrorDto::from)
}
fn ws_detach(
frame: &ClientFrame,
state: &Arc<ServerState>,
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
) -> Result<(), ErrorDto> {
let sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
if let Ok(mut owned) = owned.lock() {
if let Some(index) = owned.iter().rposition(|(session, _)| *session == sid) {
let (_, gen) = owned.remove(index);
state.ws_pty_bridge.unregister_if(&sid, gen);
}
}
Ok(())
}
async fn ws_close(
frame: &ClientFrame,
state: &Arc<ServerState>,
owned: &Arc<Mutex<Vec<(SessionId, u64)>>>,
) -> Result<(SessionId, Option<i32>), ErrorDto> {
let sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
ws_detach(frame, state, owned)?;
let output = state
.app
.close_terminal
.execute(CloseTerminalInput { session_id: sid })
.await
.map_err(ErrorDto::from)?;
Ok((sid, output.code))
}
async fn dispatch_http(
method: Method,
uri: Uri,
headers: HeaderMap,
body: Bytes,
state: Arc<ServerState>,
) -> Response<ResponseBody> {
let origin = match validate_request_origin(&headers, &state.config) {
Ok(origin) => origin,
Err(response) => return *response,
};
if has_forbidden_query_secret(&uri) {
return error_response(
StatusCode::BAD_REQUEST,
"INVALID",
"secrets in URL query strings are forbidden",
origin.as_deref(),
);
}
if method == Method::OPTIONS {
return cors_response(StatusCode::NO_CONTENT, origin.as_deref());
}
match (method, uri.path()) {
(Method::POST, "/api/pair") => pair(body, state, origin.as_deref()).await,
(Method::POST, "/api/invoke") => {
let Some(token) = session_cookie(&headers) else {
return error_response(
StatusCode::UNAUTHORIZED,
"UNAUTHORIZED",
"missing session cookie",
origin.as_deref(),
);
};
if !state.has_session(&token) {
return error_response(
StatusCode::UNAUTHORIZED,
"UNAUTHORIZED",
"invalid session cookie",
origin.as_deref(),
);
}
invoke(body, state, origin.as_deref()).await
}
(_, "/api/pair" | "/api/invoke") => error_response(
StatusCode::METHOD_NOT_ALLOWED,
"METHOD_NOT_ALLOWED",
"method not allowed",
origin.as_deref(),
),
_ => error_response(
StatusCode::NOT_FOUND,
"NOT_FOUND",
"route not found",
origin.as_deref(),
),
}
}
#[derive(Deserialize)]
struct PairRequest {
code: String,
}
async fn pair(
body: Bytes,
state: Arc<ServerState>,
origin: Option<&str>,
) -> Response<ResponseBody> {
let request = match serde_json::from_slice::<PairRequest>(&body) {
Ok(request) => request,
Err(err) => {
return error_response(
StatusCode::BAD_REQUEST,
"INVALID",
format!("invalid pairing request: {err}"),
origin,
);
}
};
if request.code != state.pairing_code() {
return error_response(
StatusCode::FORBIDDEN,
"FORBIDDEN",
"invalid pairing code",
origin,
);
}
let token = state.create_session();
let cookie = Cookie::build((SESSION_COOKIE, token))
.path("/")
.http_only(true)
.secure(state.config.secure_cookie())
.same_site(SameSite::Strict)
.build();
let mut response = json_response(StatusCode::OK, &json!({ "paired": true }), origin);
response.headers_mut().insert(
SET_COOKIE,
HeaderValue::from_str(&cookie.to_string()).expect("session cookie is header-safe"),
);
response
}
#[derive(Deserialize)]
struct InvokeRequest {
command: String,
#[serde(default)]
args: Value,
}
async fn invoke(
body: Bytes,
state: Arc<ServerState>,
origin: Option<&str>,
) -> Response<ResponseBody> {
let request = match serde_json::from_slice::<InvokeRequest>(&body) {
Ok(request) => request,
Err(err) => {
return error_response(
StatusCode::BAD_REQUEST,
"INVALID",
format!("invalid invoke request: {err}"),
origin,
);
}
};
let result = match request.command.as_str() {
"health" => invoke_health(&request.args, &state.app),
"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,
_ => Err(ErrorDto {
code: "UNKNOWN_COMMAND".to_owned(),
message: format!("unknown command: {}", request.command),
}),
};
match result {
Ok(value) => json_response(StatusCode::OK, &value, origin),
Err(error) => error_dto_response(status_for_error(&error), error, origin),
}
}
fn invoke_health(args: &Value, state: &AppState) -> Result<Value, ErrorDto> {
let request = optional_request::<HealthRequestDto>(args)?;
let output = state
.health
.execute(request.unwrap_or_default().into())
.map(HealthResponseDto::from)
.map_err(ErrorDto::from)?;
serde_json::to_value(output).map_err(serialization_error)
}
async fn invoke_list_projects(state: &AppState) -> Result<Value, ErrorDto> {
let output = state
.list_projects
.execute()
.await
.map(ProjectListDto::from)
.map_err(ErrorDto::from)?;
serde_json::to_value(output).map_err(serialization_error)
}
async fn invoke_open_project(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: "open_project requires args.projectId".to_owned(),
})?;
let project = resolve_project_readonly(project_id, state).await?;
serde_json::to_value(ProjectDto::from(project)).map_err(serialization_error)
}
async fn invoke_get_project_work_state(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: "get_project_work_state requires args.projectId".to_owned(),
})?;
let project = resolve_project_readonly(project_id, state).await?;
let output = state
.get_project_work_state
.execute(GetProjectWorkStateInput { project })
.await
.map(ProjectWorkStateDto::from)
.map_err(ErrorDto::from)?;
serde_json::to_value(output).map_err(serialization_error)
}
async fn resolve_project_readonly(project_id: &str, state: &AppState) -> Result<Project, ErrorDto> {
let id = parse_project_id(project_id)?;
state
.open_project
.execute(OpenProjectInput { project_id: id })
.await
.map(|output| output.project)
.map_err(ErrorDto::from)
}
fn optional_request<T>(args: &Value) -> Result<Option<T>, ErrorDto>
where
T: for<'de> Deserialize<'de>,
{
match args {
Value::Object(map) => match map.get("request") {
Some(value) => serde_json::from_value(value.clone())
.map(Some)
.map_err(invalid_args_error),
None if map.is_empty() => Ok(None),
None => serde_json::from_value(args.clone())
.map(Some)
.map_err(invalid_args_error),
},
Value::Null => Ok(None),
_ => Err(ErrorDto {
code: "INVALID".to_owned(),
message: "args must be an object".to_owned(),
}),
}
}
fn validate_request_origin(
headers: &HeaderMap,
config: &ServerConfig,
) -> Result<Option<String>, Box<Response<ResponseBody>>> {
let Some(origin) = headers.get(ORIGIN).and_then(|value| value.to_str().ok()) else {
return Err(Box::new(error_response(
StatusCode::FORBIDDEN,
"FORBIDDEN",
"missing Origin header",
None,
)));
};
if origin_allowed(origin, config) {
Ok(Some(origin.to_owned()))
} else {
Err(Box::new(error_response(
StatusCode::FORBIDDEN,
"FORBIDDEN",
"origin not allowed",
None,
)))
}
}
fn origin_allowed(origin: &str, config: &ServerConfig) -> bool {
if let Some(public_origin) = &config.public_origin {
return origin == public_origin;
}
if !config.listen.ip().is_loopback() {
return false;
}
let port = config.listen.port();
origin == format!("http://127.0.0.1:{port}")
|| origin == format!("http://localhost:{port}")
|| origin == format!("http://[::1]:{port}")
}
fn has_forbidden_query_secret(uri: &Uri) -> bool {
uri.query().is_some_and(|query| {
query.split('&').any(|pair| {
let key = pair.split_once('=').map_or(pair, |(key, _)| key);
matches!(
key.to_ascii_lowercase().as_str(),
"token" | "secret" | "session" | "code"
)
})
})
}
fn session_cookie(headers: &HeaderMap) -> Option<String> {
headers
.get(COOKIE)
.and_then(|value| value.to_str().ok())
.and_then(|raw| {
raw.split(';').find_map(|part| {
let (name, value) = part.trim().split_once('=')?;
(name == SESSION_COOKIE).then(|| value.to_owned())
})
})
}
fn json_response(
status: StatusCode,
value: &Value,
origin: Option<&str>,
) -> Response<ResponseBody> {
let body = serde_json::to_vec(value).expect("JSON value serializes");
let mut response = Response::new(Full::new(Bytes::from(body)));
*response.status_mut() = status;
response
.headers_mut()
.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
add_cors_headers(&mut response, origin);
response
}
fn error_dto_response(
status: StatusCode,
error: ErrorDto,
origin: Option<&str>,
) -> Response<ResponseBody> {
let value = serde_json::to_value(error).expect("ErrorDto serializes");
json_response(status, &value, origin)
}
fn error_response(
status: StatusCode,
code: impl Into<String>,
message: impl Into<String>,
origin: Option<&str>,
) -> Response<ResponseBody> {
error_dto_response(
status,
ErrorDto {
code: code.into(),
message: message.into(),
},
origin,
)
}
fn cors_response(status: StatusCode, origin: Option<&str>) -> Response<ResponseBody> {
let mut response = Response::new(Full::new(Bytes::new()));
*response.status_mut() = status;
add_cors_headers(&mut response, origin);
response
}
fn add_cors_headers(response: &mut Response<ResponseBody>, origin: Option<&str>) {
if let Some(origin) = origin {
if let Ok(origin) = HeaderValue::from_str(origin) {
response
.headers_mut()
.insert("access-control-allow-origin", origin);
response.headers_mut().insert(
"access-control-allow-credentials",
HeaderValue::from_static("true"),
);
response.headers_mut().insert(
"access-control-allow-headers",
HeaderValue::from_static("content-type"),
);
response.headers_mut().insert(
"access-control-allow-methods",
HeaderValue::from_static("POST, OPTIONS"),
);
}
}
}
fn status_for_error(error: &ErrorDto) -> StatusCode {
match error.code.as_str() {
"UNKNOWN_COMMAND" => StatusCode::BAD_REQUEST,
"INVALID" => StatusCode::BAD_REQUEST,
"NOT_FOUND" => StatusCode::NOT_FOUND,
"FORBIDDEN" => StatusCode::FORBIDDEN,
"UNAUTHORIZED" => StatusCode::UNAUTHORIZED,
_ => StatusCode::INTERNAL_SERVER_ERROR,
}
}
fn validate_origin(origin: &str) -> Result<(), String> {
if origin == "*" || origin.contains('*') {
return Err("--public-origin must be exact, not a wildcard".to_owned());
}
if origin.ends_with('/') {
return Err("--public-origin must not end with '/'".to_owned());
}
if !(origin.starts_with("https://") || origin.starts_with("http://")) {
return Err("--public-origin must be an HTTP(S) origin".to_owned());
}
Ok(())
}
fn default_app_data_dir() -> PathBuf {
if let Some(path) = env::var_os("IDEA_APP_DATA_DIR") {
return PathBuf::from(path);
}
if let Some(path) = env::var_os("XDG_DATA_HOME") {
return PathBuf::from(path).join("IdeA");
}
if let Some(home) = env::var_os("HOME") {
return PathBuf::from(home).join(".local/share/IdeA");
}
env::current_dir()
.unwrap_or_else(|_| PathBuf::from("."))
.join(".ideai/app-data")
}
fn new_pairing_code() -> String {
Uuid::new_v4()
.simple()
.to_string()
.chars()
.take(8)
.collect::<String>()
.to_ascii_uppercase()
}
fn new_session_token() -> String {
format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple())
}
fn invalid_args_error(err: serde_json::Error) -> ErrorDto {
ErrorDto {
code: "INVALID".to_owned(),
message: format!("invalid command args: {err}"),
}
}
fn serialization_error(err: serde_json::Error) -> ErrorDto {
ErrorDto {
code: "SERIALIZATION".to_owned(),
message: format!("failed to serialize response: {err}"),
}
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct ClientFrame {
id: String,
kind: String,
#[serde(default)]
payload: Value,
}
#[derive(Debug, Clone, Serialize)]
struct ServerFrame {
kind: String,
#[serde(skip_serializing_if = "Option::is_none", rename = "replyTo")]
reply_to: Option<String>,
payload: Value,
}
impl ServerFrame {
fn attached(
request_id: &str,
session: &TerminalSessionDto,
scrollback: Vec<u8>,
next_seq: u64,
gap: bool,
) -> Self {
let scrollback = if scrollback.is_empty() {
Vec::new()
} else {
vec![json!({
"seq": 0,
"bytesBase64": base64_encode(&scrollback),
})]
};
Self {
kind: "terminal.attached".to_owned(),
reply_to: Some(request_id.to_owned()),
payload: json!({
"session": {
"sessionId": session.session_id,
"rows": session.rows,
"cols": session.cols,
},
"scrollback": scrollback,
"nextSeq": next_seq,
"status": "running",
"gap": gap,
"assignedConversationId": session.assigned_conversation_id,
}),
}
}
fn output(session_id: SessionId, seq: u64, bytes: Vec<u8>) -> Self {
Self {
kind: "terminal.output".to_owned(),
reply_to: None,
payload: json!({
"sessionId": session_id.to_string(),
"seq": seq,
"bytesBase64": base64_encode(&bytes),
}),
}
}
fn status(session_id: SessionId, status: &str, exit_code: Option<i32>) -> Self {
Self {
kind: "terminal.status".to_owned(),
reply_to: None,
payload: json!({
"sessionId": session_id.to_string(),
"status": status,
"exitCode": exit_code,
}),
}
}
fn error(
request_id: Option<String>,
session_id: Option<String>,
code: impl Into<String>,
message: impl Into<String>,
) -> Self {
Self {
kind: "error".to_owned(),
reply_to: request_id,
payload: json!({
"sessionId": session_id,
"code": code.into(),
"message": message.into(),
}),
}
}
fn pong() -> Self {
Self {
kind: "pong".to_owned(),
reply_to: None,
payload: json!({}),
}
}
}
struct WsPtySink {
session_id: SessionId,
tx: mpsc::Sender<ServerFrame>,
seq: AtomicU64,
}
impl WsPtySink {
fn new(session_id: SessionId, tx: mpsc::Sender<ServerFrame>, next_seq: u64) -> Self {
Self {
session_id,
tx,
seq: AtomicU64::new(next_seq),
}
}
}
impl OutputSink<PtyChunk> for WsPtySink {
fn send(&self, item: PtyChunk) -> Result<(), OutputSinkError> {
let seq = self.seq.fetch_add(1, Ordering::Relaxed);
self.tx
.try_send(ServerFrame::output(self.session_id, seq, item))
.map_err(|err| match err {
mpsc::error::TrySendError::Full(_) => OutputSinkError::Full,
mpsc::error::TrySendError::Closed(_) => OutputSinkError::Closed,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WsOpcode {
Text,
Binary,
Close,
Ping,
Pong,
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct WsFrame {
opcode: WsOpcode,
payload: Vec<u8>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
enum WsFrameError {
Incomplete,
Protocol(String),
TooLarge,
}
impl std::fmt::Display for WsFrameError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Incomplete => f.write_str("incomplete websocket frame"),
Self::Protocol(message) => write!(f, "websocket protocol error: {message}"),
Self::TooLarge => f.write_str("websocket frame payload too large"),
}
}
}
async fn read_ws_frame<R>(reader: &mut R) -> Result<WsFrame, WsFrameError>
where
R: AsyncRead + Unpin,
{
let mut header = [0_u8; 2];
reader
.read_exact(&mut header)
.await
.map_err(|_| WsFrameError::Incomplete)?;
let len_code = header[1] & 0x7f;
let masked = header[1] & 0x80 != 0;
let mut rest = Vec::new();
let extra_len = match len_code {
126 => 2,
127 => 8,
_ => 0,
};
rest.resize(extra_len + usize::from(masked) * 4, 0);
if !rest.is_empty() {
reader
.read_exact(&mut rest)
.await
.map_err(|_| WsFrameError::Incomplete)?;
}
let mut raw = Vec::with_capacity(2 + rest.len());
raw.extend_from_slice(&header);
raw.extend_from_slice(&rest);
let payload_len = decoded_payload_len(&raw)?;
if payload_len > WS_MAX_PAYLOAD {
return Err(WsFrameError::TooLarge);
}
let mut payload = vec![0_u8; payload_len];
reader
.read_exact(&mut payload)
.await
.map_err(|_| WsFrameError::Incomplete)?;
raw.extend_from_slice(&payload);
decode_client_ws_frame(&raw)
}
fn decode_client_ws_frame(raw: &[u8]) -> Result<WsFrame, WsFrameError> {
if raw.len() < 2 {
return Err(WsFrameError::Incomplete);
}
let fin = raw[0] & 0x80 != 0;
if !fin {
return Err(WsFrameError::Protocol(
"fragmented frames are not supported".to_owned(),
));
}
let opcode = match raw[0] & 0x0f {
0x1 => WsOpcode::Text,
0x2 => WsOpcode::Binary,
0x8 => WsOpcode::Close,
0x9 => WsOpcode::Ping,
0xA => WsOpcode::Pong,
other => {
return Err(WsFrameError::Protocol(format!(
"unsupported opcode {other}"
)))
}
};
let masked = raw[1] & 0x80 != 0;
if !masked {
return Err(WsFrameError::Protocol(
"client frames must be masked".to_owned(),
));
}
let len_code = raw[1] & 0x7f;
let mut offset = 2;
let len = match len_code {
126 => {
if raw.len() < offset + 2 {
return Err(WsFrameError::Incomplete);
}
let len = u16::from_be_bytes([raw[offset], raw[offset + 1]]) as usize;
offset += 2;
len
}
127 => {
if raw.len() < offset + 8 {
return Err(WsFrameError::Incomplete);
}
let len = u64::from_be_bytes(raw[offset..offset + 8].try_into().expect("8 bytes"));
offset += 8;
usize::try_from(len).map_err(|_| WsFrameError::TooLarge)?
}
len => usize::from(len),
};
if len > WS_MAX_PAYLOAD {
return Err(WsFrameError::TooLarge);
}
if raw.len() < offset + 4 + len {
return Err(WsFrameError::Incomplete);
}
let mask = &raw[offset..offset + 4];
offset += 4;
let mut payload = raw[offset..offset + len].to_vec();
for (i, byte) in payload.iter_mut().enumerate() {
*byte ^= mask[i % 4];
}
Ok(WsFrame { opcode, payload })
}
fn decoded_payload_len(raw_header: &[u8]) -> Result<usize, WsFrameError> {
let len_code = raw_header[1] & 0x7f;
let offset = 2;
let len = match len_code {
126 => {
if raw_header.len() < offset + 2 {
return Err(WsFrameError::Incomplete);
}
u16::from_be_bytes([raw_header[offset], raw_header[offset + 1]]) as usize
}
127 => {
if raw_header.len() < offset + 8 {
return Err(WsFrameError::Incomplete);
}
let len =
u64::from_be_bytes(raw_header[offset..offset + 8].try_into().expect("8 bytes"));
usize::try_from(len).map_err(|_| WsFrameError::TooLarge)?
}
len => usize::from(len),
};
Ok(len)
}
fn encode_ws_frame(opcode: WsOpcode, payload: &[u8]) -> Vec<u8> {
let opcode = match opcode {
WsOpcode::Text => 0x1,
WsOpcode::Binary => 0x2,
WsOpcode::Close => 0x8,
WsOpcode::Ping => 0x9,
WsOpcode::Pong => 0xA,
};
let mut out = vec![0x80 | opcode];
match payload.len() {
0..=125 => out.push(payload.len() as u8),
126..=65535 => {
out.push(126);
out.extend_from_slice(&(payload.len() as u16).to_be_bytes());
}
len => {
out.push(127);
out.extend_from_slice(&(len as u64).to_be_bytes());
}
}
out.extend_from_slice(payload);
out
}
fn websocket_accept(key: &str) -> String {
let mut input = Vec::with_capacity(key.len() + WS_MAGIC.len());
input.extend_from_slice(key.as_bytes());
input.extend_from_slice(WS_MAGIC.as_bytes());
base64_encode(&sha1_digest(&input))
}
fn base64_encode(bytes: &[u8]) -> String {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD.encode(bytes)
}
fn base64_decode(raw: &str) -> Result<Vec<u8>, ErrorDto> {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD
.decode(raw)
.map_err(|err| ErrorDto {
code: "INVALID".to_owned(),
message: format!("invalid base64 bytes: {err}"),
})
}
fn sha1_digest(input: &[u8]) -> [u8; 20] {
let mut h0: u32 = 0x67452301;
let mut h1: u32 = 0xEFCDAB89;
let mut h2: u32 = 0x98BADCFE;
let mut h3: u32 = 0x10325476;
let mut h4: u32 = 0xC3D2E1F0;
let bit_len = (input.len() as u64) * 8;
let mut msg = input.to_vec();
msg.push(0x80);
while (msg.len() % 64) != 56 {
msg.push(0);
}
msg.extend_from_slice(&bit_len.to_be_bytes());
for chunk in msg.chunks(64) {
let mut w = [0_u32; 80];
for (i, word) in w.iter_mut().take(16).enumerate() {
let j = i * 4;
*word = u32::from_be_bytes([chunk[j], chunk[j + 1], chunk[j + 2], chunk[j + 3]]);
}
for i in 16..80 {
w[i] = (w[i - 3] ^ w[i - 8] ^ w[i - 14] ^ w[i - 16]).rotate_left(1);
}
let mut a = h0;
let mut b = h1;
let mut c = h2;
let mut d = h3;
let mut e = h4;
for (i, word) in w.iter().enumerate() {
let (f, k) = match i {
0..=19 => ((b & c) | ((!b) & d), 0x5A827999),
20..=39 => (b ^ c ^ d, 0x6ED9EBA1),
40..=59 => ((b & c) | (b & d) | (c & d), 0x8F1BBCDC),
_ => (b ^ c ^ d, 0xCA62C1D6),
};
let temp = a
.rotate_left(5)
.wrapping_add(f)
.wrapping_add(e)
.wrapping_add(k)
.wrapping_add(*word);
e = d;
d = c;
c = b.rotate_left(30);
b = a;
a = temp;
}
h0 = h0.wrapping_add(a);
h1 = h1.wrapping_add(b);
h2 = h2.wrapping_add(c);
h3 = h3.wrapping_add(d);
h4 = h4.wrapping_add(e);
}
let mut out = [0_u8; 20];
out[0..4].copy_from_slice(&h0.to_be_bytes());
out[4..8].copy_from_slice(&h1.to_be_bytes());
out[8..12].copy_from_slice(&h2.to_be_bytes());
out[12..16].copy_from_slice(&h3.to_be_bytes());
out[16..20].copy_from_slice(&h4.to_be_bytes());
out
}
fn payload_string(payload: &Value, key: &str) -> Result<String, ErrorDto> {
payload
.get(key)
.and_then(Value::as_str)
.map(ToOwned::to_owned)
.ok_or_else(|| ErrorDto {
code: "INVALID".to_owned(),
message: format!("payload requires {key}"),
})
}
fn payload_u16(payload: &Value, key: &str) -> Option<u16> {
payload
.get(key)
.and_then(Value::as_u64)
.and_then(|value| u16::try_from(value).ok())
}
fn payload_bytes(payload: &Value) -> Result<Vec<u8>, ErrorDto> {
if let Some(raw) = payload.get("bytesBase64").and_then(Value::as_str) {
base64_decode(raw)
} else if let Some(raw) = payload.get("bytes").and_then(Value::as_str) {
base64_decode(raw)
} else {
Err(ErrorDto {
code: "INVALID".to_owned(),
message: "payload requires bytesBase64".to_owned(),
})
}
}
fn payload_session_id(payload: &Value) -> Option<String> {
payload
.get("sessionId")
.and_then(Value::as_str)
.map(ToOwned::to_owned)
}
#[cfg(test)]
mod tests {
use super::*;
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;
fn test_config() -> ServerConfig {
ServerConfig {
listen: "127.0.0.1:17373".parse().unwrap(),
public_origin: None,
allow_remote: false,
trust_reverse_proxy: false,
app_data_dir: std::env::temp_dir().join(format!("idea-server-test-{}", Uuid::new_v4())),
}
}
fn state() -> Arc<ServerState> {
Arc::new(ServerState::new_for_test(test_config(), "PAIR1234"))
}
async fn request(
state: Arc<ServerState>,
method: Method,
uri: &str,
body: Value,
extra_headers: &[(&str, &str)],
) -> Response<ResponseBody> {
let mut builder = Request::builder()
.method(method)
.uri(uri)
.header(ORIGIN, "http://127.0.0.1:17373")
.header(CONTENT_TYPE, "application/json");
for (name, value) in extra_headers {
builder = builder.header(HeaderName::from_bytes(name.as_bytes()).unwrap(), *value);
}
handle_request(
builder
.body(Full::new(Bytes::from(serde_json::to_vec(&body).unwrap())))
.unwrap(),
state,
)
.await
}
async fn response_json(response: Response<ResponseBody>) -> (StatusCode, Value, HeaderMap) {
let status = response.status();
let headers = response.headers().clone();
let body = response.into_body().collect().await.unwrap().to_bytes();
let value = if body.is_empty() {
Value::Null
} else {
serde_json::from_slice(&body).unwrap()
};
(status, value, headers)
}
async fn pair_and_cookie(state: Arc<ServerState>) -> String {
let response = request(
state,
Method::POST,
"/api/pair",
json!({ "code": "PAIR1234" }),
&[],
)
.await;
response
.headers()
.get(SET_COOKIE)
.unwrap()
.to_str()
.unwrap()
.split(';')
.next()
.unwrap()
.to_owned()
}
async fn create_project_for_test(state: &Arc<ServerState>, name: &str) -> String {
let root = std::env::temp_dir()
.join(format!("idea-server-project-{}", Uuid::new_v4()))
.to_string_lossy()
.into_owned();
let output = state
.app
.create_project
.execute(CreateProjectInput {
name: name.to_owned(),
root,
remote: None,
default_profile_id: None,
})
.await
.expect("test project is created");
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 {
let mut headers = HeaderMap::new();
headers.insert(ORIGIN, HeaderValue::from_str(origin).unwrap());
headers.insert("upgrade", HeaderValue::from_static("websocket"));
headers.insert("connection", HeaderValue::from_static("Upgrade"));
headers.insert(
"sec-websocket-key",
HeaderValue::from_static("dGhlIHNhbXBsZSBub25jZQ=="),
);
if let Some(cookie) = cookie {
headers.insert(COOKIE, HeaderValue::from_str(cookie).unwrap());
}
headers
}
fn masked_text_frame(text: &str) -> Vec<u8> {
let payload = text.as_bytes();
let mask = [1_u8, 2, 3, 4];
let mut out = vec![0x81];
out.push(0x80 | payload.len() as u8);
out.extend_from_slice(&mask);
for (i, byte) in payload.iter().enumerate() {
out.push(byte ^ mask[i % 4]);
}
out
}
async fn recv_server_frame(rx: &mut mpsc::Receiver<ServerFrame>) -> ServerFrame {
tokio::time::timeout(Duration::from_secs(2), rx.recv())
.await
.expect("server frame received before timeout")
.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>) {
while rx.try_recv().is_ok() {}
}
fn ws_frame(id: &str, kind: &str, payload: Value) -> ClientFrame {
ClientFrame {
id: id.to_owned(),
kind: kind.to_owned(),
payload,
}
}
fn attached_session_id(frame: &ServerFrame) -> String {
assert_eq!(frame.kind, "terminal.attached");
frame.payload["session"]["sessionId"]
.as_str()
.expect("attached frame carries sessionId")
.to_owned()
}
fn open_terminal_payload(command: &str, args: Vec<&str>) -> Value {
json!({
"request": {
"cwd": std::env::temp_dir().to_string_lossy(),
"rows": 24,
"cols": 80,
"command": command,
"args": args,
}
})
}
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) {
let (tx, mut rx) = mpsc::channel(8);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"close-cleanup",
"terminal.close",
json!({ "sessionId": session_id }),
),
state,
&tx,
&owned,
)
.await;
let _ = recv_server_frame(&mut rx).await;
}
async fn wait_for_scrollback(
state: &Arc<ServerState>,
session_id: &str,
needle: &[u8],
) -> Vec<u8> {
let sid = parse_session_id(session_id).expect("test session id parses");
let handle = PtyHandle { session_id: sid };
for _ in 0..100 {
if let Ok(scrollback) = state.app.pty_port.scrollback(&handle) {
if scrollback
.windows(needle.len())
.any(|window| window == needle)
{
return scrollback;
}
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!(
"scrollback did not contain {:?}",
String::from_utf8_lossy(needle)
);
}
#[test]
fn websocket_accept_matches_rfc_example() {
assert_eq!(
websocket_accept("dGhlIHNhbXBsZSBub25jZQ=="),
"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
);
}
#[test]
fn websocket_upgrade_requires_valid_cookie() {
let state = state();
let headers = ws_headers("http://127.0.0.1:17373", None);
let response = validate_ws_upgrade(&headers, &state).unwrap_err();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[test]
fn websocket_upgrade_rejects_invalid_cookie() {
let state = state();
let headers = ws_headers(
"http://127.0.0.1:17373",
Some("idea_session=not-a-valid-session"),
);
let response = validate_ws_upgrade(&headers, &state).unwrap_err();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
#[test]
fn websocket_upgrade_requires_allowed_origin() {
let state = state();
let token = state.create_session();
let headers = ws_headers(
"https://evil.example",
Some(&format!("idea_session={token}")),
);
let response = validate_ws_upgrade(&headers, &state).unwrap_err();
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
#[test]
fn websocket_upgrade_accepts_valid_cookie_and_origin() {
let state = state();
let token = state.create_session();
let headers = ws_headers(
"http://127.0.0.1:17373",
Some(&format!("idea_session={token}")),
);
let accept = validate_ws_upgrade(&headers, &state).unwrap();
assert_eq!(accept, "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=");
}
#[test]
fn websocket_client_text_frame_decodes_masked_json() {
let raw = masked_text_frame(r#"{"kind":"ping"}"#);
let frame = decode_client_ws_frame(&raw).unwrap();
assert_eq!(frame.opcode, WsOpcode::Text);
assert_eq!(frame.payload, br#"{"kind":"ping"}"#);
}
#[test]
fn websocket_client_json_frame_round_trips_base64_bytes() {
let sid = domain::SessionId::new_random().to_string();
let value = json!({
"id": "req-1",
"kind": "input",
"payload": {
"sessionId": sid,
"bytesBase64": "AQID"
}
});
let frame: ClientFrame = serde_json::from_value(value.clone()).unwrap();
let roundtrip = serde_json::to_value(&frame).unwrap();
assert_eq!(roundtrip, value);
assert_eq!(payload_bytes(&frame.payload).unwrap(), vec![1, 2, 3]);
}
#[test]
fn websocket_server_attached_frame_uses_contract_shape() {
let sid = domain::SessionId::new_random().to_string();
let frame = ServerFrame::attached(
"req-attach",
&TerminalSessionDto {
session_id: sid.clone(),
cwd: "/tmp".to_owned(),
rows: 24,
cols: 80,
assigned_conversation_id: None,
engine_session_id: None,
cell_kind: crate::dto::CellKind::Pty,
},
vec![4, 5, 6],
1,
true,
);
let value = serde_json::to_value(frame).unwrap();
assert_eq!(value["kind"], "terminal.attached");
assert_eq!(value["replyTo"], "req-attach");
assert_eq!(value["payload"]["session"]["sessionId"], sid);
assert_eq!(value["payload"]["scrollback"][0]["seq"], 0);
assert_eq!(value["payload"]["scrollback"][0]["bytesBase64"], "BAUG");
assert_eq!(value["payload"]["nextSeq"], 1);
assert_eq!(value["payload"]["gap"], true);
}
#[tokio::test]
async fn websocket_app_ping_emits_pong() {
let state = state();
let (tx, mut rx) = mpsc::channel(4);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(ws_frame("ping-1", "ping", json!({})), &state, &tx, &owned).await;
let frame = recv_server_frame(&mut rx).await;
assert_eq!(frame.kind, "pong");
assert!(frame.payload.as_object().unwrap().is_empty());
}
#[tokio::test]
async fn websocket_open_terminal_emits_attached_with_empty_scrollback() {
let state = state();
let (tx, mut rx) = mpsc::channel(16);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"open-1",
"terminal.open",
open_terminal_payload("/bin/sh", vec!["-c", "sleep 30"]),
),
&state,
&tx,
&owned,
)
.await;
let frame = recv_server_frame(&mut rx).await;
let session_id = attached_session_id(&frame);
assert_eq!(frame.reply_to.as_deref(), Some("open-1"));
assert_eq!(frame.payload["scrollback"].as_array().unwrap().len(), 0);
assert_eq!(frame.payload["nextSeq"], 0);
close_ws_session(&state, &session_id).await;
}
#[tokio::test]
async fn websocket_attach_terminal_replays_scrollback_in_attached_ack() {
let state = state();
let (tx, mut rx) = mpsc::channel(32);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"open-replay",
"terminal.open",
open_terminal_payload("/bin/sh", vec!["-c", "printf replay-ready; sleep 30"]),
),
&state,
&tx,
&owned,
)
.await;
let opened = recv_server_frame(&mut rx).await;
let session_id = attached_session_id(&opened);
let scrollback = wait_for_scrollback(&state, &session_id, b"replay-ready").await;
assert!(scrollback.len() <= 100 * 1024);
drain_server_frames(&mut rx);
handle_client_frame(
ws_frame(
"attach-replay",
"terminal.attach",
json!({
"sessionId": session_id,
"lastSeq": 0,
"rows": 30,
"cols": 100,
}),
),
&state,
&tx,
&owned,
)
.await;
let attached = recv_server_frame(&mut rx).await;
let replay = attached.payload["scrollback"].as_array().unwrap();
assert_eq!(attached.kind, "terminal.attached");
assert_eq!(attached.reply_to.as_deref(), Some("attach-replay"));
assert_eq!(attached.payload["session"]["rows"], 30);
assert_eq!(attached.payload["session"]["cols"], 100);
assert_eq!(attached.payload["gap"], true);
assert_eq!(attached.payload["nextSeq"], 1);
assert_eq!(
base64_decode(replay[0]["bytesBase64"].as_str().unwrap()).unwrap(),
scrollback
);
close_ws_session(
&state,
attached.payload["session"]["sessionId"].as_str().unwrap(),
)
.await;
}
#[tokio::test]
async fn websocket_close_terminal_emits_exited_status_and_releases_session() {
let state = state();
let (tx, mut rx) = mpsc::channel(16);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"open-close",
"terminal.open",
open_terminal_payload("/bin/sh", vec!["-c", "sleep 30"]),
),
&state,
&tx,
&owned,
)
.await;
let opened = recv_server_frame(&mut rx).await;
let session_id = attached_session_id(&opened);
handle_client_frame(
ws_frame(
"close-1",
"terminal.close",
json!({ "sessionId": session_id }),
),
&state,
&tx,
&owned,
)
.await;
let status = recv_server_frame(&mut rx).await;
assert_eq!(status.kind, "terminal.status");
assert_eq!(status.payload["status"], "exited");
assert_eq!(status.payload["sessionId"], session_id);
assert_eq!(state.ws_pty_bridge.active_sessions(), 0);
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'];
let err = decode_client_ws_frame(&raw).unwrap_err();
assert!(matches!(err, WsFrameError::Protocol(_)));
}
#[test]
fn websocket_payload_too_large_is_rejected() {
let len = (WS_MAX_PAYLOAD + 1) as u64;
let mut raw = vec![0x81, 0x80 | 127];
raw.extend_from_slice(&len.to_be_bytes());
raw.extend_from_slice(&[1, 2, 3, 4]);
let err = decode_client_ws_frame(&raw).unwrap_err();
assert_eq!(err, WsFrameError::TooLarge);
}
#[test]
fn websocket_server_frames_are_unmasked() {
let raw = encode_ws_frame(WsOpcode::Text, b"{}");
assert_eq!(raw[0], 0x81);
assert_eq!(raw[1] & 0x80, 0, "server frames must not be masked");
}
#[tokio::test]
async fn websocket_sink_full_reports_backpressure_without_blocking() {
let (tx, mut rx) = mpsc::channel(1);
let sid = domain::SessionId::new_random();
let sink = WsPtySink::new(sid, tx, 0);
assert!(sink.send(vec![1]).is_ok());
assert_eq!(sink.send(vec![2]), Err(OutputSinkError::Full));
let frame = rx.recv().await.unwrap();
assert_eq!(frame.kind, "terminal.output");
}
#[tokio::test]
async fn websocket_output_bridge_replaces_attachment_without_double_delivery() {
let bridge = OutputBridge::<SessionId, PtyChunk>::new();
let sid = domain::SessionId::new_random();
let (old_tx, mut old_rx) = mpsc::channel(4);
let (new_tx, mut new_rx) = mpsc::channel(4);
let old_gen = bridge.register(sid, Arc::new(WsPtySink::new(sid, old_tx, 0)));
let _new_gen = bridge.register(sid, Arc::new(WsPtySink::new(sid, new_tx, 0)));
assert!(bridge.send_output(&sid, vec![9]));
assert!(old_rx.try_recv().is_err());
let frame = new_rx.recv().await.unwrap();
assert_eq!(frame.kind, "terminal.output");
assert_eq!(frame.payload["bytesBase64"], "CQ==");
bridge.unregister_if(&sid, old_gen);
assert_eq!(bridge.active_sessions(), 1);
}
#[test]
fn public_bind_requires_explicit_remote_security() {
let config = ServerConfig {
listen: "0.0.0.0:17373".parse().unwrap(),
..test_config()
};
assert!(config.validate().is_err());
let config = ServerConfig {
allow_remote: true,
public_origin: Some("https://idea.example.com".to_owned()),
trust_reverse_proxy: true,
..config
};
assert!(config.validate().is_ok());
}
#[test]
fn allow_remote_requires_https_origin_and_reverse_proxy() {
let base = ServerConfig {
listen: "0.0.0.0:17373".parse().unwrap(),
allow_remote: true,
..test_config()
};
assert!(base.validate().is_err());
let with_http = ServerConfig {
public_origin: Some("http://idea.example.com".to_owned()),
trust_reverse_proxy: true,
..base.clone()
};
assert!(with_http.validate().is_err());
let without_proxy = ServerConfig {
public_origin: Some("https://idea.example.com".to_owned()),
trust_reverse_proxy: false,
..base
};
assert!(without_proxy.validate().is_err());
}
#[tokio::test]
async fn invoke_requires_valid_cookie() {
let state = state();
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "health", "args": {} }),
&[],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
assert_eq!(body["code"], "UNAUTHORIZED");
}
#[tokio::test]
async fn invoke_rejects_invalid_session_cookie() {
let state = state();
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "health", "args": {} }),
&[("cookie", "idea_session=not-a-valid-session")],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::UNAUTHORIZED);
assert_eq!(body["code"], "UNAUTHORIZED");
}
#[tokio::test]
async fn pairing_rejects_wrong_code() {
let state = state();
let response = request(
state,
Method::POST,
"/api/pair",
json!({ "code": "WRONG999" }),
&[],
)
.await;
let (status, body, headers) = response_json(response).await;
assert_eq!(status, StatusCode::FORBIDDEN);
assert_eq!(body["code"], "FORBIDDEN");
assert!(
!headers.contains_key(SET_COOKIE),
"wrong pairing code must not issue a session cookie"
);
}
#[tokio::test]
async fn pairing_sets_http_only_strict_cookie_for_loopback_dev() {
let state = state();
let response = request(
state,
Method::POST,
"/api/pair",
json!({ "code": "PAIR1234" }),
&[],
)
.await;
let (status, body, headers) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body, json!({ "paired": true }));
let cookie = headers.get(SET_COOKIE).unwrap().to_str().unwrap();
assert!(cookie.contains("idea_session="));
assert!(cookie.contains("HttpOnly"));
assert!(cookie.contains("SameSite=Strict"));
assert!(
!cookie.contains("Secure"),
"loopback dev over HTTP deliberately avoids Secure so browsers store it"
);
}
#[tokio::test]
async fn pairing_sets_secure_cookie_for_remote_https_origin() {
let config = ServerConfig {
listen: "0.0.0.0:17373".parse().unwrap(),
public_origin: Some("https://idea.example.com".to_owned()),
allow_remote: true,
trust_reverse_proxy: true,
..test_config()
};
let state = Arc::new(ServerState::new_for_test(config, "PAIR1234"));
let mut req = Request::builder()
.method(Method::POST)
.uri("/api/pair")
.header(ORIGIN, "https://idea.example.com")
.header(CONTENT_TYPE, "application/json")
.body(Full::new(Bytes::from_static(br#"{"code":"PAIR1234"}"#)))
.unwrap();
*req.headers_mut().get_mut(CONTENT_TYPE).unwrap() =
HeaderValue::from_static("application/json");
let response = handle_request(req, state).await;
let (status, _, headers) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
let cookie = headers.get(SET_COOKIE).unwrap().to_str().unwrap();
assert!(cookie.contains("Secure"));
assert!(cookie.contains("HttpOnly"));
assert!(cookie.contains("SameSite=Strict"));
}
#[tokio::test]
async fn token_or_secret_in_query_string_is_refused() {
let state = state();
let response = request(
state,
Method::POST,
"/api/invoke?token=abc",
json!({ "command": "health", "args": {} }),
&[],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["code"], "INVALID");
}
#[tokio::test]
async fn non_allowed_origin_is_refused() {
let state = state();
let mut req = Request::builder()
.method(Method::POST)
.uri("/api/pair")
.header(ORIGIN, "https://evil.example")
.header(CONTENT_TYPE, "application/json")
.body(Full::new(Bytes::from_static(br#"{"code":"PAIR1234"}"#)))
.unwrap();
*req.headers_mut().get_mut(CONTENT_TYPE).unwrap() =
HeaderValue::from_static("application/json");
let response = handle_request(req, state).await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::FORBIDDEN);
assert_eq!(body["code"], "FORBIDDEN");
}
#[tokio::test]
async fn unknown_allowlisted_command_returns_unknown_command() {
let state = state();
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "debug_dump", "args": {} }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["code"], "UNKNOWN_COMMAND");
}
#[tokio::test]
async fn authorized_health_invoke_returns_health_report() {
let state = state();
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "health", "args": { "request": { "note": "hi" } } }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["alive"], true);
assert_eq!(body["note"], "hi");
}
#[tokio::test]
async fn authorized_list_projects_returns_tauri_project_list_contract() {
let state = state();
let project_id = create_project_for_test(&state, "Web Read").await;
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "list_projects", "args": {} }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
let projects = body
.as_array()
.expect("ProjectListDto is a transparent array");
let project = projects
.iter()
.find(|project| project["id"] == project_id)
.expect("created project is listed");
assert_eq!(project["name"], "Web Read");
assert!(project["root"]
.as_str()
.is_some_and(|root| root.contains("idea-server-project-")));
}
#[tokio::test]
async fn authorized_open_project_returns_readonly_project_dto() {
let state = state();
let project_id = create_project_for_test(&state, "Readonly Open").await;
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "open_project", "args": { "projectId": project_id } }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], project_id);
assert_eq!(body["name"], "Readonly Open");
assert!(body["root"].is_string());
}
#[tokio::test]
async fn authorized_get_project_work_state_returns_tauri_contract() {
let state = state();
let project_id = create_project_for_test(&state, "Readonly Work").await;
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "get_project_work_state", "args": { "projectId": project_id } }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::OK);
assert!(body["agents"].as_array().is_some());
assert!(body["conversations"].as_array().is_some());
}
#[tokio::test]
async fn mutation_and_pty_commands_stay_out_of_readonly_allowlist() {
let state = state();
let cookie = pair_and_cookie(Arc::clone(&state)).await;
for command in ["create_project", "open_terminal", "launch_agent"] {
let response = request(
Arc::clone(&state),
Method::POST,
"/api/invoke",
json!({ "command": command, "args": {} }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(
body["code"], "UNKNOWN_COMMAND",
"{command} must stay blocked"
);
}
}
#[tokio::test]
async fn invalid_project_id_is_mapped_to_error_dto() {
let state = state();
let cookie = pair_and_cookie(Arc::clone(&state)).await;
let response = request(
state,
Method::POST,
"/api/invoke",
json!({ "command": "open_project", "args": { "projectId": "nope" } }),
&[("cookie", &cookie)],
)
.await;
let (status, body, _) = response_json(response).await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert_eq!(body["code"], "INVALID");
}
#[test]
fn direct_project_id_parser_still_rejects_invalid_ids() {
assert!(parse_project_id("not-a-uuid").is_err());
}
}
Reference in New Issue View Git Blame Copy Permalink