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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151"""Checked group termination for the erlang supervisor.
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 (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
import signalimport threading
import asyncioimport jsonimport osimport sys
import albedo_procimport albedo_identity
TERM_MS = 250KILL_MS = 1000 # the supervisor's remaining budget is short: this is the last resort
def budget(request: dict, name: str, default: int) -> float: """Seconds for one ladder step, clamped to something a caller can wait for.""" value = request.get(name, default) return ( min(max(value, 10), default) / 1000 if isinstance(value, int) else default / 1000 )
def target(spec: object) -> albedo_proc.Group: """One requested target: a process group, or one process that leads none.""" if not isinstance(spec, dict): raise ValueError("target must be an object") pgid = spec.get("pgid") operand = pgid if isinstance(pgid, int) else spec.get("pid") if not isinstance(operand, int) or operand <= 1: raise ValueError("target needs a pgid or pid above 1") leader = spec.get("leader") whole = isinstance(pgid, int) if not whole: try: whole = os.getpgid(operand) == operand except (ProcessLookupError, PermissionError): pass return albedo_proc.Group( operand, leader if isinstance(leader, str) else None, whole )
async def supervise_async(request: object) -> dict[str, object]: if not isinstance(request, dict): raise ValueError("request must be an object") if "discover" in request: home, known = request["discover"], request.get("known", []) if ( not isinstance(home, str) or not isinstance(known, list) or not all(isinstance(path, str) for path in known) ): raise ValueError("discovery needs a home and known run directories") return {"kernels": albedo_identity.discover(home, known)} specs = request.get("targets") if not isinstance(specs, list): raise ValueError("request needs a target list") targets = [target(spec) for spec in specs] endings = await albedo_proc.terminate( targets, term=budget(request, "term_ms", TERM_MS), kill=budget(request, "kill_ms", KILL_MS), ) return { "targets": [ {**ending.as_json(), "label": spec.get("label")} for spec, ending in zip(specs, endings) ] }
def supervise(request: object) -> dict[str, object]: return asyncio.run(supervise_async(request))
async def serve() -> None: """Answer one JSON request per stdin line until end of input.
The daemon keeps one of these running, so a stop does not pay for an interpreter start. Requests run concurrently; each answer is one stdout line carrying the request's id. The first line says the interpreter is ready. """ loop = asyncio.get_running_loop() lines: asyncio.Queue[str | None] = asyncio.Queue()
def read() -> None: for line in sys.stdin: loop.call_soon_threadsafe(lines.put_nowait, line) loop.call_soon_threadsafe(lines.put_nowait, None)
threading.Thread(target=read, daemon=True).start()
async def answer(line: str) -> None: identity = None try: message = json.loads(line) identity = message["id"] reply: dict[str, object] = { "ok": True, **await supervise_async(message["request"]), } except Exception as error: reply = {"ok": False, "error": f"{type(error).__name__}: {error}"} print(json.dumps({"id": identity, "answer": reply}), flush=True)
print(json.dumps({"ready": True}), flush=True) running: set[asyncio.Task[None]] = set() while (line := await lines.get()) is not None: task = asyncio.create_task(answer(line)) running.add(task) task.add_done_callback(running.discard) await asyncio.gather(*running)
def main(request: str) -> int: try: answer: dict[str, object] = {"ok": True, **supervise(json.loads(request))} status = 0 except Exception as error: answer = {"ok": False, "error": f"{type(error).__name__}: {error}"} status = 2 print(json.dumps(answer)) return status
if __name__ == "__main__": if sys.argv[1:] == ["--serve"]: asyncio.run(serve()) sys.exit(0) # 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) request = sys.argv[1] if len(sys.argv) > 1 else "{}" sys.exit(main(sys.stdin.readline() if request == "-" else request))