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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199#!/usr/bin/env python3"""Bounded actual-80x24 PTY proof for recoverable Contracts rejection."""
import errnoimport fcntlimport jsonimport osimport pathlibimport reimport selectorsimport structimport subprocessimport sysimport termiosimport time
DEADLINE_SECONDS = 300.0CAPTURE_LIMIT = 4 * 1024 * 1024ANSI = re.compile(rb"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\)|[@-_])")PLAY_CUE = b"Recent feedback"MAP_CUE = b"TACTICAL Z 8"REJECTION_CUE = b"interaction-out-of-range:"MOVE_CUE = b"step [21, 31, 8]"
def stripped(data): return ANSI.sub(b"", data).replace(b"\r", b"")
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 terminate(child): if child.poll() is None: child.kill() try: child.wait(timeout=5) except subprocess.TimeoutExpired: child.kill() child.wait(timeout=5)
def spawn(binary, data_dir, content, seed): master, slave = os.openpty() fcntl.ioctl(slave, termios.TIOCSWINSZ, struct.pack("HHHH", 24, 80, 0, 0)) size = struct.unpack("HHHH", fcntl.ioctl(slave, termios.TIOCGWINSZ, b"\0" * 8)) child = subprocess.Popen( [ binary, "--data-dir", str(data_dir), "--content-root", str(content), "--story", "--seed", str(seed), "--name", "T15 contracts PTY delver", "--ancestry", "1", "--calling", "2", ], stdin=slave, stdout=slave, stderr=slave, close_fds=True, start_new_session=True, ) os.close(slave) os.set_blocking(master, False) return child, master, size[0] == 24 and size[1] == 80
def drive(binary, data_dir, content, mode, seed): child, master, exact_size = spawn(binary, data_dir, content, seed) selector = selectors.DefaultSelector() selector.register(master, selectors.EVENT_READ) registered = True captured = bytearray() stage = 0 search_from = 0 visible_before_followup = False deadline = time.monotonic() + DEADLINE_SECONDS try: # Child exit and PTY EOF are independent; drain until both are observed. while child.poll() is None or registered: remaining = deadline - time.monotonic() if remaining <= 0.0: raise RuntimeError("PTY child or EOF drain exceeded 300 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 child.poll() is not None or not registered: continue if stage == 0: found = plain.find(PLAY_CUE, search_from) if found >= 0: os.write(master, b"m" if mode == "map" else b"c") search_from = found + len(PLAY_CUE) stage = 1 if mode == "map" else 2 elif stage == 1: found = plain.find(MAP_CUE, search_from) if found >= 0: os.write(master, b"c") search_from = found + len(MAP_CUE) stage = 2 elif stage == 2: found = plain.find(REJECTION_CUE, search_from) if found >= 0: visible_before_followup = True # M leaves tactical mode; physical forward is world -Y at # the default north heading (legacy canonical South). os.write(master, (b"m" if mode == "map" else b"") + b"w") search_from = found + len(REJECTION_CUE) stage = 3 elif stage == 3: found = plain.find(MOVE_CUE, search_from) if found >= 0: os.write(master, b"\x1b") search_from = found + len(MOVE_CUE) stage = 4
status = child.wait(timeout=5) if stage != 4: raise RuntimeError( "child exited before visible rejection, legal move, and quit; stage=%d tail=%r" % (stage, stripped(bytes(captured))[-5000:]) ) if status != 0: raise RuntimeError( "child exit %d; tail=%r" % (status, stripped(bytes(captured))[-5000:]) ) return bytes(captured), status, exact_size, visible_before_followup finally: selector.close() os.close(master) terminate(child)
def main(): if len(sys.argv) != 6: raise RuntimeError("usage: contracts_open.py BINARY DATA CONTENT play|map SEED") binary = sys.argv[1] data_dir = pathlib.Path(sys.argv[2]) content = pathlib.Path(sys.argv[3]) mode = sys.argv[4] seed = int(sys.argv[5]) if mode not in ("play", "map"): raise RuntimeError("unknown mode: " + mode) data_dir.mkdir(parents=True, exist_ok=True)
captured, status, exact_size, visible = drive( binary, data_dir, content, mode, seed ) plain = stripped(captured) summary = { "exit": status, "window_80x24": exact_size, "rejection_visible_before_followup": visible, "session_over": b"HEXWORK session over" in plain, "run_id": only_run_id(data_dir), } sys.stdout.write(json.dumps(summary, separators=(",", ":")) + "\n")
if __name__ == "__main__": try: main() except Exception as error: sys.stderr.write(("contracts-open PTY scenario failed: %s\n" % error)[-8192:]) raise