use anyhow::{Context, Result, bail};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::path::{Path, PathBuf};
use std::time::Duration;
pub const DEFAULT_HOST: &str = "https://cloud.semlith.com";
pub const TOKEN_PREFIX: &str = "sml_live_";
pub const SHOWN_CHARS: usize = 13;
const MAX_REPLY: u64 = 64 << 20;
const CALL_TIMEOUT: Duration = Duration::from_secs(30);
pub fn credentials_path() -> Result<PathBuf> {
Ok(crate::home::home_or_error()?.join("cloud.json"))
}
pub fn signed_in() -> bool {
credentials_path().is_ok_and(|p| p.exists())
}
#[derive(Default, Serialize, Deserialize)]
pub struct Credentials {
#[serde(default)]
pub machine: String,
#[serde(default)]
pub entries: Vec<Entry>,
}
#[derive(Clone, Serialize, Deserialize)]
pub struct Entry {
pub host: String,
pub org: String,
pub token: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub plan: Option<String>,
#[serde(default)]
pub added: u64,
}
impl std::fmt::Debug for Entry {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Entry")
.field("host", &self.host)
.field("org", &self.org)
.field("token", &self.prefix())
.finish()
}
}
impl Entry {
pub fn prefix(&self) -> String {
self.token.chars().take(SHOWN_CHARS).collect()
}
pub fn host_name(&self) -> &str {
host_name(&self.host)
}
}
pub fn host_name(host: &str) -> &str {
host.split_once("://").map_or(host, |(_, rest)| rest)
}
impl Credentials {
pub fn load() -> Result<Option<Self>> {
let path = credentials_path()?;
let text = match std::fs::read_to_string(&path) {
Ok(t) => t,
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(e).with_context(|| format!("reading {}", path.display())),
};
crate::home::tighten_file(&path);
serde_json::from_str(&text).map(Some).with_context(|| {
format!(
"{} is not readable; move it aside and sign in again",
path.display()
)
})
}
pub fn save(&self) -> Result<()> {
let path = credentials_path()?;
let dir = path.parent().unwrap_or(Path::new("."));
crate::home::secure_dir(dir)?;
let temp = path.with_extension(crate::home::unique_temp_suffix());
let body = serde_json::to_string_pretty(self)? + "\n";
crate::home::write_private(&temp, body.as_bytes())
.with_context(|| format!("writing {}", temp.display()))?;
if let Err(e) = std::fs::rename(&temp, &path) {
let _ = std::fs::remove_file(&temp);
return Err(e).with_context(|| format!("writing {}", path.display()));
}
Ok(())
}
pub fn put(&mut self, entry: Entry) {
self.entries
.retain(|e| !(e.host == entry.host && e.org == entry.org));
self.entries.push(entry);
self.entries
.sort_by(|a, b| (&a.org, &a.host).cmp(&(&b.org, &b.host)));
if self.machine.is_empty() {
self.machine = random_hex(16);
}
}
pub fn find(&self, org: Option<&str>, host: Option<&str>) -> Result<&Entry> {
let host = host.map(normal_host).transpose()?;
let matching: Vec<&Entry> = self
.entries
.iter()
.filter(|e| org.is_none_or(|o| e.org == o))
.filter(|e| host.as_ref().is_none_or(|h| &e.host == h))
.collect();
match matching.as_slice() {
[one] => Ok(one),
[] => match org {
Some(org) => bail!(
"this machine is not signed in to {org}. Run `semlith cloud login {org}`."
),
None => bail!(
"this machine is not signed in to Semlith Cloud. Run `semlith cloud login <org>`."
),
},
many => bail!(
"this machine is signed in to {}; name one: {}",
many.len(),
many.iter()
.map(|e| format!("{} at {}", e.org, e.host_name()))
.collect::<Vec<_>>()
.join(", ")
),
}
}
pub fn machine_id(&self) -> String {
let hash = blake3::hash(self.machine.as_bytes()).to_hex();
format!("m_{}", &hash[..16])
}
}
pub fn entry_for(org: Option<&str>, host: Option<&str>) -> Result<Entry> {
let Some(creds) = Credentials::load()? else {
bail!("this machine is not signed in to Semlith Cloud. Run `semlith cloud login <org>`.");
};
creds.find(org, host).cloned()
}
fn random_hex(bytes: usize) -> String {
let mut raw = vec![0u8; bytes];
getrandom::fill(&mut raw).expect("the OS random source");
raw.iter().map(|b| format!("{b:02x}")).collect()
}
fn unix_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0)
}
pub fn normal_host(raw: &str) -> Result<String> {
let host = raw.trim().trim_end_matches('/');
let Some((scheme, rest)) = host.split_once("://") else {
bail!("{raw} is not a URL; a host looks like https://cloud.semlith.com");
};
if rest.is_empty() || rest.contains(['/', '?', '#', '@']) {
bail!("{raw} is not a host; give it without a path, like https://cloud.semlith.com");
}
match scheme {
"https" => {}
"http" if loopback(rest) => {}
_ => bail!("{raw} is not https; semlith sends a token to an https host only"),
}
Ok(format!("{scheme}://{rest}"))
}
fn loopback(authority: &str) -> bool {
let name = authority
.rsplit_once(':')
.filter(|(_, port)| port.chars().all(|c| c.is_ascii_digit()))
.map_or(authority, |(name, _)| name);
matches!(name, "localhost" | "127.0.0.1" | "[::1]")
}
pub fn user_agent() -> String {
format!(
"semlith/{} ({}; {})",
env!("CARGO_PKG_VERSION"),
std::env::consts::OS,
std::env::consts::ARCH
)
}
#[derive(Debug, Clone)]
pub enum Failure {
Offline,
Unreachable { host: String, why: String },
Refused {
status: u16,
code: String,
message: String,
plan: Option<String>,
min_version: Option<String>,
},
}
impl Failure {
pub fn kind(&self) -> &'static str {
match self {
Failure::Refused { .. } => "refused",
_ => "unreachable",
}
}
pub fn code(&self) -> Option<&str> {
match self {
Failure::Refused { code, .. } => Some(code),
_ => None,
}
}
pub fn status(&self) -> Option<u16> {
match self {
Failure::Refused { status, .. } => Some(*status),
_ => None,
}
}
}
impl std::fmt::Display for Failure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Failure::Offline => write!(
f,
"airgap is on (SEMLITH_AIRGAP, --airgap or the Privacy page), so semlith will not reach Semlith Cloud"
),
Failure::Unreachable { host, why } => {
write!(f, "{} could not be reached: {why}", host_name(host))
}
Failure::Refused {
status,
message,
plan,
min_version,
..
} => {
write!(f, "{}", message.trim())?;
if *status == 426 {
match min_version {
Some(v) => write!(
f,
" This semlith is {}; the cloud needs {v} or newer. Run `semlith upgrade`.",
env!("CARGO_PKG_VERSION")
)?,
None => write!(f, " Run `semlith upgrade`.")?,
}
}
if *status == 402
&& let Some(plan) = plan
{
write!(f, " It needs the {} plan.", title(plan))?;
}
Ok(())
}
}
}
}
impl std::error::Error for Failure {}
fn title(plan: &str) -> String {
let mut chars = plan.chars();
match chars.next() {
Some(c) => c.to_uppercase().chain(chars).collect(),
None => String::new(),
}
}
pub struct Reply {
pub status: u16,
pub body: Vec<u8>,
pub version: Option<String>,
}
impl Reply {
pub fn json(&self) -> Result<Value, Failure> {
serde_json::from_slice(&self.body).map_err(|e| Failure::Refused {
status: self.status,
code: "bad_reply".into(),
message: format!("The cloud answered with something that is not JSON ({e})."),
plan: None,
min_version: None,
})
}
}
#[derive(Default)]
pub struct Send<'a> {
pub json: Option<&'a Value>,
pub bytes: Option<(&'a str, Option<&'a str>, Vec<u8>)>,
pub headers: Vec<(&'static str, String)>,
pub timeout: Option<Duration>,
}
pub fn call(
host: &str,
token: Option<&str>,
method: &str,
path: &str,
send: Send<'_>,
) -> Result<Reply, Failure> {
if crate::embed::airgap() {
return Err(Failure::Offline);
}
let url = format!("{host}{path}");
crate::add::note_outbound("cloud", &url);
let agent: ureq::Agent = ureq::Agent::config_builder()
.http_status_as_error(false)
.max_redirects(0)
.max_redirects_will_error(false)
.timeout_connect(Some(Duration::from_secs(10)))
.timeout_global(Some(send.timeout.unwrap_or(CALL_TIMEOUT)))
.user_agent(user_agent())
.build()
.new_agent();
let unreachable = |e: ureq::Error| Failure::Unreachable {
host: host.to_string(),
why: e.to_string(),
};
let auth = token.map(|t| format!("Bearer {t}"));
macro_rules! headed {
($req:expr) => {{
let mut req = $req;
if !send
.headers
.iter()
.any(|(k, _)| k.eq_ignore_ascii_case("accept"))
{
req = req.header("Accept", "application/json");
}
if let Some(auth) = &auth {
req = req.header("Authorization", auth);
}
for (k, v) in &send.headers {
req = req.header(*k, v);
}
req
}};
}
let response = match method {
"GET" => headed!(agent.get(&url)).call(),
"DELETE" => headed!(agent.delete(&url)).call(),
_ => {
let builder = match method {
"PUT" => agent.put(&url),
_ => agent.post(&url),
};
let req = headed!(builder);
if let Some(body) = send.json {
req.header("Content-Type", "application/json")
.send(serde_json::to_vec(body).unwrap_or_default())
} else if let Some((kind, encoding, bytes)) = send.bytes {
let mut req = req.header("Content-Type", kind);
if let Some(encoding) = encoding {
req = req.header("Content-Encoding", encoding);
}
req.send(bytes)
} else {
req.send_empty()
}
}
};
let mut response = response.map_err(unreachable)?;
let status = response.status().as_u16();
let version = response
.headers()
.get("semlith-cloud-version")
.and_then(|v| v.to_str().ok())
.map(str::to_string);
let mut body = Vec::new();
use std::io::Read;
response
.body_mut()
.as_reader()
.take(MAX_REPLY)
.read_to_end(&mut body)
.map_err(|e| Failure::Unreachable {
host: host.to_string(),
why: format!("the reply broke off: {e}"),
})?;
if status >= 400 {
return Err(refusal(status, &body));
}
Ok(Reply {
status,
body,
version,
})
}
fn refusal(status: u16, body: &[u8]) -> Failure {
let parsed: Value = serde_json::from_slice(body).unwrap_or(Value::Null);
let error = &parsed["error"];
let text = |k: &str| error.get(k).and_then(Value::as_str).map(printable);
Failure::Refused {
status,
code: text("code").unwrap_or_else(|| format!("http_{status}")),
message: text("message").unwrap_or_else(|| match status {
401 => "The cloud did not accept this machine's token. Sign in again with `semlith cloud login`.".into(),
404 => "The cloud has no such org or store, or this token cannot reach it.".into(),
_ => format!("The cloud answered {status}."),
}),
plan: text("plan"),
min_version: text("min_version"),
}
}
pub fn call_as(entry: &Entry, method: &str, path: &str, send: Send<'_>) -> Result<Reply, Failure> {
call(&entry.host, Some(&entry.token), method, path, send)
}
pub fn seg(s: &str) -> String {
s.bytes()
.map(|b| {
if b.is_ascii_alphanumeric() || b"-_.~".contains(&b) {
(b as char).to_string()
} else {
format!("%{b:02X}")
}
})
.collect()
}
fn machine_name() -> String {
std::process::Command::new("hostname")
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.filter(|n| !n.is_empty())
.unwrap_or_else(|| "this machine".to_string())
}
fn printable(text: &str) -> String {
text.chars().filter(|c| !c.is_control()).collect()
}
fn openable<'a>(host: &str, url: &'a str) -> Option<&'a str> {
let safe = url
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b"-._~:/?=#[]".contains(&b));
(safe && url.starts_with(&format!("{host}/"))).then_some(url)
}
fn open_browser(url: &str) {
use std::io::IsTerminal;
let (program, args): (&str, &[&str]) = if cfg!(target_os = "macos") {
("open", &[])
} else if cfg!(windows) {
("rundll32", &["url.dll,FileProtocolHandler"])
} else {
("xdg-open", &[])
};
#[cfg(debug_assertions)]
if let Some(log) = std::env::var_os("SEMLITH_TEST_OPENED") {
let _ = std::fs::write(log, format!("{program} {} {url}\n", args.join(" ")));
return;
}
if !std::io::stdout().is_terminal() {
return;
}
let _ = std::process::Command::new(program)
.args(args)
.arg(url)
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.spawn();
}
pub fn login(
org: Option<&str>,
host: Option<&str>,
token: Option<&str>,
say: &mut dyn FnMut(&str),
) -> Result<Entry> {
let host = normal_host(host.unwrap_or(DEFAULT_HOST))?;
let (org, token, plan) = match token {
Some(token) => {
let token = token.trim();
if !token.starts_with(TOKEN_PREFIX) {
bail!(
"that is not a Semlith Cloud org token; one starts with {TOKEN_PREFIX} and is made on the Agents page"
);
}
let who = call(&host, Some(token), "GET", "/v1/whoami", Send::default())?.json()?;
let slug = printable(who["org"]["slug"].as_str().unwrap_or_default());
if slug.is_empty() {
bail!(
"{} did not say which org this token belongs to",
host_name(&host)
);
}
if let Some(org) = org
&& org != slug
{
bail!("that token belongs to {slug}, not {org}; nothing was saved");
}
let plan = who["org"]["plan"].as_str().map(str::to_string);
(slug, token.to_string(), plan)
}
None => {
let (slug, token) = device_flow(&host, org, say)?;
(slug, token, None)
}
};
let entry = Entry {
host,
org,
token,
plan,
added: unix_now(),
};
let mut creds = Credentials::load()?.unwrap_or_default();
creds.put(entry.clone());
creds.save()?;
Ok(entry)
}
fn device_flow(
host: &str,
org: Option<&str>,
say: &mut dyn FnMut(&str),
) -> Result<(String, String)> {
let asked = call(
host,
None,
"POST",
"/v1/cli/authorize",
Send {
json: Some(&json!({
"client": format!("semlith/{}", env!("CARGO_PKG_VERSION")),
"host_name": machine_name(),
})),
..Send::default()
},
)?
.json()?;
let device = asked["device_code"]
.as_str()
.unwrap_or_default()
.to_string();
let code = printable(asked["user_code"].as_str().unwrap_or_default());
let code = code.as_str();
let page = printable(asked["verification_uri"].as_str().unwrap_or_default());
let complete = printable(asked["verification_uri_complete"].as_str().unwrap_or(&page));
if device.is_empty() || code.is_empty() {
bail!("{} did not hand out a device code", host_name(host));
}
let mut interval = asked["interval"].as_u64().unwrap_or(5);
let expires = asked["expires_in"].as_u64().unwrap_or(600);
say(&format!(
"Open {page} and enter the code {code}{}.",
org.map(|o| format!(", then approve {o}"))
.unwrap_or_default()
));
match openable(host, &complete) {
Some(url) => open_browser(url),
None => say(&format!(
"Not opening {complete} in the browser: it is not a page on {}.",
host_name(host)
)),
}
let deadline = std::time::Instant::now() + Duration::from_secs(expires);
loop {
std::thread::sleep(Duration::from_secs(interval));
if std::time::Instant::now() > deadline {
bail!(
"the code {code} expired before it was approved; run `semlith cloud login` again"
);
}
let polled = call(
host,
None,
"POST",
"/v1/cli/token",
Send {
json: Some(&json!({ "device_code": device })),
..Send::default()
},
);
match polled {
Ok(reply) => {
let got = reply.json()?;
let token = got["token"].as_str().unwrap_or_default().to_string();
let slug = printable(got["org"].as_str().unwrap_or_default());
if !token.starts_with(TOKEN_PREFIX) || slug.is_empty() {
bail!("{} approved the code but sent no token", host_name(host));
}
if let Some(org) = org
&& org != slug
{
say(&format!(
"Approved for {slug}, not {org}; saved for {slug}."
));
}
return Ok((slug, token));
}
Err(Failure::Refused { code, .. }) if code == "authorization_pending" => {}
Err(Failure::Refused { code, .. }) if code == "slow_down" => interval += 5,
Err(Failure::Refused { code, .. }) if code == "access_denied" => {
bail!("the sign-in was denied in the browser; nothing was saved")
}
Err(Failure::Refused { status: 410, .. }) => {
bail!(
"the code {code} expired before it was approved; run `semlith cloud login` again"
)
}
Err(other) => return Err(other.into()),
}
}
}
pub fn logout(org: Option<&str>, host: Option<&str>) -> Result<Entry> {
let mut creds = Credentials::load()?.unwrap_or_default();
let gone = creds.find(org, host)?.clone();
creds
.entries
.retain(|e| !(e.host == gone.host && e.org == gone.org));
if creds.entries.is_empty() {
let path = credentials_path()?;
std::fs::remove_file(&path).with_context(|| format!("removing {}", path.display()))?;
} else {
creds.save()?;
}
Ok(gone)
}
pub fn status(entry: &Entry) -> Result<(Value, Option<String>), Failure> {
let reply = call_as(
entry,
"GET",
&format!("/v1/orgs/{}/status", seg(&entry.org)),
Send {
timeout: Some(FORWARD_TIMEOUT),
..Send::default()
},
)?;
Ok((reply.json()?, reply.version))
}
pub fn lag(seconds: Option<i64>) -> String {
match seconds {
None => "no lag known".to_string(),
Some(s) if s < 120 => format!("{s} s behind"),
Some(s) if s < 7200 => format!("{} min behind", s / 60),
Some(s) => format!("{} h behind", s / 3600),
}
}
pub fn render_status(entry: &Entry, status: &Value, version: Option<&str>) -> String {
let org = &status["org"];
let s = |v: &Value| v.as_str().unwrap_or("—").to_string();
let mut out = String::new();
let seats = match (org["seats_used"].as_i64(), org["seats"].as_i64()) {
(Some(used), Some(of)) => format!(" · {used} of {of} seats"),
(None, Some(of)) => format!(" · {of} seats"),
_ => String::new(),
};
let trial = match org["trial_ends_at"].as_str() {
Some(at) if org["trial"].as_bool() == Some(true) => {
format!(" · trial ends {}", &at[..at.len().min(10)])
}
_ => String::new(),
};
out.push_str(&format!(
"{} · {} · {} ({}){seats}{trial}\n",
s(&org["slug"]),
s(&org["name"]),
s(&org["plan"]),
s(&org["status"]),
));
out.push_str(&format!("host {}\n", entry.host));
out.push_str(&format!("token {}…\n", entry.prefix()));
if let Some(app) = org["app_url"].as_str() {
out.push_str(&format!("app {app}\n"));
}
if let Some(mcp) = status["mcp_url"].as_str() {
out.push_str(&format!("mcp {mcp}\n"));
}
if let Some(v) = version {
out.push_str(&format!("cloud {v}\n"));
}
let stores = status["stores"].as_array().cloned().unwrap_or_default();
out.push_str(&format!("stores {}\n", stores.len()));
for store in &stores {
let name = s(&store["name"]);
out.push_str(&format!(
" {name} {} · {} files · {} chunks",
s(&store["state"]),
store["files"].as_i64().unwrap_or(0),
store["chunks"].as_i64().unwrap_or(0),
));
if let Some(mcp) = status["mcp_url"].as_str() {
out.push_str(&format!(" · reach {mcp}?store={name}"));
}
out.push('\n');
for source in store["sources"].as_array().into_iter().flatten() {
out.push_str(&format!(
" {} {} {} {} {}\n",
s(&source["label"]),
s(&source["kind"]),
s(&source["revision"]),
lag(source["behind_seconds"].as_i64()),
s(&source["state"]),
));
}
}
let usage = &status["usage"];
if usage.is_object() {
let n = |k: &str| usage[k].as_i64().unwrap_or(0);
let indexed = if usage.get("index_chunks").is_some() || usage.get("cap_chunks").is_some() {
format!(
"{} of {} chunks indexed",
grouped(n("index_chunks")),
grouped(n("cap_chunks"))
)
} else {
format!(
"{} of {} index minutes",
n("index_minutes"),
n("cap_minutes")
)
};
out.push_str(&format!(
"usage {} · {} of {} stored · {indexed} · queue {} running, {} waiting\n",
s(&usage["month"]),
crate::human_bytes(n("store_bytes")),
crate::human_bytes(n("cap_bytes")),
usage["queue"]["running"].as_i64().unwrap_or(0),
usage["queue"]["waiting"].as_i64().unwrap_or(0),
));
}
out.trim_end().to_string()
}
fn grouped(n: i64) -> String {
let digits = n.unsigned_abs().to_string();
let mut out = String::new();
for (i, d) in digits.chars().enumerate() {
if i > 0 && (digits.len() - i).is_multiple_of(3) {
out.push(',');
}
out.push(d);
}
if n < 0 { format!("-{out}") } else { out }
}
pub fn mcp_url_for(entry: &Entry, reported: Option<&str>) -> String {
match reported {
Some(url) if url.starts_with(&format!("{}/", entry.host)) => url.to_string(),
_ => format!("{}/{}/mcp", entry.host, seg(&entry.org)),
}
}
pub fn connect(entry: &Entry, only: &[String]) -> Result<Vec<String>> {
let (status, _) = status(entry)?;
let listed: Vec<String> = status["stores"]
.as_array()
.into_iter()
.flatten()
.filter_map(|s| s["name"].as_str().map(str::to_string))
.collect();
for name in only {
if !listed.contains(name) {
bail!(
"{} has no store called {name} that this token can reach; it can reach: {}",
entry.org,
if listed.is_empty() {
"none".to_string()
} else {
listed.join(", ")
}
);
}
}
let mcp_url = mcp_url_for(entry, status["mcp_url"].as_str());
let mut registry = crate::home::Registry::load()?;
let mut written = Vec::new();
for store in listed
.iter()
.filter(|s| only.is_empty() || only.contains(s))
{
let name = format!("{}/{store}", entry.org);
registry.remote.insert(
name.clone(),
crate::home::Remote {
host: entry.host.clone(),
org: entry.org.clone(),
store: store.clone(),
mcp_url: mcp_url.clone(),
},
);
written.push(name);
}
registry.save()?;
Ok(written)
}
pub fn disconnect(org: &str) -> Result<Vec<String>> {
let mut registry = crate::home::Registry::load()?;
let gone: Vec<String> = registry
.remote
.iter()
.filter(|(_, r)| r.org == org)
.map(|(name, _)| name.clone())
.collect();
if gone.is_empty() {
return Ok(gone);
}
for name in &gone {
registry.remote.remove(name);
}
registry.save()?;
Ok(gone)
}
pub fn remote_stores() -> Vec<Named> {
crate::home::Registry::load()
.map(|r| r.remote.into_iter().collect())
.unwrap_or_default()
}
pub type Named = (String, crate::home::Remote);
pub fn badge(remote: &crate::home::Remote) -> String {
format!("remote · {}", remote.org)
}
pub fn remote_rows() -> Vec<Value> {
remote_stores()
.into_iter()
.map(|(name, r)| {
json!({
"name": name,
"org": r.org,
"store": r.store,
"host": r.host,
"host_name": host_name(&r.host),
"mcp_url": r.mcp_url,
"badge": badge(&r),
})
})
.collect()
}
const FORWARD_TIMEOUT: Duration = Duration::from_secs(15);
pub struct Who<'a> {
pub client: &'a str,
pub session: &'a str,
}
impl Who<'_> {
fn headers(&self) -> Vec<(&'static str, String)> {
vec![
("Semlith-Client", header_safe(self.client)),
("Semlith-Session", header_safe(self.session)),
]
}
}
fn header_safe(value: &str) -> String {
value
.chars()
.filter(|c| c.is_ascii_graphic() || *c == ' ')
.take(128)
.collect()
}
#[derive(Debug, Clone, Deserialize)]
pub struct RemoteHit {
pub score: f32,
#[serde(default)]
pub store: String,
pub path: String,
#[serde(default)]
pub start_line: u32,
#[serde(default)]
pub end_line: u32,
#[serde(default)]
pub text: String,
#[serde(default)]
pub lists: Vec<String>,
#[serde(default)]
pub symbol: Option<String>,
#[serde(default)]
pub symbol_kind: Option<String>,
#[serde(default = "yes")]
pub fresh: bool,
#[serde(default)]
pub source: Option<String>,
#[serde(default)]
pub revision: Option<String>,
#[serde(default)]
pub behind_seconds: Option<i64>,
#[serde(skip)]
pub name: String,
#[serde(skip)]
pub org: String,
}
fn yes() -> bool {
true
}
impl RemoteHit {
pub fn provenance(&self) -> String {
let mut out = format!("remote · {}", self.org);
if let Some(rev) = &self.revision {
out.push_str(&format!(" · {rev}"));
}
if self.behind_seconds.is_some() {
out.push_str(&format!(" · {}", lag(self.behind_seconds)));
}
out
}
pub fn to_hit(&self) -> crate::Hit {
crate::Hit {
score: self.score,
path: self.path.clone(),
start_line: self.start_line,
end_line: self.end_line,
text: self.text.clone(),
store: Some(format!("{} ({})", self.name, self.provenance())),
lists: self
.lists
.iter()
.filter_map(|l| match l.as_str() {
"vector" => Some("vector"),
"keyword" => Some("keyword"),
"image" => Some("image"),
"definition" => Some("definition"),
"graph" => Some("graph"),
_ => None,
})
.collect(),
image: None,
fresh: self.fresh,
symbol: self.symbol.clone(),
symbol_kind: self.symbol_kind.clone(),
symbol_line: None,
copies: Vec::new(),
}
}
pub fn to_json(&self, query: &str) -> Value {
json!({
"line": crate::mcp::line_for(&self.text, query),
"score": self.score,
"path": self.path,
"start_line": self.start_line,
"end_line": self.end_line,
"text": self.text,
"store": self.name,
"lists": self.lists,
"image": null,
"fresh": self.fresh,
"symbol": self.symbol,
"symbol_kind": self.symbol_kind,
"remote": self.org,
"badge": format!("remote · {}", self.org),
"source": self.source,
"revision": self.revision,
"behind_seconds": self.behind_seconds,
})
}
}
#[derive(Default)]
pub struct Fetched {
pub hits: Vec<RemoteHit>,
pub skipped: Vec<String>,
pub skipped_names: Vec<String>,
}
pub struct Query<'a> {
pub query: &'a str,
pub k: usize,
pub path: &'a [String],
pub ext: &'a [String],
pub lang: &'a [String],
pub prefer: Option<&'a str>,
}
pub fn skipped_line(skipped: &[String]) -> Option<String> {
(!skipped.is_empty()).then(|| format!("remote stores skipped: {}", skipped.join("; ")))
}
fn by_org(targets: &[Named]) -> Vec<(Result<Entry, String>, Vec<&Named>)> {
let creds = Credentials::load().ok().flatten();
let mut groups: Vec<((String, String), Vec<&Named>)> = Vec::new();
for target in targets {
let key = (target.1.host.clone(), target.1.org.clone());
match groups.iter_mut().find(|(k, _)| *k == key) {
Some((_, list)) => list.push(target),
None => groups.push((key, vec![target])),
}
}
groups
.into_iter()
.map(|((host, org), list)| {
let entry = creds
.as_ref()
.and_then(|c| c.entries.iter().find(|e| e.host == host && e.org == org))
.cloned()
.ok_or_else(|| format!("not signed in to {org}; run `semlith cloud login {org}`"));
(entry, list)
})
.collect()
}
fn names(list: &[&Named]) -> String {
list.iter()
.map(|(n, _)| n.as_str())
.collect::<Vec<_>>()
.join(", ")
}
fn nonempty(v: &[String]) -> Value {
if v.is_empty() { Value::Null } else { json!(v) }
}
pub fn search(targets: &[Named], q: &Query<'_>, who: &Who<'_>) -> Fetched {
let mut out = Fetched::default();
for (entry, list) in by_org(targets) {
let entry = match entry {
Ok(e) => e,
Err(why) => {
out.skipped.push(format!("{} ({why})", names(&list)));
out.skipped_names
.extend(list.iter().map(|(n, _)| n.clone()));
continue;
}
};
let body = json!({
"query": q.query,
"k": q.k,
"stores": list.iter().map(|(_, r)| r.store.as_str()).collect::<Vec<_>>(),
"path": nonempty(q.path),
"ext": nonempty(q.ext),
"lang": nonempty(q.lang),
"prefer": q.prefer,
"exact": false,
});
let reply = call_as(
&entry,
"POST",
&format!("/v1/orgs/{}/search", seg(&entry.org)),
Send {
json: Some(&body),
headers: who.headers(),
timeout: Some(FORWARD_TIMEOUT),
..Send::default()
},
)
.and_then(|r| r.json());
match reply {
Ok(answer) => {
for raw in answer["hits"].as_array().into_iter().flatten() {
let Ok(mut hit) = serde_json::from_value::<RemoteHit>(raw.clone()) else {
continue;
};
let Some((name, _)) = list.iter().find(|(_, r)| {
r.store == hit.store || (hit.store.is_empty() && list.len() == 1)
}) else {
continue;
};
hit.name = name.clone();
hit.org = entry.org.clone();
out.hits.push(hit);
}
}
Err(failure) => {
out.skipped.push(format!("{} ({failure})", names(&list)));
out.skipped_names
.extend(list.iter().map(|(n, _)| n.clone()));
}
}
}
out
}
pub fn merge(
local: Vec<crate::Hit>,
local_label: Option<&str>,
remote: &[RemoteHit],
k: usize,
) -> Vec<crate::Hit> {
let mut all: Vec<crate::Hit> = local
.into_iter()
.map(|mut h| {
if h.store.is_none() {
h.store = local_label.map(str::to_string);
}
h
})
.collect();
all.extend(remote.iter().map(RemoteHit::to_hit));
all.sort_by(|a, b| b.score.total_cmp(&a.score));
all.truncate(k);
all
}
pub fn merge_rows(local: Vec<Value>, remote: &[RemoteHit], query: &str, k: usize) -> Vec<Value> {
let mut all = local;
all.extend(remote.iter().map(|h| h.to_json(query)));
let score = |v: &Value| v["score"].as_f64().unwrap_or(0.0);
all.sort_by(|a, b| score(b).total_cmp(&score(a)));
all.truncate(k);
all
}
pub fn forward_tool(
name: &str,
remote: &crate::home::Remote,
tool: &str,
args: &Value,
who: &Who<'_>,
) -> Result<String, String> {
let creds = Credentials::load().ok().flatten();
let Some(entry) = creds.as_ref().and_then(|c| {
c.entries
.iter()
.find(|e| e.host == remote.host && e.org == remote.org)
}) else {
return Err(format!(
"{name} (not signed in to {}; run `semlith cloud login {}`)",
remote.org, remote.org
));
};
let base = mcp_url_for(entry, Some(&remote.mcp_url));
let path = base[entry.host.len()..].to_string();
let mut arguments = args.clone();
if let Some(map) = arguments.as_object_mut() {
map.insert("store".into(), json!(remote.store));
}
let body = json!({
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": { "name": tool, "arguments": arguments },
});
let mut headers = who.headers();
headers.push(("Accept", "application/json, text/event-stream".into()));
headers.push(("MCP-Protocol-Version", "2025-06-18".into()));
let reply = call_as(
entry,
"POST",
&format!("{path}?store={}", seg(&remote.store)),
Send {
json: Some(&body),
headers,
timeout: Some(FORWARD_TIMEOUT),
..Send::default()
},
)
.map_err(|f| format!("{name} ({f})"))?;
let answer = rpc_answer(&reply.body)
.ok_or_else(|| format!("{name} (the cloud's answer was not an MCP reply)"))?;
if let Some(message) = answer["error"]["message"].as_str() {
return Err(format!("{name} (refused: {message})"));
}
let text: Vec<&str> = answer["result"]["content"]
.as_array()
.into_iter()
.flatten()
.filter_map(|c| c["text"].as_str())
.collect();
Ok(text.join("\n"))
}
fn rpc_answer(body: &[u8]) -> Option<Value> {
if let Ok(v) = serde_json::from_slice::<Value>(body) {
return Some(v);
}
String::from_utf8_lossy(body)
.lines()
.filter_map(|l| l.strip_prefix("data:"))
.filter_map(|d| serde_json::from_str::<Value>(d.trim()).ok())
.rfind(|v| v.get("result").is_some() || v.get("error").is_some())
}
pub const PUSH_FILE_CAP: u64 = 1 << 20;
const UPLOAD_BATCH: u64 = 96 << 20;
const UPLOAD_TIMEOUT: Duration = Duration::from_secs(600);
const JOB_POLL: Duration = Duration::from_secs(2);
impl PushReport {
fn estimate_from(&mut self, opened: &Value) {
self.estimate_chunks = opened["estimate_chunks"].as_i64();
self.estimate_minutes = opened["estimate_minutes"].as_i64();
}
pub fn estimate_line(&self) -> Option<String> {
match (self.estimate_chunks, self.estimate_minutes) {
(Some(1), _) => Some("about 1 chunk to index".into()),
(Some(n), _) => Some(format!("about {} chunks to index", grouped(n))),
(None, Some(m)) => Some(format!(
"about {m} index minute{}",
if m == 1 { "" } else { "s" }
)),
(None, None) => None,
}
}
}
#[derive(Debug, Default, Serialize)]
pub struct PushReport {
pub files: usize,
pub sent: usize,
pub bytes_sent: u64,
pub removed: usize,
pub refused: Vec<(String, String)>,
pub push: String,
pub job: Option<i64>,
pub position: Option<i64>,
pub estimate_chunks: Option<i64>,
pub estimate_minutes: Option<i64>,
pub state: Option<String>,
}
pub fn split_target(target: &str) -> Result<(&str, &str)> {
match target.split_once('/') {
Some((org, store)) if !org.is_empty() && !store.is_empty() && !store.contains('/') => {
Ok((org, store))
}
_ => bail!("name the store as <org>/<store>, for example acme/platform"),
}
}
fn sha256_hex(bytes: &[u8]) -> String {
use sha2::Digest;
sha2::Sha256::digest(bytes)
.iter()
.map(|b| format!("{b:02x}"))
.collect()
}
pub fn push(
target: &str,
dir: &Path,
wait: bool,
prune: bool,
say: &mut dyn FnMut(&str),
) -> Result<PushReport> {
let (org, store) = split_target(target)?;
let entry = entry_for(Some(org), None)?;
if !dir.is_dir() {
bail!("{} is not a directory", dir.display());
}
let root = crate::canonical(dir);
let (files, refused) = crate::push_files(&root, PUSH_FILE_CAP);
let rel = |p: &Path| -> String {
p.strip_prefix(&root)
.unwrap_or(p)
.components()
.map(|c| c.as_os_str().to_string_lossy().into_owned())
.collect::<Vec<_>>()
.join("/")
};
let mut report = PushReport {
refused: refused
.iter()
.map(|(p, why)| (crate::plain(&rel(p)), why.clone()))
.collect(),
..PushReport::default()
};
let mut manifest = Vec::with_capacity(files.len());
let mut by_path: std::collections::HashMap<String, (PathBuf, String)> = Default::default();
for path in &files {
let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?;
let name = rel(path);
let hash = sha256_hex(&bytes);
manifest.push(json!({ "path": name, "sha256": hash, "bytes": bytes.len() }));
by_path.insert(name, (path.clone(), hash));
}
report.files = manifest.len();
say(&format!(
"{} file{} to {target} from {}{}",
manifest.len(),
if manifest.len() == 1 { "" } else { "s" },
root.display(),
if refused.is_empty() {
String::new()
} else {
format!(" · {} kept back by semlith's own rules", refused.len())
}
));
let mut body = json!({ "files": manifest, "prune": prune });
if let Some(commit) = git_head(&root) {
body["commit"] = json!(commit);
}
let opened = call_as(
&entry,
"POST",
&format!("/v1/orgs/{}/stores/{}/pushes", seg(org), seg(store)),
Send {
json: Some(&body),
..Send::default()
},
)?
.json()?;
report.push = opened["push"].as_str().unwrap_or_default().to_string();
if report.push.is_empty() {
bail!("{} opened no push", entry.host_name());
}
report.removed = opened["remove"].as_array().map_or(0, Vec::len);
report.estimate_from(&opened);
let need: Vec<&str> = opened["need"]
.as_array()
.into_iter()
.flatten()
.filter_map(Value::as_str)
.filter(|p| by_path.contains_key(*p))
.collect();
let need_total = need.len();
let mut batch: Vec<(String, Vec<u8>)> = Vec::new();
let mut batch_bytes = 0u64;
let mut send_batch =
|batch: &mut Vec<(String, Vec<u8>)>, report: &mut PushReport| -> Result<()> {
if batch.is_empty() {
return Ok(());
}
let raw: u64 = batch.iter().map(|(_, b)| b.len() as u64).sum();
let archive = tar_gz(batch)?;
call_as(
&entry,
"PUT",
&format!("/v1/orgs/{}/pushes/{}/files", seg(org), seg(&report.push)),
Send {
bytes: Some(("application/x-tar", Some("gzip"), archive)),
timeout: Some(UPLOAD_TIMEOUT),
..Send::default()
},
)?;
report.sent += batch.len();
report.bytes_sent += raw;
say(&format!(
"sent {} of {} changed files ({})",
report.sent,
need_total,
crate::human_bytes(report.bytes_sent as i64)
));
batch.clear();
Ok(())
};
for name in &need {
let (path, hash) = &by_path[*name];
let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?;
if &sha256_hex(&bytes) != hash {
bail!(
"{} changed while it was being pushed; push again",
path.display()
);
}
let cost = bytes.len() as u64 + 1024;
if batch_bytes + cost > UPLOAD_BATCH {
send_batch(&mut batch, &mut report)?;
batch_bytes = 0;
}
batch_bytes += cost;
batch.push((name.to_string(), bytes));
}
send_batch(&mut batch, &mut report)?;
if need.is_empty() {
say("nothing changed since the last push; no file was sent");
}
let committed = call_as(
&entry,
"POST",
&format!("/v1/orgs/{}/pushes/{}/commit", seg(org), seg(&report.push)),
Send::default(),
)?
.json()?;
report.job = committed["job"].as_i64();
report.position = committed["position"].as_i64();
if let Some(job) = report.job {
say(&format!(
"job {job} queued{}",
report
.position
.map(|p| format!(", {p} ahead of it"))
.unwrap_or_default()
));
if wait {
let mut last = String::new();
loop {
let state = call_as(
&entry,
"GET",
&format!("/v1/orgs/{}/jobs/{job}", seg(org)),
Send::default(),
)?
.json()?;
let now = state["state"].as_str().unwrap_or("unknown").to_string();
if now != last {
say(&format!(
"job {job}: {now}{}",
state["chunks"]
.as_i64()
.map(|c| format!(" · {c} chunks"))
.unwrap_or_default()
));
last = now.clone();
}
if matches!(now.as_str(), "done" | "stopped" | "error") {
if now == "error"
&& let Some(outcome) = state["outcome"].as_str()
{
say(outcome);
}
report.state = Some(now);
break;
}
std::thread::sleep(JOB_POLL);
}
}
}
Ok(report)
}
fn git_head(dir: &Path) -> Option<String> {
let out = crate::git(dir)
.args(["rev-parse", "HEAD"])
.stderr(std::process::Stdio::null())
.output()
.ok()?;
let head = String::from_utf8_lossy(&out.stdout).trim().to_string();
(out.status.success() && head.len() >= 7).then_some(head)
}
fn tar_gz(files: &[(String, Vec<u8>)]) -> Result<Vec<u8>> {
use std::io::Write;
let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
for (name, bytes) in files {
if name.len() > 100 {
let record = pax_record("path", name);
gz.write_all(&tar_header("PaxHeader", record.len() as u64, b'x'))?;
gz.write_all(&padded(record.as_bytes()))?;
}
gz.write_all(&tar_header(name, bytes.len() as u64, b'0'))?;
gz.write_all(&padded(bytes))?;
}
gz.write_all(&[0u8; 1024])?;
Ok(gz.finish()?)
}
fn pax_record(key: &str, value: &str) -> String {
let body = format!(" {key}={value}\n");
let mut len = body.len() + 1;
while format!("{len}{body}").len() != len {
len += 1;
}
format!("{len}{body}")
}
fn padded(bytes: &[u8]) -> Vec<u8> {
let mut out = bytes.to_vec();
out.resize(bytes.len().div_ceil(512) * 512, 0);
out
}
fn tar_header(name: &str, size: u64, kind: u8) -> [u8; 512] {
let mut h = [0u8; 512];
let name = name.as_bytes();
let cut = name.len().min(100);
h[..cut].copy_from_slice(&name[..cut]);
let octal = |h: &mut [u8; 512], at: usize, width: usize, value: u64| {
let text = format!("{value:0w$o}", w = width - 1);
h[at..at + width - 1].copy_from_slice(text.as_bytes());
};
octal(&mut h, 100, 8, 0o644);
octal(&mut h, 108, 8, 0);
octal(&mut h, 116, 8, 0);
octal(&mut h, 124, 12, size);
octal(&mut h, 136, 12, 0);
h[148..156].copy_from_slice(b" ");
h[156] = kind;
h[257..263].copy_from_slice(b"ustar\0");
h[263..265].copy_from_slice(b"00");
let sum: u32 = h.iter().map(|b| *b as u32).sum();
let text = format!("{sum:06o}\0 ");
h[148..156].copy_from_slice(text.as_bytes());
h
}
pub const SYNC_BATCH: usize = 1000;
pub const SYNC_EVERY: Duration = Duration::from_secs(60);
pub fn sync_body(machine: &str, store: &str, rows: &[crate::store::SyncRow]) -> Value {
json!({
"machine": machine,
"store": store,
"rows": rows.iter().map(|r| json!({
"local_id": r.id,
"at": crate::clock::utc_rfc3339(r.at),
"client": r.client,
"session": r.session,
"tool": r.tool,
"bytes": null,
"tokens": r.excerpt_tokens,
})).collect::<Vec<_>>(),
})
}
pub fn send_rows(
entry: &Entry,
machine: &str,
store: &str,
rows: &[crate::store::SyncRow],
) -> Result<Value, Failure> {
call_as(
entry,
"POST",
&format!("/v1/orgs/{}/ledger", seg(&entry.org)),
Send {
json: Some(&sync_body(machine, store, rows)),
timeout: Some(FORWARD_TIMEOUT),
..Send::default()
},
)?
.json()
}
pub fn sync_store(
entry: &Entry,
machine: &str,
store: &str,
db: &rusqlite::Connection,
since: i64,
) -> Result<usize> {
let rows = crate::store::unsynced(db, since, SYNC_BATCH)?;
if rows.is_empty() {
return Ok(0);
}
send_rows(entry, machine, store, &rows)?;
let ids: Vec<i64> = rows.iter().map(|r| r.id).collect();
crate::store::mark_synced(db, &ids, unix_now() as i64)?;
Ok(rows.len())
}
#[derive(Debug, Clone, Default, Serialize)]
pub struct SyncState {
pub last_sent: Option<u64>,
pub last_store: Option<String>,
pub last_rows: usize,
pub last_error: Option<String>,
}
static SYNC: std::sync::Mutex<Option<SyncState>> = std::sync::Mutex::new(None);
static SYNCING: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);
fn note_sync(change: impl FnOnce(&mut SyncState)) {
let mut state = SYNC.lock().unwrap_or_else(|e| e.into_inner());
change(state.get_or_insert_with(SyncState::default));
}
pub fn spawn_sync(state: std::sync::Arc<crate::daemon::State>) {
use std::sync::atomic::Ordering;
if SYNCING.swap(true, Ordering::SeqCst) {
return;
}
std::thread::spawn(move || {
while !crate::watch::STOP.load(Ordering::Relaxed) {
let started = std::time::Instant::now();
while started.elapsed() < SYNC_EVERY {
if crate::watch::STOP.load(Ordering::Relaxed) {
return;
}
std::thread::sleep(Duration::from_millis(500));
}
sync_pass(&state);
}
});
}
fn sync_pass(state: &crate::daemon::State) {
let Ok(Some(creds)) = Credentials::load() else {
return;
};
let Ok(registry) = crate::home::Registry::load() else {
return;
};
for (name, entry) in ®istry.stores {
let Some(sync) = &entry.settings.cloud_sync else {
continue;
};
let Some(cred) = creds.entries.iter().find(|e| e.org == sync.org) else {
note_sync(|s| s.last_error = Some(format!("not signed in to {}", sync.org)));
continue;
};
if state.open_fleet().is_err() {
return;
}
let rows = {
let fleet = state.fleet.lock().unwrap_or_else(|e| e.into_inner());
let Some(store) = fleet
.as_ref()
.and_then(|f| f.each().find(|(l, _)| l == name).map(|(_, s)| s))
else {
continue;
};
crate::store::unsynced(store.db(), sync.since, SYNC_BATCH).unwrap_or_default()
};
if rows.is_empty() {
continue;
}
match send_rows(cred, &creds.machine_id(), name, &rows) {
Ok(_) => {
let ids: Vec<i64> = rows.iter().map(|r| r.id).collect();
let fleet = state.fleet.lock().unwrap_or_else(|e| e.into_inner());
if let Some(store) = fleet
.as_ref()
.and_then(|f| f.each().find(|(l, _)| l == name).map(|(_, s)| s))
{
let _ = crate::store::mark_synced(store.db(), &ids, unix_now() as i64);
}
note_sync(|s| {
s.last_sent = Some(unix_now());
s.last_store = Some(name.clone());
s.last_rows = rows.len();
s.last_error = None;
});
}
Err(e) => note_sync(|s| s.last_error = Some(e.to_string())),
}
return;
}
}
pub fn set_sync(
store: &str,
on: bool,
org: Option<&str>,
) -> Result<Option<crate::home::CloudSync>> {
crate::home::refuse_remote(store)?;
let sync = if on {
let entry = entry_for(org, None)?;
Some(crate::home::CloudSync {
org: entry.org,
since: unix_now() as i64,
})
} else {
None
};
crate::home::update_store_settings(store, move |s| match (&s.cloud_sync, sync) {
(Some(old), Some(new)) if old.org == new.org => {}
(_, sync) => s.cloud_sync = sync,
})
.map(|s| s.cloud_sync)
}
pub fn sync_view() -> Value {
let registry = crate::home::Registry::load().unwrap_or_default();
let on: Vec<Value> = registry
.stores
.iter()
.filter_map(|(name, e)| {
e.settings
.cloud_sync
.as_ref()
.map(|s| json!({ "store": name, "org": s.org, "since": s.since }))
})
.collect();
let last = SYNC
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
.unwrap_or_default();
json!({
"on": on,
"stores": registry.stores.len(),
"last_sent": last.last_sent,
"last_store": last.last_store,
"last_rows": last.last_rows,
"last_error": last.last_error,
})
}
pub const REPORT_FORMATS: [&str; 5] = ["md", "csv", "json", "html", "pdf"];
pub struct ReportAsk<'a> {
pub kind: &'a str,
pub format: &'a str,
pub window: Option<&'a str>,
pub model: Option<&'a str>,
pub stores: &'a [String],
}
pub fn report(entry: &Entry, ask: &ReportAsk<'_>) -> Result<Vec<u8>> {
if !REPORT_FORMATS.contains(&ask.format) {
bail!(
"{} is not a report format; one of {}",
ask.format,
REPORT_FORMATS.join(", ")
);
}
let mut query = format!("format={}", seg(ask.format));
if let Some(window) = ask.window {
query.push_str(&format!("&window={}", seg(window)));
}
if let Some(model) = ask.model {
query.push_str(&format!("&model={}", seg(model)));
}
if !ask.stores.is_empty() {
let list: Vec<String> = ask.stores.iter().map(|s| seg(s)).collect();
query.push_str(&format!("&stores={}", list.join(",")));
}
let reply = call_as(
entry,
"GET",
&format!(
"/v1/orgs/{}/reports/{}?{query}",
seg(&entry.org),
seg(ask.kind)
),
Send {
headers: vec![("Accept", "*/*".into())],
timeout: Some(Duration::from_secs(120)),
..Send::default()
},
)?;
Ok(reply.body)
}
pub fn replay(entry: &Entry, session: &crate::replay::Session) -> Result<i64> {
let items: Vec<Value> = session
.recent
.iter()
.map(|a| {
json!({
"at": a.at,
"tool": a.tool,
"query": a.query,
"outcome": if a.outcome == "unknown" { "none" } else { a.outcome },
})
})
.collect();
let sent = call_as(
entry,
"POST",
&format!("/v1/orgs/{}/replay", seg(&entry.org)),
Send {
json: Some(&json!({
"session": session.id,
"client": crate::replay::CLIENT,
"items": items,
})),
..Send::default()
},
);
match sent {
Ok(reply) => Ok(reply.json()?["accepted"].as_i64().unwrap_or(0)),
Err(Failure::Refused { code, .. }) if code == "replay_off" => bail!(
"Session replay is off for {}: an admin turns it on in the cloud app under \
Ledger › Session replay. Nothing was kept.",
entry.org
),
Err(other) => Err(other.into()),
}
}
pub fn replay_sessions() -> Result<(PathBuf, crate::replay::Replay)> {
if !crate::home::Settings::load().replay_on() {
bail!(
"Session replay is off on this machine (the Privacy page's switch), so semlith \
reads no transcript; turn it on to pick one to send"
);
}
let dir = crate::replay::transcripts_dir()?;
let found = crate::replay::read(&dir, crate::replay::FILES)?;
Ok((dir, found))
}
pub fn status_view() -> Value {
let creds = Credentials::load().ok().flatten();
let orgs: Vec<Value> = creds
.iter()
.flat_map(|c| c.entries.iter())
.map(|e| match status(e) {
Ok((status, version)) => json!({
"org": e.org, "host": e.host, "reach": "connected",
"why": format!("{} answered just now", e.host_name()),
"version": version, "status": status,
}),
Err(f) => json!({
"org": e.org, "host": e.host, "reach": f.kind(),
"why": f.to_string(), "code": f.code(),
}),
})
.collect();
json!({ "orgs": orgs })
}
pub fn doctor() -> Vec<Value> {
let Some(creds) = Credentials::load().ok().flatten() else {
return Vec::new();
};
creds
.entries
.iter()
.map(|e| {
let asked = call_as(
e,
"GET",
"/v1/whoami",
Send {
timeout: Some(FORWARD_TIMEOUT),
..Send::default()
},
);
match asked {
Ok(reply) => json!({
"org": e.org, "host": e.host, "reach": "connected",
"why": format!("{} answered with this token", e.host_name()),
"version": reply.version,
}),
Err(f) => {
json!({ "org": e.org, "host": e.host, "reach": f.kind(), "why": f.to_string() })
}
}
})
.collect()
}
pub fn local_view() -> Value {
let creds = Credentials::load().ok().flatten();
let orgs: Vec<Value> = creds
.iter()
.flat_map(|c| c.entries.iter())
.map(|e| {
json!({
"org": e.org,
"host": e.host,
"host_name": e.host_name(),
"plan": e.plan,
"prefix": e.prefix(),
})
})
.collect();
json!({
"signed_in": !orgs.is_empty(),
"orgs": orgs,
"remote": remote_rows(),
"sync": sync_view(),
})
}
#[cfg(test)]
#[path = "../tests/common/stub.rs"]
mod stub;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn only_a_plain_page_on_the_host_is_opened() {
let host = "https://cloud.semlith.com";
let good = "https://cloud.semlith.com/cli?code=WDJB-MJHT";
assert_eq!(openable(host, good), Some(good));
for planted in [
"/System/Applications/Calculator.app",
"file:///etc/passwd",
"https://cloud.semlith.com.evil.example/cli",
"https://evil.example/cli?code=A",
"https://cloud.semlith.com/cli?code=A&calc",
"https://cloud.semlith.com/cli?code=A|calc",
"https://cloud.semlith.com/cli?code=%41",
"https://cloud.semlith.com/cli code",
"https://cloud.semlith.com/cli\"x",
"https://cloud.semlith.com",
] {
assert_eq!(openable(host, planted), None, "{planted}");
}
assert_eq!(
printable("a\u{1b}]0;x\u{7}b\u{1b}[2J\r\n\u{9b}c"),
"a]0;xb[2Jc"
);
}
fn local(score: f32, path: &str) -> crate::Hit {
RemoteHit {
name: "x".into(),
org: "x".into(),
..remote(score, path)
}
.to_hit()
}
fn remote(score: f32, path: &str) -> RemoteHit {
serde_json::from_value(json!({
"score": score, "store": "platform", "path": path, "start_line": 1, "end_line": 2,
"text": "fn order_total() { a long line of code that costs tokens to show }",
"revision": "a41c9e2", "behind_seconds": 38
}))
.map(|mut h: RemoteHit| {
h.name = "acme/platform".into();
h.org = "acme".into();
h
})
.unwrap()
}
#[test]
fn a_merge_orders_by_score_cuts_to_k_and_labels_both_sides() {
let mut mine = local(0.03, "notes/a.md");
mine.store = None;
let merged = merge(
vec![mine, local(0.01, "notes/b.md")],
Some("notes"),
&[remote(0.02, "r/c.rs"), remote(0.05, "r/d.rs")],
3,
);
let order: Vec<(&str, f32)> = merged.iter().map(|h| (h.path.as_str(), h.score)).collect();
assert_eq!(
order,
[("r/d.rs", 0.05), ("notes/a.md", 0.03), ("r/c.rs", 0.02)]
);
assert_eq!(merged[1].store.as_deref(), Some("notes"));
assert_eq!(
merged[0].store.as_deref(),
Some("acme/platform (remote · acme · a41c9e2 · 38 s behind)")
);
}
#[test]
fn the_token_budget_applies_to_the_merged_list() {
let many: Vec<RemoteHit> = (0..40)
.map(|i| remote(1.0 - i as f32 / 100.0, &format!("r/{i}.rs")))
.collect();
let merged = merge(Vec::new(), None, &many, 40);
let reply = crate::mcp::search_reply(
&crate::fleet::Fleet::empty(),
&merged,
"order total",
crate::Prefer::default(),
200,
);
let kept = reply.kept.iter().filter(|k| **k).count();
assert!(kept < 40, "{kept} of 40 kept under 200 tokens");
assert!(
reply.text.contains(&format!("truncated: {kept} of 40")),
"{}",
reply.text
);
}
#[test]
fn the_upload_is_a_tar_a_reader_can_unpack() {
use std::io::Read;
let long = format!("{}/deep.md", "d".repeat(120));
let files = vec![
("docs/runbook.md".to_string(), b"hello".to_vec()),
(long.clone(), b"long".to_vec()),
];
let mut raw = Vec::new();
flate2::read::GzDecoder::new(&tar_gz(&files).unwrap()[..])
.read_to_end(&mut raw)
.unwrap();
let mut at = 0;
let mut found = Vec::new();
let mut pax_path: Option<String> = None;
while at + 512 <= raw.len() && raw[at] != 0 {
let h = &raw[at..at + 512];
let stored: u32 = h
.iter()
.enumerate()
.map(|(i, b)| {
if (148..156).contains(&i) {
32
} else {
*b as u32
}
})
.sum();
let said = u32::from_str_radix(std::str::from_utf8(&h[148..154]).unwrap(), 8).unwrap();
assert_eq!(stored, said, "header checksum");
let size = u64::from_str_radix(std::str::from_utf8(&h[124..135]).unwrap(), 8).unwrap()
as usize;
let name = String::from_utf8_lossy(&h[..100])
.trim_end_matches('\0')
.to_string();
let body = &raw[at + 512..at + 512 + size];
if h[156] == b'x' {
let text = String::from_utf8_lossy(body);
pax_path = text
.split_once("path=")
.map(|(_, p)| p.trim_end().to_string());
assert_eq!(
text.split_once(' ').unwrap().0.parse::<usize>().unwrap(),
text.len()
);
} else {
found.push((pax_path.take().unwrap_or(name), body.to_vec()));
}
at += 512 + size.div_ceil(512) * 512;
}
assert_eq!(found, files);
}
fn ledger(n: usize) -> (tempfile::TempDir, rusqlite::Connection) {
let dir = tempfile::tempdir().unwrap();
let db = crate::store::open(&dir.path().join("store.db")).unwrap();
let _w = crate::store::Writing::begin(&db).unwrap();
for i in 0..n {
db.execute(
"INSERT INTO retrievals (at, client, query, hits, micros, excerpt_tokens,
whole_file_tokens, prev, hash, session, tool)
VALUES (?1, 'claude-code', ?2, 3, 10, 120, 900, '', '', 's-1', 'semlith_search')",
rusqlite::params![1_000 + i as i64, format!("PRIVATE QUERY TEXT {i}")],
)
.unwrap();
}
drop(_w);
(dir, db)
}
fn entry_at(url: &str) -> Entry {
Entry {
host: url.to_string(),
org: "acme".into(),
token: format!("{TOKEN_PREFIX}test"),
plan: None,
added: 0,
}
}
#[test]
fn ledger_sync_batches_resends_idempotently_and_never_sends_text() {
let fail_first = std::sync::atomic::AtomicBool::new(true);
let host = stub::Stub::start(move |seen| {
assert_eq!(seen.path, "/v1/orgs/acme/ledger");
if fail_first.swap(false, std::sync::atomic::Ordering::SeqCst) {
stub::error(503, "unavailable", "Postgres is away.")
} else {
stub::json(200, json!({ "accepted": 1, "duplicates": 0 }))
}
});
let (_dir, db) = ledger(1_500);
let entry = entry_at(&host.url);
assert!(sync_store(&entry, "m_1", "notes", &db, 0).is_err());
assert_eq!(crate::store::unsynced_count(&db, 0).unwrap(), 1_500);
assert_eq!(
sync_store(&entry, "m_1", "notes", &db, 0).unwrap(),
SYNC_BATCH
);
assert_eq!(sync_store(&entry, "m_1", "notes", &db, 0).unwrap(), 500);
assert_eq!(sync_store(&entry, "m_1", "notes", &db, 0).unwrap(), 0);
let seen = host.seen();
assert_eq!(seen.len(), 3, "an empty batch is no request");
let ids = |i: usize| -> Vec<i64> {
seen[i].json()["rows"]
.as_array()
.unwrap()
.iter()
.map(|r| r["local_id"].as_i64().unwrap())
.collect()
};
assert_eq!(ids(0), ids(1), "the resend carries the same rows");
assert_eq!(ids(1).len(), 1_000);
assert_eq!(ids(2).len(), 500);
assert_eq!(seen[0].json()["rows"][0]["at"], "1970-01-01T00:16:40Z");
for s in &seen {
assert!(
!s.text().contains("PRIVATE QUERY TEXT"),
"query text left the machine"
);
assert!(
!s.text().contains("query"),
"a query field left the machine"
);
let body = s.json();
assert_eq!(body["machine"], "m_1");
assert_eq!(body["store"], "notes");
assert_eq!(body["rows"][0]["tool"], "semlith_search");
}
}
#[test]
fn ledger_sync_starts_from_when_it_was_turned_on() {
let (_dir, db) = ledger(10);
assert_eq!(crate::store::unsynced(&db, 1_005, 1_000).unwrap().len(), 5);
}
#[test]
fn a_host_is_https_or_loopback() {
assert_eq!(
normal_host("https://cloud.semlith.com/").unwrap(),
"https://cloud.semlith.com"
);
assert_eq!(
normal_host("http://127.0.0.1:8080").unwrap(),
"http://127.0.0.1:8080"
);
assert_eq!(
normal_host("http://localhost:3000").unwrap(),
"http://localhost:3000"
);
assert!(normal_host("http://cloud.semlith.com").is_err());
assert!(normal_host("https://cloud.semlith.com/acme").is_err());
assert!(normal_host("https://user@cloud.semlith.com").is_err());
assert!(normal_host("cloud.semlith.com").is_err());
}
#[test]
fn an_entry_never_prints_its_token() {
let entry = Entry {
host: "https://cloud.semlith.com".into(),
org: "acme".into(),
token: format!("{TOKEN_PREFIX}7f3cSECRETSECRETSECRET"),
plan: None,
added: 0,
};
let shown = format!("{entry:?}");
assert!(!shown.contains("SECRET"), "{shown}");
assert!(shown.contains("sml_live_7f3c"), "{shown}");
}
#[test]
fn a_refusal_prints_the_hosts_sentence_and_the_next_step() {
let old = refusal(
426,
br#"{"error":{"code":"client_too_old","message":"This semlith is too old.","min_version":"0.37.0"}}"#,
);
assert!(old.to_string().contains("Run `semlith upgrade`."), "{old}");
let plan = refusal(
402,
br#"{"error":{"code":"plan_required","message":"Pushing needs a plan.","plan":"pro"}}"#,
);
assert_eq!(
plan.to_string(),
"Pushing needs a plan. It needs the Pro plan."
);
assert_eq!(plan.kind(), "refused");
}
#[test]
fn status_reads_indexed_chunks_and_falls_back_to_minutes_for_an_older_host() {
let entry = entry_at("https://cloud.example");
let status = |usage: Value| json!({ "org": { "slug": "acme", "plan": "pro", "status": "active" }, "usage": usage });
let chunks = render_status(
&entry,
&status(
json!({ "month": "2026-10", "store_bytes": 1000, "cap_bytes": 8000,
"index_chunks": 12400, "cap_chunks": 100000, "queue": { "running": 0, "waiting": 0 } }),
),
None,
);
assert!(
chunks.contains("12,400 of 100,000 chunks indexed"),
"{chunks}"
);
assert!(!chunks.contains("minutes"), "{chunks}");
let minutes = render_status(
&entry,
&status(
json!({ "month": "2026-10", "store_bytes": 1000, "cap_bytes": 8000,
"index_minutes": 2, "cap_minutes": 100, "queue": { "running": 0, "waiting": 0 } }),
),
None,
);
assert!(minutes.contains("2 of 100 index minutes"), "{minutes}");
assert!(!minutes.contains("chunks"), "{minutes}");
}
#[test]
fn a_push_estimate_reads_chunks_or_the_older_minutes() {
let mut report = PushReport::default();
report.estimate_from(&json!({ "push": "p", "estimate_chunks": 1400 }));
assert_eq!(
(report.estimate_chunks, report.estimate_minutes),
(Some(1400), None)
);
assert_eq!(
report.estimate_line().as_deref(),
Some("about 1,400 chunks to index")
);
let mut report = PushReport::default();
report.estimate_from(&json!({ "push": "p", "estimate_minutes": 3 }));
assert_eq!(
(report.estimate_chunks, report.estimate_minutes),
(None, Some(3))
);
assert_eq!(
report.estimate_line().as_deref(),
Some("about 3 index minutes")
);
let mut report = PushReport::default();
report.estimate_from(&json!({ "push": "p", "estimate_chunks": 1 }));
assert_eq!(
report.estimate_line().as_deref(),
Some("about 1 chunk to index")
);
report.estimate_from(&json!({ "push": "p" }));
assert_eq!(report.estimate_line(), None);
}
}