okami.mom/albedo

git clone https://git.nkp.pet/okami.mom/albedo

nekomimi.petkernel: keep output once, bound snapshots and replays, drop the outbox scanf479db5

main
branchesmainall refs →
2.0 KiB73 linesraw
1"""Opt-in: a real ssh target end to end. Needs key auth and python3 remotely.
2
3    python3 test/manual/remote_ssh.py [user@host]
4
5Exercises staging, kernel boot, tool calls, relay, degradation is NOT tested
6here (that needs a broken target), and control-master reuse across calls.
7"""
8
9import asyncio
10import os
11import sys
12from pathlib import Path
13
14ROOT = Path(__file__).resolve().parents[2]
15sys.path.insert(0, str(ROOT / "priv" / "python"))
16from albedo_api import PythonApi  # noqa: E402
17from albedo_capture import Capture  # noqa: E402
18from albedo_plugins import remote  # noqa: E402
19
20
21async def main(host: str) -> int:
22    loop = asyncio.get_running_loop()
23
24    async def daemon_host(method, args):
25        return [{"relay": True, "method": method}]
26
27    api = PythonApi(
28        loop,
29        daemon_host,
30        RuntimeError,
31        Capture,
32        65536,
33        lambda event: None,
34        lambda close: None,
35        lambda cls: None,
36        2,
37        ["run", "files", "work", "skills", "remote"],
38    )
39    remote.setup(api)
40    rem_mod = remote.Remote()
41    rem = await rem_mod.connect(host=host)
42    print("connected:", repr(rem))
43    tools = await rem.tools()
44    print("remote tools:", tools["names"])
45
46    # the local-run shape: sync call, sync state, one await for completion
47    job = rem.run("uname", "-s")
48    print("mid-run tail:", repr(job.tail()), "poll:", job.poll())
49    await job
50    print("run:", job.tail().strip(), "exit:", job.poll())
51
52    listing = await rem.files.ls(".")
53    print("files.ls entries:", len(listing))
54    relay = await rem.work.list()
55    print("relay answer:", relay[0])
56
57    import time
58
59    started = time.monotonic()
60    again = rem.run("true")
61    await again
62    print(f"second call over the control socket: {time.monotonic() - started:.2f}s")
63
64    await rem.close()
65    print("closed:", rem.closed)
66    return 0
67
68
69if __name__ == "__main__":
70    target = (
71        sys.argv[1] if len(sys.argv) > 1 else os.environ.get("ALBEDO_SSH", "localhost")
72    )
73    raise SystemExit(asyncio.run(main(target)))