use std::time::{Duration, Instant};
use super::{ClientId, PendingWait, Server};
use crate::agent::AgentState;
use crate::input::KeySpec;
use crate::layout::PaneId;
use crate::pane::PaneSpec;
use crate::protocol::{self, Request, Response, ServerMsg, WorkspaceInfo};
type Reply = Option<Result<Response, String>>;
impl Server {
pub(super) fn request(&mut self, client: ClientId, req: Request) -> Reply {
let reply = self.answer(client, req);
self.drop_empty_workspaces();
reply
}
fn answer(&mut self, client: ClientId, req: Request) -> Reply {
Some(match req {
Request::RunbookPlan { path, target, params, pane } => {
let pane = match pane.as_deref().map(|p| self.resolve(p)).transpose() {
Ok(p) => p,
Err(e) => return Some(Err(e)),
};
self.prepare_runbook(&path, target.as_deref(), params, pane)
.map(|(text, token)| Response::Plan { text, token })
.map_err(|e| format!("{e:#}"))
}
Request::RunbookRun { token } => match self.run_prepared(&token, client) {
Ok(()) => return None,
Err(e) => Err(format!("{e:#}")),
},
Request::Subscribe { events, since, boot } => match self.subscribe(client, events, since, boot) {
Ok(()) => return None,
Err(e) => Err(e),
},
Request::Ping => Ok(Response::Pong {
pid: std::process::id(),
session: self.session.clone(),
version: env!("CARGO_PKG_VERSION").into(),
}),
Request::ListPanes => {
let mut v: Vec<PaneId> = Vec::new();
for ws in &self.workspaces {
v.extend(ws.panes());
}
Ok(Response::Panes(v.into_iter().map(|p| self.pane_info(p)).collect()))
}
Request::ListWorkspaces => Ok(Response::Workspaces(self.workspace_infos())),
Request::OpenWorkspace { name } => {
self.open_workspace(&name).map(|_| Response::Ok).map_err(|e| format!("{e:#}"))
}
Request::NewPane { target, command, cwd, name, placement, preview, focus, close_on_exit } => {
let target = match target.as_deref().map(|t| self.resolve(t)).transpose() {
Ok(t) => t,
Err(e) => return Some(Err(e)),
};
let ws = match target.and_then(|t| self.workspace_of(t)) {
Some((wi, _)) => self.workspaces[wi].id,
None => match self
.primary
.and_then(|p| self.clients.get(&p))
.map(|c| c.workspace)
.or_else(|| self.workspaces.first().map(|w| w.id))
{
Some(ws) => ws,
None => self.new_workspace("main".into(), None),
},
};
let previous = match target.and_then(|t| self.workspace_of(t)) {
Some((wi, ti)) => self.workspaces[wi].tabs[ti].focus,
None => self
.ws_index(ws)
.and_then(|wi| self.workspaces[wi].tabs.get(self.workspaces[wi].active))
.and_then(|t| t.focus),
};
let shown = self.ws_index(ws).map(|wi| {
let w = &self.workspaces[wi];
(w.active, w.tabs.get(w.active).map(|t| t.show_floating))
});
let spec = PaneSpec { command, cwd, preview, close_on_exit, ..Default::default() };
match self.spawn_pane_in(ws, target, spec, placement, name) {
Ok(id) => {
if !focus {
if let Some(prev) = previous
&& let Some((wi, ti)) = self.workspace_of(prev)
{
self.workspaces[wi].tabs[ti].set_focus(prev);
}
if let (Some(wi), Some((active, show_floating))) = (self.ws_index(ws), shown)
&& active < self.workspaces[wi].tabs.len()
{
let w = &mut self.workspaces[wi];
w.active = active;
if let Some(show) = show_floating {
w.tabs[active].show_floating = show && !w.tabs[active].floating.is_empty();
}
}
self.relayout();
}
self.mark_dirty();
Ok(Response::Pane(Box::new(self.pane_info(id))))
}
Err(e) => Err(format!("{e:#}")),
}
}
Request::ClosePane { pane } => self.resolve(&pane).map(|p| {
self.close_pane(p);
Response::Ok
}),
Request::FocusPane { pane } => self.resolve(&pane).map(|p| {
match self.primary {
Some(c) => self.reveal(c, p),
None => {
if let Some((wi, ti)) = self.workspace_of(p) {
self.workspaces[wi].active = ti;
self.workspaces[wi].tabs[ti].set_focus(p);
self.relayout();
}
}
}
Response::Ok
}),
Request::RenamePane { pane, name } => self.resolve(&pane).map(|p| {
if let Some(pn) = self.panes.get_mut(&p) {
pn.name = (!name.is_empty()).then_some(name);
}
self.mark_persist();
self.mark_dirty();
Response::Ok
}),
Request::SendText { pane, text, production } => self.resolve(&pane).and_then(|p| {
self.allow_input(p, production)?;
let pn = self.panes.get_mut(&p).ok_or("no such pane")?;
if pn.exit_code.is_some() {
return Err("the pane's program has exited".into());
}
pn.write_input(text.into_bytes());
Ok(Response::Ok)
}),
Request::SendKeys { pane, keys, production } => self.resolve(&pane).and_then(|p| {
self.allow_input(p, production)?;
let mut specs = Vec::new();
for k in &keys {
specs.push(KeySpec::parse(k).map_err(|e| format!("{k:?}: {e}"))?);
}
let pn = self.panes.get_mut(&p).ok_or("no such pane")?;
let modes = pn.input_modes();
let mut bytes = Vec::new();
for s in specs {
let key = crate::input::Key { code: s.code, mods: s.mods, raw: Vec::new() };
bytes.extend(crate::input::encode_key(&key, modes));
}
pn.write_input(bytes);
Ok(Response::Ok)
}),
Request::ReadPane { pane, lines } => {
self.resolve(&pane).map(|p| Response::Text(self.panes[&p].text(lines)))
}
Request::Wait { pane, until, timeout_ms } => match self.resolve(&pane) {
Ok(p) => {
let until = if until.is_empty() {
vec!["blocked".into(), "done".into(), "idle".into(), "exit".into()]
} else {
until
};
for u in &until {
if !matches!(u.as_str(), "exit" | "quiet") && AgentState::parse(u).is_none() {
return Some(Err(format!(
"unknown condition {u:?} (use working, blocked, done, idle, exit, quiet)"
)));
}
}
self.add_wait(client, p, until, timeout_ms.map(Duration::from_millis));
return None;
}
Err(e) => Err(e),
},
Request::AgentReport { pane, agent, state, session_id, message } => self.resolve(&pane).map(|p| {
let running = self.panes[&p].foreground_argv().as_deref().and_then(crate::agent::detect_agent);
if running.is_some_and(|r| r != agent) {
return Response::Ok;
}
let before = self.panes[&p].agent.state;
let pn = self.panes.get_mut(&p).unwrap();
let a = &mut pn.agent;
if state == Some(AgentState::Unknown) {
a.hooked = false;
a.kind = None;
a.set(AgentState::Unknown);
} else {
a.kind = Some(agent.clone());
a.hooked = crate::agent::hooks_cover_lifecycle(&agent);
if let Some(id) = session_id {
a.session_id = Some(id);
}
if let Some(s) = state {
a.set(s);
}
if message.is_some() {
a.message = message;
}
}
self.on_agent_change(p, before);
self.mark_persist();
self.mark_dirty();
Response::Ok
}),
Request::Notify { title, body } => {
self.notify(&title, &body, None);
Ok(Response::Ok)
}
Request::SaveWorkspace { workspace, name } => {
let wi = match workspace {
Some(w) => self.workspaces.iter().position(|x| x.name == w),
None => self
.primary
.and_then(|p| self.clients.get(&p))
.and_then(|c| self.ws_index(c.workspace))
.or(Some(0)),
};
match wi.filter(|i| *i < self.workspaces.len()) {
None => Err("no such workspace".into()),
Some(wi) => {
let name = name.unwrap_or_else(|| crate::workspace::file_name_from(&self.workspaces[wi].name));
self.save_definition(wi, &name).map(Response::Path).map_err(|e| format!("{e:#}"))
}
}
}
Request::OpenSsh { target, group, placement, near, command } => {
if group {
let group = target.strip_prefix('@').unwrap_or(&target);
return Some(self.open_group(group, &command).map(|_| Response::Ok).map_err(|e| format!("{e:#}")));
}
let near = match near.as_deref().map(|n| self.resolve(n)).transpose() {
Ok(n) => n,
Err(e) => return Some(Err(e)),
};
let Some(server) = self.resolve_server(&target) else {
return Some(Err(format!("not a server: {target}")));
};
let ws = match near.and_then(|n| self.workspace_of(n)) {
Some((wi, _)) => self.workspaces[wi].id,
None => match self.primary.and_then(|p| self.clients.get(&p)).map(|c| c.workspace) {
Some(ws) => ws,
None => match self.workspaces.first() {
Some(w) => w.id,
None => self.new_workspace("main".into(), None),
},
},
};
self.open_ssh_running(ws, near, &server, placement, &command)
.map(|p| Response::Pane(Box::new(self.pane_info(p))))
.map_err(|e| format!("{e:#}"))
}
Request::OpenAgent { agent, account, placement, near, cwd, args } => {
let near = match near.as_deref().map(|n| self.resolve(n)).transpose() {
Ok(n) => n,
Err(e) => return Some(Err(e)),
};
let ws = self.agent_workspace(near);
let dir = ws.and_then(|ws| self.agent_cwd(ws, near, cwd.clone()));
let undecided =
account.is_none() && dir.as_ref().is_some_and(|d| self.accounts.undecided(&agent, Some(d)));
let new_worktree = dir
.as_ref()
.is_some_and(|d| crate::worktree_setup::in_worktrees_dir(d) && !self.git.contains_key(d));
if let Some(dir) = dir.filter(|_| undecided || new_worktree) {
self.background(
move || {
crate::git::project_path(&dir);
let info = new_worktree.then(|| crate::git::status(&dir)).flatten();
(dir, info)
},
move |server, (dir, info)| {
if info.is_some() {
server.git_result(dir, info.map(Box::new));
}
let result = server.open_agent(near, &agent, account.as_deref(), placement, cwd, &args);
server.send_msg(client, &ServerMsg::Response { result });
server.clients.remove(&client);
server.drop_empty_workspaces();
},
);
return None;
}
self.open_agent(near, &agent, account.as_deref(), placement, cwd, &args)
}
Request::OpenFiles { left, right, near } => {
let near = match near.as_deref().map(|n| self.resolve(n)).transpose() {
Ok(n) => n,
Err(e) => return Some(Err(e)),
};
let ws = match near.and_then(|n| self.workspace_of(n)) {
Some((wi, _)) => self.workspaces[wi].id,
None => match self.primary.and_then(|p| self.clients.get(&p)).map(|c| c.workspace) {
Some(ws) => ws,
None => match self.workspaces.first() {
Some(w) => w.id,
None => self.new_workspace("main".into(), None),
},
},
};
self.open_files(ws, left, right)
.map(|p| Response::Pane(Box::new(self.pane_info(p))))
.map_err(|e| format!("{e:#}"))
}
Request::EditRemote { target, line, near, production } => {
let near = match near.as_deref().map(|n| self.resolve(n)).transpose() {
Ok(n) => n,
Err(e) => return Some(Err(e)),
};
let (server, path) = match crate::transfer::parse_endpoint(&target, &self.known_servers) {
Ok(crate::transfer::Endpoint::Remote { server, path }) => (server, path),
Ok(crate::transfer::Endpoint::Local(_)) => {
return Some(Err(format!("{target} is not on a server: kraal edit takes SERVER:PATH")));
}
Err(e) => return Some(Err(format!("{e:#}"))),
};
if path.is_empty() {
return Some(Err("which file? kraal edit takes SERVER:PATH".into()));
}
if !production && let Some(why) = self.server_production(&server) {
return Some(Err(format!(
"{} is production ({why}): pass --production to edit there",
server.name
)));
}
let ws = match near.and_then(|n| self.workspace_of(n)) {
Some((wi, _)) => Some(self.workspaces[wi].id),
None => self
.primary
.and_then(|p| self.clients.get(&p))
.map(|c| c.workspace)
.or_else(|| self.workspaces.first().map(|w| w.id)),
};
let shows = |c: &&super::Client| c.attached && Some(c.workspace) == ws;
let viewer = self.primary.filter(|p| self.clients.get(p).is_some_and(|c| shows(&c)));
let viewer = viewer.or_else(|| self.clients.values().find(shows).map(|c| c.id)).or(self.primary);
let mut ssh = self.ssh_options(&server);
ssh.push(server.destination());
let target = super::edits::EditTarget { server: server.name.clone(), ssh, path, line, ws, near };
self.start_edit(viewer, target, Some(client));
return None;
}
Request::TunnelAdd { server, remote, local } => self
.add_tunnel(&server, &remote, local)
.map(|t| Response::Tunnels(vec![t]))
.map_err(|e| format!("{e:#}")),
Request::TunnelRemove { key } => {
self.remove_tunnel(&key).map(|_| Response::Ok).map_err(|e| format!("{e:#}"))
}
Request::TunnelList => Ok(Response::Tunnels(self.tunnel_infos())),
Request::ReloadConfig => self.reload_config().map(|warnings| Response::Text(warnings.join("\n"))),
Request::KillServer => {
self.quitting = true;
self.kill_waiters.push(client);
return None;
}
})
}
fn agent_workspace(&self, near: Option<PaneId>) -> Option<u32> {
near.and_then(|n| self.workspace_of(n))
.map(|(wi, _)| self.workspaces[wi].id)
.or_else(|| self.primary.and_then(|p| self.clients.get(&p)).map(|c| c.workspace))
.or_else(|| self.workspaces.first().map(|w| w.id))
}
fn open_agent(
&mut self,
near: Option<PaneId>,
agent: &str,
account: Option<&str>,
placement: crate::protocol::Placement,
cwd: Option<std::path::PathBuf>,
args: &[String],
) -> Result<Response, String> {
let ws = match self.agent_workspace(near) {
Some(ws) => ws,
None => self.new_workspace("main".into(), None),
};
self.launch_agent_with(ws, near, agent, account, placement, cwd, args)
.map(|p| Response::Pane(Box::new(self.pane_info(p))))
.map_err(|e| format!("{e:#}"))
}
fn allow_input(&self, pane: PaneId, production: bool) -> Result<(), String> {
match self.pane_production(pane) {
Some(why) if !production => {
Err(format!("{} is production ({why}); add --production to send to it", self.pane_label(pane)))
}
_ => Ok(()),
}
}
pub(super) fn resolve(&self, r: &str) -> Result<PaneId, String> {
if let Ok(id) = r.parse::<PaneId>() {
return if self.panes.contains_key(&id) { Ok(id) } else { Err(format!("no pane with id {id}")) };
}
let (ws, name) = match r.split_once(':') {
Some((w, n)) => (Some(w), n),
None => (None, r),
};
let matches: Vec<PaneId> = self
.workspaces
.iter()
.filter(|w| ws.is_none_or(|n| w.name == n))
.flat_map(|w| w.panes())
.filter(|p| {
self.panes.get(p).is_some_and(|pn| pn.name.as_deref() == Some(name) || pn.display_name() == name)
})
.collect();
match matches.as_slice() {
[one] => Ok(*one),
[] => Err(format!("no pane named {r:?}")),
_ => Err(format!("{} panes are named {r:?}; use an id (kraal panes)", matches.len())),
}
}
fn workspace_infos(&self) -> Vec<WorkspaceInfo> {
let mut out = Vec::new();
for ws in &self.workspaces {
let panes = ws.panes();
let count =
|s: AgentState| panes.iter().filter(|p| self.panes.get(p).is_some_and(|x| x.agent.state == s)).count();
out.push(WorkspaceInfo {
name: ws.name.clone(),
open: true,
definition: ws.definition.clone(),
tabs: ws.tabs.len(),
panes: panes.len(),
working: count(AgentState::Working),
blocked: count(AgentState::Blocked),
done: count(AgentState::Done),
});
}
for (path, def) in self.definitions() {
let Ok(def) = def else { continue };
if self.workspaces.iter().any(|w| w.name == def.name) {
continue;
}
out.push(WorkspaceInfo {
name: def.name.clone(),
open: false,
definition: Some(path),
tabs: def.tabs.len(),
panes: def.panes().len(),
working: 0,
blocked: 0,
done: 0,
});
}
out
}
}
impl Server {
pub(super) fn add_wait(&mut self, client: ClientId, pane: PaneId, until: Vec<String>, timeout: Option<Duration>) {
let Some(c) = self.clients.remove(&client) else { return };
self.waits.push(PendingWait { client, out: c.out, pane, until, deadline: timeout.map(|t| Instant::now() + t) });
self.check_waits();
}
pub(super) fn check_waits(&mut self) {
let now = Instant::now();
let mut i = 0;
while i < self.waits.len() {
let w = &self.waits[i];
let result = match self.panes.get(&w.pane) {
None => Some(Err("pane closed".to_string())),
Some(p) => {
let state = p.agent.state;
let hit = w.until.iter().any(|u| match u.as_str() {
"exit" => p.exit_code.is_some(),
"quiet" => p.last_output.elapsed() > Duration::from_secs(2),
other => AgentState::parse(other) == Some(state),
});
if hit {
Some(Ok(protocol::Response::Pane(Box::new(self.pane_info(w.pane)))))
} else if w.deadline.is_some_and(|d| d <= now) {
Some(Err("timeout".to_string()))
} else {
None
}
}
};
match result {
Some(result) => {
let w = self.waits.remove(i);
let mut buf = Vec::new();
if protocol::write_json(&mut buf, &ServerMsg::Response { result }).is_ok() {
let _ = w.out.send_timeout(buf, Duration::from_millis(200));
}
}
None => i += 1,
}
}
}
}