Publish Beam Boot / Super Beam source

324ecd2 verified 7 days ago

16 kB

"""Beam plays SUPERBEAM: the hosted Beam model picks every action from an ASCII frame.
The game runs locally in the reference CHIP-8 emulator (beamboot/superbeam_gym.py).
A small Modal function proxies chat requests to the HF router; the HF token stays
in the Modal secret `huggingface-secret` and is never printed.
  OMP_NUM_THREADS=1 .venv/bin/modal run scripts/beam_play.py               # real run
  OMP_NUM_THREADS=1 .venv/bin/modal run scripts/beam_play.py --dry-run     # scripted stand-in, no HF calls
Budget: <= --max-requests chat requests per attempt (default 80) and
<= --total-cap overall (default 200, i.e. up to 3 attempts).  Every HTTP attempt,
including retries, counts toward the cap.
Writes results/beam_play_run.json (every request/response, actions, rewards,
frames) and results/beam_key_schedule.json (best attempt's key schedule, for
replay on the in-weights backend).
"""
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request

import modal

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
RUN_OUT = os.path.join(ROOT, "results", "beam_play_run.json")
SCHED_OUT = os.path.join(ROOT, "results", "beam_key_schedule.json")
MODEL = "reflection/Beam-1.0-501B-A23B:baseten"

app = modal.App("beamboot-beam-play")


@app.function(image=modal.Image.debian_slim(python_version="3.12"),
              secrets=[modal.Secret.from_name("huggingface-secret")], timeout=240)
def beam_chat(messages, tools, model=MODEL, max_tokens=6144):
    token = (os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
             or os.environ.get("HUGGINGFACE_TOKEN") or os.environ.get("HUGGINGFACEHUB_API_TOKEN"))
    if not token:
        raise RuntimeError("Hugging Face token was not found under a supported environment "
                           "variable name in the Modal secret")
    payload = {"model": model, "messages": messages, "max_tokens": max_tokens}
    if tools:
        payload.update(tools=tools, tool_choice="auto")
    req = urllib.request.Request("https://router.huggingface.co/v1/chat/completions",
                                 data=json.dumps(payload).encode(),
                                 headers={"Authorization": "Bearer " + token,
                                          "Content-Type": "application/json"})
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=220) as response:
            out = json.load(response)
    except urllib.error.HTTPError as error:
        raise RuntimeError(f"Hugging Face router returned HTTP {error.code}: "
                           f"{error.read(1000).decode(errors='replace')}") from None
    out["_latency_s"] = round(time.time() - t0, 2)
    return out


# ----------------------------------------------------------------- prompt
ACTION_NAMES = ["noop", "left", "right", "jump", "right+jump", "left+jump"]
TOOL = {"type": "function", "function": {
    "name": "act",
    "description": "Press the controller for the next 4 frames.",
    "parameters": {"type": "object",
                   "properties": {"action": {"type": "string", "enum": ACTION_NAMES}},
                   "required": ["action"], "additionalProperties": False}}}

SYSTEM = """You are playing SUPERBEAM, an original side-scrolling platformer running on a CHIP-8 computer (64x32 pixel screen).
Each turn you get the screen as ASCII art plus a few facts, then you choose ONE action by calling the `act` tool.
Legend: P = you, E = walker enemy, # = ground, H = pipe (solid block), o = coin, F = flag pole, C = castle, . = empty.
Rules:
- Goal: reach the flag pole far to the right. The camera scrolls as you move right.
- Falling into a gap in the ground (columns where the ground rows are '.') kills you.
- Pipes are solid: you cannot walk through them, you can stand on top. A pipe is 8 px tall and can be
  jumped onto from the ground. A tall pipe is 16 px tall: climb it from a pipe next to it. You cannot
  jump while standing on a tall pipe; walk off it.
- Coins: touching one collects it (+points). Low coins: walk into them. Mid coins: jump from the ground
  or walk on a pipe. High coins: only from a pipe top (jump) or by standing on a tall pipe.
- Walkers walk and turn around at gaps and pipes. Touching one from the side kills you; landing on top stomps it.
- One action = 4 frames. Running moves you 8 px per action. A jump rises 10 px, lasts about 9 frames
  (~2 actions) and can only start while standing (on ground or a pipe).
- There is NO momentum: you only move sideways on turns where you choose right/left, in the air too.
  While airborne, choosing "noop" makes you fall straight down (often into the gap). To clear a gap:
  "right+jump" once, then keep choosing "right" every turn until you land (~18 px forward in total).
  To get onto a pipe: walk up to it, "right+jump", then keep choosing "right".
- Good timing: jump over a gap on the turn when it is 1 px ahead of you or less (it may already be
  "under you" while you still stand: you only fall once your back edge leaves the ground); jumping
  earlier lands in a 16 px gap. Jump when a walker is about 4-20 px ahead (landing on it stomps it).
  Otherwise keep running right.
Actions: noop, left, right, jump, right+jump, left+jump.
If you cannot call the tool, reply with exactly one action word and nothing else."""


def describe(env, obs):
    """Short factual legend derived from the game's registers/RAM (SuperBeamEnv.describe): your
    position and footing, gaps, solid blocks (pipes) and coins on screen, walkers, coins, progress."""
    return env.describe(obs)


def view(env):
    """Rows 0..25 of the labelled ASCII frame (rows 26-29 repeat the ground, 30-31 are the coin bar)."""
    rows = env.ascii(legend=True).splitlines()
    return "\n".join(rows[0:26])


def parse_action(message):
    for call in message.get("tool_calls") or []:
        try:
            if call["function"]["name"] == "act":
                a = json.loads(call["function"]["arguments"]).get("action")
                if a in ACTION_NAMES:
                    return a, "tool"
        except (KeyError, ValueError, TypeError, AttributeError):
            pass
    text = (message.get("content") or "").strip().lower()
    words = re.findall(r"(?:left|right)\+jump|noop|left|right|jump", text)
    if words:
        return words[-1], "text"
    return None, "invalid"


# ----------------------------------------------------------------- play loop
def play_attempt(env, chat, max_requests, history_turns=2, log=print):
    obs, info = env.reset()
    obs, _, _, _, info = env.step(0)            # one noop so the first frame is drawn
    history, history_turns_list, records, used = [], [], [], 0
    while used < max_requests:
        facts = describe(env, obs)
        user = {"role": "user", "content": "Screen (rows 0-25, 64 columns):\n%s\n%s\nChoose your action."
                % (view(env), facts)}
        messages = [{"role": "system", "content": SYSTEM}] + history + [user]
        response, error = None, None
        for _try in range(3):
            if used >= max_requests:
                break
            used += 1
            try:
                response = chat(messages, [TOOL])
                break
            except Exception as e:  # provider/network error: record, maybe retry
                error = str(e)[:2000]
                log("request %d failed: %s" % (used, error[:300]))
                if "HTTP 4" in error and "HTTP 429" not in error:
                    break
                time.sleep(3)
        if response is None:
            records.append({"request": used, "messages": messages, "error": error})
            if error and "HTTP 4" in error and "HTTP 429" not in error:
                return records, used, "provider_error: " + error[:500]
            continue
        msg = response["choices"][0]["message"]
        action, how = parse_action(msg)
        a = ACTION_NAMES.index(action) if action else 0
        frame_before = view(env)
        obs, reward, term, trunc, info = env.step(a)
        rec = {"request": used, "messages": messages, "response": response,
               "action": action or "noop", "parsed_from": how, "reward": reward,
               "frame": frame_before, "progress": info["best_progress"],
               "frames": info["frames"], "terminated": term, "truncated": trunc}
        records.append(rec)
        log("req %2d  %-10s (%s)  progress %5.1f%%  frames %3d  reward %+6.2f%s"
            % (used, action or "noop", how, 100 * info["best_progress"], info["frames"], reward,
               "  WIN" if info["won"] else ("  DIED" if info["dead"] else "")))
        # rolling history: assistant reply (with reasoning_content if returned) + result
        assistant = {k: msg[k] for k in ("role", "content", "reasoning_content", "tool_calls") if k in msg}
        assistant.setdefault("role", "assistant")
        result = "done; progress now %.0f%%" % (100 * info["best_progress"])
        # older turns keep only the facts line (no ASCII) to keep requests short
        turn = [{"role": "user", "content": "(earlier turn) " + facts}, assistant]
        calls = msg.get("tool_calls") or []
        for call in calls:
            turn.append({"role": "tool", "tool_call_id": call.get("id", ""), "content": result})
        if not calls:
            turn[1] = {"role": "assistant", "content": action or "noop"}
            if msg.get("reasoning_content"):   # provider rejects assistant turns without their reasoning
                turn[1]["reasoning_content"] = msg["reasoning_content"]
        history_turns_list = (history_turns_list + [turn])[-history_turns:] if history_turns else []
        history = [m for t in history_turns_list for m in t]
        if term or trunc:
            break
    status = "won" if env.won else ("died" if env.dead else
                                    ("truncated" if env.frames >= env.max_frames else "request_cap"))
    return records, used, status


def run(chat, max_requests=80, total_cap=200, frame_skip=4, max_attempts=3, model=MODEL, log=print,
        save=True, backend="emulator", level="classic"):
    global RUN_OUT, SCHED_OUT
    level = int(level) if str(level).isdigit() else level
    tag = "" if level == "classic" else "_level%s" % level
    if backend != "emulator":          # live run against the in-weights computer: separate files
        RUN_OUT = os.path.join(ROOT, "results", "beam_live_weights%s_run.json" % tag)
        SCHED_OUT = os.path.join(ROOT, "results", "beam_live_weights%s_schedule.json" % tag)
    elif tag:                          # held-out / generated level evaluation
        RUN_OUT = os.path.join(ROOT, "results", "levels", "beam%s_run.json" % tag)
        SCHED_OUT = os.path.join(ROOT, "results", "levels", "beam%s_schedule.json" % tag)
    sys.path.insert(0, ROOT)
    from beamboot.superbeam_gym import SuperBeamEnv, WIN_X, START_X, IPF
    est = -(-(WIN_X - START_X) // (2 * frame_skip))
    log("estimate: level needs >= %d frames = >= %d actions at frame_skip=%d (cap %d requests/attempt, %d total)"
        % (-(-(WIN_X - START_X) // 2), est, frame_skip, max_requests, total_cap))
    if est > max_requests:
        raise SystemExit("frame_skip too small for the request cap")
    attempts, total = [], 0
    for k in range(max_attempts):
        budget = min(max_requests, total_cap - total)
        if budget <= 0:
            break
        env = SuperBeamEnv(backend=backend, frame_skip=frame_skip, max_frames=600, level=level,
                           device="cuda" if backend != "emulator" else "cpu")
        log("=== attempt %d (budget %d requests, backend %s) ===" % (k + 1, budget, backend))
        t0 = time.time()
        records, used, status = play_attempt(env, chat, budget, log=log)
        total += used
        attempts.append({"attempt": k + 1, "status": status, "requests_used": used,
                         "won": env.won, "dead": env.dead, "progress": env._info()["best_progress"],
                         "frames": env.frames, "actions": [ACTION_NAMES[a] for a in env.actions],
                         "key_schedule": env.key_schedule(), "requests": records,
                         "backend": backend, "wall_seconds": round(time.time() - t0, 1)})
        if backend != "emulator":
            st = env.backend.session.stats
            attempts[-1]["weights"] = {
                "frame_rows": env.backend.session.frame_rows,
                "per_frame": [{"f": x.frame, "tokens": x.frame_tokens, "accepted_all": x.accepted_all,
                               "first_reject": x.first_reject, "s": round(x.seconds, 3)} for x in st],
                "frames_all_draft_accepted": sum(x.accepted_all for x in st),
                "compute_seconds": round(sum(x.seconds for x in st), 1)}
        log("attempt %d: %s, progress %.1f%%, %d requests" % (k + 1, status,
                                                               100 * attempts[-1]["progress"], used))
        if env.won or status.startswith("provider_error"):
            break
    best = max(attempts, key=lambda a: (a["won"], a["progress"]))
    # verify the best schedule replays to the same outcome
    venv = SuperBeamEnv(frame_skip=1, max_frames=600, level=level)
    vinfo = venv.replay_schedule(best["key_schedule"], n_frames=best["frames"])
    out = {"model": model, "level": level, "frame_skip": frame_skip, "ipf": IPF,
           "max_requests_per_attempt": max_requests,
           "total_cap": total_cap, "total_requests": total, "best_attempt": best["attempt"],
           "best": {k: best[k] for k in ("status", "won", "progress", "frames", "requests_used")},
           "replay_check": vinfo, "attempts": attempts}
    log("best: attempt %d %s progress %.1f%% | total requests %d | replay %s"
        % (best["attempt"], best["status"], 100 * best["progress"], total, vinfo))
    if not save:
        return out
    os.makedirs(os.path.dirname(RUN_OUT), exist_ok=True)
    with open(RUN_OUT, "w") as f:
        json.dump(out, f, indent=1, default=str)
    with open(SCHED_OUT, "w") as f:
        json.dump({"rom": "roms/superbeam.ch8", "ipf": IPF, "frame_skip": frame_skip,
                   "frames": best["frames"], "won": best["won"], "progress": best["progress"],
                   "format": "[[frame, key, 'down'|'up'], ...] applied before that frame's instructions",
                   "events": best["key_schedule"]}, f, indent=1)
    log("wrote %s and %s" % (os.path.relpath(RUN_OUT, ROOT), os.path.relpath(SCHED_OUT, ROOT)))
    return out


def _dry_chat(messages, tools):
    """Offline stand-in: reads the facts line like the prompt suggests (no HF call)."""
    text = messages[-1]["content"]
    m = re.search(r"standing on (ground|a pipe)", text)
    terrain, walkers = text.split("Walkers:")[0], text.split("Walkers:")[-1].split("Coins collected")[0]
    near = [int(v) for v in re.findall(r"\((-?\d+) px ahead\)", terrain)]
    foes = [int(v) for v in re.findall(r"\((-?\d+) px ahead\)", walkers)]
    act = "right"
    if m and (any(d <= 1 for d in near) or "under you" in terrain.split("Solid")[0]
              or any(4 <= d <= 20 for d in foes)):
        act = "right+jump"
    return {"choices": [{"message": {"role": "assistant", "content": None, "tool_calls": [
        {"id": "dry", "type": "function", "function": {"name": "act",
                                                       "arguments": json.dumps({"action": act})}}]}}]}


@app.local_entrypoint()
def main(dry_run: bool = False, max_requests: int = 80, total_cap: int = 200,
         frame_skip: int = 4, model: str = MODEL, backend: str = "emulator", max_attempts: int = 3,
         level: str = "classic"):
    if dry_run:
        run(_dry_chat, max_requests, total_cap, frame_skip, max_attempts, model="dry-run", save=False,
            backend=backend, level=level)
        return
    run(lambda msgs, tools: beam_chat.remote(msgs, tools, model), max_requests, total_cap,
        frame_skip, max_attempts, model=model, backend=backend, level=level)

	"""Beam plays SUPERBEAM: the hosted Beam model picks every action from an ASCII frame.

	The game runs locally in the reference CHIP-8 emulator (beamboot/superbeam_gym.py).
	A small Modal function proxies chat requests to the HF router; the HF token stays
	in the Modal secret `huggingface-secret` and is never printed.

	OMP_NUM_THREADS=1 .venv/bin/modal run scripts/beam_play.py # real run
	OMP_NUM_THREADS=1 .venv/bin/modal run scripts/beam_play.py --dry-run # scripted stand-in, no HF calls

	Budget: <= --max-requests chat requests per attempt (default 80) and
	<= --total-cap overall (default 200, i.e. up to 3 attempts). Every HTTP attempt,
	including retries, counts toward the cap.

	Writes results/beam_play_run.json (every request/response, actions, rewards,
	frames) and results/beam_key_schedule.json (best attempt's key schedule, for
	replay on the in-weights backend).
	"""
	import json
	import os
	import re
	import sys
	import time
	import urllib.error
	import urllib.request

	import modal

	ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
	RUN_OUT = os.path.join(ROOT, "results", "beam_play_run.json")
	SCHED_OUT = os.path.join(ROOT, "results", "beam_key_schedule.json")
	MODEL = "reflection/Beam-1.0-501B-A23B:baseten"

	app = modal.App("beamboot-beam-play")


	@app.function(image=modal.Image.debian_slim(python_version="3.12"),
	secrets=[modal.Secret.from_name("huggingface-secret")], timeout=240)
	def beam_chat(messages, tools, model=MODEL, max_tokens=6144):
	token = (os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
	or os.environ.get("HUGGINGFACE_TOKEN") or os.environ.get("HUGGINGFACEHUB_API_TOKEN"))
	if not token:
	raise RuntimeError("Hugging Face token was not found under a supported environment "
	"variable name in the Modal secret")
	payload = {"model": model, "messages": messages, "max_tokens": max_tokens}
	if tools:
	payload.update(tools=tools, tool_choice="auto")
	req = urllib.request.Request("https://router.huggingface.co/v1/chat/completions",
	data=json.dumps(payload).encode(),
	headers={"Authorization": "Bearer " + token,
	"Content-Type": "application/json"})
	t0 = time.time()
	try:
	with urllib.request.urlopen(req, timeout=220) as response:
	out = json.load(response)
	except urllib.error.HTTPError as error:
	raise RuntimeError(f"Hugging Face router returned HTTP {error.code}: "
	f"{error.read(1000).decode(errors='replace')}") from None
	out["_latency_s"] = round(time.time() - t0, 2)
	return out


	# ----------------------------------------------------------------- prompt
	ACTION_NAMES = ["noop", "left", "right", "jump", "right+jump", "left+jump"]
	TOOL = {"type": "function", "function": {
	"name": "act",
	"description": "Press the controller for the next 4 frames.",
	"parameters": {"type": "object",
	"properties": {"action": {"type": "string", "enum": ACTION_NAMES}},
	"required": ["action"], "additionalProperties": False}}}

	SYSTEM = """You are playing SUPERBEAM, an original side-scrolling platformer running on a CHIP-8 computer (64x32 pixel screen).
	Each turn you get the screen as ASCII art plus a few facts, then you choose ONE action by calling the `act` tool.
	Legend: P = you, E = walker enemy, # = ground, H = pipe (solid block), o = coin, F = flag pole, C = castle, . = empty.
	Rules:
	- Goal: reach the flag pole far to the right. The camera scrolls as you move right.
	- Falling into a gap in the ground (columns where the ground rows are '.') kills you.
	- Pipes are solid: you cannot walk through them, you can stand on top. A pipe is 8 px tall and can be
	jumped onto from the ground. A tall pipe is 16 px tall: climb it from a pipe next to it. You cannot
	jump while standing on a tall pipe; walk off it.
	- Coins: touching one collects it (+points). Low coins: walk into them. Mid coins: jump from the ground
	or walk on a pipe. High coins: only from a pipe top (jump) or by standing on a tall pipe.
	- Walkers walk and turn around at gaps and pipes. Touching one from the side kills you; landing on top stomps it.
	- One action = 4 frames. Running moves you 8 px per action. A jump rises 10 px, lasts about 9 frames
	(~2 actions) and can only start while standing (on ground or a pipe).
	- There is NO momentum: you only move sideways on turns where you choose right/left, in the air too.
	While airborne, choosing "noop" makes you fall straight down (often into the gap). To clear a gap:
	"right+jump" once, then keep choosing "right" every turn until you land (~18 px forward in total).
	To get onto a pipe: walk up to it, "right+jump", then keep choosing "right".
	- Good timing: jump over a gap on the turn when it is 1 px ahead of you or less (it may already be
	"under you" while you still stand: you only fall once your back edge leaves the ground); jumping
	earlier lands in a 16 px gap. Jump when a walker is about 4-20 px ahead (landing on it stomps it).
	Otherwise keep running right.
	Actions: noop, left, right, jump, right+jump, left+jump.
	If you cannot call the tool, reply with exactly one action word and nothing else."""


	def describe(env, obs):
	"""Short factual legend derived from the game's registers/RAM (SuperBeamEnv.describe): your
	position and footing, gaps, solid blocks (pipes) and coins on screen, walkers, coins, progress."""
	return env.describe(obs)


	def view(env):
	"""Rows 0..25 of the labelled ASCII frame (rows 26-29 repeat the ground, 30-31 are the coin bar)."""
	rows = env.ascii(legend=True).splitlines()
	return "\n".join(rows[0:26])


	def parse_action(message):
	for call in message.get("tool_calls") or []:
	try:
	if call["function"]["name"] == "act":
	a = json.loads(call["function"]["arguments"]).get("action")
	if a in ACTION_NAMES:
	return a, "tool"
	except (KeyError, ValueError, TypeError, AttributeError):
	pass
	text = (message.get("content") or "").strip().lower()
	words = re.findall(r"(?:left\|right)\+jump\|noop\|left\|right\|jump", text)
	if words:
	return words[-1], "text"
	return None, "invalid"


	# ----------------------------------------------------------------- play loop
	def play_attempt(env, chat, max_requests, history_turns=2, log=print):
	obs, info = env.reset()
	obs, _, _, _, info = env.step(0) # one noop so the first frame is drawn
	history, history_turns_list, records, used = [], [], [], 0
	while used < max_requests:
	facts = describe(env, obs)
	user = {"role": "user", "content": "Screen (rows 0-25, 64 columns):\n%s\n%s\nChoose your action."
	% (view(env), facts)}
	messages = [{"role": "system", "content": SYSTEM}] + history + [user]
	response, error = None, None
	for _try in range(3):
	if used >= max_requests:
	break
	used += 1
	try:
	response = chat(messages, [TOOL])
	break
	except Exception as e: # provider/network error: record, maybe retry
	error = str(e)[:2000]
	log("request %d failed: %s" % (used, error[:300]))
	if "HTTP 4" in error and "HTTP 429" not in error:
	break
	time.sleep(3)
	if response is None:
	records.append({"request": used, "messages": messages, "error": error})
	if error and "HTTP 4" in error and "HTTP 429" not in error:
	return records, used, "provider_error: " + error[:500]
	continue
	msg = response["choices"][0]["message"]
	action, how = parse_action(msg)
	a = ACTION_NAMES.index(action) if action else 0
	frame_before = view(env)
	obs, reward, term, trunc, info = env.step(a)
	rec = {"request": used, "messages": messages, "response": response,
	"action": action or "noop", "parsed_from": how, "reward": reward,
	"frame": frame_before, "progress": info["best_progress"],
	"frames": info["frames"], "terminated": term, "truncated": trunc}
	records.append(rec)
	log("req %2d %-10s (%s) progress %5.1f%% frames %3d reward %+6.2f%s"
	% (used, action or "noop", how, 100 * info["best_progress"], info["frames"], reward,
	" WIN" if info["won"] else (" DIED" if info["dead"] else "")))
	# rolling history: assistant reply (with reasoning_content if returned) + result
	assistant = {k: msg[k] for k in ("role", "content", "reasoning_content", "tool_calls") if k in msg}
	assistant.setdefault("role", "assistant")
	result = "done; progress now %.0f%%" % (100 * info["best_progress"])
	# older turns keep only the facts line (no ASCII) to keep requests short
	turn = [{"role": "user", "content": "(earlier turn) " + facts}, assistant]
	calls = msg.get("tool_calls") or []
	for call in calls:
	turn.append({"role": "tool", "tool_call_id": call.get("id", ""), "content": result})
	if not calls:
	turn[1] = {"role": "assistant", "content": action or "noop"}
	if msg.get("reasoning_content"): # provider rejects assistant turns without their reasoning
	turn[1]["reasoning_content"] = msg["reasoning_content"]
	history_turns_list = (history_turns_list + [turn])[-history_turns:] if history_turns else []
	history = [m for t in history_turns_list for m in t]
	if term or trunc:
	break
	status = "won" if env.won else ("died" if env.dead else
	("truncated" if env.frames >= env.max_frames else "request_cap"))
	return records, used, status


	def run(chat, max_requests=80, total_cap=200, frame_skip=4, max_attempts=3, model=MODEL, log=print,
	save=True, backend="emulator", level="classic"):
	global RUN_OUT, SCHED_OUT
	level = int(level) if str(level).isdigit() else level
	tag = "" if level == "classic" else "_level%s" % level
	if backend != "emulator": # live run against the in-weights computer: separate files
	RUN_OUT = os.path.join(ROOT, "results", "beam_live_weights%s_run.json" % tag)
	SCHED_OUT = os.path.join(ROOT, "results", "beam_live_weights%s_schedule.json" % tag)
	elif tag: # held-out / generated level evaluation
	RUN_OUT = os.path.join(ROOT, "results", "levels", "beam%s_run.json" % tag)
	SCHED_OUT = os.path.join(ROOT, "results", "levels", "beam%s_schedule.json" % tag)
	sys.path.insert(0, ROOT)
	from beamboot.superbeam_gym import SuperBeamEnv, WIN_X, START_X, IPF
	est = -(-(WIN_X - START_X) // (2 * frame_skip))
	log("estimate: level needs >= %d frames = >= %d actions at frame_skip=%d (cap %d requests/attempt, %d total)"
	% (-(-(WIN_X - START_X) // 2), est, frame_skip, max_requests, total_cap))
	if est > max_requests:
	raise SystemExit("frame_skip too small for the request cap")
	attempts, total = [], 0
	for k in range(max_attempts):
	budget = min(max_requests, total_cap - total)
	if budget <= 0:
	break
	env = SuperBeamEnv(backend=backend, frame_skip=frame_skip, max_frames=600, level=level,
	device="cuda" if backend != "emulator" else "cpu")
	log("=== attempt %d (budget %d requests, backend %s) ===" % (k + 1, budget, backend))
	t0 = time.time()
	records, used, status = play_attempt(env, chat, budget, log=log)
	total += used
	attempts.append({"attempt": k + 1, "status": status, "requests_used": used,
	"won": env.won, "dead": env.dead, "progress": env._info()["best_progress"],
	"frames": env.frames, "actions": [ACTION_NAMES[a] for a in env.actions],
	"key_schedule": env.key_schedule(), "requests": records,
	"backend": backend, "wall_seconds": round(time.time() - t0, 1)})
	if backend != "emulator":
	st = env.backend.session.stats
	attempts[-1]["weights"] = {
	"frame_rows": env.backend.session.frame_rows,
	"per_frame": [{"f": x.frame, "tokens": x.frame_tokens, "accepted_all": x.accepted_all,
	"first_reject": x.first_reject, "s": round(x.seconds, 3)} for x in st],
	"frames_all_draft_accepted": sum(x.accepted_all for x in st),
	"compute_seconds": round(sum(x.seconds for x in st), 1)}
	log("attempt %d: %s, progress %.1f%%, %d requests" % (k + 1, status,
	100 * attempts[-1]["progress"], used))
	if env.won or status.startswith("provider_error"):
	break
	best = max(attempts, key=lambda a: (a["won"], a["progress"]))
	# verify the best schedule replays to the same outcome
	venv = SuperBeamEnv(frame_skip=1, max_frames=600, level=level)
	vinfo = venv.replay_schedule(best["key_schedule"], n_frames=best["frames"])
	out = {"model": model, "level": level, "frame_skip": frame_skip, "ipf": IPF,
	"max_requests_per_attempt": max_requests,
	"total_cap": total_cap, "total_requests": total, "best_attempt": best["attempt"],
	"best": {k: best[k] for k in ("status", "won", "progress", "frames", "requests_used")},
	"replay_check": vinfo, "attempts": attempts}
	log("best: attempt %d %s progress %.1f%% \| total requests %d \| replay %s"
	% (best["attempt"], best["status"], 100 * best["progress"], total, vinfo))
	if not save:
	return out
	os.makedirs(os.path.dirname(RUN_OUT), exist_ok=True)
	with open(RUN_OUT, "w") as f:
	json.dump(out, f, indent=1, default=str)
	with open(SCHED_OUT, "w") as f:
	json.dump({"rom": "roms/superbeam.ch8", "ipf": IPF, "frame_skip": frame_skip,
	"frames": best["frames"], "won": best["won"], "progress": best["progress"],
	"format": "[[frame, key, 'down'\|'up'], ...] applied before that frame's instructions",
	"events": best["key_schedule"]}, f, indent=1)
	log("wrote %s and %s" % (os.path.relpath(RUN_OUT, ROOT), os.path.relpath(SCHED_OUT, ROOT)))
	return out


	def _dry_chat(messages, tools):
	"""Offline stand-in: reads the facts line like the prompt suggests (no HF call)."""
	text = messages[-1]["content"]
	m = re.search(r"standing on (ground\|a pipe)", text)
	terrain, walkers = text.split("Walkers:")[0], text.split("Walkers:")[-1].split("Coins collected")[0]
	near = [int(v) for v in re.findall(r"\((-?\d+) px ahead\)", terrain)]
	foes = [int(v) for v in re.findall(r"\((-?\d+) px ahead\)", walkers)]
	act = "right"
	if m and (any(d <= 1 for d in near) or "under you" in terrain.split("Solid")[0]
	or any(4 <= d <= 20 for d in foes)):
	act = "right+jump"
	return {"choices": [{"message": {"role": "assistant", "content": None, "tool_calls": [
	{"id": "dry", "type": "function", "function": {"name": "act",
	"arguments": json.dumps({"action": act})}}]}}]}


	@app.local_entrypoint()
	def main(dry_run: bool = False, max_requests: int = 80, total_cap: int = 200,
	frame_skip: int = 4, model: str = MODEL, backend: str = "emulator", max_attempts: int = 3,
	level: str = "classic"):
	if dry_run:
	run(_dry_chat, max_requests, total_cap, frame_skip, max_attempts, model="dry-run", save=False,
	backend=backend, level=level)
	return
	run(lambda msgs, tools: beam_chat.remote(msgs, tools, model), max_requests, total_cap,
	frame_skip, max_attempts, model=model, backend=backend, level=level)