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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247"""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, tracked, deadline, counted}`, `dir` absoluteand normalised. `route` is list, repo, preview or project, or exists, whichanswers `directory` alone. The file route takes `{route, path, limit}` insteadand answers `{size, modified, data}` (base64, left out past `limit`) or`{error}`. `tracked` names, per kind, the planned commandthat lists a preview's tracked files and the separator its output uses.
Snapshot: `{directory, entries: {dir: [[name, dir, modified, symlink]]}, 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): path = child(directory, name) symlink = os.path.islink(path) try: info = os.stat(path) listed.append( [name, stat.S_ISDIR(info.st_mode), int(info.st_mtime), symlink] ) except OSError: listed.append([name, False, 0, symlink]) 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 read(path: str, limit: int) -> dict[str, object]: """A regular file's size, mtime and bytes: at most one byte past `limit`, so a file that grew since the stat still reads as too large.""" try: info = os.stat(path) except (FileNotFoundError, NotADirectoryError): return {"error": "does not exist"} except OSError: return {"error": "cannot be read"} if stat.S_ISDIR(info.st_mode): return {"error": "is a directory"} if not stat.S_ISREG(info.st_mode): return {"error": "is not a regular file"} found: dict[str, object] = {"size": info.st_size, "modified": int(info.st_mtime)} if info.st_size > limit: return found try: with open(path, "rb") as handle: data = handle.read(limit + 1) except OSError: return {"error": "cannot be read"} found["data"] = base64.b64encode(data).decode() return found
def gather(request: dict) -> dict[str, object]: if request["route"] == "file": return read(request["path"], int(request["limit"])) route, directory = request["route"], request["dir"] snapshot: dict[str, object] = {"directory": os.path.isdir(directory)} if not snapshot["directory"] or route == "exists": 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, _, _symlink 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, _, _symlink 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": relatives: list[str] = [] tracked = request.get("tracked", {}).get(kind) if tracked is not None: program, args, _ = tracked["command"] listing = outputs.get(key(program, args)) or "" relatives = listing.split(tracked["separator"]) sizes: dict[str, int] = {} for relative in relatives[: 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())