commit 65fc75e50918138916feebe14c51c62335c7eab1 Author: Shivering Ocelot Date: Thu Sep 3 19:43:46 2026 -0500 initial spec diff --git a/design.md b/design.md new file mode 100644 index 0000000..501f16d --- /dev/null +++ b/design.md @@ -0,0 +1,37 @@ +# Remote Center + +`remote-center` is a slim application, easily runs in any system. It connects to a configurable horde host, can be given an optional name (defaults to "remote-center") and a working path. +The application connects to the hub api reliably, connects it's main remote, and exposes all it's functions. +It's not in the hub repository. Users can either install `remote-center` and launch it with any working path, or clone the remote center repository and run it with itself to serve as the user's working copy. + +A **Space** is a "sub-remote" of the remote center. They are named, and have an assigned working subpath of the center's working path. +This is the structure of all remote types: +- Center: has it's own distinct set of functions, working directory is not configurable, configurable name with default +- Space: subdirectory of the center's working path, configurable and required name + - spaces share functions that can be defined in the center's space functions + - every space can define their own unique functions + +Functions are declared by writing them in expected file directories, which the center scans. +All spaces connect automatically to the hub when the center starts up, and are always live, until deleted. +Functions are user-written, directly executable from the client, and stored in the remote directories. They are executed with the fewest layer's of abstraction possible, even by the package manager itself if it's possible (but also remember the application must be small, use the simplest and most robust method available, not necessarily the most secured). + +All configuration lives in the '.remote' subdirectory, for all remotes. Functions specifically live in .remote as well, but they are meant to be relatively small and contained, working closely with the remote-center's api. For less limited environment, functions can execute scripts or read any data from the remote's working directory. +Because functions live as scripts in the remote's directories, the initial space functions are templates, not remote source code. This also means that new remote centers must be installed, by copying an initial '.remote' from a template. +These are the initial functions for all remote types: +- Center functions: create_space(name), clone_space(space_name or git_url, optional new space name) +- Space functions: basic file operations, create and delete local functions (no execution command) + +## Implementation + +Application uses `deno` with only 2 commands: 'init' to copy the template directory into the local remote, and 'start' to start in the current directory or a given one. +File structure: +- src/: main.ts (cli, config), runtime.ts (remotes, functions), horde.ts (backend hub connection) +- templates/functions/: create_space.ts and clone_space.ts +- templates/space_functions/: all space functions +- spaces/: untracked directory where spaces are stored +- readme.md +- .remote/config.json: holds host, token, name +- .remote/functions/: every function is a module in this directory, loaded by deno +Functions are loaded as modules, and the .remote/functions directory is directly watched to track function updates, falling back to a scanning loop every 5 seconds. +Function modules export `spec` with the metadata of the function, and a default function that takes `args` and returns a json value. +The 'HORDE_HOST' is used if present, overriding any existing config, and defaults to 'wss://hub.hordelab.com' if not defined at all. \ No newline at end of file diff --git a/reference/backend.js b/reference/backend.js new file mode 100644 index 0000000..0486e2f --- /dev/null +++ b/reference/backend.js @@ -0,0 +1,498 @@ + +import * as ui from './ui'; + +export default class Stream { + constructor (id) { + this.id = id; + this.partialContent = ""; + this.eof = false; + + this.listeners = { + data: [], + end: [], + }; + + let onResponse = (msg) => { + if (this.eof) return; + + if (msg.eof) { + this.eof = true; + this.handle('end'); + } else { + this.partialContent += msg.data; + this.handle('data', msg.data); + } + }; + + this.contentPromise = new Promise((resolve, reject) => { + this.handle("end", () => { + resolve(this.partialContent); + }); + }); + } + + feed (chunk) { + if (this.eof) return; + this.partialContent += chunk; + this.handle('data', chunk); + } + + close () { + this.eof = true; + this.handle('end', this.partialContent); + } + + getContent () { + return this.contentPromise; + } + + handle (event, ...args) { + for (let fn of this.listeners[event]) { + fn(...args) + } + } + + on (event, callback) { + this.listeners[event].push(callback); + + if (event == "data" && this.partialContent) { + callback(this.partialContent); + } + if (event == "end" && this.eof) { + callback(); + } + } +} + +export class ChatInfo { + constructor({ id, name }) { + this.id = id; + this.name = name; + } +} + +// Basic objects and configuration +let socket = null; +let backendUrl = null; + +// State +let nextNonce = 0; +let reconnectTimeout = null; // For managing the reconnect timer +let cache = { + chats: null, + agents: null, +}; + +// Auth +let loginToken = null; +let loginTokenDeferred = createDeferred(); +let authDeferred = createDeferred(); +let credentialsDeferred = createDeferred(); + +// Handlers +const pendingActions = {}; // Maps nonce to { resolve, reject } +const chatListeners = {}; // Maps chatId to an array of callback functions +const actionListeners = {}; // Listeners for other arbitrary actions +const orphanHandlers = []; // Queue of callbacks for orphan server messages +const connectListeners = []; // Listeners for horde connection + +function createDeferred() { + let resolve, reject; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return {promise, resolve, reject}; +} + +function createDeferredStream(callback) { + let resolve, reject; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + let deferred = {promise, resolve, reject}; + return {promise, resolve (value) { callback(value, deferred) }, reject} +} + +function initState () { + backendUrl = window.location.origin; + const tk = localStorage.getItem('loginToken'); + if (tk) { + resolveToken(tk); + + // Clear token if it doesn't work + authDeferred.promise.then(success => { + console.log(`initState.authDeferred; success=${success}`); + if (!success) { + localStorage.removeItem('loginToken'); + } + }); + } else { + authDeferred.resolve(false); + } + connect(); +} + +initState(); + + +// --- Authentication --- + +// Passes the token to everyone who needs it, and returns it +function resolveToken (tk) { + loginToken = tk; + loginTokenDeferred.resolve(tk); + return tk; +} + +// Waits for the result of the last authentication attempt +export async function authResult() { + // No token has been set, login or signup have to be called + if (!loginToken) return false; + return await authDeferred.promise; +} + +// Creates an account and returns a login token +export async function signup(username, email, password) { + try { + const response = await postRequest('/user/signup', { + username, + email, + password, + }); + // Reset authentication status + authDeferred = createDeferred(); + resolveToken(response.token); + return await authResult(); + } catch (error) { + ui.alert('Signup error: ' + (error.message || String(error))); + return false; + } +} + +// Returns a login token for the given credentials +export async function login(username, password) { + try { + credentialsDeferred.resolve({username, password}); + const response = await postRequest('/user/login', { + username, + password, + }); + // Reset authentication status + authDeferred = createDeferred(); + resolveToken(response.token); + return await authResult(); + } catch (error) { + ui.alert('Login error: ' + (error.message || String(error))); + return false; + } +} + +export async function newToken () { + const response = await postRequest('/user/login', { + token: loginToken || await loginTokenDeferred.promise, + }); + return response.token; +} + + +// --- Communication --- + +export function sendAction(action, body = {}, callback = null) { + if (socket?.readyState !== WebSocket.OPEN) { + return Promise.reject(new Error("WebSocket is not open. Cannot send action.")); + } + const nonce = nextNonce++; + const deferred = callback ? createDeferredStream(callback) : createDeferred(); + + pendingActions[nonce] = deferred; + deferred.promise.finally(() => { + delete pendingActions[nonce]; // Clean up + }); + + socket.send(JSON.stringify({ + action, + nonce, + body, + })); + return deferred.promise; +} + +export function ping() { + if (socket && socket.readyState === WebSocket.OPEN) { + const data = JSON.stringify({ path: 'ping' }); + socket.send(data); + console.log('Ping sent:', data); + } else { + console.warn('Cannot send ping: WebSocket is not open.'); + } +} + +function handleIncomingMessage(rawData) { + let packet; + try { + packet = JSON.parse(rawData); + } catch (e) { + console.error("Failed to parse incoming message as JSON:", rawData, e); + return; + } + + if (packet.nonce != null) { + const nonce = packet.nonce; + const actionPromise = pendingActions[nonce]; + if (actionPromise) { + if (packet.error) { + actionPromise.reject(new Error(packet.error)); + } else { + actionPromise.resolve(packet); + } + } else { + console.warn(`Orphan response for action nonce ${nonce}. No pending action found.`); + } + } else if (packet.action === 'chat-message') { + // This can be in a 'chat-message' listener in the future + const chatId = packet.chatId; + const listeners = chatListeners[chatId]; + if (listeners) { + listeners.forEach(callback => { + try { + callback(packet); + } catch (e) { + console.error(`Error in chat message listener for chat ${chatId}:`, e); + } + }); + } else { + console.log(`No active listeners for chatId ${chatId}`); + } + } else if (packet.action) { + let listeners = actionListeners[packet.action]; + if (listeners && listeners.length) { + for (let f of listeners) { + f(packet); + } + } + } else if (orphanHandlers.length) { + const f = orphanHandlers.shift(); + f(packet); + } else if (packet.error) { + console.error(`Backend Error: ${packet.error}`); + } else { + console.warn('ignored server message', packet); + } +} + +async function postRequest(path, body) { + const response = await fetch(`${backendUrl}${path}`, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + }, + body: JSON.stringify(body), + }); + + if (!response.ok) { + const errorText = await response.text(); + throw new Error(`Response Error: ${response.status}\n${errorText}`); + } + + const responseText = await response.text(); + const responseData = JSON.parse(responseText); + return responseData; +} + + +// --- Data Management --- + +// Returns a function to unsubscribe +export function onChatMessages(chatId, callback) { + if (typeof callback !== 'function') { + throw new TypeError("chatMessages: callback must be a function."); + } + + if (!chatListeners[chatId]) { + chatListeners[chatId] = []; + } + const listeners = chatListeners[chatId]; + listeners.push(callback); + + // Return an unsubscribe function + return () => { + const index = listeners.indexOf(callback); + if (index > -1) { + listeners.splice(index, 1); + } + if (listeners.length === 0) { + delete chatListeners[chatId]; + } + }; +} + +export function onAction(action, callback) { + if (typeof callback !== 'function') { + throw new TypeError("onAction: callback must be a function."); + } + + let listeners = actionListeners[action]; + if (!listeners) { + listeners = []; + actionListeners[action] = listeners; + } + listeners.push(callback); + + // Return an unsubscribe function + return () => { + const index = listeners.indexOf(callback); + if (index > -1) { + listeners.splice(index, 1); + } + }; +} + +export async function getChats() { + if (cache.chats === null) { + const response = await sendAction("get-chats", {}); + if (response && response.chats) { + cache.chats = response.chats.map(msg => new ChatInfo({ + id: msg.id, + name: msg.name || ``, + })); + } else { + throw new Error("Invalid response for get-chats: " + JSON.stringify(response)); + } + } + return cache.chats; +} + +export async function getChat(chatId) { + const allChats = await getChats(); + return allChats.find(it => it.id == chatId); +} + +export async function getAgents() { + if (cache.agents === null) { + const response = await sendAction("get-agents", {}); + if (response && response.agents) { + cache.agents = response.agents.map(msg => ({ + id: msg.id, + name: msg.name || ``, + agentname: msg.agentname, + description: msg.description, + picture: msg.picture, + kind: msg.kind, + })); + } else { + throw new Error("Invalid response for get-agents: " + JSON.stringify(response)); + } + } + return cache.agents; +} + +export async function getAgent(agentname) { + const allAgents = await getAgents(); + return allAgents.find(it => it.agentname == agentname); +} + +export function readStream(streamId) { + let stream = new Stream(streamId); + sendAction("read-stream", {streamId}, (packet, deferred) => { + if (packet.eof) { + stream.close(); + deferred.resolve(); + } else if (packet.data != null) { + stream.feed(packet.data); + } + }); + return stream; +} + +export function invalidateCache({ chats=false, agents=false } = {}) { + if (chats) cache.chats = null; + if (agents) cache.agents = null; +} + + + +// --- Connection --- + +export function onConnect (callback) { + connectListeners.push(callback); + if (socket && socket.readyState === WebSocket.OPEN) { + callback(); + } + // TODO: return cancel function +} + +function reconnect() { + if (reconnectTimeout) { + clearTimeout(reconnectTimeout); + } + reconnectTimeout = setTimeout(() => { + connect(); + }, 1000); // Wait 1 second before attempting to reconnect +} + +function connect() { + if (socket && (socket.readyState === WebSocket.OPEN || socket.readyState === WebSocket.CONNECTING)) { + console.log('WebSocket is already connected or connecting.'); + return; + } + + socket = new WebSocket(`${backendUrl}/ws`); + + socket.onopen = async (event) => { + console.log('Connected to the WebSocket server'); + invalidateCache({ chat: true }); + + console.log('Attempting to log in...'); + try { + const tk = loginToken || await loginTokenDeferred.promise; + socket.send(JSON.stringify({ token: tk })); + + // Wait for the login response as an orphan server message + const response = await new Promise(resolve => { + orphanHandlers.push(resolve); + }); + + if (response && !response.error) { + // Successful connection, store token long term + authDeferred.resolve(true); + localStorage.setItem('loginToken', loginToken); + + for (let cb of connectListeners) cb(); + } else { + authDeferred.resolve(false); + console.error("Socket authentication failed:", response ? response.error : "Unknown error"); + socket.close(); // Close to trigger reconnect and retry login + } + } catch (error) { + console.error("Error during WebSocket login:", error); + authDeferred.resolve(false); + socket.close(); // Close to trigger reconnect + } + }; + + socket.onerror = (event) => { + console.error('WebSocket Error:', event); + reconnect(); + }; + + socket.onclose = (event) => { + console.log('Disconnected. Attempting to reconnect...', event.reason); + reconnect(); + }; + + socket.onmessage = (event) => { + handleIncomingMessage(event.data); + }; +} + +function disconnect() { + if (socket) { + socket.close(1000, "Client initiated disconnect"); + socket = null; + if (reconnectTimeout) { + clearTimeout(reconnectTimeout); // Stop any pending reconnects + } + } +} diff --git a/reference/horde.rs b/reference/horde.rs new file mode 100644 index 0000000..ca4fd7b --- /dev/null +++ b/reference/horde.rs @@ -0,0 +1,424 @@ +use futures::prelude::*; +use smol::channel; +use smol::lock::{Mutex, RwLock, OnceCell}; +use smol::net::TcpStream; +use std::collections::HashMap; +use std::sync::Arc; + +use async_tungstenite::tungstenite::Error; +use async_tungstenite::{async_tls::client_async_tls, tungstenite::protocol::Message}; +use tinyjson::JsonValue; + +use crate::JsonObject; + +#[derive(Debug, Clone)] +struct Incoming { + nonce: Option, + value: JsonValue, +} + +struct NonceInfo { + queue: Vec, + listener: Option, +} + +#[derive(Debug, Clone)] +enum Credential { + Empty, + Token(String), + User{username: String, password: String}, +} + +/// Central connection controller +struct Controller { + nonce: std::sync::atomic::AtomicU32, + sender: channel::Sender, + nonce_data: Mutex>, + credential: RwLock, +} + +/// Shareable Handle to a Horde connection +#[derive(Clone)] +pub struct Horde { + controller: Arc, +} + +pub struct Request<'a> { + nonce: u32, + horde: &'a Horde, + action: &'a str, + body: JsonObject, +} + +enum RequestListener { + Value(channel::Sender), + Stream(channel::Sender), +} + +impl RequestListener { + // Creates a Value variant, returning the receiver side + fn value() -> (Self, channel::Receiver) { + let (sender, receiver) = channel::unbounded(); + (RequestListener::Value(sender), receiver) + } + + // Creates a Stream variant, returning the receiver side + fn stream() -> (Self, channel::Receiver) { + let (sender, receiver) = channel::unbounded(); + (RequestListener::Stream(sender), receiver) + } + + // Sends data to the listener. In case of a value listener, also takes it from the option. + // Returns whether the option had a stream to feed. + async fn feed(opt_self: &mut Option, value: JsonValue) { + match opt_self { + Some(RequestListener::Stream(sender)) => { + sender.send(value).await.unwrap(); + } + Some(RequestListener::Value(sender)) => { + sender.send(value).await.unwrap(); + //*opt = None; + } + None => panic!("Cannot feed to None"), + } + } +} + +type SinkBox = Box + Unpin + Send>; +enum ConnectionWriterState { + Active(SinkBox), + Inactive(Arc>), +} + +struct ConnectionWriter(Mutex); + +impl ConnectionWriter { + async fn send (&self, msg: Message) { + let mut guard = self.0.lock().await; + match &mut*guard { + ConnectionWriterState::Active(sender) => { + // I have to clone it :( + if let Err(_) = sender.send(msg.clone()).await { + // Make the connection inactive + let cell_arc = Arc::new(OnceCell::new()); + *guard = ConnectionWriterState::Inactive(cell_arc.clone()); + + // Drop the reference and wait for it to become active + drop(guard); + cell_arc.wait().await; + + // Recursively try again + self.send(msg); + } + }, + ConnectionWriterState::Inactive(..) => { + unreachable!() + } + } + } + + async fn replace (&self, sender: SinkBox) { + let mut guard = self.0.lock().await; + let previous = std::mem::replace(&mut*guard, ConnectionWriterState::Active(sender)); + + // Signal the writer that the connection was replaced + if let ConnectionWriterState::Inactive(cell) = previous { + let _ = cell.set(()).await; // Signal the writer that the connection was replaced + } + } +} + +impl Request<'_> { + pub fn add_field(&mut self, key: &str, value: impl Into) { + self.body.insert(key.to_owned(), value.into()); + } + + /// Sends the request and returns its nonce for follow up + async fn send_only(self) { + let body = self.body; + let json_payload: JsonValue = JsonValue::from(HashMap::from([ + ("nonce".to_owned(), (self.nonce as f64).into()), + ("action".to_owned(), JsonValue::from(self.action.to_owned())), + ("body".to_owned(), JsonValue::from(body)), + ])); + + let text_payload = json_payload.stringify().unwrap(); + self.horde + .controller + .sender + .send(Message::Text(text_payload)) + .await + .unwrap(); + } + + pub async fn send_recv_value(self) -> JsonObject { + let horde = self.horde; + + let (listener, recv) = RequestListener::value(); + horde.listen(self.nonce, listener).await; + + self.send_only().await; + + recv.recv().await.unwrap().try_into().unwrap() + } + + pub async fn send_recv_stream(self) -> impl smol::stream::Stream { + let horde = self.horde; + + let (listener, recv) = RequestListener::stream(); + horde.listen(self.nonce, listener).await; + + self.send_only().await; + recv + } + + pub async fn send_forget(self) { + self.send_only().await; + } +} + +impl Horde { + pub async fn connect(host: &str) -> Result { + + // Format hosts for each connection + let ws_host = format!("{}/ws", host); + let tcp_host = if host.starts_with("ws://") { + host.split_at_checked(5).unwrap().1 + } else if host.starts_with("wss://") { + host.split_at_checked(6).unwrap().1 + } else { + &host + }.to_owned(); + + let tcp_stream = TcpStream::connect(&tcp_host).await.unwrap(); + let (ws_stream, _) = client_async_tls(&ws_host, tcp_stream).await.unwrap(); + let (ws_sender, mut ws_receiver) = ws_stream.split(); + + let connection_out = Arc::new(ConnectionWriter(Mutex::new( + ConnectionWriterState::Active(Box::new(ws_sender)) + ))); + + let (sender, receiver) = channel::unbounded(); + + let controller = Arc::new(Controller { + nonce: 0.into(), + sender, + nonce_data: Mutex::new(HashMap::new()), + credential: RwLock::new(Credential::Empty), + }); + + let horde = Horde { controller: controller.clone() }; + + // Forward received messages to the websocket connection + let out_clone = connection_out.clone(); + smol::spawn(async move { + while let Ok(msg) = receiver.recv().await { + out_clone.send(msg).await; + } + }) + .detach(); + + let horde_clone = horde.clone(); + smol::spawn(async move { + loop { + { + // Cannot read from a stream unless it is pinned + // Pin it only in the context where it's receiving + smol::pin!(ws_receiver); + + while let Some(Ok(msg)) = ws_receiver.next().await { + match msg { + Message::Text(text) => { + if let Some(incoming) = Horde::parse_incoming(&text) { + let mut listeners = controller.nonce_data.lock().await; + + if let Some(nonce) = incoming.nonce { + if let Some(info) = listeners.get_mut(&nonce) { + if info.listener.is_some() { + RequestListener::feed(&mut info.listener, incoming.value).await + } else { + info.queue.push(incoming.value); + } + } else { + listeners.insert( + nonce, + NonceInfo { + queue: vec![incoming.value], + listener: None, + }, + ); + } + } else { + // Message without nonce + } + } else { + println!("Bad message: {}", text); + } + }, + _ => { + println!("Other Message: {:?}", msg); + } + } + } + } + + let tcp_stream = TcpStream::connect(&tcp_host).await.unwrap(); + let (ws_stream, _) = client_async_tls(&ws_host, tcp_stream).await.unwrap(); + let (ws_sender, mut new_receiver) = ws_stream.split(); + + ws_receiver = new_receiver; + connection_out.replace(Box::new(ws_sender)).await; + + horde.auth(horde.controller.credential.read().await.clone()).await; + } + }) + .detach(); + + Ok(horde_clone) + } + pub async fn with_token(host: &str, token: String) -> Result { + let horde = Self::connect(host).await?; + let credential = Credential::Token(token); + horde.set_credential(credential.clone()).await; + horde.auth(credential).await; + Ok(horde) + } + pub async fn connect_user(host: &str, username: String, password: String) -> Result { + let horde = Self::connect(host).await?; + let credential = Credential::User{username, password}; + horde.set_credential(credential.clone()).await; + horde.auth(credential).await; + Ok(horde) + } + + async fn set_credential (&self, credential: Credential) { + let mut writer = self.controller.credential.write().await; + *writer = credential.clone(); + } + + async fn auth (&self, credential: Credential) { + match credential { + Credential::Token(token) => { + let req_obj: JsonValue = JsonObject::from([ + ("token".to_owned(), token.into()), + ]).into(); + + let text_payload = req_obj.stringify().unwrap(); + self.controller + .sender + .send(Message::Text(text_payload)) + .await + .unwrap(); + }, + Credential::User{username, password} => { + let req_obj: JsonValue = JsonObject::from([ + ("username".to_owned(), username.into()), + ("password".to_owned(), password.into()), + ]).into(); + + let text_payload = req_obj.stringify().unwrap(); + self.controller + .sender + .send(Message::Text(text_payload)) + .await + .unwrap(); + }, + Credential::Empty => { + println!("Empty credentials"); + }, + } + } + + pub fn request<'a>(&'a self, action: &'a str) -> Request<'a> { + Request { + nonce: self + .controller + .nonce + .fetch_add(1, std::sync::atomic::Ordering::SeqCst), + horde: &self, + action, + body: JsonObject::new(), + } + } + + fn parse_incoming(message: &str) -> Option { + let json_parsed: JsonValue = message.parse().ok()?; + + let mut response = json_parsed.get::()?.clone(); + let nonce = response.remove("nonce") + .and_then(|jval| jval.get::().copied()) + .map(|x| x as u32); + + Some(Incoming { + nonce, value: response.into(), + }) + } + + async fn listen(&self, nonce: u32, listener: RequestListener) { + let mut nonce_data = self.controller.nonce_data.lock().await; + + // Check if there are values in the queue for the specified nonce + if let Some(info) = nonce_data.get_mut(&nonce) { + let mut maybe_listener = Some(listener); + + if info.listener.is_some() { + panic!("Multiple listeners for same nonce: {}", nonce); + } + + while !info.queue.is_empty() && maybe_listener.is_some() { + let value = info.queue.remove(0); + RequestListener::feed(&mut maybe_listener, value).await; + } + + info.listener = maybe_listener; + } else { + nonce_data.insert( + nonce, + NonceInfo { + queue: vec![], + listener: Some(listener), + }, + ); + } + } + + pub async fn handle(&self, action: &str, inputs: JsonObject) -> JsonObject { + let mut req = self.request(action); + req.body = inputs; + + let res = req.send_recv_value().await; + //println!("handle response: {:?}", res); + + if let Some(message_value) = res.get("error") { + panic!("Hanlde Error: {:?}", message_value); + } + + return res; + + //res.get("output").unwrap().get::().unwrap().clone() + } + + pub async fn read_stream(&self, stream_id: JsonValue) -> impl Stream { + let mut request = self.request("read-stream"); + request.add_field("streamId", stream_id); + + let receiver = request.send_recv_stream().await; + + async_stream::stream! { + smol::pin!(receiver); + while let Some(mut packet) = receiver.next().await { + let packet_obj = packet.get_mut::().unwrap(); + + // Detect end of stream + if let Some(value) = packet_obj.get("eof") { + if *value.get().unwrap() { + break; + } + } + + let data = packet_obj.remove("data").unwrap(); + yield String::try_from(data).unwrap(); + } + } + } +} diff --git a/reference/remotes.md b/reference/remotes.md new file mode 100644 index 0000000..2b61760 --- /dev/null +++ b/reference/remotes.md @@ -0,0 +1,87 @@ +# Remotes + +## Overview + +A **remote** is a collection of functions exposed to an AI agent through the WebSocket session. Remotes are scoped to a user and live in memory during the server's lifetime. + +## Architecture & Function Lifecycle + +Remote functions are not standard Python methods attached to the agent. Instead, they are definitions (schemas) wrapped into adapters that fit the expert agent's tool loop. + +1. **Definition (`backend/src/remote.py`)**: + * A **`RemoteFunction`** represents a function exposed by a remote. It contains the function's name, description, and its OpenAI-compatible schema for input parameters. + * A **`RemoteToolSpec`** wraps a `RemoteFunction` to make it compatible with the agent's tool system. It exposes the required interface: `.name`, `.description`, `.to_openai_dict()` (for the LLM), and `.execute_async(**args)` (for execution). + * A **`RemoteConnection`** manages the connection state. It holds the function definitions and an internal queue of pending calls. + +2. **Injection**: + * When a run starts (see **Agent Integration** below), the expert agent requests tools for specific remotes. The server resolves these remote names into `RemoteToolSpec` objects and adds them to the run's input tools list. + +3. **Execution**: + * When the AI decides to use a remote tool, the expert agent's loop calls `execute_async` on the `RemoteToolSpec`. + * This triggers `RemoteConnection.call()`, which sends a request over the WebSocket to the session that owns the remote. + * The remote executes the function code (for local remotes) or accepts an RPC call (for external remotes). + * Results are sent back through `remote-call-result` actions. Pending calls time out after 30 seconds if no response is received. + +4. **Lifecycle**: + * Remotes are established when a user connects them via the UI (`remote-connect`). + * They persist for the session lifetime unless manually disconnected (`remote-disconnect`) or if the session serving them closes. + * Collisions are handled by scoping connections to `(user_id, name)`. + +## API Reference + +All endpoints are actions sent to the backend via the WebSocket. + +### `remote-connect` +Registers a new remote connection for the current user. + +**Payload**: +* `remotename` (string): The unique name for this remote collection. +* `functions` (array of objects): List of function specs. Each object must contain: + * `name` (string): The function identifier. + * `description` (string): Text visible to the AI. + * `parameters` (object): OpenAI tool parameter schema (`type`, `properties`, `required`). + +### `remote-disconnect` +Removes a connected remote. + +**Payload**: +* `remotename` (string): The name of the remote to remove. + +### `remote-list` +Requests the list of remotes currently registered for the authenticated user. + +**Payload**: (Empty) + +### `remote-call` +Invokes a specific function on a remote. + +**Payload**: +* `remotename` (string): The target remote. +* `function` (string): The function name. +* `args` (object): The arguments to pass to the function, matching the schema defined in `remote-connect`. + +### `remote-call-result` +Sent by the server to the active session to deliver the outcome of a pending `remote-call`. + +**Payload**: +* `requestId` (string): A unique identifier for this specific request. +* `result` (any): The return value of the function. +* `error` (string): An error message if execution failed. + +## Agent Integration + +Remotes are integrated into the agent workflow in two stages: receiving context and executing tools. + +### Chat Integration (`Agent.received_message`) +When a user sends a message to a chat thread, the frontend sends a special system resource to mark which remotes should be available for that conversation. + +* In `backend/src/agent.py`, the `received_message` method scans the thread's resources. +* It looks for items with `subkind: 'remote'`. +* It extracts the content of these resources (the remote names) and passes them into the agent run context. + +### Execution Integration (`ExpertAgent.run`) +In `backend/src/agents/expert.py`, the `run` method prepares for execution. + +* It retrieves the list of remotes from the run input. +* It calls `app.getRemoteTools(user_id, remotes)` to resolve these names into active tool specifications. +* These tools are injected into the inference loop. When the AI selects a tool, the expert agent executes it. If a remote fails (timeout or error), the failure is returned as a string error, allowing the AI to retry or handle the failure naturally. \ No newline at end of file