Something went wrong. Try again.
HEXWORK - A Lantern in the Deep: 3D ASCII terminal roguelike RPG in Rust.
Entirely designed and built by Muse Spark 1.3 to evaluate if I like the model.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204#!/usr/bin/env python3"""Bounded stdlib PTY proof that toggle-walk drives real paced world ticks.
A raw PTY cannot synthesize `KeyEventKind::Repeat`, so held-key pulsing iscovered by the deterministic fake-clock tests. This scenario exercises theprotocol-independent half: a discrete tap sets the walk direction, the explicittoggle-walk key then travels that direction with no key release, and theordinary stop key (`Space`) halts it before `Esc` quits.
Raw PTY bytes are a stream of ANSI diffs, not frames, so the durable world tickcount is read back from the saved run rather than scraped from the screen."""
import errnoimport fcntlimport jsonimport osimport pathlibimport reimport selectorsimport structimport subprocessimport sysimport termiosimport time
DEADLINE_SECONDS = 240.0CAPTURE_LIMIT = 4 * 1024 * 1024ANSI = re.compile(rb"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\)|[@-_])")MENU_CUE = b"Cancel: [Esc]"PLAY_CUE = b"Recent feedback"SESSION_OVER = re.compile(rb"HEXWORK session over")SESSION_TICK = re.compile(rb"tick (\d+)")ACTING = b"ACTING"WALK_SECONDS = 0.6STOP_SECONDS = 0.2
def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"")
def spawn(binary, data_dir, content, seed): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) child = subprocess.Popen( [ binary, "--data-dir", str(data_dir), "--content-root", str(content), "--seed", str(seed), ], stdin=slave, stdout=slave, stderr=slave, close_fds=True, start_new_session=True, ) os.close(slave) os.set_blocking(master, False) return child, master
def terminate(child): if child.poll() is None: child.kill() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill() child.wait(timeout=5)
def only_run_id(data_dir): values = sorted(path.name for path in (data_dir / "runs").iterdir() if path.is_dir()) if len(values) != 1: raise RuntimeError("expected exactly one run, got %r" % values) return values[0]
def drive(binary, data_dir, content, seed): child, master = spawn(binary, data_dir, content, seed) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True captured = bytearray() stage = 0 search_from = 0 walk_at = None stop_at = None acting_seen = False deadline = time.monotonic() + DEADLINE_SECONDS try: while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0.0: raise RuntimeError("PTY child or EOF drain exceeded 240 second deadline") if registered: for _, _ in selector.select(timeout=min(0.25, remaining)): try: chunk = os.read(master, 65536) except OSError as error: if error.errno == errno.EIO: chunk = b"" elif error.errno in (errno.EAGAIN, errno.EWOULDBLOCK): continue else: raise if not chunk: selector.unregister(master) registered = False break if len(captured) + len(chunk) > CAPTURE_LIMIT: raise RuntimeError("PTY capture exceeded 4 MiB") captured.extend(chunk) else: try: child.wait(timeout=min(0.25, remaining)) except subprocess.TimeoutExpired: pass
plain = stripped(bytes(captured)) if ACTING in plain: acting_seen = True if child.poll() is not None or not registered: continue if stage == 0: found = plain.find(MENU_CUE, search_from) if found >= 0: os.write(master, b"1") search_from = found + len(MENU_CUE) stage = 1 elif stage == 1: found = plain.find(MENU_CUE, search_from) if found >= 0: os.write(master, b"1") search_from = found + len(MENU_CUE) stage = 2 elif stage == 2: found = plain.find(PLAY_CUE, search_from) if found >= 0: # A discrete tap sets the last-moved direction; the explicit # toggle then walks that direction with no key release. os.write(master, b"wt") search_from = found + len(PLAY_CUE) walk_at = time.monotonic() stage = 3 elif stage == 3: if time.monotonic() - walk_at >= WALK_SECONDS: os.write(master, b" ") # Space stops the walk in place of Wait. stop_at = time.monotonic() stage = 4 elif stage == 4: if time.monotonic() - stop_at >= STOP_SECONDS: os.write(master, b"\x1b") # Esc quits and saves. stage = 5
status = child.wait(timeout=5) plain = stripped(bytes(captured)) if stage != 5: raise RuntimeError( "child exited before paced walk completed; stage=%d tail=%r" % (stage, plain[-4000:]) ) if status != 0: raise RuntimeError("child exit %d; tail=%r" % (status, plain[-4000:])) ticks = SESSION_TICK.findall(plain) return { "exit": status, "session_over": bool(SESSION_OVER.search(plain)), "final_tick": int(ticks[-1]) if ticks else -1, "acting_seen": acting_seen, "run_id": only_run_id(data_dir), } finally: selector.close() os.close(master) terminate(child)
def main(): if len(sys.argv) != 5: raise RuntimeError("usage: pace_scenario.py BINARY DATA CONTENT SEED") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) seed = int(sys.argv[4]) data_dir.mkdir(parents=True, exist_ok=True) summary = drive(binary, data_dir, content, seed) sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n") if not summary["session_over"]: raise RuntimeError("session did not end cleanly: %r" % summary)
if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("pace PTY scenario failed: %s\n" % error)[-8192:]) raise