johnandersen777.bsky.social/atlibs-agents
agents and atirc modules for atlibs: launch agents from a composition, chat on atirc; pages and the agents service
0
0
// agents-service: the agents XRPC service on the Rust host.
//
// serve.json the ingress module registers a did:key with xrpc.fedproxy.com and forwards
// com.fedfork.agents.{launchAgent,getAgent,shutdownAgent} to the handlers below
// handlers check the caller's atproto service auth (atlibs_host::serviceauth: aud = this
// service, lxm = the method, iss in ATLIBS_AGENTS_ISSUERS), then run
// composition.json agents + requester + allow-bidders, as the signed-in account, each instance held to
// its own module's session-record scope (records/<session>.json beside the module)
//
// flags (the operator account, explicit; no account is read from the environment):
// --app-password <handle> the account the service runs as (required)
// --password-file <path|-> its app password from a file, or `-` for stdin (first line, trimmed)
// --password-env <NAME> its app password from the env var NAMED here (never in argv)
// --pds <url> createSession service (default https://bsky.social)
// exactly one of --password-file / --password-env.
// env: ATLIBS_SERVER_STATE, ATLIBS_IROH_KEY (as atlibs-host),
// ATLIBS_AGENTS_ISSUERS (optional: only these DIDs, or `did:<method>:*` prefixes, may call; default
// anyone who signs in), ATLIBS_AGENTS_USER_DATA (required: a
// user_data file, `node ../atlibs-iroh/examples/iroh.mjs user-data $ATLIBS_IROH_KEY.pub`), ATLIBS_AGENTS_BIDDERS
// (default bidders), ATLIBS_RELAY (https://xrpc.fedproxy.com), ATLIBS_AGENTS_LIFETIME_MS (serve this
// long, default until Ctrl-C), ATLIBS_AGENTS_MAX_PER_CALLER (launches a caller may have at a time,
// default 2), ATLIBS_AGENTS_MAX_LAUNCHES (launches the service runs at a time, all callers: default 8),
// ATLIBS_REPO (where modules/, records/ and target/ are; default this checkout; the requester and
// policy-open modules come from ../atlibs-socialweb-computer-requester, ingress from ../atlibs-fedproxy,
// iroh-user-data (a hook default of the requester's user-data) from ../atlibs-iroh).
use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::time::Duration;
use atlibs_host::compose::{self, CallFor, Composition, Handler, Info};
use atlibs_host::crypto::ingress_host;
use atlibs_host::host::{DidkeyCall, Fetch, HostCall, Session};
use atlibs_host::oauth::ScopeCall;
use atlibs_host::transport::{TransportCall, Transports};
use atlibs_host::run::{Call, Ledger, Module, Options, Request, Sub};
use atlibs_host::serviceauth::{Auth, Replay, Resolve, resolver, verify};
use atlibs_host::store::{Store, uuid};
use atlibs_host::wasm::{WALL_TIME_MS, Wasm};
use atlibs_host::{BoxFut, Signal, now_ms, pretty};
use serde_json::{Map, Value, json};
use tokio::sync::{oneshot, watch};
const LAUNCH: &str = "com.fedfork.agents.launch";
const LAUNCH_AGENT: &str = "com.fedfork.agents.launchAgent";
const GET_AGENT: &str = "com.fedfork.agents.getAgent";
const SHUTDOWN_AGENT: &str = "com.fedfork.agents.shutdownAgent";
// What a caller may choose; user_data, the default bidders and the guest command
// (the agents module's default, `atirc agent run`) are the service's.
const CLIENT_FIELDS: [&str; 5] = ["name", "via", "bidders", "exp_s", "tag"];
// The agent token's life when the caller names none; the agents module caps exp_s at 24 h. The
// service shuts a launch down when it ends.
const DEFAULT_EXP_S: u64 = 7200;
// Launches a caller may have at a time (ATLIBS_AGENTS_MAX_PER_CALLER).
const MAX_PER_CALLER: usize = 2;
// Launches the service runs at a time, all callers together (ATLIBS_AGENTS_MAX_LAUNCHES): the
// service account pays for every VM, and a new identity is cheap.
const MAX_LAUNCHES: usize = 8;
// The relay answers 502 after 30 s; a slower shutdown answers "stopping" and keeps going.
const SHUTDOWN_WAIT: Duration = Duration::from_secs(25);
// ---- compositions --------------------------------------------------------------------------
// Where a module key's modules/<key>/module.json may be: this repo, then the sibling checkouts that own the
// modules the compositions borrow (requester, policy-open; ingress; iroh-user-data). Each has its own records/ and target/.
const SIBLINGS: [&str; 3] = ["../atlibs-socialweb-computer-requester", "../atlibs-fedproxy", "../atlibs-iroh"];
/// The checkout holding modules/<key>: `repo` or a sibling of it.
fn root_of(repo: &Path, key: &str) -> Result<PathBuf, String> {
std::iter::once(repo.to_path_buf())
.chain(SIBLINGS.iter().map(|s| repo.join(s)))
.find(|r| r.join(format!("modules/{key}/module.json")).is_file())
.ok_or_else(|| format!("{key}: no modules/{key}/module.json in {} or {}", repo.display(), SIBLINGS.join(", ")))
}
/// A module's session record as (rkey, scope): records/<session>.json in its checkout (the CLI's module_of
/// reads the same file); None when module.json names no session.
fn scope_of(root: &Path, spec: &Value) -> Result<Option<(String, String)>, String> {
let Some(session) = spec["session"].as_str() else { return Ok(None) };
let path = root.join(format!("records/{session}.json"));
let record: Value = serde_json::from_slice(&std::fs::read(&path).map_err(|e| format!("{}: {e}", path.display()))?).map_err(|e| format!("{}: {e}", path.display()))?;
let scope = record["clientMetadata"]["scope"].as_str().ok_or_else(|| format!("{}: no clientMetadata.scope", path.display()))?;
Ok(Some((session.to_string(), scope.to_string())))
}
/// A composition file under modules/agents with each instance's module (modules/<key>/module.json,
/// target/wasm32-unknown-unknown/release/<key>.wasm and its session's records/<session>.json, from root_of),
/// hook defaults (the requester's policy) included (compose::load), checked against the transports the
/// service serves.
async fn load(repo: &Path, file: &str, provides: &HashSet<String>) -> Result<Composition, String> {
let manifest: Value = serde_json::from_slice(&std::fs::read(repo.join("modules/agents").join(file)).map_err(|e| format!("{file}: {e}"))?).map_err(|e| format!("{file}: {e}"))?;
let repo = repo.to_path_buf();
let loader: compose::ModuleLoader = Arc::new(move |key: String| {
let repo = repo.clone();
Box::pin(async move {
let read = |p: PathBuf| std::fs::read(&p).map_err(|e| format!("{}: {e}", p.display()));
if !atlibs_host::run::module_key_ok(&key) {
return Err(format!("{key}: not a module key"));
}
let root = root_of(&repo, &key)?;
let spec: Value = serde_json::from_slice(&read(root.join(format!("modules/{key}/module.json")))?).map_err(|e| format!("{key}: {e}"))?;
let path = root.join(format!("target/wasm32-unknown-unknown/release/{}.wasm", key.replace('-', "_")));
let mut module = Module::from_manifest(read(path.clone())?, &spec)?;
// Identity by path, as the CLI: the composition's ledger owner (and so which records shutdown may
// delete) stays the same across rebuilds of the wasm.
module.id = Some(std::fs::canonicalize(&path).unwrap_or(path).display().to_string());
module.key = Some(key);
module.scope = scope_of(&root, &spec)?;
Ok(module)
})
});
let comp = compose::load(&manifest, &loader).await?;
compose::check(&comp, provides)?;
Ok(comp)
}
/// Passes calls through and hands over the launch record's {uri, cid} as soon as the agents
/// module's createRecord of it returns: the launch id, before the VM.
struct Launched {
inner: Arc<dyn Call>,
tx: Mutex<Option<oneshot::Sender<Value>>>,
}
impl Call for Launched {
fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
Box::pin(async move {
let launch = nsid == "com.atproto.repo.createRecord" && request.input.as_ref().is_some_and(|i| i["collection"] == LAUNCH);
let out = self.inner.call(nsid, request).await;
if let (true, Ok(Some(created))) = (launch, &out) {
if let Some(tx) = self.tx.lock().unwrap().take() {
let _ = tx.send(created.clone());
}
}
out
})
}
fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
self.inner.subscribe(nsid, request)
}
}
// A launch this process started and has not shut down: whose it is, and whether its run ended.
struct Launch {
caller: String,
done: watch::Receiver<bool>,
}
pub struct Service {
pub comp: Arc<Composition>,
pub wasm: Arc<Wasm>,
pub ledger: Option<Arc<dyn Ledger>>,
/// by instance name: the signed-in account's call (HostCall + attest) held to that instance's own
/// module's session scope (ScopeCall, as the CLI's Host::call_for), then the transports (iroh)
calls: HashMap<String, Arc<dyn Call>>,
pub did: String,
pub defaults: Map<String, Value>,
pub auth: Auth,
pub resolve: Resolve,
// service-auth tokens already used: each is good for one request
replay: Replay,
pub max_per_caller: usize,
pub max_launches: usize,
pub shutdown_wait: Duration,
// ponytail: in memory, keyed by launch rkey (a slot key while the record is being made); a
// restart forgets the caps and lifetimes of earlier launches (their tokens still expire).
launches: Mutex<HashMap<String, Launch>>,
}
fn client_error(e: String) -> String {
if e.split_once(' ').is_some_and(|(code, _)| code.len() == 3 && code.parse::<u16>().is_ok()) { e } else { format!("400 InvalidRequest: {e}") }
}
// exp_s as a number (form fields arrive as strings); None when it is not one (the module refuses it).
fn exp_s_of(body: &Value) -> Option<u64> {
match body.get("exp_s") {
None | Some(Value::Null) => Some(DEFAULT_EXP_S),
Some(v) => v.as_u64().or_else(|| v.as_str().and_then(|s| s.trim().parse().ok())),
}
}
impl Service {
#[allow(clippy::too_many_arguments)]
/// `account`: the signed-in account's unscoped call; each instance gets it only through its own scope.
pub fn new(comp: Composition, wasm: Arc<Wasm>, ledger: Option<Arc<dyn Ledger>>, account: Arc<dyn Call>, transports: Transports, did: String, defaults: Map<String, Value>, auth: Auth, resolve: Resolve) -> Arc<Self> {
let calls = comp
.modules
.iter()
.map(|(name, module)| {
let scoped: Arc<dyn Call> = Arc::new(ScopeCall { module: module.key.clone().unwrap_or_else(|| name.clone()), session: module.scope.clone(), inner: account.clone() });
let listed = module.calls.as_ref().is_some_and(|c| !c.is_empty());
(name.clone(), Arc::new(TransportCall { transports: transports.clone(), listed, inner: scoped }) as Arc<dyn Call>)
})
.collect();
Arc::new(Self { comp: Arc::new(comp), wasm, ledger, calls, did, defaults, auth, resolve, replay: Replay::default(), max_per_caller: MAX_PER_CALLER, max_launches: MAX_LAUNCHES, shutdown_wait: SHUTDOWN_WAIT, launches: Mutex::default() })
}
// One root run of composition.json's agents instance; the run's error is Err.
async fn run(&self, entry: &str, mut input: Map<String, Value>, launched: Option<oneshot::Sender<Value>>) -> Result<Value, String> {
input.insert("did".into(), json!(self.did));
let agents: Arc<dyn Call> = Arc::new(Launched { inner: self.calls["agents"].clone(), tx: Mutex::new(launched) });
let (calls, comp) = (self.calls.clone(), self.comp.clone());
let call_for: CallFor = Arc::new(move |name: &str| if name == "agents" { Some(agents.clone()) } else { calls.get(name).cloned() });
let ran = compose::run_compose(comp, self.wasm.clone(), self.ledger.clone(), "agents", entry, input, call_for, HashMap::new(), None, Options::default()).await?;
match ran.get("error").and_then(Value::as_str) {
Some(e) => Err(e.to_string()),
None => Ok(ran["result"].clone()),
}
}
async fn launch(self: Arc<Self>, caller: &str, body: &Value) -> Result<Value, String> {
let mut input = self.defaults.clone();
input.insert("caller".into(), json!(caller));
for key in CLIENT_FIELDS {
if let Some(v) = body.get(key).filter(|v| !v.is_null()) {
input.insert(key.into(), v.clone());
}
}
let exp_s = exp_s_of(body);
if let Some(exp_s) = exp_s {
input.insert("exp_s".into(), json!(exp_s));
}
let (done_tx, done_rx) = watch::channel(false);
// A slot for this launch, taken before any record so concurrent launches cannot pass the cap.
let slot = format!("slot {}", uuid());
{
let mut launches = self.launches.lock().unwrap();
if launches.values().filter(|l| l.caller == caller).count() >= self.max_per_caller {
return Err(format!("429 RateLimitExceeded: at most {} launches at a time; shut one down first", self.max_per_caller));
}
if launches.len() >= self.max_launches {
return Err(format!("429 RateLimitExceeded: the service runs at most {} launches at a time; try again later", self.max_launches));
}
launches.insert(slot.clone(), Launch { caller: caller.into(), done: done_rx });
}
let (tx, rx) = oneshot::channel();
let me = self.clone();
let task = tokio::spawn(async move {
let ran = me.run("launch", input, Some(tx)).await;
let _ = done_tx.send(true);
if let Err(e) = &ran {
eprintln!("agents-service: launch: {e}");
}
ran
});
let Ok(created) = rx.await else {
// No launch record: the run ended before it (bad input).
self.launches.lock().unwrap().remove(&slot);
return Err(client_error(task.await.map_err(|e| e.to_string())?.err().unwrap_or_else(|| "the launch made no record".into())));
};
let uri = created["uri"].as_str().unwrap_or_default().to_string();
let rkey = uri.rsplit('/').next().unwrap_or_default().to_string();
eprintln!("agents-service: launching {uri}");
{
let mut launches = self.launches.lock().unwrap();
if let Some(launch) = launches.remove(&slot) {
launches.insert(rkey.clone(), launch);
}
}
// A launch that did not get to running tore itself down: it no longer counts. One that runs
// is shut down when its agent token expires.
let me = self.clone();
let (who, key) = (caller.to_string(), rkey.clone());
let lifetime = Duration::from_secs(exp_s.unwrap_or(DEFAULT_EXP_S));
tokio::spawn(async move {
let started = tokio::time::Instant::now();
if !matches!(task.await, Ok(Ok(v)) if v["status"] == "running") {
me.launches.lock().unwrap().remove(&key);
return;
}
tokio::time::sleep_until(started + lifetime).await;
me.reap(&who, &key).await;
});
Ok(json!({ "rkey": rkey, "uri": uri, "status": "launching" }))
}
// Shuts a launch down at the end of its life, unless it already was.
async fn reap(self: Arc<Self>, caller: &str, rkey: &str) {
if !self.launches.lock().unwrap().contains_key(rkey) {
return;
}
eprintln!("agents-service: {rkey} reached its lifetime; shutting down");
if let Err(e) = self.clone().shutdown(caller, &json!({ "rkey": rkey })).await {
eprintln!("agents-service: shutting down {rkey}: {e}");
}
}
async fn shutdown(self: Arc<Self>, caller: &str, body: &Value) -> Result<Value, String> {
let rkey = body["rkey"].as_str().filter(|r| !r.is_empty()).ok_or("400 InvalidRequest: rkey is required")?.to_string();
// Another caller's launch in this process is refused at once (the module checks the record
// for the rest); one still starting finishes first, so its VM is in the record.
let running = match self.launches.lock().unwrap().get(&rkey) {
Some(l) if l.caller != caller => return Err(format!("403 Forbidden: launch {rkey} belongs to another caller")),
Some(l) => Some(l.done.clone()),
None => None,
};
let input = Map::from_iter([("rkey".into(), json!(rkey)), ("delete".into(), json!(body["delete"] == true)), ("caller".into(), json!(caller))]);
let me = self.clone();
let key = rkey.clone();
let task = tokio::spawn(async move {
if let Some(mut done) = running {
let _ = done.wait_for(|d| *d).await;
}
let out = me.run("shutdown", input, None).await;
if out.is_ok() {
me.launches.lock().unwrap().remove(&key);
}
out
});
match tokio::time::timeout(self.shutdown_wait, task).await {
Ok(joined) => joined.map_err(|e| e.to_string())?.map_err(client_error),
Err(_) => Ok(json!({ "rkey": rkey, "status": "stopping" })),
}
}
/// One forwarded request {method, params, body, authorization}: errors are "<status> <Name>: <msg>".
pub async fn handle(self: Arc<Self>, nsid: &str, request: Value) -> Result<Value, String> {
let iss = verify(&self.auth, request["authorization"].as_str(), nsid, now_ms() / 1000, &self.resolve, &self.replay)
.await
.map_err(|e| format!("401 AuthenticationRequired: {e}"))?;
let method = request["method"].as_str().unwrap_or_default();
let (want, ok) = if nsid == GET_AGENT { ("GET", method == "GET") } else { ("POST", method == "POST") };
if !ok {
return Err(format!("405 InvalidRequest: {nsid} takes {want}"));
}
eprintln!("agents-service: {nsid} by {iss}");
match nsid {
LAUNCH_AGENT => self.launch(&iss, &request["body"]).await,
SHUTDOWN_AGENT => self.shutdown(&iss, &request["body"]).await,
GET_AGENT => {
let rkey = request["params"]["rkey"].as_str().filter(|r| !r.is_empty()).ok_or("400 InvalidRequest: rkey is required")?;
self.run("status", Map::from_iter([("rkey".into(), json!(rkey)), ("caller".into(), json!(iss))]), None).await.map_err(client_error)
}
_ => Err(format!("404 MethodNotImplemented: {nsid}")),
}
}
pub fn handlers(self: &Arc<Self>) -> HashMap<String, Handler> {
[("launchAgent", LAUNCH_AGENT), ("getAgent", GET_AGENT), ("shutdownAgent", SHUTDOWN_AGENT)]
.into_iter()
.map(|(name, nsid)| {
let me = self.clone();
let handler: Handler = Arc::new(move |input: Value, _info: Info| {
let me = me.clone();
Box::pin(async move { me.handle(nsid, input).await })
});
(name.to_string(), handler)
})
.collect()
}
}
// ---- main ----------------------------------------------------------------------------------
fn env(key: &str) -> Option<String> {
std::env::var(key).ok().filter(|v| !v.trim().is_empty())
}
const USAGE: &str = "usage: agents-service --app-password <handle> (--password-file <path|-> | --password-env <NAME>) [--pds <url>]";
/// The operator account: --app-password <handle> with exactly one password source.
// No Debug: the password is never printed.
struct Operator {
handle: String,
password: String,
pds: String,
}
impl Operator {
/// `read(path)` reads a password file (`-` = stdin); `var(name)` reads the named env var.
fn parse(args: &[String], read: impl Fn(&str) -> Result<String, String>, var: impl Fn(&str) -> Option<String>) -> Result<Operator, String> {
let (mut handle, mut file, mut name, mut pds) = (None, None, None, None);
let mut it = args.iter();
while let Some(flag) = it.next() {
let slot = match flag.as_str() {
"--app-password" => &mut handle,
"--password-file" => &mut file,
"--password-env" => &mut name,
"--pds" => &mut pds,
_ => return Err(format!("unknown argument {flag}\n{USAGE}")),
};
let value = it.next().filter(|v| !v.is_empty()).ok_or(format!("{flag} needs a value\n{USAGE}"))?;
if slot.replace(value.clone()).is_some() {
return Err(format!("{flag} given twice\n{USAGE}"));
}
}
let handle = handle.ok_or(format!("--app-password <handle> is required (the operator account)\n{USAGE}"))?;
let password = match (file, name) {
(Some(path), None) => read(&path).map_err(|e| format!("--password-file {path}: {e}"))?.lines().next().unwrap_or("").trim().to_string(),
(None, Some(name)) => var(&name).ok_or(format!("--password-env {name}: the variable is unset"))?.trim().to_string(),
_ => return Err(format!("--app-password needs exactly one of --password-file / --password-env\n{USAGE}")),
};
if password.is_empty() {
return Err("the app password is empty".into());
}
Ok(Operator { handle, password, pds: pds.unwrap_or_else(|| "https://bsky.social".into()) })
}
}
fn read_password(path: &str) -> Result<String, String> {
if path == "-" {
let mut line = String::new();
std::io::stdin().read_line(&mut line).map_err(|e| e.to_string())?;
return Ok(line);
}
std::fs::read_to_string(path).map_err(|e| e.to_string())
}
async fn serve(operator: Operator, signal: Signal) -> Result<Value, String> {
let need = |key: &str| env(key).ok_or(format!("{key} is required"));
let repo = env("ATLIBS_REPO").map(PathBuf::from).unwrap_or_else(|| Path::new(env!("CARGO_MANIFEST_DIR")).join(".."));
let state_dir = env("ATLIBS_SERVER_STATE").map(PathBuf::from).unwrap_or_else(|| PathBuf::from(std::env::var("HOME").unwrap_or_default()).join(".atlibs-server"));
let relay = env("ATLIBS_RELAY").unwrap_or_else(|| "https://xrpc.fedproxy.com".into());
let issuers: Vec<String> = env("ATLIBS_AGENTS_ISSUERS").unwrap_or_default().split([' ', ',']).filter(|s| s.starts_with("did:")).map(String::from).collect();
let user_data = std::fs::read_to_string(need("ATLIBS_AGENTS_USER_DATA")?).map_err(|e| format!("ATLIBS_AGENTS_USER_DATA: {e}"))?;
let store = Store::open(&state_dir, "atlibs-server")?;
let client = reqwest::Client::builder().user_agent("atlibs-agents-service").build().map_err(|e| e.to_string())?;
let fetch: Arc<dyn Fetch> = Arc::new(client.clone());
let session = Session::login(&operator.handle, Some(operator.password), &operator.pds, fetch.clone(), Some(store.clone())).await?;
let iroh_key = env("ATLIBS_IROH_KEY").map(PathBuf::from).unwrap_or_else(|| state_dir.join("iroh_ed25519"));
let account: Arc<dyn Call> = Arc::new(HostCall { fetch: fetch.clone(), session: Some(session.clone()), attest: Some(store.clone()) });
let transports = vec![atlibs_iroh::transport(iroh_key)];
let provides = atlibs_host::transport::provides(&transports);
// The service's did:key: the ingress identity and the aud callers mint service auth for.
let serve_path = repo.join("modules/agents/serve.json");
let key_id = format!("{}#ingress", std::fs::canonicalize(&serve_path).unwrap_or(serve_path).display());
let key = store.didkey(&key_id).await?;
let did_key = key.did.clone();
let host = ingress_host(&did_key, &relay);
eprintln!("agents-service for {} at https://{host}/xrpc/{{{LAUNCH_AGENT},{GET_AGENT},{SHUTDOWN_AGENT}}} (service DID {did_key})", session.did());
let ingress: Arc<dyn Call> = Arc::new(DidkeyCall { key, services: vec![relay.clone()], fetch: fetch.clone(), inner: Arc::new(HostCall { fetch, session: None, attest: None }), store: Some(store.clone()) });
let mut defaults = Map::from_iter([("user_data".into(), json!(user_data)), ("via".into(), json!("iroh"))]);
if let Some(bidders) = env("ATLIBS_AGENTS_BIDDERS") {
defaults.insert("bidders".into(), json!(bidders.split([' ', ',']).filter(|s| !s.is_empty()).collect::<Vec<_>>()));
}
let wasm = Wasm::new(WALL_TIME_MS);
let ledger = Some(store.ledger("managed_records"));
let auth = Auth { audiences: vec![did_key.clone(), format!("did:web:{host}")], issuers };
let mut service = Service::new(load(&repo, "composition.json", &provides).await?, wasm.clone(), ledger.clone(), account, transports, session.did(), defaults, auth, resolver(client));
if let Some(n) = env("ATLIBS_AGENTS_MAX_PER_CALLER") {
Arc::get_mut(&mut service).unwrap().max_per_caller = n.parse().map_err(|_| "ATLIBS_AGENTS_MAX_PER_CALLER must be a number")?;
}
if let Some(n) = env("ATLIBS_AGENTS_MAX_LAUNCHES") {
Arc::get_mut(&mut service).unwrap().max_launches = n.parse().map_err(|_| "ATLIBS_AGENTS_MAX_LAUNCHES must be a number")?;
}
let mut input = Map::from_iter([("relay".into(), json!(relay))]);
if let Some(ms) = env("ATLIBS_AGENTS_LIFETIME_MS") {
input.insert("lifetimeMs".into(), json!(ms));
}
let call_for: CallFor = Arc::new(move |name: &str| (name == "ingress").then(|| ingress.clone()));
let serve = Arc::new(load(&repo, "serve.json", &provides).await?);
compose::run_compose(serve, wasm, ledger, "ingress", "serve", input, call_for, service.handlers(), None, Options { signal: Some(signal), ..Default::default() }).await
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
if let Err(e) = atlibs_host::wasm::concurrency() {
eprintln!("{e}");
std::process::exit(1);
}
let args: Vec<String> = std::env::args().skip(1).collect();
let operator = match Operator::parse(&args, read_password, |name| std::env::var(name).ok()) {
Ok(operator) => operator,
Err(e) => {
eprintln!("{e}");
std::process::exit(2);
}
};
let signal = Signal::new();
let stop = signal.clone();
tokio::spawn(async move {
if tokio::signal::ctrl_c().await.is_ok() {
stop.abort("aborted");
}
});
match serve(operator, signal).await {
Ok(ran) => println!("{}", pretty(&ran)),
Err(e) => {
eprintln!("{e}");
std::process::exit(1);
}
}
}
#[cfg(test)]
mod tests;