Files
mattlqxandClaude Code 11d42dbb36 feat: log the game's high-score report as a gameplay event
When the game builds its high-score table report for Stern's agent
(/api/v1/game/high_score_events - the endpoint the Insider gate already sits
on), the logger writes an HSTD line with all four players' scores. This is
the report the game sends when a game has changed the table, so the line
marks a high-score event with no new per-game RE. The entries themselves
(initials, position) live deeper in the report body and stay uncaptured for
now - that needs locating the report builder or the in-memory table.

Tests: the endpoint trigger and the HSTD line lift into the host harness
(the endpoint pointer is a 32-bit register, so the harness maps its string
below 4 GiB and the tests run on the Linux leg like the port-gate ones).

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-10-22 22:31:15 -04:00

491 lines
18 KiB
Python

"""Multi-game port selection: which port file is this game's.
- port_select tries each candidate port in order, resetting the port tables between them, and
keeps the first whose core sites verify against the running game - that is how the machine
says which game it is on. A wrong candidate hooks nothing (the gate), so trying it is safe.
- ports_scan_buf reads the kernel's getdents64 records: every *.port in a ports directory is a
candidate, nothing else is.
- gl_switch_log writes the human buttons to the switch log, named by the port's own `switch`
lines - the switch names are no longer baked into the runtime per game.
Lifted verbatim from pad_mode_runtime.c and compiled for the HOST (the lift/_host_run helpers
are the decryptor repo tests', tests/test_spike2_mode_roster.py). Desk only: no card, no game.
"""
import pathlib
import shutil
import subprocess
import sys
import pytest
RUNTIME = pathlib.Path(__file__).resolve().parents[1] / "pad_mode_runtime.c"
def _src():
return RUNTIME.read_text(encoding="utf-8")
def _lift(src, signature):
"""A function's text, verbatim from the runtime (brace-matched from its signature)."""
i = src.index(signature)
depth = 0
for k in range(src.index("{", i), len(src)):
depth += {"{": 1, "}": -1}.get(src[k], 0)
if depth == 0:
return src[i:k + 1]
raise AssertionError("unbalanced: " + signature)
def _between(src, start, end):
i = src.index(start)
return src[i:src.index(end, i)]
def _cc():
for c in ("gcc", "cc", "clang"):
if shutil.which(c):
return c
pytest.skip("no host C compiler")
def _host_run(tmp_path, code, flags=(), args=()):
cc = _cc()
(tmp_path / "t.c").write_text(code)
r = subprocess.run([cc, "-std=gnu17", "-Wall", "-Wno-unused-parameter", *flags,
"-o", str(tmp_path / "t"), str(tmp_path / "t.c")],
capture_output=True, text=True)
assert r.returncode == 0, r.stderr
return subprocess.run([str(tmp_path / "t"), *args], capture_output=True, text=True,
timeout=30).stdout
def _port_section(src):
"""The port reader, lifted: its tables, port_line and port_read, then the lookups."""
helpers = "\n".join(_lift(src, s) for s in (
"static int str_eq(", "static void str_copy(", "static int hexval(", "static uint64_t number("))
reader = _between(src, "/* The port is read in PORT_CHUNK pieces",
"static struct site *site(const char *name)")
lookups = "\n".join(_lift(src, s) for s in (
"static struct site *site(const char *name)", "static unsigned fn(const char *name)",
"static long named(", "static unsigned data(const char *name)", "long pm_port_value("))
return helpers + "\n" + reader + "\n" + lookups
_SELECT_PRELUDE = r"""
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <stdint.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#define N_BUS_IDS 208
static void port_line(const char *s);
static void port_reset(void);
static int port_select(const char *const *files, int n);
static void event_line(const char *s) { (void)s; }
static void lamp_line(const char *s) { (void)s; }
static void rule_line(const char *s) { (void)s; } /* not this test's */
/* the runtime's parse-state counters outside the port struct; the tables themselves are stubbed */
static int n_event_names, n_site_events;
static int n_lamps;
static int n_stock_rules;
static void say(const char *fmt, ...)
{
va_list ap;
va_start(ap, fmt);
printf("SAY ");
vprintf(fmt, ap);
va_end(ap);
putchar('\n');
}
"""
_SELECT_TAIL = r"""
/* the game's "code": words the sites point at. The port's site addresses are 32 bits (ARM), so
* the code page is mapped at a low fixed address instead of living in this binary's data. */
#include <sys/mman.h>
static unsigned *code;
static void line(const char *fmt, ...)
{
char b[256];
va_list ap;
va_start(ap, fmt);
vsnprintf(b, sizeof b, fmt, ap);
va_end(ap);
port_line(b);
}
static void site_at(const char *name, int i)
{
line("site %s 0x%lx 0x%x 0x%x", name, (unsigned long)&code[i], code[i], code[i + 1]);
}
static void core_lines(void)
{
site_at("tick", 0);
site_at("shot_dispatch", 2);
site_at("ball_end", 4);
site_at("score_add", 6);
line("data cur_player 0x5b322c");
line("data scores 0x5b3578");
}
/* one candidate port file: game <name>, a matching core only when good. The sites carry THIS
* process's code addresses - the file is selected in this same process, as on the machine. */
static void write_port(const char *path, const char *game, int good)
{
FILE *f = fopen(path, "w");
fprintf(f, "game %s\nversion 1.0\n", game);
if (good) {
fprintf(f, "site tick 0x%lx 0x%x 0x%x\n", (unsigned long)&code[0], code[0], code[1]);
fprintf(f, "site shot_dispatch 0x%lx 0x%x 0x%x\n", (unsigned long)&code[2], code[2], code[3]);
fprintf(f, "site ball_end 0x%lx 0x%x 0x%x\n", (unsigned long)&code[4], code[4], code[5]);
fprintf(f, "site score_add 0x%lx 0x%x 0x%x\n", (unsigned long)&code[6], code[6], code[7]);
} else {
/* the file's site names the right address with the wrong words: the gate refuses it */
fprintf(f, "site tick 0x%lx 0x11111111 0x22222222\n", (unsigned long)&code[0]);
fprintf(f, "site shot_dispatch 0x%lx 0x11111111 0x22222222\n", (unsigned long)&code[2]);
fprintf(f, "site ball_end 0x%lx 0x11111111 0x22222222\n", (unsigned long)&code[4]);
fprintf(f, "site score_add 0x%lx 0x11111111 0x22222222\n", (unsigned long)&code[6]);
}
fprintf(f, "data cur_player 0x5b322c\ndata scores 0x5b3578\n");
fprintf(f, "shot 0x1 X1\nshot 0x2 Left slingshot\n");
fprintf(f, "event game_start 0x4d\nevent ball_start 0x25\n");
fclose(f);
}
int main(int argc, char **argv)
{
const char *what = argv[1];
char maps_text[128];
static const char *files[8];
int i, n = 0;
code = mmap((void *)0x08000000, 4096, PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON | MAP_FIXED, -1, 0);
if (code == (unsigned *)-1) { printf("mmap failed\n"); return 1; }
{
unsigned words[] = {
0xe92d4038, 0xe3a00037, /* 0: tick */
0xe3500b02, 0xe92d4008, /* 2: shot_dispatch */
0xe92d4008, 0xe3a00000, /* 4: ball_end */
0xe30817fe, 0xe340108b, /* 6: score_add */
};
memcpy(code, words, sizeof words);
}
snprintf(maps_text, sizeof maps_text, "%lx-%lx r-xp 00000000 00:00 0 /games/x/game",
(unsigned long)code, (unsigned long)(code + 64));
maps_line(maps_text, maps_text + strlen(maps_text));
if (!strcmp(what, "select")) {
int ok;
n = atoi(argv[2]); /* then, per candidate: <path> <game> <good> */
for (i = 0; i < n; i++) {
write_port(argv[3 + 3 * i], argv[4 + 3 * i], atoi(argv[5 + 3 * i]));
files[i] = argv[3 + 3 * i];
}
ok = port_select(files, n);
printf("SELECT %d\n", ok);
printf("GAME %s\n", port.game);
printf("SITES %d SHOTS %d EVENTS %d\n", port.n_site, port.n_shot, n_event_names);
} else { /* reset */
printf("RESET game=%s sites=%d shots=%d events=%d can=%u\n",
port.game, port.n_site, port.n_shot, n_event_names, can);
line("game leftover");
line("version 9.9");
core_lines();
line("shot 0x1 X1");
line("shot 0x2 Left slingshot");
line("shot 0x4 third");
line("event game_start 0x4d");
line("event ball_start 0x25");
line("event ball_end 0x34");
can = 0xffffffffu;
port_reset();
printf("RESET game=%s sites=%d shots=%d events=%d can=%u\n",
port.game, port.n_site, port.n_shot, n_event_names, can);
core_lines(); /* clean tables still gate a good port in */
printf("RESETGATE %d\n", port_gate());
}
return 0;
}
"""
def _select_source():
src = _src()
maps = _between(src, "#define N_MAPS", "/* ---- rule 2:")
gate = "\n".join(_lift(src, s) for s in (
"static int words_match(struct site *x)", "static void core_of_port(",
"static int site_check(", "static int port_gate(void)\n{"))
maps = maps.replace("static void maps_line(const char *s, const char *e)",
"static void maps_line(const char *s, const char *e_)")
maps = maps.replace(" unsigned long lo = 0, hi = 0;\n const char *q = s, *p;",
" const char *e = e_ ? e_ : s + strlen(s);\n"
" unsigned long lo = 0, hi = 0;\n const char *q = s, *p;", 1)
extra = "\n".join(_lift(src, s) for s in (
"static void port_reset(void)",
"static int port_select(const char *const *files, int n)"))
return (_SELECT_PRELUDE + _port_section(src) + "\n" + maps + "\n" + gate + "\n" + extra
+ _SELECT_TAIL)
def test_port_select_keeps_the_port_that_matches(tmp_path):
if sys.platform == "darwin":
pytest.skip("the port's site addresses are 32 bits and darwin arm64 maps nothing below "
"4 GiB - the Linux leg runs these")
a, b = str(tmp_path / "a.port"), str(tmp_path / "b.port")
out = _host_run(tmp_path, _select_source(),
args=("select", "2", a, "wrong_game", "0", b, "right_game", "1"))
assert "SELECT 1" in out, out
assert "GAME right_game" in out, out
assert "SAY port %s: says wrong_game 1.0 - not this game's port, skipped" % a in out, out
assert "SITES 4 SHOTS 2 EVENTS 0" in out, out # event_line is stubbed in this harness
def test_port_select_prefers_the_earlier_candidate(tmp_path):
if sys.platform == "darwin":
pytest.skip("32-bit site addresses: see the first select test")
a, b = str(tmp_path / "a.port"), str(tmp_path / "b.port")
out = _host_run(tmp_path, _select_source(),
args=("select", "2", a, "a_game", "1", b, "b_game", "1"))
assert "SELECT 1" in out and "GAME a_game" in out, out
def test_port_select_refuses_when_no_port_matches(tmp_path):
if sys.platform == "darwin":
pytest.skip("32-bit site addresses: see the first select test")
a = str(tmp_path / "a.port")
out = _host_run(tmp_path, _select_source(), args=("select", "1", a, "wrong_game", "0"))
assert "SELECT 0" in out, out
assert "not this game's port, skipped" in out, out
def test_port_select_resets_the_tables_between_candidates(tmp_path):
"""A candidate's lines must not leak into the next one's tables: after a full port is read, a
reset leaves every table empty (game, sites, shots, events, can) and still gates a good port."""
if sys.platform == "darwin":
pytest.skip("32-bit site addresses: see the first select test")
out = _host_run(tmp_path, _select_source(), args=("reset",))
got = [l for l in out.splitlines() if l.startswith("RESET ")]
assert got == ["RESET game= sites=0 shots=0 events=0 can=0",
"RESET game= sites=0 shots=0 events=0 can=0"], out
assert "RESETGATE 1" in out, out
_SCAN_MAIN = r"""
int main(void)
{
static char buf[512];
char cand[8][128];
long n, i;
char *at = buf;
const char *names[] = { "king_kong_le-0.97.port", "readme.txt", ".", "..", "godzilla_le-1.16.port" };
unsigned char types[] = { 8, 8, 4, 4, 8 };
for (i = 0; i < 5; i++) {
unsigned long long ino = (unsigned long long)i + 1;
long long off = (long long)i + 1;
unsigned short reclen = (unsigned short)(((strlen(names[i]) + 1 + 19) + 7) & ~7L);
memcpy(at, &ino, 8);
memcpy(at + 8, &off, 8);
memcpy(at + 16, &reclen, 2);
at[18] = types[i];
strcpy(at + 19, names[i]);
at += reclen;
}
n = ports_scan_buf(buf, (long)(at - buf), "/usr/local/padmode/ports", cand, 8, 0);
printf("SCAN %ld\n", n);
for (i = 0; i < n; i++) printf("CAND %s\n", cand[i]);
return 0;
}
"""
def _scan_source():
src = _src()
prelude = r"""
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <stdint.h>
static void say(const char *fmt, ...) { (void)fmt; }
int pm_snprintf(char *out, unsigned long cap, const char *fmt, ...)
{
va_list ap;
int n;
va_start(ap, fmt);
n = vsnprintf(out, cap, fmt, ap);
va_end(ap);
return n;
}
static long ports_scan_buf(char *buf, long len, const char *dir, char (*cand)[128], long cap, long n);
"""
helpers = _lift(src, "static int str_eq(")
dirent = _between(src, "/* the kernel's getdents64 record", "static long ports_scan(")
return prelude + helpers + "\n" + dirent + _SCAN_MAIN
def test_ports_scan_buf_takes_only_port_files(tmp_path):
out = _host_run(tmp_path, _scan_source())
lines = out.splitlines()
assert "SCAN 2" in out, out
assert "CAND /usr/local/padmode/ports/king_kong_le-0.97.port" in lines, out
assert "CAND /usr/local/padmode/ports/godzilla_le-1.16.port" in lines, out
_SWITCH_PRELUDE = r"""
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <stdint.h>
#include <fcntl.h>
#include <unistd.h>
static void port_line(const char *s);
static void gl_switch_log(unsigned id, unsigned level, int hit, int playing);
static const char *GL_SWITCH_LOG;
static void say(const char *fmt, ...) { (void)fmt; }
static void gl_rfc3339(char *buf, unsigned cap) { snprintf(buf, cap, "2026-10-13T12:00:00.000Z"); }
int pm_snprintf(char *out, unsigned long cap, const char *fmt, ...)
{
va_list ap;
int n;
va_start(ap, fmt);
n = vsnprintf(out, cap, fmt, ap);
va_end(ap);
return n;
}
static void event_line(const char *s) { (void)s; }
static void lamp_line(const char *s) { (void)s; }
static void rule_line(const char *s) { (void)s; }
static void line(const char *fmt, ...)
{
char b[256];
va_list ap;
va_start(ap, fmt);
vsnprintf(b, sizeof b, fmt, ap);
va_end(ap);
port_line(b);
}
int main(int argc, char **argv)
{
GL_SWITCH_LOG = argv[1];
line("game king_kong_le");
line("version 0.97");
line("switch 34 0x1000000000000000 LOCKDOWN BUTTON");
line("switch 36 0x2000000000000000 START BUTTON");
line("switch 62 0x4000000000000000 RIGHT FLIPPER BUTTON");
line("switch 63 0x8000000000000000 LEFT FLIPPER BUTTON");
gl_switch_log(62, 1, 1, 1);
gl_switch_log(50, 1, 1, 1); /* not a button the port names: nothing */
gl_switch_log(36, 0, 0, 0);
return 0;
}
"""
def _switch_source():
src = _src()
helpers = "\n".join(_lift(src, s) for s in (
"static int str_eq(", "static void str_copy(", "static int hexval(", "static uint64_t number("))
reader = _between(src, "/* The port is read in PORT_CHUNK pieces",
"static struct site *site(const char *name)")
lookups = _lift(src, "static struct site *site(const char *name)")
fn = _lift(src, "static void gl_switch_log(")
return _SWITCH_PRELUDE + helpers + "\n" + reader + "\n" + lookups + "\n" + fn
def test_gl_switch_log_writes_the_ports_buttons(tmp_path):
log = tmp_path / "switch.log"
out = _host_run(tmp_path, _switch_source(), args=(str(log),))
assert out == ""
text = log.read_text(encoding="utf-8").splitlines()
assert text == [
'2026-10-13T12:00:00.000Z SW 62 "RIGHT FLIPPER BUTTON" level=1 hit=1 game=1',
'2026-10-13T12:00:00.000Z SW 36 "START BUTTON" level=0 hit=0 game=0',
], text
# ---- the high-score report event ------------------------------------------------------
_HSTD_PRELUDE = r"""
#define _GNU_SOURCE
#include <stdio.h>
#include <string.h>
#include <stdarg.h>
#include <stdint.h>
#include <sys/mman.h>
static void event_line(const char *s) { (void)s; }
static void lamp_line(const char *s) { (void)s; }
static void rule_line(const char *s) { (void)s; }
static void port_line(const char *s) { (void)s; }
static void say(const char *fmt, ...) { (void)fmt; }
static void on_agent_header(unsigned *r);
static void gl_rfc3339(char *buf, unsigned cap) { snprintf(buf, cap, "2026-10-22T23:59:01.123Z"); }
int pm_snprintf(char *out, unsigned long cap, const char *fmt, ...)
{
va_list ap;
int n;
va_start(ap, fmt);
n = vsnprintf(out, cap, fmt, ap);
va_end(ap);
return n;
}
void pm_log(const char *fmt, ...)
{
char b[256];
va_list ap;
va_start(ap, fmt);
vsnprintf(b, sizeof b, fmt, ap);
va_end(ap);
printf("PMLOG %s\n", b);
}
static void pm_commas(char *out, unsigned cap, unsigned long long v)
{
snprintf(out, cap, "%llu", (unsigned long long)v);
}
static unsigned long long pm_score(unsigned player) { return player == 1 ? 137615960 : 0; }
int main(int argc, char **argv)
{
static unsigned r[4];
/* the runtime reads the endpoint pointer as a 32-bit register: the path must live below 4 GiB */
char *path = mmap((void *)0x08000000, 4096, PROT_READ | PROT_WRITE,
MAP_PRIVATE | MAP_ANON | MAP_FIXED, -1, 0);
if (path == (char *)-1) { printf("mmap failed\n"); return 1; }
snprintf(path, 4096, "%s", argv[1]);
r[1] = (unsigned)(unsigned long)path;
on_agent_header(r);
return 0;
}
"""
def _hstd_source():
src = _src()
helpers = "\n".join(_lift(src, s) for s in (
"static int str_eq(", "static void str_copy(", "static int hexval(", "static uint64_t number("))
insider = _between(src, "#define INSIDER_ENDPOINT_MAX", "static void on_agent_header(unsigned *r)")
fn = "\n".join(_lift(src, s) for s in (
"static void gl_hstd_report(void)", "static void on_agent_header(unsigned *r)"))
return _HSTD_PRELUDE + helpers + "\n" + insider + "\n" + fn
def test_a_high_score_report_logs_an_hstd_event_with_the_scores(tmp_path):
if sys.platform == "darwin":
pytest.skip("the runtime's pointers are 32-bit registers; darwin arm64 maps nothing "
"below 4 GiB - the Linux leg runs these")
out = _host_run(tmp_path, _hstd_source(),
args=("/api/v1/game/high_score_events",))
assert "HSTD high-score report" in out, out
assert "P1=137615960 P2=0 P3=0 P4=0" in out, out
def test_other_agent_reports_log_no_hstd_event(tmp_path):
if sys.platform == "darwin":
pytest.skip("32-bit pointer registers: see the first HSTD test")
for path in ("/api/v3/game/session_start", "/ingest/v1/game/game_events", "junk", ""):
out = _host_run(tmp_path, _hstd_source(), args=(path,))
assert "HSTD" not in out, "%s logged an HSTD event" % path