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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169"""Run the replay gate once as perf-agent, publish it, then exit with its status."""
import contextlibimport jsonimport osfrom pathlib import Pathimport subprocessimport tempfile
from isolation import IsolationError, Sandboxfrom records import Repo, is_main_pushfrom render import SCANNER, build_run_site, read_result
SHARED_ROOT = "/dev/shm" if Path("/dev/shm").is_dir() else "/tmp"WISP = "github:WaveringAna/wisp.place-monorepo/c85fdc2ff4568018defcb05c08adfc527918144c#wispctl"
def git(*argv): return subprocess.run(["git", *argv], capture_output=True, text=True)
def commit(): return os.environ.get("TANGLED_SHA") or git("rev-parse", "HEAD").stdout.strip()
@contextlib.contextmanagerdef measuring(): """Yield (root, shared, sandbox, runner_env) for runs as perf-agent.
Builds the CLI and a daemon snapshot perf-agent can execute, and stops every perf-agent process before returning, so no secret reaches an argv beside it. """ os.environ["ALBEDO_NO_BROWSER"] = "1" subprocess.run( ["go", "-C", "cli", "build", "-o", "bin/albedo", "./cmd/albedo"], check=True ) try: sandbox = Sandbox.open(protected_pids=(os.getpid(), os.getppid())) except IsolationError as error: # Fail closed: model-controlled code never runs beside this uid's secrets. raise SystemExit(f"perf isolation probe failed: {error}") from None with ( tempfile.TemporaryDirectory(prefix="albedo-perf-ci-") as temporary, # Not under TMPDIR: nix develop makes that a 0700 directory of this uid. tempfile.TemporaryDirectory( prefix="albedo-perf-snapshot-", dir=SHARED_ROOT ) as snapshot, ): root, shared = Path(temporary), Path(snapshot) # perf-agent reads the snapshot; it holds compiled code only. shared.chmod(0o755) try: daemon = subprocess.check_output( ["test/snapshot-daemon.sh", str(shared / "daemon")], text=True ).strip() if not sandbox.can_execute(daemon): raise SystemExit( f"perf-agent cannot execute the daemon snapshot {daemon}" ) yield ( root, shared, sandbox, {"ALBEDO_TEST_DAEMON": daemon, "PERF_COMMIT": commit()}, ) finally: sandbox.close()
def baseline(shared, sha): """The latest main-push record whose commit is an ancestor of this one.""" if git("rev-parse", "--is-shallow-repository").stdout.strip() == "true": git("fetch", "--unshallow", "origin") repo = Repo() runs = sorted( (run for run in repo.runs() if is_main_push(run) and run["commit"] != sha), key=lambda run: (run["committedAt"], run["createdAt"]), reverse=True, ) for run in runs: if git("merge-base", "--is-ancestor", run["commit"], sha).returncode == 0: path = shared / "baseline.json" data = repo.blob(run["result"]) SCANNER.check(data, label="baseline") path.write_bytes(data) path.chmod(0o644) print(f"baseline: {run['commit']} ({run['site']})", flush=True) return path print("baseline: none recorded for an ancestor; budgets only", flush=True) return None
def replay(root, shared, sandbox, runner_env): """Replay as perf-agent; copy its result into root and return its exit code.""" output = sandbox.directory / "results.json" base = baseline(shared, runner_env["PERF_COMMIT"]) measured = sandbox.run( [ "python3", "test/perf/run.py", "--mode", "replay", "--runs", "5", "--out", str(output), *(["--baseline", str(base)] if base else []), ], runner_env, # Spindle stops a workflow at 40 minutes; fail here first, with a reason. timeout=1800, ) data = sandbox.read(output) if data is None: raise SystemExit(measured.returncode or 1) SCANNER.check(data, label="replay result") (root / "results.json").write_bytes(data) return measured.returncode
def main(): subprocess.run( ["python3", "-m", "unittest", "discover", "-s", "test/perf", "-p", "*_test.py"], check=True, ) for test in ("report_test.py", "gateway_test.py"): subprocess.run(["python3", f"test/perf/report/{test}"], check=True) with measuring() as (root, shared, sandbox, runner_env): measurement = replay(root, shared, sandbox, runner_env) result_path = root / "results.json" result = read_result(result_path) # run.py writes after every scenario, so a crash leaves a partial # result with no gated numbers. It is not a run; publish nothing. if not result["gated"]: scenarios = ", ".join(result["scenarios"]) or "none" raise SystemExit( f"replay did not finish (scenarios done: {scenarios}); nothing published" ) # Logs remain useful on fork PRs where no site can be published. print(json.dumps(result, indent=2), flush=True) # Render here too, so a page that cannot be built fails without secrets. build_run_site( root / "site", result, {"did": "", "collection": "", "rkey": "", "pds": None}, ) publication = 0 if os.environ.get("WISP_APP_PASSWORD"): publication = subprocess.run( [ "nix", "shell", WISP, "-c", "python3", "test/perf/report/publish.py", "replay", str(result_path), ] ).returncode else: print("publish skipped: WISP_APP_PASSWORD is unset", flush=True) # Neither a successful deploy nor an unavailable publisher can hide red. raise SystemExit(measurement or (result["verdict"] != "pass") or publication)
if __name__ == "__main__": main()