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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269"""One human command: no TTY, no kernel, bounded sanitized packet output.
The BEAM port owns our stdin. EOF ends the entire command group, including onan unclean daemon death. One acknowledged frame at a time bounds pipe/mailboxes."""
from __future__ import annotations
import asyncioimport codecsimport jsonimport osfrom pathlib import Pathimport pwdimport selectimport structimport subprocessimport sysimport threading
import albedo_proc
CAP = 16 * 1024 * 1024
def read_exact(size: int) -> bytes: parts = bytearray() while len(parts) < size: chunk = os.read(0, size - len(parts)) if not chunk: break parts.extend(chunk) return bytes(parts)
def packet() -> bytes | None: header = read_exact(4) if not header: return None if len(header) != 4: raise EOFError("incomplete shell control frame") size = struct.unpack(">I", header)[0] if size > 262144: raise ValueError("oversized shell control frame") data = read_exact(size) if len(data) != size: raise EOFError("incomplete shell control frame") return data
class Sanitizer: """Incremental UTF-8 and ECMA-48 stripping, with constant parser state."""
def __init__(self) -> None: self.decoder = codecs.getincrementaldecoder("utf-8")("replace") self.state = "text"
def feed(self, data: bytes, final: bool = False) -> str: kept = [] for char in self.decoder.decode(data, final): n = ord(char) if self.state == "esc": self.state = ( "csi" if char == "[" else "string" if char in "]PX^_" else "text" ) elif self.state == "csi": if 0x40 <= n <= 0x7E: self.state = "text" elif self.state == "string": if char == "\x07" or n == 0x9C: self.state = "text" elif char == "\x1b": self.state = "string_esc" elif self.state == "string_esc": self.state = "text" if char == "\\" else "string" elif char == "\x1b": self.state = "esc" elif n == 0x9B: self.state = "csi" elif n in (0x90, 0x98, 0x9D, 0x9E, 0x9F): self.state = "string" elif char in "\n\t" or n >= 0x20 and not 0x7F <= n <= 0x9F: kept.append(char) return "".join(kept)
def shell_args(command: str) -> tuple[list[str], str]: shell = os.environ.get("SHELL") if not shell: try: shell = pwd.getpwuid(os.getuid()).pw_shell except KeyError: shell = None shell = shell or "/bin/sh" name = Path(shell).name if name in ("csh", "tcsh"): return ["-" + name, "-c", command], shell if name in ("nu", "nushell"): return [shell, "--login", "-c", command], shell return [shell, "-l", "-c", command], shell
def main() -> None: initial = packet() if initial is None: return config = json.loads(initial) ack = threading.Event() disconnected = threading.Event() cancelled = threading.Event() lock = threading.Lock() child: subprocess.Popen[bytes] | None = None group: albedo_proc.Group | None = None cleanup_error: str | None = None
def end_group() -> None: nonlocal group, cleanup_error with lock: if group is not None: ending = asyncio.run(albedo_proc.terminate([group]))[0] if ending.gone: group = None else: cleanup_error = ending.report()
def controls() -> None: try: while (data := packet()) is not None: if data == b"stop": cancelled.set() end_group() elif data == b"ack": ack.set() except (OSError, EOFError, ValueError): pass disconnected.set() cancelled.set() ack.set() end_group()
def emit(value: dict[str, object]) -> None: if disconnected.is_set(): raise BrokenPipeError ack.clear() data = json.dumps(value, ensure_ascii=False).encode() sys.stdout.buffer.write(struct.pack(">I", len(data)) + data) sys.stdout.buffer.flush() ack.wait() if disconnected.is_set(): raise BrokenPipeError
# Start the child before the control reader so an immediate EOF cannot race # group publication and leave a just-started child without cleanup. try: args, executable = shell_args(config["command"]) child = subprocess.Popen( args, executable=executable, cwd=config["cwd"], stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, start_new_session=True, ) group = albedo_proc.Group(child.pid, albedo_proc.leader_token(child.pid)) threading.Thread(target=controls, daemon=True).start() assert child.stdout is not None fd = child.stdout.fileno() os.set_blocking(fd, False) sanitizer = Sanitizer() captured = 0 truncated = False
def capture(data: bytes, final: bool = False) -> None: nonlocal captured, truncated text = sanitizer.feed(data, final) encoded = text.encode() kept = ( "" if truncated else encoded[: max(0, CAP - captured)].decode("utf-8", "ignore") ) dropped = len(kept.encode()) < len(encoded) changed = dropped and not truncated truncated = truncated or dropped captured += len(kept.encode()) if kept or changed: emit({"type": "output", "text": kept, "truncated": truncated})
output_eof = False while True: if cleanup_error is not None: raise RuntimeError(cleanup_error) if child.poll() is not None: # A shell may exit while a background descendant holds stdout. # End all descendants before bounded final draining. end_group() if cleanup_error is not None: raise RuntimeError(cleanup_error) while True: try: data = os.read(fd, 16384) except BlockingIOError: break if not data: break capture(data) break if output_eof: select.select([], [], [], 0.1) continue ready, _, _ = select.select([fd], [], [], 0.1) if ready: data = os.read(fd, 16384) if data: capture(data) else: # EOF does not mean the process stopped: it can close both # output descriptors and keep running. Keep control alive. output_eof = True capture(b"", final=True) end_group() if cleanup_error is not None: raise RuntimeError(cleanup_error) code = child.wait() emit( { "type": "ended", "status": "cancelled" if cancelled.is_set() else "completed", "exit_code": code, "error": None, } ) except BrokenPipeError: pass except Exception as error: threading.Thread( target=controls, daemon=True ).start() if child is None else None try: emit( { "type": "ended", "status": "failed", "exit_code": None, "error": Sanitizer() .feed(str(error).encode(), final=True) .encode()[:4096] .decode("utf-8", "ignore"), } ) except BrokenPipeError: pass finally: end_group() if child is not None: try: child.wait(timeout=3) except subprocess.TimeoutExpired: # The checked cleanup failure is explicit in the terminal frame; # never keep a supervisor stuck forever on an unsignalable child. pass if child.stdout is not None: child.stdout.close()
if __name__ == "__main__": main()