use std::time::Duration;
use reqwest::Method;
use serde_json::{json, Value};
use sylphx_apps_sdk_core::{
parse_connection_url, Credential, RetryConfig, SdkError, Transport, TransportConfig,
};
pub struct RuntimeClient {
transport: Transport,
}
impl RuntimeClient {
pub fn from_secret_url(secret_url: Option<&String>) -> anyhow::Result<Self> {
let raw = secret_url
.map(|s| s.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.or_else(|| {
std::env::var("SYLPHX_SECRET_URL")
.ok()
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
});
let Some(url) = raw else {
anyhow::bail!(
"runtime credential required. Set SYLPHX_SECRET_URL (sylphx://sk_…@<slug>.api.sylphx.com) \
or pass --secret-url. Values are inject/vault-path only — do not paste secrets into chat."
);
};
if !url.starts_with("sylphx://") {
anyhow::bail!(
"SYLPHX_SECRET_URL must be a sylphx:// connection URL (not a bare sk_*). \
Runtime services receive this via platform inject; agents should read vault/env only."
);
}
let parsed = parse_connection_url(&url).map_err(|e| anyhow::anyhow!(e.to_string()))?;
Ok(Self {
transport: Transport::new(TransportConfig {
base_url: parsed.api_base_url,
credential: Credential::app_secret(parsed.credential)
.map_err(|e| anyhow::anyhow!(e.to_string()))?,
timeout: Duration::from_secs(30),
retry: RetryConfig::default(),
})
.map_err(|e| anyhow::anyhow!(e.to_string()))?,
})
}
pub async fn work_container_index(&self) -> Result<Value, SdkError> {
self.transport.get_json("/work/container", &[]).await
}
pub async fn work_container_create(&self, body: &Value) -> Result<Value, SdkError> {
self.transport
.post_json::<Value, Value>("/work/container", body, &[])
.await
}
pub async fn work_container_get(&self, id: &str) -> Result<Value, SdkError> {
let path = format!("/work/container/{id}");
self.transport.get_json(&path, &[]).await
}
pub async fn objects_list_buckets(&self) -> Result<Value, SdkError> {
self.transport.get_json("/objects/buckets", &[]).await
}
pub async fn objects_list(
&self,
bucket: Option<&str>,
prefix: Option<&str>,
limit: Option<u32>,
) -> Result<Value, SdkError> {
let mut path = String::from("/objects");
let mut sep = '?';
if let Some(b) = bucket {
path.push(sep);
sep = '&';
path.push_str("bucket=");
path.push_str(b);
}
if let Some(pfx) = prefix {
path.push(sep);
sep = '&';
path.push_str("prefix=");
path.push_str(pfx);
}
if let Some(n) = limit {
path.push(sep);
path.push_str(&format!("limit={n}"));
}
self.transport.get_json(&path, &[]).await
}
pub async fn objects_put(
&self,
key: &str,
body: &[u8],
content_type: Option<&str>,
bucket: Option<&str>,
) -> Result<Value, SdkError> {
let mut path = format!("/objects/{}", key.trim_start_matches('/'));
if let Some(b) = bucket {
path.push('?');
path.push_str("bucket=");
path.push_str(b);
}
let ct = content_type.unwrap_or("application/octet-stream");
let headers = [("content-type", ct)];
let (_status, bytes) = self
.transport
.raw_bytes_request(Method::PUT, &path, Some(body), &headers)
.await?;
if bytes.is_empty() {
return Ok(json!({"ok": true, "key": key}));
}
serde_json::from_slice(&bytes).map_err(|e| SdkError::Decode(e.to_string()))
}
pub async fn objects_get_bytes(
&self,
key: &str,
bucket: Option<&str>,
) -> Result<Vec<u8>, SdkError> {
let mut path = format!("/objects/{}", key.trim_start_matches('/'));
if let Some(b) = bucket {
path.push('?');
path.push_str("bucket=");
path.push_str(b);
}
let (_status, bytes) = self
.transport
.raw_bytes_request(Method::GET, &path, None, &[])
.await?;
Ok(bytes)
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::time::{Duration, Instant};
fn request_target(req: &str) -> &str {
req.lines()
.next()
.unwrap_or("")
.split_whitespace()
.nth(1)
.unwrap_or("")
}
fn request_host(req: &str) -> &str {
req.lines()
.find_map(|line| {
let (name, value) = line.split_once(':')?;
name.eq_ignore_ascii_case("host").then(|| value.trim())
})
.unwrap_or("")
}
fn leftover_schedule_target(target: &str) -> bool {
target.contains("/schedules")
}
fn read_http_headers(stream: &mut std::net::TcpStream) -> String {
let mut buf = Vec::new();
let mut tmp = [0u8; 512];
loop {
match stream.read(&mut tmp) {
Ok(0) => break,
Ok(n) => {
buf.extend_from_slice(&tmp[..n]);
if buf.windows(4).any(|window| window == b"\r\n\r\n") {
break;
}
}
Err(_) => break,
}
}
String::from_utf8_lossy(&buf).into_owned()
}
fn spawn_json_server(
n: usize,
route: fn(&str) -> (u16, &'static str),
) -> (
String,
std::sync::mpsc::Receiver<String>,
std::thread::JoinHandle<()>,
) {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let addr = listener.local_addr().unwrap();
let (tx, rx) = std::sync::mpsc::channel();
let join = std::thread::spawn(move || {
let deadline = Instant::now() + Duration::from_secs(4);
let mut got = 0usize;
while got < n && Instant::now() < deadline {
match listener.accept() {
Ok((mut stream, _)) => {
let _ = stream.set_read_timeout(Some(Duration::from_secs(2)));
let _ = stream.set_write_timeout(Some(Duration::from_secs(2)));
let req = read_http_headers(&mut stream);
let (status, body) = route(&req);
let reason = match status {
200 => "OK",
410 => "Gone",
_ => "Not Found",
};
let resp = format!(
"HTTP/1.1 {status} {reason}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
body.len()
);
let _ = stream.write_all(resp.as_bytes());
let _ = tx.send(req);
got += 1;
}
Err(err) if err.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(5));
}
Err(_) => break,
}
}
});
(format!("http://{addr}/v1"), rx, join)
}
fn runtime_on(base: &str) -> RuntimeClient {
RuntimeClient {
transport: Transport::new(TransportConfig {
base_url: base.to_string(),
credential: Credential::app_secret("sk_test_deadbeef").expect("secret"),
timeout: Duration::from_secs(2),
retry: RetryConfig {
max_attempts: 1,
base_delay: Duration::from_millis(1),
},
})
.expect("transport"),
}
}
#[test]
fn from_secret_url_uses_caller_host_not_invented_peel() {
let hex = "a".repeat(32);
let secret_url = format!("sylphx://sk_prod_{hex}@demo.api.example.test/v1");
let client = RuntimeClient::from_secret_url(Some(&secret_url)).expect("runtime");
assert_eq!(
client.transport.base_url(),
"https://demo.api.example.test/v1"
);
assert!(
!client.transport.base_url().contains(".api.sylphx.com"),
"leftover runtime invented a peel hostname from the caller host: {}",
client.transport.base_url()
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn work_container_index_targets_work_container_not_schedule() {
let (base, rx, join) = spawn_json_server(1, |req| {
let target = request_target(req);
if leftover_schedule_target(target) {
(
410,
r#"{"error":{"code":"gone","message":"leftover customer schedule"}}"#,
)
} else if target.starts_with("/v1/work/container") {
(200, r#"{"object":"list","data":[]}"#)
} else {
(404, r#"{"error":{"code":"not_found"}}"#)
}
});
let client = runtime_on(&base);
let body = tokio::time::timeout(Duration::from_secs(3), client.work_container_index())
.await
.expect("work.container did not reach caller runtime api_base")
.expect("work container index");
assert_eq!(body["object"], "list");
let req = rx
.recv_timeout(Duration::from_secs(2))
.expect("runtime request");
let target = request_target(&req);
let host = request_host(&req);
assert_eq!(target, "/v1/work/container");
assert!(
host.starts_with("127.0.0.1:"),
"leftover runtime invented a peel hostname, Host={host}"
);
assert!(
!leftover_schedule_target(target),
"leftover runtime targeted customer Schedule: {target}"
);
join.join().expect("runtime server");
}
}