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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225"""Opt-in macOS benchmark: live task with the actual CLI attached to a PTY.Uses the active ~/.albedo provider in an isolated home; never prints credentials."""import argparseimport contextlibimport fcntlimport jsonimport osfrom pathlib import Pathimport ptyimport selectorsimport signalimport statisticsimport structimport subprocessimport termiosimport timeimport urllib.request
ROOT = Path(__file__).resolve().parents[2]
def cpu_seconds(value): parts = value.split(":") return sum(float(part) * 60 ** position for position, part in enumerate(reversed(parts)))
def processes(): text = subprocess.check_output(["ps", "-axo", "pid=,ppid=,rss=,time=,comm="], text=True) rows = {} for line in text.splitlines(): fields = line.split(None, 4) if len(fields) == 5: pid, parent, rss, cpu, command = fields rows[int(pid)] = dict(parent=int(parent), rss_kib=int(rss), cpu=cpu_seconds(cpu), command=command) return rows
def summarize(samples): result = {} for phase in sorted({sample["phase"] for sample in samples}): selected = [sample for sample in samples if sample["phase"] == phase] elapsed = sum(sample["interval"] for sample in selected) groups = {} for group in ("daemon", "python", "cli", "tools", "total"): values = [sample["groups"][group] for sample in selected] groups[group] = { "mean_cpu_pct": round(sum(value["cpu_seconds"] for value in values) / elapsed * 100, 2), "peak_cpu_pct": round(max(value["cpu_seconds"] / sample["interval"] * 100 for value, sample in zip(values, selected)), 2), "mean_rss_mib": round(sum(value["rss_kib"] * sample["interval"] for value, sample in zip(values, selected)) / elapsed / 1024, 2), "peak_rss_mib": round(max(value["rss_kib"] for value in values) / 1024, 2), } result[phase] = {"seconds": round(elapsed, 2), "samples": len(selected), "groups": groups} return result
def run(output, prompt, timeout): output.mkdir(parents=True, exist_ok=False, mode=0o700) home = output / "home" home.mkdir(mode=0o700) workspace = output / "workspace" workspace.mkdir() saved = json.loads((Path.home() / ".albedo/config.json").read_text()) name = saved.get("active", "default") provider = saved.get("providers", {"default": saved})[name] config = home / "config.json" fd = os.open(config, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as file: json.dump({"active": name, "providers": {name: provider}}, file) metadata = {"provider": name, "model": provider["model"], "protocol": provider["protocol"], "terminal": {"columns": 120, "rows": 40}, "sample_seconds": 1, "cpu_basis": "100% = one logical core; deltas of sampled ps CPU time", "memory_basis": "RSS; sums may count shared pages more than once; short-lived processes can be missed"} del saved, provider (output / "task.txt").write_text(prompt) (output / "metadata.json").write_text(json.dumps(metadata, indent=2)) env = {**os.environ, "ALBEDO_HOME": str(home), "TERM": "xterm-256color"} connection = None cli = None master = None samples = [] owners = {} previous = {} terminal_bytes = 0 try: created = subprocess.run(["node", str(ROOT / "cli/bin/albedo.mjs"), "new", str(workspace)], cwd=ROOT, env=env, text=True, capture_output=True, timeout=60, check=True) session = json.loads(created.stdout)["session"] connection = json.loads((home / "daemon.json").read_text()) def api(path, body=None): req = urllib.request.Request(f"http://127.0.0.1:{connection['port']}" + path, headers={"Authorization": "Bearer " + connection["token"], "Content-Type": "application/json"}, data=None if body is None else json.dumps(body).encode()) with urllib.request.urlopen(req, timeout=15) as response: return json.load(response) master, slave = pty.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 40, 120, 0, 0)) cli = subprocess.Popen(["node", str(ROOT / "cli/bin/albedo.mjs"), "resume", session], cwd=workspace, env=env, stdin=slave, stdout=slave, stderr=slave, start_new_session=True) os.close(slave) os.set_blocking(master, False) owners[connection["pid"]] = "daemon" owners[cli.pid] = "cli" selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) phase = "idle_before" started = last_sample = time.monotonic() phase_start = started next_sample = started + 1 submitted = None with (output / "terminal.log").open("wb") as terminal, (output / "samples.jsonl").open("w") as trace: while True: for key, _ in selector.select(max(0, next_sample - time.monotonic())): try: chunk = os.read(key.fd, 65536) except (BlockingIOError, OSError): chunk = b"" if chunk: terminal.write(chunk) terminal_bytes += len(chunk) now = time.monotonic() if cli.poll() is not None: raise RuntimeError("attached cli exited during benchmark; inspect terminal.log") if now < next_sample: continue rows = processes() for pid, row in rows.items(): if pid in owners: continue lineage = [] ancestor = pid while ancestor in rows and ancestor not in owners and ancestor not in lineage: lineage.append(ancestor) ancestor = rows[ancestor]["parent"] if ancestor not in owners: continue owner = owners[ancestor] for descendant in reversed(lineage): if owner == "python" or owner == "tools": owner = "tools" elif owner == "daemon" and "python" in Path(rows[descendant]["command"]).name.lower(): owner = "python" owners[descendant] = owner groups = {group: {"rss_kib": 0, "cpu_seconds": 0} for group in ("daemon", "python", "cli", "tools", "total")} active_pids = {} for pid, group in owners.items(): if pid not in rows: continue row = rows[pid] cpu = max(0, row["cpu"] - previous.get(pid, row["cpu"] if not samples else 0)) previous[pid] = row["cpu"] groups[group]["rss_kib"] += row["rss_kib"] groups[group]["cpu_seconds"] += cpu active_pids[str(pid)] = group for group in ("daemon", "python", "cli", "tools"): for metric in ("rss_kib", "cpu_seconds"): groups["total"][metric] += groups[group][metric] status = api(f"/sessions/{session}/status") sample = {"elapsed": now - started, "interval": now - last_sample, "phase": phase, "groups": groups, "pids": active_pids, "status": status, "terminal_bytes": terminal_bytes} samples.append(sample) trace.write(json.dumps(sample) + "\n") trace.flush() if len(samples) % 15 == 0: print(json.dumps({"elapsed": round(now-started), "phase": phase, "rss_mib": round(groups["total"]["rss_kib"]/1024, 1), "terminal_bytes": terminal_bytes}), flush=True) last_sample = now next_sample = now + 1 if phase == "idle_before" and now - phase_start >= 8: if terminal_bytes == 0: raise RuntimeError("attached cli produced no terminal output") api(f"/sessions/{session}/events", {"content": prompt}) phase = "active" submitted = phase_start = time.monotonic() print(json.dumps({"event": "submitted", "session": session}), flush=True) elif phase == "active" and not status["running"]: phase = "idle_after" phase_start = now elif phase == "active" and submitted and now-submitted > timeout: api(f"/sessions/{session}/interrupt", {}) raise TimeoutError("live task exceeded benchmark deadline") elif phase == "idle_after" and now - phase_start >= 8: break req = urllib.request.Request(f"http://127.0.0.1:{connection['port']}/sessions/{session}/stream", headers={"Authorization": "Bearer " + connection["token"]}) with urllib.request.urlopen(req, timeout=15) as response: while line := response.readline(): if line.startswith(b"data: "): events = json.loads(line[6:])["events"] break else: raise RuntimeError("no final session snapshot") (output / "events.json").write_text(json.dumps(events, indent=2)) summary = {"metadata": metadata, "session": session, "terminal_bytes": terminal_bytes, "tool_calls": sum(event.get("type") == "tool" for event in events), "final_messages": [event["text"] for event in events if event.get("type") == "message"][-2:], "phases": summarize(samples)} (output / "summary.json").write_text(json.dumps(summary, indent=2)) print(json.dumps(summary, indent=2), flush=True) finally: if samples and not (output / "summary.json").exists(): (output / "partial-summary.json").write_text(json.dumps(summarize(samples), indent=2)) if cli and cli.poll() is None: with contextlib.suppress(ProcessLookupError): os.killpg(cli.pid, signal.SIGTERM) try: cli.wait(timeout=5) except subprocess.TimeoutExpired: os.killpg(cli.pid, signal.SIGKILL) cli.wait() if master is not None: os.close(master) if connection: with contextlib.suppress(Exception): api("/shutdown", {}) config.unlink(missing_ok=True)
if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("output", type=Path) parser.add_argument("prompt", type=Path) parser.add_argument("--timeout", type=int, default=1200) args = parser.parse_args() run(args.output.resolve(), args.prompt.read_text(), args.timeout)