kernel: swap stale kernels at idle and run kernels on remote hosts over ssh
#55: a kernel on an older bundle, protocol or module set is swapped at the next idle moment (snapshot, boot the new one beside it, restore, stop the old), with a note naming what carried; live jobs hold the swap until /kernel upgrade forces it. #56: sessions at host:/path boot their kernel there through the same bridge over ssh, on the ControlMasters remote.connect() uses; reaping runs on the host, heavy slots are per host, an unreachable host is a reattach rather than a lost kernel, and login profiles that print cannot corrupt the frames. #57: /fs routes, AGENTS.md and project skills read remote workspaces in one ssh round trip, interpreted by the local code path; GET /hosts lists recent and ssh-config hosts. Signed-off-by: dawn <90008@klbr.net>
41 unmodified lines424344454647484950513 unmodified lines41 unmodified lines- `/model [model] [provider] [effort]` — show the selection (any caller) or switch it (user only, idle session; a model call is always mid-turn, so switching is refused and the model is told to ask the user). A given effort must be one the new model supports; without one, the current effort carries over when the new model supports it.- `/context [section] [page]` — the prepared model request: a summary, or one bounded section page.- `/raise-cap [on|off] [model]` — raise a model's context window to the provider's maximum, or restore its default (user only). Without a state it toggles; without a model it uses the session's. The choice is saved per model in `raisedCaps` and applies to every session on that model from its next request. See [raised caps](models.md#raised-caps).
the python extension contributes `/kernel [upgrade]` — whether the session's kernel runs older code than the daemon (bundle, module set, or protocol) and what keeps it, or, with `upgrade` (user only, between turns), swap it now past its live jobs (`KernelReport` / `KernelUpgrade` state ops). See [kernel](kernel.md#version-skew).
skills contributes one command per cataloged skill; `/tree`, `/fork`, `/login`, and the other picker-style entries stay presentation-only in the CLI.
## CLI page lifecycle
3 unmodified lines177 unmodified linesExpand all17817918018118218318418518618735 unmodified lines177 unmodified linesExpand all## remote
the `remote` extension boots this session's Python kernel on a remote host over one SSH connection, so every harness tool is callable on it, and it is enabled by default. `rem = await remote.connect()` stages albedo's python bundle on the target, starts the kernel there through SSH with `ControlMaster`/`ControlPersist` (one multiplexed TCP connection: no re-authentication per call), and answers the remote kernel's host-route calls against this session's daemon.
the ssh options, control path, agent socket discovery, login-shell wrapper and bundle staging live in `priv/python/albedo_ssh.py`, which the daemon's own remote kernels use too (kernel.md, "remote kernels"), so both ride the same control sockets and the same staged bundle under `~/.albedo-remote/<digest>`. a session whose workspace is itself on another host needs no `remote.connect()`: its kernel already runs there.
machine tools run on the remote kernel: `rem.run(program, *args)` starts a supervised job on the host immediately and synchronously, like local run — the handle exists right away, `job.tail()`, `job.id`, `job.exit_code`, `job.duration`, `job.waited`, and `job.timed_out` answer synchronously from the mirrored output stream, `await job` waits for completion, and `await job.stop()` stops it. pipes run entirely on the host, chained before the one await: `job = rem.run("journalctl", "-u", "nginx", "--no-pager").pipe("rg", "error")`, `await job`, then `job.tail(lines=40)`. `tail(lines=)` and `exit_code` answer from the mirror; other job methods, like `head(lines=)`, are remote calls to await, and properties such as `job.pipeline` do not cross. session tools run where the daemon runs: `await rem.work.*` and `await rem.skills.*` relay over the connection to the local daemon's ledger and catalog. both kernels load the same content-hashed bundle, so a tool's remote shape matches its local one. every other call returns a reference that settles on its first await: values cross as real objects — dataclasses and list subclasses are rebuilt, so remote output prints exactly like local output — and a result that cannot cross comes back as a live reference whose methods are further remote calls. references passed back into remote calls stay references, not copies, including ones still in flight. one rule remains: a call whose local counterpart is synchronous (`rem.files.read`) still needs `await`, because the value itself crosses the network; handles and their state never do.
images are the exception to "every harness tool runs remotely": a remote call is not a cell, so the remote kernel has no result to carry one. `await rem.show_image(path_or_bytes)` runs locally instead: it reads a remote path over the control connection (at most one byte past the 5 MiB a cell may carry) and attaches the bytes to the local cell through `api.attach_image`, so the image meets the same format, count, size and edge checks as `show_image`. it works in degraded mode too.
35 unmodified lines19 unmodified lines202122232424252627282910 unmodified lines40414243434444454647484945 unmodified lines959697989899100101991001011021031041051051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541552 unmodified lines15815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427519 unmodified lines over the kernel's stdin, never argv) and waits for its socket; `attach <run_dir>` connects to a running one. it announces its own bundle hash (`{"bridge": {"bundle"}}`), then copies bytes both ways until either side closes. exit status 3 means no kernel is there. killing it never touches the kernel. #56 runs the same argv over ssh. touches the kernel. a remote kernel's bridge is the same argv run over ssh (see remote kernels below).- **port owner** (`albedo_python.erl`): one erlang process per kernel, as before. it owns the job groups the kernel reports and the `albedo_signal.py` termination ladder, aimed at the pid/pgid/leader the kernel declares in its hello.10 unmodified lines`{"hello": {protocol, bundle, epoch, ack, pid, pgid, leader, ready, jobs,external, slots, dropped}}`, or `{"refused": reason}`. `protocol` (1) and thehello/snapshot/shutdown frames never change shape. `bundle` is`albedo_bundle.digest()`, the same content hash the remote plugin stagesunder; a hello whose bundle or protocol differs from the bridge's is onlylogged for now (#55 acts on it). `jobs` (live `job_start` frames), `external`under; a hello whose bundle or protocol differs from the bridge's makes thekernel stale (see version skew). `jobs` (live `job_start` frames), `external`and `slots` (job slots still waited for) let a fresh port owner take over jobownership and admission.
## session layer45 unmodified lines- daemon start: `runtime.resume_kernels` reattaches every recorded kernel in the background, one at a time through the boot slots, so late results and job wakes arrive without waiting for the session. a session that asks for its kernel meanwhile waits for that attach, and gets a fresh boot if there was nothing to attach to. an attached kernel is `runtime.resumed`, and the session keeps its namespace without the disk restore. a recorded kernel whose workspace or module set no longer fits is stopped and replaced (#55 will swap it at idle instead). was nothing to attach to. an attached kernel's `runtime.origin` is `Resumed`, and the session keeps its namespace without the disk restore. a recorded kernel whose module set no longer fits is kept and marked stale; one in another workspace is stopped and replaced.
explicit drops still end the kernel at once: reset, release by the idlesweep, `/cd`, session close, extension reloads (the replacement boots besidethe live kernel, never attached to it).the live kernel, never attached to it), and the old kernel after a swap.
## version skew
a kernel is stale when it runs another bundle than the bridge that reached it,speaks another protocol (both read from the hello), or booted with anothermodule set than its session now has (found at reattach). `kernel.stale`answers the reason and whether the swap was forced; the session status carries`kernel: {stale, reason?, link}`, and the tui shows `kernel older` beside theheader counts. `link` is how the session reaches its kernel: `none` before itneeded one, `booting` while one opens (or a daemon start's attach is awaited),`attached`, `reattaching` while the bridge is down and the port owner retries(`albedo_python:linked/1`), and `lost` once it gave up. the tui fades a remotehost in the header while booting or reattaching, and colors it as an erroronce lost.
the swap happens at the session's next idle moment: `session_namespace.ready`,which every turn, compaction and background call goes through, lets go of akernel that `runtime.upgradable` says nothing keeps, so the session asks theruntime again (parking its work), and the runtime's open runs `kernel.upgrade`off the actor while the session waits, as for a boot:
1. snapshot the old namespace to `namespace.state` in the old kernel's run directory (a kernel with a cell still running answers busy at once);2. boot a fresh kernel on the current bundle and module set beside it;3. restore into it from that file;4. stop the old kernel, ending any jobs it still had, and its run directory.
the session adopts it with origin `Upgraded`: the model's next message carries"The python kernel was upgraded to the new python bundle. Restored: a, b. Notcarried: x (why)." and the transcript a note. imports and definitions thatwere not saved are gone, as after a disk restore. a swap that cannot happennow (busy, a namespace that would not save) hands the old kernel back(`Kept`), still stale, and the next idle moment tries again.
live jobs keep an older bundle or module set until they end (the job's wakeis usually that idle moment). `/kernel` reports the staleness and the livejobs; `/kernel upgrade` (user only, between turns) forces the swap now,stopping the jobs as a restart did before kernels were detached.
protocol skew does not wait for jobs: a kernel on another protocol cannot besupervised. its outbox is dropped, not replayed; the swap asks for a snapshotand a shutdown through the frozen frames, carries nothing if the snapshotfails, and the stop's signal ladder ends the recorded process group eitherway.
## lifetime
a kernel with nothing attached for its grace period reaps its job groups(plugin cleanup) and exits, unless a job or cell is still live. the grace is2 unmodified linesexits within a second of its run directory disappearing, which is how testhomes clean up after themselves. the e2e harness sets a 30 s grace, the gleamtest home 20 s.
## remote kernels
a session whose workspace is `[user@]host:/path` (workspaces.md) runs itskernel on that host. nothing above the bridge changes: the outbox, reattachwith backoff and `resume_kernels` treat an ssh drop as a bridge that exited.
- **one ssh layer**: `priv/python/albedo_ssh.py` builds the argv (BatchMode, `ConnectTimeout=10`, `StrictHostKeyChecking=accept-new`, `ControlMaster=auto`, `ControlPath=/tmp/albedo-ssh-cm/%C`, `ControlPersist=600`), finds an agent socket, wraps commands in the remote login shell and stages the bundle. the model's `remote` plugin imports it, so the daemon and `remote.connect()` share masters and the staged bundle.- **clean stdout**: every remote command is `/bin/sh -c 'exec 3>&1 1>&2; exec "$SHELL" -l -c "exec 1>&3 3>&-; <command>"'`, quoted with shlex in `albedo_ssh.in_login_shell` only. whatever the login profile prints (a motd, a greeting) goes to stderr, and the command gets the real stdout back, so the bridge's frames, the remote plugin's kernel and the probe's lines stay clean whatever the user's shell is.- **fixed command lines**: the probe (and, with no network, `albedo_ssh.py commands <target> <home>`) answers the complete remote commands the daemon runs: `bridge` (`albedo_bridge.py --frame`), `signal` (`albedo_signal.py -`) and `remove`. their inputs go on stdin, never in the command line, so nothing in erlang quotes for a shell: the bridge's first frame is `{"bridge": {"argv": [start|attach, run_dir, modules?], "cwd": path}}` (exit 4 when `cwd` is no folder), the ladder reads its request line, `remove` the run directory.- **probe** (`albedo_ssh.py probe <target>`, run by `albedo_ssh.erl`): one command over the master prints os, arch, home, cores, the python3 version and whether `~/.albedo-remote/<digest[:16]>` is staged; python older than 3.11 is `unsupported`, a missing bundle is staged (tar over the same stream, unpacked beside the target and moved into place, so a bundle directory that exists is complete). ssh's own failure (exit 255) is `needs_auth` when its words say a person could get in (permission denied, host key verification, passphrase, keyboard-interactive), otherwise `unreachable`. answers are cached in the daemon, ready ones for a minute and failures for five seconds; a second caller joins a probe in flight. `albedo/harness/ssh.gleam` is the gleam face.- **boot**: `kernel.place` probes (or reuses the cached probe) before the bridge starts, so a warmed host boots fast. the port owner runs the `bridge` command over ssh. a fresh boot's exit 4 reads as "not a folder there", 255 as an ssh failure. run directories live under the remote home, `~/.albedo-remote/run/<kernel id>`, recorded absolute.- **a host out of reach is not a kernel gone**: only an attach the bridge ends with exit 3 (nothing at the run directory) or a refused attach proves a remote kernel gone. ssh failing (255), any other exit, or a startup timeout while attaching is a dropped connection: the session gets the kernel at once, reattaching, and the port owner keeps attaching in the background, backing off to every 30 s, until the kernel's grace (plus a minute) has surely passed since it lost the connection; then it forgets the record quietly, without reaching for the host. when the probe itself fails at boot, a recorded kernel is attached with commands rebuilt from its run directory (which names the home), so a wifi blip during a daemon restart keeps the namespace. a turn in a session with a kernel on record goes through while its host is unreachable or needs a sign-in, waiting like any reattach. only a fresh boot needs the host now.- **ownership**: the pids a remote kernel declares are that host's. the `albedo_signal.py` ladder for its kernel and job groups runs there over ssh (the `signal` command, 15 s), never locally; if ssh can't reach the host the targets are left to the kernel's own grace exit, which reaps its jobs on the host, and the error says so. only the bridge (a local ssh client) is ever signalled here. the run directory goes with `rm -rf` over ssh. `os_pid` answers nothing for a remote kernel, so the reaper's `ps` never reads an unrelated local pid.- **heavy slots**: a remote kernel's jobs queue in that host's own pool (`albedo_job_slots:ensure(Host, Cpus)`, started and watched by the local pool), sized to the cores the probe found (`ALBEDO_MAX_REMOTE_JOBS` overrides; two while a host attached without a probe is still unknown, corrected by the next probe) with no load adjustment; admission is the same protocol.- **turns**: a session that holds a kernel (attached, or reattaching through its outbox) never probes on a turn. one that must boot probes for up to 3 s: still warming is left to the boot, which waits for the probe; `needs_auth`, unreachable or unsupported leave the turn admitted and waiting, blocked with the probe's words as its receipt's `blockingReason`, retried every 15 s like any blocked input (operations.md).- **the model** gets a context line (python extension, `place.gleam`): "your python kernel and run jobs execute on chernobog (Linux aarch64)…", with the remote home.
### `/hosts`
- `GET /hosts/:host` (`[user@]host`): the cached probe, or `warming` while one runs (a stale or missing answer starts one).- `POST /hosts/:host/warm`: drops a cached answer and probes now, waiting up to a minute.
both answer `{host, state, detail}` with `state` one of `ready`, `warming`,`needs_auth` (with `control_path`), `unreachable`, `unsupported`, and `os`,`arch`, `home` when ready. `/health` lists `remote_hosts`. `GET /hosts`(the picker's host completion) is in workspaces.md. the same probe alsoanswers the `gather` command the folder browser and project readers run.
### signing in
BatchMode stays on for the daemon. when a remote session's waiting turn isblocked, the chat asks `GET /hosts/:host`; if that says `needs_auth` and thedaemon shares the tui's machine, the chat offers ctrl+l: it runs`ssh -M -fN -o ControlPath=<control_path> -o ControlPersist=600 <host>`through `tea.ExecProcess`, so ssh asks in the terminal, then warms the host,and the waiting turn starts on the daemon's next try; the daemon rides thatmaster from then on. with a daemon elsewhere the notice says to run`ssh <host>` on its machine. the offer goes once the turn starts or goesaway, or the session moves.
## not yet
- after a daemon restart, job groups of a kernel that died hard (SIGKILL) while nobody was attached are not reaped: the recorded pid could be reused.- job slots a reattached kernel's running jobs held are not re-counted in the new daemon's pool; only slots still waited for are asked for again.- profile output a remote login shell prints reaches the daemon's own stderr log on every remote command.16 unmodified lines17181920212221222324252627287 unmodified lines36373839383940414041424344454344454647484950514647484950515253545556575859605556575859616263646566676869707172737475766566777879808182834 unmodified lines888990917980929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313471 unmodified lines2062072082091621631641651662102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482495 unmodified lines16 unmodified lines remote when it does not start with `/` or `~` and the part before the first `:` holds no `/`. user and host start with a letter or digit, so neither is ever an ssh option. the path must be absolute and is normalised (`//`, `.`, `..`, trailing slash), so one remote directory has one key. a remote `~` is refused: the remote home is only known once albedo connects there. one key. a stored remote path is always absolute: `host:~` and `host:~/x` given to create or move a session are resolved against that host's home (one probe over ssh, bounded, kernel.md) and stored absolute; an unreachable host refuses with ssh's words.
session info carries the parsed location beside the `workspace` string:
```json7 unmodified lines`label` is how clients show the host: the alias, with `user@` dropped whenit is the user `ssh -G <host>` reports (local config only, no network),cached per host for the daemon's life. without ssh the user stays.
nothing runs at a remote location yet. what is only a key works as it is:the work ledger and paperclips scope, recent folders, session search bycwd, family moves. what needs the remote filesystem refuses or skips byname, and never looks for the path on the daemon's own disk:a remote session's python kernel and `run` jobs run on its host(kernel.md, "remote kernels"). what is only a key works as it is: the workledger and paperclips scope, recent folders, session search by cwd, familymoves. what the daemon itself reads from a workspace goes through thelocation, and never looks for a remote path on its own disk:
- a turn in a remote session answers 409 `kernels on chernobog aren't available yet` (the kernel refuses to boot there too).- `/fs/list`, `/fs/repo` and `/fs/preview` answer 400 `folders on chernobog aren't available yet`.- AGENTS.md, SYSTEM.md and the other instruction files, and skills, are read from the home directories only; the instructions extension warns `project instruction files on chernobog aren't available yet` and the skills catalog lists `project skills on chernobog aren't available yet` among its diagnostics.- `/fs/list`, `/fs/repo` and `/fs/preview` take `host:/abs` and `host:~` too (see browsing).- AGENTS.md, SYSTEM.md and the other instruction files, and project skills, are read from a daemon-side mirror of the remote workspace's project files (`$ALBEDO_HOME/mirror/<digest>/`: root `*.md`, `.agents/*.md`, `.albedo/*.md`, and the `.agents/skills` and `.albedo/skills` trees, at most 512 files and 8 MiB, each read up to one byte past the 1 MiB instruction limit). one gather over ssh fills it, at most every 5 s, and the local readers read it as the workspace, so the same rules apply. a skill's `location` is its mirror path. while the host is out of reach only the home directories count, and the instructions warn and the skills catalog diagnoses `project … skipped: can't reach chernobog: …`.- memory is a daemon-side file keyed by the workspace string, so a remote workspace gets its own.
the tui shows a remote workspace as `label:/path`: the chat header reads`✦ albedo on chernobog:/home/mayer/proj/albedo`, folds only the path, keepsthe host in every layout that shows a place, and colors the host with astable hue derived from its label, lifted to the theme's contrast. recentfolders and the sessions view lead remote entries with their host.the tui shows a remote workspace as `label:path`, the path folded underthat host's own home once a listing or probe reported it: the chat headerreads `✦ albedo on chernobog:~/proj/albedo`, folds only the path, keeps thehost in every layout that shows a place, and colors the host with a stablehue derived from its label, lifted to the theme's contrast. the host goesfaint while the session's kernel boots or reattaches (the status line says`connecting to chernobog…`) and takes the error color once it is lost (theport owner gave up and forgot the kernel, so the status line says the nextturn starts a fresh one), fromthe session status's `kernel.link` (kernel.md). recent folders and thesessions view lead remote entries with their host.
## moving a session
`POST /sessions/:id/workspace {"workspace": "/abs/dir"}` answers the session'sinfo. the session must be idle and the target an existing absolute directory,or a remote location (stored canonically; it can't be checked before albedoconnects to the host). `POST /sessions` takes the same forms.or a remote location (stored canonically; an absolute one isn't checkeduntil a kernel boots there, a `~` one is resolved over ssh first).`POST /sessions` takes the same forms.the kernel is dropped (python variables start fresh), the transcript stays,and a note records the move.
every descendant of the session (children, their children; open or closed)4 unmodified lines
## browsing
all three routes take `path`: absolute, or starting with `~`, which expandsto the daemon's home. anything else is a 400, a remote location included; a path that is not an existingdirectory is a 404. errors are `{"error": "..."}`. `/health` liststo the daemon's home, or a remote `[user@]host:/abs` or `[user@]host:~/x`(`~` is that host's home). anything else is a 400; a path that is not anexisting directory is a 404. errors are `{"error": "..."}`. `/health` lists`workspace_browser` when these routes exist.
a remote path is gathered on its host in one ssh round trip per request(`priv/python/albedo_gather.py`, after the host's probe, kernel.md): thedirectory entries with their stats, which `.jj`/`.git` markers exist, thevery vcs commands `vcs.plan` names (run there with the same 2 s deadline),and the sizes of tracked files. the daemon reads that snapshot with thesame code it reads its own disk with (`folders.gleam`, `vcs.gleam` over a`Shell`), so a remote answer is the local answer for the same tree, exceptthat `path` and `home` are canonical location strings(`mayer@chernobog:/home/mayer/proj`, `mayer@chernobog:/home/mayer`) so aclient can fold `~`. a repository's `root` stays the plain path on thathost. a host without a ready probe answers 503 `{error, host, state}`(`needs_auth` with `control_path`, `unreachable`, `unsupported`,`warming`), so the picker can show it on the row.
### `GET /hosts`
the hosts the picker completes:
```json{"hosts": [{"host": "mayer@chernobog", "label": "chernobog", "source": "recent", "state": "ready"}, {"host": "devbox", "label": "devbox", "source": "config"}]}```
- `recent` hosts come from session workspaces, ranked as the picker ranks recent folders (per session 4 within the hour, halving past a day, a week and a month, summed per host).- `config` hosts are the `Host` names of the daemon user's `~/.ssh/config` and the files it `Include`s, wildcard patterns (`*`, `?`, `!`) left out, in file order, after the recent ones and without repeats.- `host` is the canonical `[user@]host` for `/hosts/:host` and `host:/path`; `label` the display form. `state` is the cached probe's when one is fresh; listing never probes.
### `GET /fs/list?path=P`
the directories directly inside P, for the picker's list.
71 unmodified lines
## the picker
`/cd` opens it full screen; `/cd <path>` moves at once. an empty query listsrecent folders (the workspaces of your sessions, ranked by how often and howrecently they were used). typing a path lists that folder's directories,filtered by the last segment. tab enters the highlighted folder, shift+tab goesup, enter moves there, esc goes back. the same picker replaces the plain textbox shown when a session's workspace has gone missing.recent folders (the workspaces of your sessions, local and remote, ranked byhow often and how recently they were used). typing a path lists thatfolder's directories, filtered by the last segment. tab enters thehighlighted folder, shift+tab goes up, enter moves there, esc goes back. thesame picker replaces the plain text box shown when a session's workspace hasgone missing.
the query takes scp syntax for another host, so a host is picked like afolder (`cli/internal/tui/folder_picker.go`, `parseQuery`):
- `chernobog:` lists that host's home, `chernobog:proj/` and `chernobog:~/proj/` browse under it, `chernobog:/srv/` from its root. everything after the last slash filters, as locally. the listing comes from `/fs/list` with the location; its canonical `path` and `home` fold rows to `chernobog:~/proj/albedo`.- hosts complete from `GET /hosts` (recent first, then ssh config), with the hosts of the sessions already listed as a fallback. a bare word still filters the recent folders exactly as before and only adds the hosts whose name it starts, after the folders, so `cher` offers `chernobog:` without turning a local filter fuzzy. text with an `@` and no `:` can only be `user@host`: it lists hosts alone, fuzzy on the host part, and the typed user goes with the completion. tab or enter on a host completes to `host:`.- highlighting a remote row (a recent remote folder, a host, a folder in a remote listing), or typing a remote listing, warms its host once per picker: `POST /hosts/:host/warm`, then `GET /hosts/:host` every 750 ms until the probe settles. the row shows it quietly: faint while warming, plain once ready, the probe's detail in the error style when unreachable or unsupported, `sign in · ctrl+l` when it needs a person. ctrl+l runs the same `ssh -M -fN` handoff a refused turn offers (kernel.md, "signing in"), with the `control_path` the probe answered, then warms the host again. repositories and the preview of a remote row wait for ready; until then the pane shows the host's state, and when ready its os, arch and ~.- enter never picks a dead host: on an unreachable, unsupported or needs-auth host it stays and says why in the footer.
in the sessions view, ctrl+f (or `~` or `/` typed into an empty search) opensthe same picker to browse sessions by folder: enter starts a new session inthe highlighted folder instead of moving one. in either mode → steps into thehighlighted folder's sessions, where enter opens one and ← goes back.5 unmodified lines12345678910111213141519 unmodified lines35363738394041424357 unmodified lines10110210310410510610710810911011111211311411511611711811912012112212312412512612722 unmodified lines"""Copy a detached kernel's frames between this process's stdio and its socket.
albedo_bridge.py start <run_dir> <modules-json> albedo_bridge.py attach <run_dir> albedo_bridge.py --frame
`--frame` takes the same arguments from its first stdin frame instead,`{"bridge": {"argv": [...], "cwd": dir}}`, and starts in `cwd` (exit status 4when it is no directory): the remote form, so the daemon's ssh command linenever carries a path or the module list.
The daemon talks to this process as it once talked to the kernel itself, over4-byte-length-framed stdio, and writes the attach frame first. `start` readsthat frame to learn the kernel's token, starts the albedo_kernel.py beside thisfile in its own session (the token travels over the kernel's stdin, never19 unmodified linesimport albedo_bundleimport albedo_link
GONE = 3NO_FOLDER = 4START_TIMEOUT = 10.0 # seconds a starting kernel has to bind its socket
def connect(run_dir: str) -> socket.socket | None:57 unmodified lines pass os._exit(0)
def framed() -> list[str]: """The arguments a remote daemon sent as the first frame.""" first = albedo_link.read_frame(stdin) spec = first.get("bridge") if isinstance(first, dict) else None argv = spec.get("argv") if isinstance(spec, dict) else None if not isinstance(argv, list) or not all(isinstance(a, str) for a in argv): sys.exit(2) cwd = spec.get("cwd") if isinstance(spec, dict) else None if isinstance(cwd, str): try: os.chdir(cwd) except OSError: sys.exit(NO_FOLDER) return [sys.argv[0], *argv]
def main(argv: list[str]) -> None: if argv[1:] == ["--frame"]: argv = framed() mode, run_dir = argv[1], argv[2] if mode == "start": attach = albedo_link.read_frame(stdin) if not isinstance(attach, dict) or not isinstance(attach.get("attach"), dict):22 unmodified lines123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210"""What the daemon's folder browser and project readers need from a remote host.
albedo_gather.py
reads one JSON request line on stdin and prints one JSON snapshot. The daemoninterprets the snapshot with the same code it uses on its own disk(albedo/daemon/folders.gleam, albedo/harness/vcs.gleam), so this side onlycollects: directory entries, which repository markers exist, the vcscommands the daemon planned, file sizes, and project files. Everything comesback in one ssh round trip per question.
Request: `{route, dir, plans, separators, deadline, counted}`, `dir` absoluteand normalised. `route` is list, repo, preview or project.
Snapshot: `{directory, entries: {dir: [[name, dir, modified]]}, exists:[path], outputs: {key: stdout}, sizes: {path: bytes}, files: {relative:base64}}`. An output's key is its program and arguments joined by NUL; onlycommands that exit 0 within the deadline answer."""
from __future__ import annotations
import base64import jsonimport osimport statimport subprocessimport sysfrom concurrent.futures import ThreadPoolExecutor
MARKERS = (".jj", ".git")SUBDIRECTORIES = 256 # child directories whose entries a preview listsPROJECT_FILES = 512PROJECT_BYTES = 8 * 1024 * 1024FILE_BYTES = 1024 * 1024 + 1 # one past the instruction limit, so it still warns
def child(directory: str, name: str) -> str: return "/" + name if directory == "/" else f"{directory}/{name}"
def shown(name: str) -> bool: """Names the daemon can show and send back: valid UTF-8.""" try: name.encode() return True except UnicodeEncodeError: return False
def entries(directory: str) -> list[list[object]]: try: names = os.listdir(directory) except OSError: return [] listed: list[list[object]] = [] for name in filter(shown, names): try: info = os.stat(child(directory, name)) listed.append([name, stat.S_ISDIR(info.st_mode), int(info.st_mtime)]) except OSError: listed.append([name, False, 0]) return listed
def marks(directory: str, exists: list[str]) -> str | None: """Record the markers at `directory`; the kind it roots, .jj first.""" found = None for marker in MARKERS: path = child(directory, marker) if os.path.lexists(path) and (os.path.isfile(path) or os.path.isdir(path)): exists.append(path) found = found or marker return found
def parent(directory: str) -> str | None: return None if directory == "/" else os.path.dirname(directory) or "/"
def checkout(directory: str, exists: list[str]) -> tuple[str, str] | None: """The nearest enclosing repository: (kind, root).""" current: str | None = directory while current is not None: found = marks(current, exists) if found is not None: return found[1:], current current = parent(current) return None
def key(program: str, args: list[str]) -> str: return "\0".join([program, *args])
def run( command: list, root: str, directory: str, deadline: float) -> tuple[str, str | None]: program, args, at = command try: done = subprocess.run( [program, *args], cwd=root if at == "root" else directory, stdin=subprocess.DEVNULL, capture_output=True, timeout=deadline, ) except (OSError, subprocess.TimeoutExpired): return key(program, args), None output = done.stdout.decode(errors="replace") if done.returncode == 0 else None return key(program, args), output
def project(directory: str) -> dict[str, str]: """Project instruction files and skills, relative to `directory`.""" picked: list[str] = [] for relative in ("", ".agents", ".albedo"): base = child(directory, relative) if relative else directory try: names = sorted(filter(shown, os.listdir(base))) except OSError: continue for name in names: if name.lower().endswith(".md") and os.path.isfile(child(base, name)): picked.append(f"{relative}/{name}" if relative else name) for relative in (".agents/skills", ".albedo/skills"): for root, directories, names in os.walk(child(directory, relative)): directories.sort() for name in sorted(filter(shown, names)): path = os.path.join(root, name) if os.path.isfile(path) and len(picked) < PROJECT_FILES: picked.append(os.path.relpath(path, directory)) files: dict[str, str] = {} budget = PROJECT_BYTES for relative in picked: try: with open(child(directory, relative), "rb") as handle: data = handle.read(min(FILE_BYTES, budget)) except OSError: continue budget -= len(data) files[relative] = base64.b64encode(data).decode() if budget <= 0: break return files
def gather(request: dict) -> dict[str, object]: route, directory = request["route"], request["dir"] snapshot: dict[str, object] = {"directory": os.path.isdir(directory)} if not snapshot["directory"]: return snapshot if route == "project": snapshot["files"] = project(directory) return snapshot exists: list[str] = [] listed = {directory: entries(directory)} snapshot.update(entries=listed, exists=exists) if route == "list": for name, is_directory, _ in listed[directory]: if is_directory: marks(child(directory, str(name)), exists) return snapshot found = checkout(directory, exists) if route == "preview": children = [ child(directory, str(name)) for name, is_directory, _ in listed[directory] if is_directory and not str(name).startswith(".") ] for path in children[:SUBDIRECTORIES]: listed[path] = entries(path) if found is None: return snapshot kind, root = found deadline = float(request.get("deadline", 2.0)) plan = request.get("plans", {}).get(kind, []) with ThreadPoolExecutor(max_workers=max(1, len(plan))) as pool: answers = dict(pool.map(lambda c: run(c, root, directory, deadline), plan)) outputs = {k: v for k, v in answers.items() if v is not None} snapshot["outputs"] = outputs if route == "preview": tracked = next((c for c in plan if c[2] == "dir"), None) separator = request.get("separators", {}).get(kind, "\n") listing = outputs.get(key(tracked[0], tracked[1])) if tracked else None sizes: dict[str, int] = {} for relative in (listing or "").split(separator)[ : int(request.get("counted", 0)) ]: if not relative: continue path = child(directory, relative) try: info = os.stat(path) except OSError: continue if stat.S_ISREG(info.st_mode): sizes[path] = info.st_size snapshot["sizes"] = sizes return snapshot
def main() -> int: request = json.loads(sys.stdin.readline()) print(json.dumps(gather(request))) return 0
if __name__ == "__main__": sys.exit(main())251 unmodified linesExpand all252253254255256257258259260261262853 unmodified linesExpand all11161117111811191120112111221117111811191120112111221123112411251126112711281129113011311132513 unmodified linesExpand all251 unmodified linesExpand all } ) elif message["type"] == "release": LIVE.pop(message["handle"], None) elif message["type"] == "snapshot" and (active is not None or not QUEUE.empty()): # A namespace mid-cell is no state to carry: the owner tries again later. send(state_reply(message, {"error": "a cell is still running"})) elif message["type"] == "execute" and message["id"] in FINISHED: send(FINISHED[message["id"]]) elif message["type"] == "execute" and message["id"] in EXECUTING: pass # already queued or running; its done frame answers both853 unmodified linesExpand all forget_output(key) counts[held.kind] -= 1
def state_reply( message: albedo_api.State, state: dict[str, object] | None = None) -> dict[str, object]:def state_reply(message: albedo_api.State) -> dict[str, object]: state = ( save_state(message["path"]) if message["type"] == "snapshot" else load_state(message["path"]) ) if state is None: state = ( save_state(message["path"]) if message["type"] == "snapshot" else load_state(message["path"]) ) return { "type": "done", "id": message["id"], "status": "error" if "error" in state else "ok",513 unmodified linesExpand all1 unmodified line23456678910111277 unmodified lines909192939294951 unmodified line
Erlang runs this as its own process, so supervision keeps making progress whenthe kernel cannot (a native cell holding the GIL, a wedged interpreter), and sothe supervisor never parses kill(1) error text or guesses an exit status. Oneargv carries the request; one JSON object on stdout carries the verdict.argv carries the request (or `-` and one stdin line, the form run over sshfor a remote kernel's processes); one JSON object on stdout carries theverdict."""
from __future__ import annotations
77 unmodified lines # Default SIGALRM terminates even a wedged interpreter; no Python handler. # The owning port waits longer than this before reporting a helper timeout. signal.signal(signal.SIGALRM, signal.SIG_DFL) signal.alarm(3) sys.exit(main(sys.argv[1] if len(sys.argv) > 1 else "{}")) request = sys.argv[1] if len(sys.argv) > 1 else "{}" sys.exit(main(sys.stdin.readline() if request == "-" else request))123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322"""One ssh layer for albedo: the daemon's remote kernels and the model's remote plugin.
Both ride the same multiplexed control connections (`/tmp/albedo-ssh-cm/%C`),so a host warmed by one is warm for the other. BatchMode stays on: a hostthat needs a passphrase, a second factor or a new host key answers`needs_auth`, and the tui opens the master in the terminal instead.
albedo_ssh.py probe <[user@]host> albedo_ssh.py commands <[user@]host> <home> albedo_ssh.py hosts
`probe` prints one JSON object: `{state, detail, os, arch, home, cpus,python}` and, once the home is known, the `commands` answer: `{argv, control,bundle, bridge, signal, remove, gather, auth_sock}`, the complete remotecommands the daemon runs there. state is ready, needs_auth, unreachable or unsupported. Aready probe has checked python >= 3.11 and staged the content-hashed bundleunder `~/.albedo-remote/<digest>`. `commands` needs no network, for a hostknown only from a kernel's record while ssh is down. `hosts` lists the `Host`names of ~/.ssh/config and its includes, patterns left out."""
from __future__ import annotations
import globimport ioimport jsonimport osimport shleximport subprocessimport sysimport tarfileimport tempfile
import albedo_bundle
CONNECT_TIMEOUT = 10 # seconds ssh may take to reach a hostPROBE_TIMEOUT = 30 # seconds for the whole probe commandSTAGE_TIMEOUT = 120 # seconds to move the bundle over one ssh streamMINIMUM_PYTHON = (3, 11)STAGED_DIR = ".albedo-remote"
def control_dir() -> str: """Control sockets live under /tmp when it fits, else the temp dir: unix socket paths are capped (~104 bytes on macOS) and %C hashes to 40 more, so neither ALBEDO_HOME nor macOS's deep $TMPDIR may be on the path.""" control = os.path.join("/tmp", "albedo-ssh-cm") if len(control) + 42 > 104: control = os.path.join(tempfile.gettempdir(), "albedo-ssh-cm") return control
def base() -> list[str]: """The ssh argv prefix: one multiplexed control connection per target.""" control = control_dir() os.makedirs(control, exist_ok=True) return [ "ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={CONNECT_TIMEOUT}", "-o", "StrictHostKeyChecking=accept-new", "-o", "ControlMaster=auto", "-o", f"ControlPath={control}/%C", "-o", "ControlPersist=600", ]
def agent_socket() -> str | None: """SSH_AUTH_SOCK, or a standard agent socket when it was not inherited, so hardware keys, 1Password, and ssh-agent work out of the box.""" if "SSH_AUTH_SOCK" in os.environ: return os.environ["SSH_AUTH_SOCK"] for candidate in ( os.path.expanduser("~/.ssh/agent.sock"), os.path.expanduser( "~/Library/Group Containers/2BUA8C4S2C.com.1password/t/agent.sock" ), ): if os.path.exists(candidate): return candidate return None
def env() -> dict[str, str]: """The local environment for ssh client processes.""" environment = dict(os.environ) socket = agent_socket() if socket is not None: environment["SSH_AUTH_SOCK"] = socket return environment
def in_login_shell(script: str) -> str: """The remote command that runs `script` under the user's login shell.
Non-interactive SSH commands run under `$SHELL -c` without sourcing `/etc/profile` or user profiles, which leaves PATH minimal (missing Nix, Homebrew, or user tool paths); a login shell finds python3 and the user's tools. Profiles often print (a motd, a greeting), which would corrupt a framed stdout, so the login shell starts with stdout on stderr and the real stdout kept on fd 3, and `script` gets it back. The outer `/bin/sh` makes the redirections work whatever the user's shell is. """ login = f"exec 1>&3 3>&-; {script}" outer = f'exec 3>&1 1>&2; exec "${{SHELL:-/bin/sh}}" -l -c {shlex.quote(login)}' return f"/bin/sh -c {shlex.quote(outer)}"
def commands(target: str, home: str) -> dict[str, object]: """Everything the daemon runs on a host whose home is known, as complete remote commands: it never quotes anything itself. Variable inputs travel on stdin (the bridge's first frame, the ladder's request line, the run directory to remove), never in the command line.""" bundle = f"{home}/{staged_name()}" python = "exec python3 -u " answer: dict[str, object] = { "host": target, "argv": [*base(), target], "control": f"{control_dir()}/%C", "bundle": bundle, "bridge": in_login_shell( "ALBEDO_JOB_ADMISSION=1 " + python + shlex.quote(f"{bundle}/albedo_bridge.py") + " --frame" ), "signal": in_login_shell( python + shlex.quote(f"{bundle}/albedo_signal.py") + " -" ), "remove": in_login_shell('read -r run && rm -rf -- "$run"'), "gather": in_login_shell(python + shlex.quote(f"{bundle}/albedo_gather.py")), } socket = agent_socket() if socket is not None: answer["auth_sock"] = socket return answer
def staged_name(digest: str | None = None) -> str: """The bundle's directory under the remote home.""" return f"{STAGED_DIR}/{(digest or albedo_bundle.digest())[:16]}"
def archive() -> bytes: data = io.BytesIO() with tarfile.open(fileobj=data, mode="w") as tar: tar.add( str(albedo_bundle.ROOT), arcname=".", filter=lambda info: None if "__pycache__" in info.name else info, ) return data.getvalue()
def stage_script(remote: str) -> str: """Unpack beside the target, then move into place, so a bundle directory that exists is always complete.""" return ( f'mkdir -p "{remote}.part.$$" && tar -C "{remote}.part.$$" -xf - && ' f'{{ mv "{remote}.part.$$" "{remote}" 2>/dev/null || rm -rf "{remote}.part.$$"; }}' )
def run( target: str, script: str, timeout: float, data: bytes | None = None) -> subprocess.CompletedProcess[bytes]: return subprocess.run( [*base(), target, script], input=data, stdin=None if data is not None else subprocess.DEVNULL, capture_output=True, timeout=timeout, env=env(), )
def failure(stderr: str) -> tuple[str, str]: """What ssh's own exit (255) means: a host we could reach with a person's help, or one we could not reach at all.""" text = stderr.strip() lines = [line for line in text.splitlines() if line.strip()] detail = lines[-1] if lines else "ssh failed" lowered = text.lower() interactive = ( "permission denied", "host key verification failed", "passphrase", "keyboard-interactive", "verification code", "too many authentication failures", ) if any(phrase in lowered for phrase in interactive): return "needs_auth", detail return "unreachable", detail
PROBE = ( "uname -s; uname -m; printf '%s\\n' \"$HOME\"; " "getconf _NPROCESSORS_ONLN 2>/dev/null || nproc 2>/dev/null || echo 1; " "python3 -c 'import sys; print(\"%d.%d\" % sys.version_info[:2])' 2>/dev/null " "|| echo none; " 'test -f "$HOME/{bundle}/albedo_bridge.py" && echo staged || echo missing')
def probe(target: str) -> dict[str, object]: answer: dict[str, object] = { "host": target, "argv": [*base(), target], "control": f"{control_dir()}/%C", } bundle = staged_name() try: done = run(target, in_login_shell(PROBE.format(bundle=bundle)), PROBE_TIMEOUT) except subprocess.TimeoutExpired: return {**answer, "state": "unreachable", "detail": "ssh timed out"} except OSError as error: return {**answer, "state": "unreachable", "detail": f"ssh: {error}"} fields = done.stdout.decode(errors="replace").splitlines() if done.returncode == 255 or len(fields) != 6: state, detail = failure(done.stderr.decode(errors="replace")) return {**answer, "state": state, "detail": detail} system, arch, home, cpus, python, staged = (field.strip() for field in fields) answer.update( commands(target, home), os=system, arch=arch, home=home, cpus=int(cpus) if cpus.isdigit() and int(cpus) > 0 else 1, python=python, ) version = ( tuple(int(part) for part in python.split(".")) if python[:1].isdigit() else None ) if version is None: return { **answer, "state": "unsupported", "detail": "needs python >= 3.11 (python3 not found)", } if version < MINIMUM_PYTHON: return { **answer, "state": "unsupported", "detail": f"needs python >= 3.11 (found {python})", } if staged != "staged": try: staging = run( target, in_login_shell(stage_script(f"$HOME/{bundle}")), STAGE_TIMEOUT, archive(), ) except subprocess.TimeoutExpired: return {**answer, "state": "unreachable", "detail": "staging timed out"} if staging.returncode != 0: detail = staging.stderr.decode(errors="replace").strip()[-500:] return { **answer, "state": "unreachable", "detail": f"staging failed: {detail}", } return {**answer, "state": "ready", "detail": ""}
def config_hosts(path: str | None = None, seen: set[str] | None = None) -> list[str]: """`Host` names in ~/.ssh/config and the files it includes, in order, without patterns (`*`, `?`, `!`): the hosts a person can name.""" path = path or os.path.expanduser("~/.ssh/config") seen = set() if seen is None else seen if path in seen: return [] seen.add(path) try: with open(path, encoding="utf-8", errors="replace") as handle: lines = handle.read().splitlines() except OSError: return [] names: list[str] = [] for line in lines: words = shlex.split(line, comments=True) if line.strip() else [] if len(words) < 2: continue keyword = words[0].lower() if keyword == "host": names += [w for w in words[1:] if not any(c in w for c in "*?!")] elif keyword == "include": for pattern in words[1:]: pattern = os.path.expanduser(pattern) if not os.path.isabs(pattern): pattern = os.path.join(os.path.expanduser("~/.ssh"), pattern) for included in sorted(glob.glob(pattern)): names += config_hosts(included, seen) return list(dict.fromkeys(names))
def main(argv: list[str]) -> int: if len(argv) == 3 and argv[1] == "probe": print(json.dumps(probe(argv[2]))) return 0 if len(argv) == 2 and argv[1] == "hosts": print(json.dumps(config_hosts())) return 0 if len(argv) == 4 and argv[1] == "commands": print(json.dumps(commands(argv[2], argv[3]))) return 0 print( "usage: albedo_ssh.py probe <[user@]host> | commands <[user@]host> <home>", file=sys.stderr, ) return 2
if __name__ == "__main__": sys.exit(main(sys.argv))