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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183#!/usr/bin/env python3"""Bounded stdlib PTY proof for Phase 12 physical-key remapping."""
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\\)|[@-_])")
def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"")
def artifact_absent(data_dir): return not (data_dir / "lineage-vault.json").exists() and not (data_dir / "runs").exists()
def spawn(binary, data_dir, content, extra): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 100, 0, 0)) argv = [binary, "--data-dir", str(data_dir), "--content-root", str(content)] + list(extra) child = subprocess.Popen( argv, 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 drive(binary, data_dir, content, extra, steps): child, master = spawn(binary, data_dir, content, extra) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) captured = bytearray() step = 0 search_from = 0 master_registered = True deadline = time.monotonic() + DEADLINE_SECONDS try: while child.poll() is None or master_registered: remaining = deadline - time.monotonic() if remaining <= 0.0: raise RuntimeError("PTY child or output drain exceeded 240 second deadline") if master_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) 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)) while child.poll() is None and master_registered and step < len(steps): found = plain.find(steps[step][0], search_from) if found < 0: break callback = steps[step][2] if callback is not None: callback() os.write(master, steps[step][1]) search_from = found + len(steps[step][0]) step += 1 if step != len(steps): raise RuntimeError("child exited before expected prompt; tail=" + repr(stripped(bytes(captured))[-2000:])) status = child.wait(timeout=5) if status != 0: raise RuntimeError("child exit %d; tail=%r" % (status, stripped(bytes(captured))[-2000:])) return bytes(captured) finally: selector.close() os.close(master) terminate(child)
def run_id(data_dir): runs = data_dir / "runs" values = sorted(path.name for path in runs.iterdir() if path.is_dir()) if len(values) != 1: raise RuntimeError("expected exactly one allocated run, got %r" % values) return values[0]
def main(): if len(sys.argv) != 5: raise RuntimeError("usage: keymap_scenario.py BINARY DATA CONTENT remapped|default") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) mode = sys.argv[4] data_dir.mkdir(parents=True, exist_ok=True)
prompt_checks = []
def before_choice(): prompt_checks.append(artifact_absent(data_dir))
if mode == "remapped": ancestry_key, calling_key, move_quit, play_cue, menu_cue = b"7", b"8", b"fv", b"v quit", b"Cancel: [v]" elif mode == "default": ancestry_key, calling_key, move_quit, play_cue, menu_cue = b"1", b"2", b"a\x1b", b"Esc quit", b"Cancel: [Esc]" else: raise RuntimeError("unknown scenario mode: " + mode)
first = drive( binary, data_dir, content, [], [ (menu_cue, ancestry_key, before_choice), (menu_cue, calling_key, before_choice), (play_cue, move_quit, None), ], ) allocated = run_id(data_dir) first_save = data_dir / "runs" / allocated / "run.save" (data_dir / "first-run.save").write_bytes(first_save.read_bytes()) load_move_quit = b"dv" if mode == "remapped" else b"d\x1b" second = drive( binary, data_dir, content, ["--load", allocated], [(play_cue, load_move_quit, None)], ) summary = { "prompt_before_allocation": bool(prompt_checks) and all(prompt_checks), "fallback_reported": b"using embedded defaults" in first, "load_completed": b"HEXWORK session over" in stripped(second), "run_id": allocated, } sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n")
if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("keymap PTY scenario failed: %s\n" % error)[-8192:]) raise