use crate::{
CLI_VERSION,
api::ApiClient,
auth::AuthManager,
config::{ConfigStore, ProjectEntry, WorktreeSyncState},
error::{ErrorCode, ExeoraError},
policy::{CommandPolicy, effective_policy, policy_allows},
protocol::{
HEARTBEAT_INTERVAL_MS, HEARTBEAT_REQUEST, HEARTBEAT_TIMEOUT_MS, MAX_RESULT_BYTES,
PRESENCE_SIGNAL_INTERVAL_MS, PROTOCOL_VERSION, ToolName, now_ms,
},
tools::ToolEngine,
workspace::WorkspaceEngine,
};
use anyhow::{Context, Result, anyhow};
use base64::{Engine as _, engine::general_purpose::STANDARD};
use futures_util::{SinkExt, StreamExt};
use serde_json::{Value, json};
use std::{
collections::{HashMap, HashSet},
path::{Path, PathBuf},
sync::Arc,
time::Duration,
};
use tokio::sync::{Mutex, mpsc};
use tokio_tungstenite::{
connect_async,
tungstenite::{Error as WebSocketError, Message, client::IntoClientRequest, http::HeaderValue},
};
use tokio_util::sync::CancellationToken;
use url::Url;
struct ActiveCall {
cancel: CancellationToken,
root: PathBuf,
}
#[derive(Debug)]
struct ResolvedTarget {
project: ProjectEntry,
root: PathBuf,
worktree_slug: Option<String>,
}
type InFlight = Arc<Mutex<HashMap<String, ActiveCall>>>;
pub async fn connect_forever(
config: &ConfigStore,
api: &ApiClient,
auth: Arc<AuthManager>,
device_id: String,
projects: Vec<ProjectEntry>,
json_output: bool,
) -> Result<()> {
let _awake = acquire_keep_awake(json_output);
let engine = Arc::new(ToolEngine::new()?);
let workspace = Arc::new(WorkspaceEngine::new());
let config_path = config.path().to_path_buf();
let gateway = config.gateway_url();
let mut delay = Duration::from_secs(1);
let stop = CancellationToken::new();
let signal_stop = stop.clone();
tokio::spawn(async move {
let _ = tokio::signal::ctrl_c().await;
signal_stop.cancel();
});
loop {
if stop.is_cancelled() {
break;
}
let outcome = connect_once(
&gateway,
&device_id,
&projects,
config_path.clone(),
auth.clone(),
api.clone(),
engine.clone(),
workspace.clone(),
stop.clone(),
json_output,
)
.await;
workspace.kill_all().await;
engine.kill_all().await;
match outcome {
Ok(ConnectOutcome::Stopped) => break,
Ok(ConnectOutcome::Rejected(reason)) => return Err(anyhow!(reason)),
Ok(ConnectOutcome::Disconnected) => {
delay = Duration::from_secs(1);
emit_event(
json_output,
"close",
json!({ "reason": format!("Disconnected. Reconnecting in {}s.", delay.as_secs()) }),
);
tokio::select! { _ = tokio::time::sleep(delay) => {}, _ = stop.cancelled() => break }
}
Err(error) => {
emit_event(
json_output,
"close",
json!({ "reason": format!("{error}. Reconnecting in {}s.", delay.as_secs()) }),
);
tokio::select! { _ = tokio::time::sleep(delay) => {}, _ = stop.cancelled() => break }
delay = (delay * 2).min(Duration::from_secs(30));
}
}
}
engine.kill_all().await;
workspace.kill_all().await;
if !json_output {
println!("Disconnected.");
}
Ok(())
}
fn acquire_keep_awake(json_output: bool) -> Option<keepawake::KeepAwake> {
match keepawake::Builder::default()
.idle(true)
.display(true)
.reason("Exeora is serving remote tool calls")
.app_name("Exeora")
.app_reverse_domain("dev.exeora.cli")
.create()
{
Ok(awake) => {
emit_event(json_output, "awake", awake_event(true, None));
if !json_output {
println!("✓ Keeping the system and display awake while connect runs.");
}
Some(awake)
}
Err(error) => {
let reason = error.to_string();
emit_event(json_output, "awake", awake_event(false, Some(&reason)));
if !json_output {
eprintln!(
"warning: Could not keep the system and display awake: {reason}. Continuing to connect."
);
}
None
}
}
}
fn awake_event(active: bool, reason: Option<&str>) -> Value {
let mut fields = json!({
"active": active,
"system": active,
"display": active,
});
if let (Some(fields), Some(reason)) = (fields.as_object_mut(), reason) {
fields.insert("reason".to_owned(), json!(reason));
}
fields
}
enum ConnectOutcome {
Stopped,
Rejected(String),
Disconnected,
}
fn handshake_rejection(status: u16, body: Option<&[u8]>) -> String {
let detail = body
.and_then(|bytes| std::str::from_utf8(bytes).ok())
.map(str::trim)
.filter(|text| !text.is_empty())
.map(|text| {
serde_json::from_str::<Value>(text)
.ok()
.and_then(|value| {
let error = value.get("error")?.as_str()?;
let scopes = value
.get("requiredScopes")
.and_then(Value::as_array)
.map(|entries| {
entries
.iter()
.filter_map(Value::as_str)
.collect::<Vec<_>>()
.join(", ")
})
.filter(|scopes| !scopes.is_empty());
Some(scopes.map_or_else(
|| error.to_owned(),
|scopes| format!("{error}; required scopes: {scopes}"),
))
})
.unwrap_or_else(|| text.chars().take(200).collect())
})
.unwrap_or_else(|| "the gateway refused this machine".to_owned());
format!("Relay rejected the connection ({status}): {detail}")
}
#[allow(clippy::too_many_arguments)]
async fn connect_once(
gateway: &str,
device_id: &str,
projects: &[ProjectEntry],
config_path: PathBuf,
auth: Arc<AuthManager>,
api: crate::api::ApiClient,
engine: Arc<ToolEngine>,
workspace: Arc<WorkspaceEngine>,
stop: CancellationToken,
json_output: bool,
) -> Result<ConnectOutcome> {
let token = auth.access_token().await?;
let mut url = Url::parse(gateway)?.join(&format!("/api/relay/{device_id}"))?;
url.set_scheme(if url.scheme() == "https" { "wss" } else { "ws" })
.map_err(|_| anyhow!("invalid relay URL"))?;
let mut request = url.as_str().into_client_request()?;
request.headers_mut().insert(
"authorization",
HeaderValue::from_str(&format!("Bearer {token}"))?,
);
let (mut socket, _) = match connect_async(request).await {
Ok(connection) => connection,
Err(WebSocketError::Http(response))
if matches!(response.status().as_u16(), 401 | 403 | 404) =>
{
return Ok(ConnectOutcome::Rejected(handshake_rejection(
response.status().as_u16(),
response.body().as_deref(),
)));
}
Err(error) => return Err(error).context("Could not connect to the Exeora relay"),
};
let can_prompt = !json_output
&& std::io::IsTerminal::is_terminal(&std::io::stdin())
&& std::io::IsTerminal::is_terminal(&std::io::stdout());
socket.send(Message::Text(serde_json::to_string(&json!({
"type": "hello", "protocolVersion": PROTOCOL_VERSION, "deviceId": device_id,
"cliVersion": CLI_VERSION, "platform": platform(),
"projects": projects.iter().map(|project| json!({ "id": project.id, "slug": project.slug })).collect::<Vec<_>>(),
"capabilities": {
"prompt": can_prompt,
"tools": ToolName::ALL.iter().map(ToString::to_string).collect::<Vec<_>>(),
"features": ["source-control-v1", "terminal-v1"],
"worktreeRouting": true,
},
}))?.into())).await?;
emit_event(json_output, "open", json!({}));
if !json_output {
println!("✓ Connected. Waiting for tool calls.");
}
let (out_tx, mut out_rx) = mpsc::unbounded_channel::<Value>();
let (terminal_tx, mut terminal_rx) = mpsc::channel::<Value>(256);
let in_flight: InFlight = Arc::new(Mutex::new(HashMap::new()));
let mut tick = tokio::time::interval(Duration::from_millis(HEARTBEAT_INTERVAL_MS));
tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut heartbeat_auto = false;
let mut last_ack = now_ms();
let mut last_presence = now_ms();
let mut roots_tick = tokio::time::interval(Duration::from_secs(1));
roots_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
let mut known_roots = served_roots(&config_path);
loop {
tokio::select! {
_ = stop.cancelled() => {
let _ = socket.close(None).await;
cancel_all(&in_flight).await;
engine.kill_all().await;
workspace.kill_all().await;
return Ok(ConnectOutcome::Stopped);
}
_ = tick.tick() => {
let now = now_ms();
if heartbeat_auto && now.saturating_sub(last_ack) > HEARTBEAT_TIMEOUT_MS {
let _ = socket.close(None).await;
break;
}
let frame = if heartbeat_auto { HEARTBEAT_REQUEST.to_owned() } else { json!({ "type": "heartbeat", "at": now }).to_string() };
socket.send(Message::Text(frame.into())).await?;
if heartbeat_auto && now.saturating_sub(last_presence) >= PRESENCE_SIGNAL_INTERVAL_MS {
socket.send(Message::Text(json!({ "type": "presence", "at": now }).to_string().into())).await?;
last_presence = now;
}
}
Some(outgoing) = out_rx.recv() => {
socket.send(Message::Text(outgoing.to_string().into())).await?;
}
Some(outgoing) = terminal_rx.recv() => {
socket.send(Message::Text(outgoing.to_string().into())).await?;
}
incoming = socket.next() => {
let Some(incoming) = incoming else { break; };
match incoming? {
Message::Text(text) => {
let Ok(message) = serde_json::from_str::<Value>(&text) else { continue; };
match message.get("type").and_then(Value::as_str) {
Some("heartbeat.ack") => last_ack = now_ms(),
Some("hello.ack") => {
heartbeat_auto = message.get("heartbeatMode").and_then(Value::as_str) == Some("auto");
last_ack = now_ms();
if let Some(latest) = message.get("latestCliVersion").and_then(Value::as_str)
&& is_outdated(CLI_VERSION, latest) {
let notice = format!("A newer Exeora CLI is available ({CLI_VERSION} → {latest}). Run `exeora upgrade`.");
emit_event(json_output, "notice", json!({ "message": notice }));
if !json_output { println!("{notice}"); }
}
}
Some("cancel") => {
if let Some(id) = message.get("requestId").and_then(Value::as_str)
&& let Some(call) = in_flight.lock().await.get(id) {
call.cancel.cancel();
}
}
Some("approval.request") => {
let tx = out_tx.clone();
tokio::spawn(handle_approval(message, tx, can_prompt, json_output));
}
Some("approval.resolved") => {}
Some("shutdown") => {
let reason = message.get("reason").and_then(Value::as_str).unwrap_or("The gateway closed the connection.");
cancel_all(&in_flight).await;
engine.kill_all().await;
workspace.kill_all().await;
return Ok(ConnectOutcome::Rejected(reason.to_owned()));
}
Some("tool.call") => {
spawn_tool_call(message, config_path.clone(), engine.clone(), in_flight.clone(), out_tx.clone(), json_output).await;
}
Some("workspace.call") => {
spawn_workspace_call(message, config_path.clone(), api.clone(), workspace.clone(), in_flight.clone(), out_tx.clone()).await;
}
Some("terminal.open") | Some("terminal.input") | Some("terminal.resize") | Some("terminal.close") => {
handle_terminal_message(message, config_path.clone(), workspace.clone(), terminal_tx.clone()).await;
}
_ => {}
}
}
Message::Ping(data) => socket.send(Message::Pong(data)).await?,
Message::Close(frame) => {
if let Some(frame) = frame
&& frame.code == tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Policy {
return Ok(ConnectOutcome::Rejected(frame.reason.to_string()));
}
break;
}
_ => {}
}
}
_ = roots_tick.tick() => {
reconcile_roots(&config_path, &engine, &workspace, &in_flight, &mut known_roots).await;
}
}
}
cancel_all(&in_flight).await;
engine.kill_all().await;
workspace.kill_all().await;
Ok(ConnectOutcome::Disconnected)
}
async fn spawn_tool_call(
message: Value,
config_path: PathBuf,
engine: Arc<ToolEngine>,
in_flight: InFlight,
outgoing: mpsc::UnboundedSender<Value>,
json_output: bool,
) {
let Some(request_id) = message
.get("requestId")
.and_then(Value::as_str)
.map(str::to_owned)
else {
return;
};
let Some(project_id) = message.get("projectId").and_then(Value::as_str) else {
return;
};
let started = now_ms();
let send_error = |code: ErrorCode, text: &str| {
let _ = outgoing.send(result_frame(
&request_id,
started,
Err(ExeoraError::new(code, text)),
));
};
if message
.get("expiresAt")
.and_then(Value::as_u64)
.is_some_and(|expires| now_ms() > expires)
{
send_error(
ErrorCode::ToolTimeout,
"The request expired before it was received.",
);
return;
}
let target = match resolve_target(
&config_path,
project_id,
message.get("worktreeId").and_then(Value::as_str),
message.get("worktreeSlug").and_then(Value::as_str),
) {
Ok(target) => target,
Err(error) => {
send_error(error.code, &error.message);
return;
}
};
let ResolvedTarget {
project,
root,
worktree_slug,
} = target;
let Some(tool_name) = message
.get("tool")
.and_then(Value::as_str)
.map(str::to_owned)
else {
send_error(ErrorCode::UnknownTool, "Unsupported tool.");
return;
};
let Ok(tool) = tool_name.parse::<ToolName>() else {
send_error(ErrorCode::UnknownTool, "Unsupported tool.");
return;
};
let arguments = message
.get("arguments")
.cloned()
.unwrap_or_else(|| json!({}));
let remote = message
.get("policy")
.cloned()
.and_then(|value| serde_json::from_value::<CommandPolicy>(value).ok());
let (policy, problem) = effective_policy(&root, remote);
if let Some(problem) = problem {
emit_event(json_output, "error", json!({ "message": problem }));
}
let verdict = policy_allows(&policy, tool, &arguments);
if !verdict.allowed {
send_error(
ErrorCode::Forbidden,
verdict
.reason
.as_deref()
.unwrap_or("This project does not allow that."),
);
return;
}
let cancel = CancellationToken::new();
in_flight.lock().await.insert(
request_id.clone(),
ActiveCall {
cancel: cancel.clone(),
root: root.clone(),
},
);
emit_event(
json_output,
"call",
json!({ "tool": tool_name, "project": project.slug, "worktree": worktree_slug, "client": describe_client(message.get("client")) }),
);
if !json_output {
println!(
"→ {tool_name} ({}/{})",
project.slug,
worktree_slug.as_deref().unwrap_or("main")
);
}
tokio::spawn(async move {
let result = engine
.execute_for_project(&root, &project.id, tool, arguments, cancel)
.await;
in_flight.lock().await.remove(&request_id);
let elapsed = now_ms().saturating_sub(started);
let frame = result_frame(&request_id, started, result);
let ok = frame
.pointer("/result/ok")
.and_then(Value::as_bool)
.unwrap_or(false);
let _ = outgoing.send(frame);
emit_event(
json_output,
"result",
json!({ "tool": tool_name, "ok": ok, "durationMs": elapsed }),
);
if !json_output {
println!("{} {tool_name} {elapsed}ms", if ok { "✓" } else { "✗" });
}
});
}
async fn handle_approval(
message: Value,
outgoing: mpsc::UnboundedSender<Value>,
can_prompt: bool,
json_output: bool,
) {
let Some(id) = message.get("id").and_then(Value::as_str).map(str::to_owned) else {
return;
};
if !can_prompt {
let _ = outgoing.send(json!({ "type": "approval.answer", "id": id, "approved": false }));
return;
}
let prompt = message
.get("prompt")
.and_then(Value::as_str)
.unwrap_or("Allow this tool call?")
.to_owned();
let approved = tokio::task::spawn_blocking(move || {
cliclack::confirm(prompt)
.initial_value(false)
.interact()
.unwrap_or(false)
})
.await
.unwrap_or(false);
emit_event(
json_output,
"approval",
json!({ "id": id, "approved": approved }),
);
let _ = outgoing.send(json!({ "type": "approval.answer", "id": id, "approved": approved }));
}
async fn spawn_workspace_call(
message: Value,
config_path: PathBuf,
api: crate::api::ApiClient,
workspace: Arc<WorkspaceEngine>,
in_flight: InFlight,
outgoing: mpsc::UnboundedSender<Value>,
) {
let Some(request_id) = message
.get("requestId")
.and_then(Value::as_str)
.map(str::to_owned)
else {
return;
};
let Some(project_id) = message.get("projectId").and_then(Value::as_str) else {
return;
};
let started = now_ms();
let send_error = |error: ExeoraError| {
let _ = outgoing.send(workspace_result_frame(&request_id, started, Err(error)));
};
if message
.get("expiresAt")
.and_then(Value::as_u64)
.is_some_and(|expires| started > expires)
{
send_error(ExeoraError::new(
ErrorCode::ToolTimeout,
"The workspace request expired before it was received.",
));
return;
}
let target = match resolve_target(
&config_path,
project_id,
message.get("worktreeId").and_then(Value::as_str),
message.get("worktreeSlug").and_then(Value::as_str),
) {
Ok(target) => target,
Err(error) => {
send_error(error);
return;
}
};
let action = message.get("action").cloned().unwrap_or_else(|| json!({}));
let cancel = CancellationToken::new();
in_flight.lock().await.insert(
request_id.clone(),
ActiveCall {
cancel: cancel.clone(),
root: target.root.clone(),
},
);
let create_worktree = action.get("action").and_then(Value::as_str) == Some("worktree_create");
tokio::spawn(async move {
let result = if create_worktree {
create_workspace_worktree(
&config_path,
&api,
&target.project.id,
&target.root,
action,
workspace.as_ref(),
cancel,
)
.await
} else {
workspace.execute(&target.root, action, cancel).await
};
in_flight.lock().await.remove(&request_id);
let _ = outgoing.send(workspace_result_frame(&request_id, started, result));
});
}
async fn create_workspace_worktree(
config_path: &std::path::Path,
api: &crate::api::ApiClient,
project_id: &str,
source_root: &Path,
action: Value,
workspace: &WorkspaceEngine,
cancel: CancellationToken,
) -> Result<Value, ExeoraError> {
let branch = action
.get("branch")
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.ok_or_else(|| {
ExeoraError::new(ErrorCode::InvalidArguments, "A branch name is required.")
})?;
let mut config = ConfigStore::load_from(config_path.to_path_buf()).map_err(|_| {
ExeoraError::new(
ErrorCode::InternalError,
"Could not reload the local Exeora configuration.",
)
})?;
let project = config.find_project(project_id).cloned().ok_or_else(|| {
ExeoraError::new(
ErrorCode::UnknownProject,
"This machine does not serve that project. Run `exeora project add` there.",
)
})?;
let entry = crate::worktrees::create(
&config,
&project,
crate::worktrees::CreateWorktree {
branch: branch.to_owned(),
from: action
.get("from")
.and_then(Value::as_str)
.map(str::to_owned),
reuse_existing_branch: action
.get("reuseExistingBranch")
.and_then(Value::as_bool)
.unwrap_or(false),
name: action
.get("name")
.and_then(Value::as_str)
.map(str::to_owned),
slug: action
.get("slug")
.and_then(Value::as_str)
.map(str::to_owned),
path: None,
source: Some(source_root.to_path_buf()),
},
)
.map_err(|error| ExeoraError::new(ErrorCode::InvalidArguments, error.to_string()))?;
config.upsert_worktree(entry.clone());
config.save().map_err(|error| {
ExeoraError::new(
ErrorCode::InternalError,
format!("Could not save the local worktree record: {error}"),
)
})?;
let mut entry = entry;
api.put_worktree(&entry.project_id, &entry)
.await
.map_err(|error| {
ExeoraError::new(
ErrorCode::InternalError,
format!("Created the Git worktree but could not register it with Exeora: {error}"),
)
})?;
entry.sync_state = WorktreeSyncState::Active;
config.upsert_worktree(entry.clone());
let _ = config.save();
let status = workspace
.execute(source_root, json!({ "action": "status" }), cancel)
.await?;
Ok(json!({
"kind": "mutation",
"stdout": "",
"stderr": "",
"status": status,
"worktree": {
"id": entry.id,
"slug": entry.slug,
"name": entry.name,
"branch": entry.branch,
"localPath": entry.root,
}
}))
}
async fn handle_terminal_message(
message: Value,
config_path: PathBuf,
workspace: Arc<WorkspaceEngine>,
outgoing: mpsc::Sender<Value>,
) {
let Some(kind) = message.get("type").and_then(Value::as_str) else {
return;
};
let Some(session_id) = message
.get("sessionId")
.and_then(Value::as_str)
.map(str::to_owned)
else {
return;
};
let result = match kind {
"terminal.open" => {
let Some(project_id) = message.get("projectId").and_then(Value::as_str) else {
send_terminal_error(&outgoing, &session_id, "A terminal project is required.")
.await;
return;
};
let target = match resolve_target(
&config_path,
project_id,
message.get("worktreeId").and_then(Value::as_str),
message.get("worktreeSlug").and_then(Value::as_str),
) {
Ok(target) => target,
Err(error) => {
send_terminal_error(&outgoing, &session_id, &error.message).await;
return;
}
};
let cols = message
.get("cols")
.and_then(Value::as_u64)
.and_then(|value| u16::try_from(value).ok());
let rows = message
.get("rows")
.and_then(Value::as_u64)
.and_then(|value| u16::try_from(value).ok());
match (cols, rows) {
(Some(cols), Some(rows)) => {
workspace
.terminal_open(
session_id.clone(),
&target.root,
cols,
rows,
outgoing.clone(),
)
.await
}
_ => Err(ExeoraError::new(
ErrorCode::InvalidArguments,
"Invalid terminal size.",
)),
}
}
"terminal.input" => match message
.get("data")
.and_then(Value::as_str)
.and_then(|data| STANDARD.decode(data).ok())
{
Some(data) => workspace.terminal_input(&session_id, &data).await,
None => Err(ExeoraError::new(
ErrorCode::InvalidArguments,
"Invalid terminal input encoding.",
)),
},
"terminal.resize" => {
let cols = message
.get("cols")
.and_then(Value::as_u64)
.and_then(|value| u16::try_from(value).ok());
let rows = message
.get("rows")
.and_then(Value::as_u64)
.and_then(|value| u16::try_from(value).ok());
match (cols, rows) {
(Some(cols), Some(rows)) => {
workspace.terminal_resize(&session_id, cols, rows).await
}
_ => Err(ExeoraError::new(
ErrorCode::InvalidArguments,
"Invalid terminal size.",
)),
}
}
"terminal.close" => {
workspace.terminal_close(&session_id).await;
Ok(())
}
_ => return,
};
if let Err(error) = result {
send_terminal_error(&outgoing, &session_id, &error.message).await;
}
}
async fn send_terminal_error(outgoing: &mpsc::Sender<Value>, session_id: &str, message: &str) {
let _ = outgoing
.send(json!({ "type": "terminal.error", "sessionId": session_id, "message": message }))
.await;
}
fn resolve_target(
config_path: &Path,
project_id: &str,
worktree_id: Option<&str>,
worktree_slug: Option<&str>,
) -> Result<ResolvedTarget, ExeoraError> {
let config = ConfigStore::load_from(config_path.to_path_buf()).map_err(|_| {
ExeoraError::new(
ErrorCode::InternalError,
"Could not reload the local Exeora configuration.",
)
})?;
let project = config.find_project(project_id).cloned().ok_or_else(|| {
ExeoraError::new(
ErrorCode::UnknownProject,
"This machine does not serve that project. Run `exeora project add` there.",
)
})?;
let Some(worktree_id) = worktree_id else {
if worktree_slug.is_some() {
return Err(ExeoraError::new(
ErrorCode::WorktreeUnavailable,
"The worktree target is incomplete.",
));
}
if !project.root.is_dir() {
return Err(ExeoraError::new(
ErrorCode::PathNotFound,
"The project directory is unavailable on this machine.",
));
}
return Ok(ResolvedTarget {
root: std::fs::canonicalize(&project.root).unwrap_or_else(|_| project.root.clone()),
project,
worktree_slug: None,
});
};
let worktree = config
.data()
.worktrees
.iter()
.find(|entry| {
entry.id == worktree_id
&& entry.project_id == project.id
&& entry.sync_state == WorktreeSyncState::Active
&& worktree_slug.is_none_or(|slug| slug == entry.slug)
})
.ok_or_else(|| {
ExeoraError::new(
ErrorCode::WorktreeUnavailable,
"That worktree is no longer connected or available on this machine.",
)
})?;
if !worktree.root.is_dir() {
return Err(ExeoraError::new(
ErrorCode::WorktreeUnavailable,
"That worktree is registered but its directory is unavailable.",
));
}
Ok(ResolvedTarget {
project,
root: std::fs::canonicalize(&worktree.root).unwrap_or_else(|_| worktree.root.clone()),
worktree_slug: Some(worktree.slug.clone()),
})
}
fn workspace_result_frame(
request_id: &str,
started: u64,
result: Result<Value, ExeoraError>,
) -> Value {
let result = match result {
Ok(value)
if serde_json::to_vec(&value).is_ok_and(|bytes| bytes.len() <= MAX_RESULT_BYTES) =>
{
json!({ "ok": true, "value": value })
}
Ok(_) => json!({
"ok": false,
"error": {
"code": ErrorCode::ToolFailed.as_str(),
"message": "Workspace result exceeded the protocol limit.",
}
}),
Err(error) => {
json!({ "ok": false, "error": { "code": error.code.as_str(), "message": error.message } })
}
};
json!({ "type": "workspace.result", "requestId": request_id, "durationMs": now_ms().saturating_sub(started), "result": result })
}
fn result_frame(request_id: &str, started: u64, result: Result<Value, ExeoraError>) -> Value {
let result = match result {
Ok(value)
if serde_json::to_vec(&value).is_ok_and(|bytes| bytes.len() <= MAX_RESULT_BYTES) =>
{
json!({ "ok": true, "value": value })
}
Ok(_) => json!({
"ok": false,
"error": {
"code": ErrorCode::ToolFailed.as_str(),
"message": format!("Tool result exceeded the {MAX_RESULT_BYTES}-byte protocol limit. Narrow the request and try again."),
}
}),
Err(error) => {
json!({ "ok": false, "error": { "code": error.code.as_str(), "message": error.message } })
}
};
json!({ "type": "tool.result", "requestId": request_id, "durationMs": now_ms().saturating_sub(started), "result": result })
}
async fn cancel_all(in_flight: &InFlight) {
let mut calls = in_flight.lock().await;
for call in calls.values() {
call.cancel.cancel();
}
calls.clear();
}
fn served_roots(config_path: &std::path::Path) -> HashSet<PathBuf> {
let Ok(config) = ConfigStore::load_from(config_path.to_path_buf()) else {
return HashSet::new();
};
let mut allowed: HashSet<PathBuf> = config
.data()
.projects
.iter()
.filter_map(|entry| std::fs::canonicalize(&entry.root).ok())
.collect();
allowed.extend(
config
.data()
.worktrees
.iter()
.filter(|entry| entry.sync_state == WorktreeSyncState::Active)
.filter_map(|entry| std::fs::canonicalize(&entry.root).ok()),
);
allowed
}
async fn reconcile_roots(
config_path: &std::path::Path,
engine: &ToolEngine,
workspace: &WorkspaceEngine,
in_flight: &InFlight,
known_roots: &mut HashSet<PathBuf>,
) {
let allowed = served_roots(config_path);
let mut removed: HashSet<PathBuf> = known_roots.difference(&allowed).cloned().collect();
removed.extend({
let calls = in_flight.lock().await;
calls
.values()
.map(|call| std::fs::canonicalize(&call.root).unwrap_or_else(|_| call.root.clone()))
.filter(|root| !allowed.contains(root))
.collect::<Vec<_>>()
});
*known_roots = allowed;
if removed.is_empty() {
return;
}
{
let calls = in_flight.lock().await;
for call in calls.values() {
let root = std::fs::canonicalize(&call.root).unwrap_or_else(|_| call.root.clone());
if removed.contains(&root) {
call.cancel.cancel();
}
}
}
for root in removed {
engine.kill_root(&root).await;
workspace.kill_root(&root).await;
}
}
fn describe_client(value: Option<&Value>) -> Option<String> {
let value = value?;
match (
value.get("name").and_then(Value::as_str),
value.get("version").and_then(Value::as_str),
) {
(Some(name), Some(version)) => Some(format!("{name} {version}")),
(Some(name), None) => Some(name.to_owned()),
(None, Some(version)) => Some(version.to_owned()),
_ => None,
}
}
fn emit_event(json_output: bool, event: &str, fields: Value) {
if !json_output {
return;
}
let mut value = json!({ "at": now_ms(), "event": event });
if let (Some(target), Some(source)) = (value.as_object_mut(), fields.as_object()) {
target.extend(source.clone());
}
println!("{value}");
}
fn is_outdated(current: &str, latest: &str) -> bool {
match (
semver::Version::parse(current),
semver::Version::parse(latest),
) {
(Ok(current), Ok(latest)) => current < latest,
_ => false,
}
}
fn platform() -> &'static str {
if cfg!(target_os = "windows") {
"win32"
} else if cfg!(target_os = "macos") {
"darwin"
} else {
"linux"
}
}
#[cfg(test)]
mod tests {
use super::{awake_event, handshake_rejection, resolve_target, result_frame};
use crate::{
config::{ConfigStore, ProjectEntry, WorktreeEntry, WorktreeSyncState},
error::ErrorCode,
protocol::MAX_RESULT_BYTES,
};
use serde_json::json;
use std::fs;
use tempfile::tempdir;
#[test]
fn rejects_an_oversized_tool_result_before_it_reaches_the_socket() {
let frame = result_frame(
"req_test",
0,
Ok(json!({ "content": "x".repeat(MAX_RESULT_BYTES) })),
);
assert_eq!(frame["result"]["ok"], false);
assert_eq!(frame["result"]["error"]["code"], "TOOL_FAILED");
}
#[test]
fn explains_a_rejected_relay_handshake_instead_of_hiding_it_in_retries() {
let message = handshake_rejection(
403,
Some(br#"{"error":"insufficient_scope","requiredScopes":["executor:connect"]}"#),
);
assert_eq!(
message,
"Relay rejected the connection (403): insufficient_scope; required scopes: executor:connect"
);
}
#[test]
fn reports_keep_awake_state_without_breaking_json_streams() {
assert_eq!(
awake_event(true, None),
json!({ "active": true, "system": true, "display": true })
);
assert_eq!(
awake_event(false, Some("not supported")),
json!({
"active": false,
"system": false,
"display": false,
"reason": "not supported",
})
);
}
#[test]
fn resolves_only_the_active_worktree_with_the_matching_stable_identity() {
let directory = tempdir().unwrap();
let main = directory.path().join("main");
let feature = directory.path().join("feature");
let pending = directory.path().join("pending");
fs::create_dir_all(&main).unwrap();
fs::create_dir_all(&feature).unwrap();
fs::create_dir_all(&pending).unwrap();
let config_path = directory.path().join("config.json");
let mut config = ConfigStore::load_from(config_path.clone()).unwrap();
config.upsert_project(ProjectEntry {
id: "prj_1".to_owned(),
slug: "project".to_owned(),
name: "Project".to_owned(),
root: main.clone(),
});
for (id, slug, root, sync_state) in [
(
"wtr_active",
"feature",
feature.clone(),
WorktreeSyncState::Active,
),
(
"wtr_pending",
"pending",
pending,
WorktreeSyncState::PendingUpsert,
),
] {
config.upsert_worktree(WorktreeEntry {
id: id.to_owned(),
project_id: "prj_1".to_owned(),
slug: slug.to_owned(),
name: slug.to_owned(),
branch: Some(slug.to_owned()),
git_root: main.clone(),
root,
managed: true,
sync_state,
});
}
config.save().unwrap();
let resolved =
resolve_target(&config_path, "prj_1", Some("wtr_active"), Some("feature")).unwrap();
assert_eq!(resolved.root, fs::canonicalize(feature).unwrap());
assert_eq!(resolved.worktree_slug.as_deref(), Some("feature"));
for (id, slug) in [
("wtr_active", Some("renamed")),
("wtr_pending", Some("pending")),
("wtr_missing", None),
] {
let error = resolve_target(&config_path, "prj_1", Some(id), slug).unwrap_err();
assert_eq!(error.code, ErrorCode::WorktreeUnavailable);
}
}
}