"""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 daemon interprets the snapshot with the same code it uses on its own disk (albedo/daemon/folders.gleam, albedo/harness/vcs.gleam), so this side only collects: directory entries, which repository markers exist, the vcs commands the daemon planned, file sizes, and project files. Everything comes back in one ssh round trip per question. Request: `{route, dir, plans, tracked, deadline, counted}`, `dir` absolute and normalised. `route` is list, repo, preview or project, or exists, which answers `directory` alone. `tracked` names, per kind, the planned command that 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; only commands that exit 0 within the deadline answer. """ from __future__ import annotations import base64 import json import os import stat import subprocess import sys from concurrent.futures import ThreadPoolExecutor MARKERS = (".jj", ".git") SUBDIRECTORIES = 256 # child directories whose entries a preview lists PROJECT_FILES = 512 PROJECT_BYTES = 8 * 1024 * 1024 FILE_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 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"] 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())