Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371"""The kernel's half of the durable session layer.
A detached kernel listens on a unix socket in its run directory. Whoeverattaches (a bridge the daemon runs, locally today and over ssh later) presentsthe kernel's secret token; the newest attach wins and cuts the previousconnection off. Every frame after the handshake travels as``{"seq": n, "ack": m, "frame": {...}}``: the kernel keeps what it sent untilthe daemon acknowledges it and resends the rest on the next attach, and itdrops any daemon frame it has already seen, so a reconnect neither loses norrepeats a message. A kernel run over plain stdio (the remote plugin's targets)keeps the raw frames and ends with its pipe."""
from __future__ import annotations
import collectionsimport hmacimport jsonimport osimport socketimport structimport threadingimport timefrom collections.abc import Callablefrom typing import Any, BinaryIO
PROTOCOL = 1 # framing version; hello fields describe the current bundleMAX_FRAME = 8 * 1024 * 1024SOCKET = "kernel.sock"ATTACH_TIMEOUT = 10.0 # seconds a fresh connection has to present its tokenOUTBOX_FRAMES = 4096OUTBOX_BYTES = 64 * 1024 * 1024# Frames whose newest copy supersedes every older unacknowledged one, by the# field naming what they describe. Everything else must be delivered.COALESCE = {"mirror": "handle", "trace": "id", "jobs": None, "cells": None}
def encode(value: object) -> bytes: return json.dumps(value, ensure_ascii=True, separators=(",", ":")).encode()
def envelope(seq: int, ack: int, frame: bytes) -> bytes: return b'{"seq":%d,"ack":%d,"frame":%s}' % (seq, ack, frame)
def count_of(value: object) -> int | None: """A frame's count: a non-negative integer, else None.""" if isinstance(value, int) and not isinstance(value, bool) and value >= 0: return value return None
def write_all(write: Callable[[Any], int | None], data: bytes) -> None: remaining = memoryview(data) while remaining: written = write(remaining) remaining = remaining[len(remaining) if written is None else written :]
def write_frame(write: Callable[[Any], int | None], data: bytes) -> None: write_all(write, struct.pack(">I", len(data)) + data)
def read_exact(read: Callable[[int], bytes], size: int) -> bytes: data = bytearray() while len(data) < size: chunk = read(size - len(data)) if not chunk: raise EOFError() data.extend(chunk) return bytes(data)
def read_frame(read: Callable[[int], bytes]) -> object: size = struct.unpack(">I", read_exact(read, 4))[0] if size > MAX_FRAME: raise ValueError("control frame too large") return json.loads(read_exact(read, size))
class Outbox: """Frames sent and not yet acknowledged, oldest first.
Coalescing frames replace their older unacknowledged copy, so a flood of output mirrors costs one entry. Past the bound, the oldest coalescing frames go first; a must-deliver frame is dropped only when nothing else is left, and `dropped` counts every such loss for the next hello. """
def __init__(self, frames: int = OUTBOX_FRAMES, size: int = OUTBOX_BYTES): self.limit_frames = frames self.limit_bytes = size self.seq = 0 self.size = 0 self.dropped = 0 self.entries: collections.OrderedDict[int, tuple[bytes, object]] = ( collections.OrderedDict() ) self.keys: dict[object, int] = {}
def add(self, frame: dict[str, object]) -> tuple[int, bytes]: data = encode(frame) self.seq += 1 kind = frame.get("type") key: object = None if isinstance(kind, str) and kind in COALESCE: field = COALESCE[kind] key = (kind, frame.get(field) if field else None) older = self.keys.pop(key, None) if older is not None: self._remove(older) self.entries[self.seq] = (data, key) self.size += len(data) if key is not None: self.keys[key] = self.seq self._bound() return self.seq, data
def ack(self, upto: int) -> None: while self.entries: seq = next(iter(self.entries)) if seq > upto: return self._remove(seq)
def pending(self) -> list[tuple[int, bytes]]: return [(seq, data) for seq, (data, _) in self.entries.items()]
def _remove(self, seq: int) -> None: entry = self.entries.pop(seq, None) if entry is None: return data, key = entry self.size -= len(data) if key is not None and self.keys.get(key) == seq: del self.keys[key]
def _bound(self) -> None: def over() -> bool: return len(self.entries) > self.limit_frames or self.size > self.limit_bytes
if not over(): return # `keys` indexes every coalescing entry (one per live handle or cell), # so shedding them oldest first never walks the whole outbox. while over() and self.keys: self._remove(min(self.keys.values())) while over() and len(self.entries) > 1: self._remove(next(iter(self.entries))) self.dropped += 1
class Inbound: """The newest daemon frame applied; anything at or below it is a replay."""
def __init__(self) -> None: self.last = 0
def accept(self, seq: int) -> bool: if seq <= self.last: return False self.last = seq return True
class JobBook: """Job groups this kernel still owns, read from the frames it sends.
The same bookkeeping the daemon keeps: a started job's group is owned until a job frame proves it gone. A reattaching daemon learns the groups from the hello. The kernel stays alive past its grace while a job is live. """
def __init__(self) -> None: self.started: dict[str, dict[str, object]] = {} self.external = 0 self.awaited_external = 0
def observe(self, frame: dict[str, object]) -> None: kind, id = frame.get("type"), frame.get("id") if kind == "job_start" and isinstance(id, str): pgid = frame.get("pgid") if isinstance(pgid, int) and not isinstance(pgid, bool) and pgid > 1: self.started[id] = frame elif kind == "job" and isinstance(id, str): cleanup = frame.get("cleanup") if isinstance(cleanup, dict) and cleanup.get("gone") is True: self.started.pop(id, None) elif kind == "jobs": live = count_of(frame.get("live")) awaited = count_of(frame.get("awaited", 0)) if live is not None and awaited is not None: self.external, self.awaited_external = live, awaited
def live(self) -> int: return len(self.started) + self.external
class StdioLink: """Raw frames over the inherited stdin/stdout; the pipe closing ends us."""
def __init__(self) -> None: self.control_in: BinaryIO = os.fdopen(os.dup(0), "rb", buffering=0) self.control_out: BinaryIO = os.fdopen(os.dup(1), "wb", buffering=0) self.lock = threading.Lock() self.jobs = JobBook()
def send(self, frame: dict[str, object]) -> None: data = encode(frame) with self.lock: self.jobs.observe(frame) write_frame(self.control_out.write, data)
def serve(self, dispatch: Callable[[object], None]) -> None: """Deliver frames until the pipe closes; raises when it does.""" while True: dispatch(read_frame(self.control_in.read))
def attached(self) -> bool: return True
class SocketLink: """The session layer over a unix socket in the kernel's run directory."""
def __init__( self, run_dir: str, token: str, hello: Callable[[], dict[str, object]], *, bundle: str, grace: float, ) -> None: self.run_dir = run_dir self.token = token self.hello = hello self.bundle = bundle self.grace = grace self.lock = threading.RLock() self.outbox = Outbox() self.inbound = Inbound() self.jobs = JobBook() self.epoch = 0 self.connection: socket.socket | None = None self.detached_at = time.monotonic() self.listener = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) # Bound relative to the run directory: sun_path holds ~104 bytes and a # temporary ALBEDO_HOME alone can use most of that. previous = os.getcwd() os.chdir(run_dir) try: if os.path.exists(SOCKET): os.unlink(SOCKET) self.listener.bind(SOCKET) os.chmod(SOCKET, 0o600) finally: os.chdir(previous) self.listener.listen(4)
def send(self, frame: dict[str, object]) -> None: with self.lock: self.jobs.observe(frame) seq, data = self.outbox.add(frame) self._write(envelope(seq, self.inbound.last, data))
def attached(self) -> bool: return self.connection is not None
def idle_for(self) -> float: """Seconds with nothing attached; zero while attached.""" if self.connection is not None: return 0.0 return time.monotonic() - self.detached_at
def serve(self, dispatch: Callable[[object], None]) -> None: """Accept attaches forever; each connection reads on its own thread.""" while True: connection, _ = self.listener.accept() threading.Thread( target=self._session, args=(connection, dispatch), daemon=True ).start()
def _write(self, data: bytes) -> None: connection = self.connection if connection is None: return try: write_frame(connection.send, data) except OSError: self._detach(connection)
def _detach(self, connection: socket.socket) -> None: with self.lock: if self.connection is connection: self.connection = None self.detached_at = time.monotonic() try: connection.close() except OSError: pass
def _attach(self, connection: socket.socket) -> bool: connection.settimeout(ATTACH_TIMEOUT) message = read_frame(connection.recv) attach = message.get("attach") if isinstance(message, dict) else None token = attach.get("token") if isinstance(attach, dict) else None if not isinstance(attach, dict) or not isinstance(token, str): write_frame(connection.send, encode({"refused": "expected attach"})) return False if not hmac.compare_digest(token.encode(), self.token.encode()): write_frame(connection.send, encode({"refused": "wrong token"})) return False connection.settimeout(None) grace = attach.get("grace") ack = attach.get("ack") with self.lock: if isinstance(grace, int | float) and not isinstance(grace, bool): self.grace = max(float(grace), 0.0) previous, self.connection = self.connection, connection self.epoch += 1 if previous is not None: try: previous.shutdown(socket.SHUT_RDWR) except OSError: pass if isinstance(ack, int) and not isinstance(ack, bool): self.outbox.ack(ack) hello = { **self.hello(), "protocol": PROTOCOL, "bundle": self.bundle, "epoch": self.epoch, "ack": self.inbound.last, "jobs": list(self.jobs.started.values()), "external": self.jobs.external, "awaited_external": self.jobs.awaited_external, "dropped": self.outbox.dropped, } write_frame(connection.send, encode({"hello": hello})) for seq, data in self.outbox.pending(): write_frame(connection.send, envelope(seq, self.inbound.last, data)) return True
def _session( self, connection: socket.socket, dispatch: Callable[[object], None] ) -> None: try: if not self._attach(connection): return while True: message = read_frame(connection.recv) if not isinstance(message, dict): raise ValueError("expected an envelope") ack = message.get("ack") seq = message.get("seq") with self.lock: if isinstance(ack, int) and not isinstance(ack, bool): self.outbox.ack(ack) fresh = isinstance(seq, int) and self.inbound.accept(seq) if fresh: dispatch(message.get("frame")) if isinstance(seq, int): with self.lock: if self.connection is connection: self._write(encode({"ack": self.inbound.last})) except (OSError, EOFError, ValueError): pass finally: self._detach(connection)