Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990"""Opt-in: a real ssh target end to end. Needs key auth and python3 remotely.
python3 test/manual/remote_ssh.py [user@host]
Exercises staging, kernel boot, tool calls, relay, degradation is NOT testedhere (that needs a broken target), and control-master reuse across calls."""
import asyncioimport osimport sysfrom pathlib import Path
ROOT = Path(__file__).resolve().parents[2]sys.path.insert(0, str(ROOT / "priv" / "python"))from albedo_api import PythonApi # noqa: E402from albedo_plugins import remote # noqa: E402
class FakeCapture: def __init__(self, id): self.id, self.seen, self.tail_data = id, 0, bytearray() self.raw_data = bytearray() self.raw_seen = 0 self.data = bytearray()
def write(self, text): data = text.encode() self.data.extend(data) self.seen += len(data) self.tail_data.extend(data)
def read(self, offset: int = 0, limit: int = 4000) -> str: return bytes(self.data[offset : offset + limit]).decode(errors="ignore")
async def main(host: str) -> int: loop = asyncio.get_running_loop()
async def daemon_host(method, args): return [{"relay": True, "method": method}]
api = PythonApi( loop, daemon_host, RuntimeError, FakeCapture, 65536, lambda event: None, lambda close: None, lambda cls: None, 2, ["run", "files", "work", "skills", "remote"], ) remote.setup(api) rem_mod = remote.Remote() rem = await rem_mod.connect(host=host) print("connected:", repr(rem)) tools = await rem.tools() print("remote tools:", tools["names"])
# the local-run shape: sync call, sync state, one await for completion job = rem.run("uname", "-s") print("mid-run tail:", repr(job.tail()), "poll:", job.poll()) await job print("run:", job.tail().strip(), "exit:", job.poll())
listing = await rem.files.ls(".") print("files.ls entries:", len(listing)) relay = await rem.work.list() print("relay answer:", relay[0])
import time
started = time.monotonic() again = rem.run("true") await again print(f"second call over the control socket: {time.monotonic() - started:.2f}s")
await rem.close() print("closed:", rem.closed) return 0
if __name__ == "__main__": target = ( sys.argv[1] if len(sys.argv) > 1 else os.environ.get("ALBEDO_SSH", "localhost") ) raise SystemExit(asyncio.run(main(target)))