From e69b7e2ad61c0734f06fdb13d7887f74d8a4a12d Mon Sep 17 00:00:00 2001 From: type-two Date: Sun, 5 Jul 2026 16:18:35 +1000 Subject: [PATCH] Accounts, socio-economic feeds, and hub fixes (track deployed work) Captures backend work already running on godstrument.pro but never committed: - auth.py: invite-only accounts + per-user presets (SQLite, scrypt, signed stateless sessions), mounted by hub.py at /api/* - hub.py: /api static+API handler, readonly public mode, full route spec in hello - socio-economic / market / fire / debt / crypto workers + normalize/transform - .gitignore: never commit godstrument_users.db* or auth_secret - test_fixes.py: framework-free self-check for the review fixes Co-Authored-By: Claude Fable 5 --- .claude/launch.json | 18 ++ .gitignore | 6 + auth.py | 468 +++++++++++++++++++++++++++++++++++++++ config.json | 3 +- hub.py | 182 ++++++++++++--- normalize.py | 7 +- run.py | 38 +++- test_fixes.py | 151 +++++++++++++ transform.py | 39 ++-- workers/world_bank.py | 15 +- workers/world_crypto.py | 29 +-- workers/world_dealgod.py | 6 +- workers/world_debt.py | 18 +- workers/world_econ.py | 20 +- workers/world_fire.py | 16 +- workers/world_planes.py | 15 +- 16 files changed, 931 insertions(+), 100 deletions(-) create mode 100644 auth.py create mode 100644 test_fixes.py diff --git a/.claude/launch.json b/.claude/launch.json index 1bfe9da..836c9d0 100644 --- a/.claude/launch.json +++ b/.claude/launch.json @@ -6,6 +6,24 @@ "runtimeExecutable": "/Users/m3ultra/Documents/godstrument/.venv/bin/python", "runtimeArgs": ["run.py", "--no-browser", "--profile", "warp", "--city", "melbourne", "--dealgod"], "port": 8088 + }, + { + "name": "godstrument-local", + "runtimeExecutable": ".venv/bin/python", + "runtimeArgs": ["run.py", "--no-browser", "--profile", "minimal"], + "port": 8088 + }, + { + "name": "godstrument-local-ro", + "runtimeExecutable": "sh", + "runtimeArgs": ["-c", "GODSTRUMENT_READONLY=1 .venv/bin/python run.py --no-browser --profile minimal"], + "port": 8088 + }, + { + "name": "viz-static", + "runtimeExecutable": "python3", + "runtimeArgs": ["-m", "http.server", "8899", "--directory", "viz"], + "port": 8899 } ] } diff --git a/.gitignore b/.gitignore index 66b75a5..d05d4ce 100644 --- a/.gitignore +++ b/.gitignore @@ -11,5 +11,11 @@ godstrument.db-shm # user-saved templates (runtime, per-machine) patches/ +# accounts db + session secret — NEVER commit (user credentials) +godstrument_users.db +godstrument_users.db-wal +godstrument_users.db-shm +auth_secret + # os .DS_Store diff --git a/auth.py b/auth.py new file mode 100644 index 0000000..340f783 --- /dev/null +++ b/auth.py @@ -0,0 +1,468 @@ +""" +auth.py — invite-only accounts + per-user presets for the Godstrument. + +The planet is shared; the *instrument* is personal. This stores each user's +saved patches (routes, tweaks, godtime, synth) so a logged-in browser can load +its own instrument off the shared feed. Pure stdlib: SQLite + scrypt hashing + +signed stateless session cookies. No frameworks, no new services. + + python auth.py --init # create the db + python auth.py --mint-invite [N] # print N new invite codes (default 1) + python auth.py --users # list accounts + python auth.py --selftest # assert the whole flow works + +The hub mounts handle_api() at /api/* (see hub.py). Secret comes from +GODSTRUMENT_SECRET or an auto-generated auth_secret file next to the db. +""" + +from __future__ import annotations +import argparse +import base64 +import hashlib +import hmac +import json +import os +import re +import secrets +import sqlite3 +import threading +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +DEFAULT_DB = os.path.join(HERE, "godstrument_users.db") +EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") +USERNAME_RE = re.compile(r"^[A-Za-z0-9_.\-]{2,24}$") +MIN_PW = 8 +SESSION_DAYS = 30 +# whoever signs up with this email is an admin (set in the systemd unit) +ADMIN_EMAIL = (os.environ.get("GODSTRUMENT_ADMIN") or "").strip().lower() + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY, username TEXT UNIQUE NOT NULL, email TEXT UNIQUE NOT NULL, + pw_hash TEXT NOT NULL, salt TEXT NOT NULL, created REAL); +CREATE TABLE IF NOT EXISTS invites ( + code TEXT PRIMARY KEY, created REAL, used_by INTEGER); +CREATE TABLE IF NOT EXISTS presets ( + user_id INTEGER, name TEXT, data TEXT, updated REAL, + PRIMARY KEY (user_id, name)); +CREATE TABLE IF NOT EXISTS feedback ( + id INTEGER PRIMARY KEY, user_id INTEGER, username TEXT, text TEXT, created REAL); +""" + +# ---- basic in-memory login throttle (friends-scale; resets on restart) ------ +_ATTEMPTS: dict[str, list[float]] = {} +_ATT_LOCK = threading.Lock() +_MAX_TRIES, _WINDOW = 8, 300.0 # 8 failures / 5 min per email + + +def _throttled(email: str) -> bool: + now = time.time() + with _ATT_LOCK: + q = [t for t in _ATTEMPTS.get(email, []) if now - t < _WINDOW] + _ATTEMPTS[email] = q + return len(q) >= _MAX_TRIES + + +def _note_fail(email: str): + with _ATT_LOCK: + _ATTEMPTS.setdefault(email, []).append(time.time()) + + +# ---- db --------------------------------------------------------------------- +def _con(db: str) -> sqlite3.Connection: + con = sqlite3.connect(db) + con.execute("PRAGMA journal_mode=WAL") + con.executescript(SCHEMA) + try: # migrate: add is_admin to older dbs + con.execute("ALTER TABLE users ADD COLUMN is_admin INTEGER DEFAULT 0") + con.commit() + except sqlite3.OperationalError: + pass # column already exists + return con + + +def init_db(db: str = DEFAULT_DB): + _con(db).close() + + +# ---- secret / password hashing ---------------------------------------------- +def _secret(db: str = DEFAULT_DB) -> bytes: + env = os.environ.get("GODSTRUMENT_SECRET") + if env: + return env.encode() + path = os.path.join(os.path.dirname(os.path.abspath(db)), "auth_secret") + if not os.path.exists(path): + with open(os.open(path, os.O_CREAT | os.O_WRONLY, 0o600), "w") as f: + f.write(secrets.token_hex(32)) + with open(path) as f: + return f.read().strip().encode() + + +def hash_pw(pw: str, salt: str | None = None) -> tuple[str, str]: + salt = salt or secrets.token_hex(16) + h = hashlib.scrypt(pw.encode(), salt=bytes.fromhex(salt), + n=16384, r=8, p=1, dklen=32, maxmem=64 * 1024 * 1024) + return salt, h.hex() + + +def verify_pw(pw: str, salt: str, expected: str) -> bool: + _, h = hash_pw(pw, salt) + return hmac.compare_digest(h, expected) + + +# ---- stateless signed session token ----------------------------------------- +def make_token(user_id: int, db: str = DEFAULT_DB, days: int = SESSION_DAYS) -> str: + exp = int(time.time()) + days * 86400 + payload = f"{user_id}:{exp}" + sig = hmac.new(_secret(db), payload.encode(), hashlib.sha256).hexdigest() + return base64.urlsafe_b64encode(f"{payload}:{sig}".encode()).decode() + + +def read_token(token: str, db: str = DEFAULT_DB) -> int | None: + try: + raw = base64.urlsafe_b64decode(token.encode()).decode() + uid, exp, sig = raw.split(":") + good = hmac.new(_secret(db), f"{uid}:{exp}".encode(), hashlib.sha256).hexdigest() + if not hmac.compare_digest(sig, good): + return None + if int(exp) < time.time(): + return None + return int(uid) + except (ValueError, TypeError, AttributeError): + return None + + +# ---- account operations ----------------------------------------------------- +def mint_invite(db: str = DEFAULT_DB) -> str: + code = secrets.token_urlsafe(9) + con = _con(db) + con.execute("INSERT INTO invites VALUES (?,?,?)", (code, time.time(), None)) + con.commit(); con.close() + return code + + +def signup(username: str, email: str, pw: str, invite: str, + db: str = DEFAULT_DB) -> tuple[bool, str, int | None]: + username = (username or "").strip() + email = (email or "").strip().lower() + if not USERNAME_RE.match(username): + return False, "username must be 2-24 letters, numbers, . _ or -", None + if not EMAIL_RE.match(email): + return False, "invalid email", None + if len(pw or "") < MIN_PW: + return False, f"password must be at least {MIN_PW} characters", None + con = _con(db) + try: + row = con.execute("SELECT used_by FROM invites WHERE code=?", (invite or "",)).fetchone() + if row is None: + return False, "unknown invite code", None + if row[0] is not None: + return False, "invite code already used", None + if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone(): + return False, "email already registered", None + if con.execute("SELECT 1 FROM users WHERE username=? COLLATE NOCASE", (username,)).fetchone(): + return False, "username already taken", None + salt, ph = hash_pw(pw) + cur = con.execute("INSERT INTO users (username,email,pw_hash,salt,created) VALUES (?,?,?,?,?)", + (username, email, ph, salt, time.time())) + uid = cur.lastrowid + con.execute("UPDATE invites SET used_by=? WHERE code=?", (uid, invite)) + con.commit() + return True, "ok", uid + finally: + con.close() + + +def login(identifier: str, pw: str, db: str = DEFAULT_DB) -> int | None: + identifier = (identifier or "").strip() + if _throttled(identifier.lower()): + return None + con = _con(db) + try: # log in with email OR username + row = con.execute("SELECT id,salt,pw_hash FROM users WHERE email=? OR username=? COLLATE NOCASE", + (identifier.lower(), identifier)).fetchone() + finally: + con.close() + if row and verify_pw(pw or "", row[1], row[2]): + return row[0] + _note_fail(identifier.lower()) + return None + + +def user_info(user_id: int, db: str = DEFAULT_DB) -> dict | None: + con = _con(db) + try: + row = con.execute("SELECT username,email,is_admin FROM users WHERE id=?", (user_id,)).fetchone() + finally: + con.close() + if not row: + return None + admin = bool(row[2] or (ADMIN_EMAIL and row[1] == ADMIN_EMAIL)) + return {"username": row[0], "email": row[1], "admin": admin} + + +def _is_admin(user_id: int, db: str) -> bool: + info = user_info(user_id, db) + return bool(info and info.get("admin")) + + +def make_admin(email: str, db: str = DEFAULT_DB) -> bool: + con = _con(db) + cur = con.execute("UPDATE users SET is_admin=1 WHERE email=?", ((email or "").strip().lower(),)) + con.commit(); con.close() + return cur.rowcount > 0 + + +# ---- feedback (from friends testing) + admin stats -------------------------- +def submit_feedback(user_id: int, username: str, text: str, db: str = DEFAULT_DB) -> bool: + text = (text or "").strip()[:4000] + if not text: + return False + con = _con(db) + con.execute("INSERT INTO feedback (user_id,username,text,created) VALUES (?,?,?,?)", + (user_id, username, text, time.time())) + con.commit(); con.close() + return True + + +def admin_stats(db: str = DEFAULT_DB) -> dict: + con = _con(db) + try: + users = con.execute("SELECT COUNT(*) FROM users").fetchone()[0] + unused = con.execute("SELECT COUNT(*) FROM invites WHERE used_by IS NULL").fetchone()[0] + presets = con.execute("SELECT COUNT(*) FROM presets").fetchone()[0] + accounts = [{"username": u, "email": e, + "joined": time.strftime("%Y-%m-%d", time.localtime(c))} + for u, e, c in con.execute("SELECT username,email,created FROM users ORDER BY created DESC")] + fb = [{"username": u, "text": t, + "when": time.strftime("%Y-%m-%d %H:%M", time.localtime(c))} + for u, t, c in con.execute("SELECT username,text,created FROM feedback ORDER BY created DESC LIMIT 100")] + finally: + con.close() + return {"users": users, "unused_codes": unused, "presets": presets, + "accounts": accounts, "feedback": fb} + + +# ---- presets (a full personal patch, stored as one JSON blob) --------------- +def save_preset(user_id: int, name: str, data, db: str = DEFAULT_DB) -> bool: + name = (name or "").strip()[:60] + if not name: + return False + blob = json.dumps(data) + if len(blob) > 256 * 1024: # sanity cap; a patch is small + return False + con = _con(db) + con.execute("INSERT INTO presets VALUES (?,?,?,?) " + "ON CONFLICT(user_id,name) DO UPDATE SET data=excluded.data, updated=excluded.updated", + (user_id, name, blob, time.time())) + con.commit(); con.close() + return True + + +def list_presets(user_id: int, db: str = DEFAULT_DB) -> list[dict]: + con = _con(db) + try: + rows = con.execute("SELECT name,updated FROM presets WHERE user_id=? ORDER BY updated DESC", + (user_id,)).fetchall() + finally: + con.close() + return [{"name": n, "updated": u} for n, u in rows] + + +def load_preset(user_id: int, name: str, db: str = DEFAULT_DB): + con = _con(db) + try: + row = con.execute("SELECT data FROM presets WHERE user_id=? AND name=?", + (user_id, name)).fetchone() + finally: + con.close() + return json.loads(row[0]) if row else None + + +def delete_preset(user_id: int, name: str, db: str = DEFAULT_DB) -> bool: + con = _con(db) + cur = con.execute("DELETE FROM presets WHERE user_id=? AND name=?", (user_id, name)) + con.commit(); con.close() + return cur.rowcount > 0 + + +# ---- HTTP dispatch (called by hub.py's request handler) --------------------- +COOKIE = "gs_session" + + +def _cookie(token: str, clear: bool = False) -> str: + if clear: + return f"{COOKIE}=; Path=/; Max-Age=0; HttpOnly; Secure; SameSite=Lax" + age = SESSION_DAYS * 86400 + return f"{COOKIE}={token}; Path=/; Max-Age={age}; HttpOnly; Secure; SameSite=Lax" + + +def _uid_from_cookies(cookie_header: str, db: str) -> int | None: + for part in (cookie_header or "").split(";"): + if "=" in part: + k, v = part.strip().split("=", 1) + if k == COOKIE: + return read_token(v, db) + return None + + +def handle_api(method: str, path: str, body: bytes, cookie_header: str, + db: str = DEFAULT_DB): + """Returns (status, response_dict, set_cookie_or_None). Pure enough to test.""" + try: + payload = json.loads(body or b"{}") if body else {} + except (ValueError, TypeError): + payload = {} + uid = _uid_from_cookies(cookie_header, db) + + if path == "/api/me": + info = user_info(uid, db) if uid else None + return (200, info, None) if info else (401, {"error": "not logged in"}, None) + + if path == "/api/signup" and method == "POST": + ok, msg, new_uid = signup(payload.get("username"), payload.get("email"), + payload.get("password"), payload.get("invite"), db) + if ok: + return 200, user_info(new_uid, db), _cookie(make_token(new_uid, db)) + return 400, {"error": msg}, None + + if path == "/api/login" and method == "POST": + u = login(payload.get("identifier") or payload.get("email"), payload.get("password"), db) + if u: + return 200, user_info(u, db), _cookie(make_token(u, db)) + return 401, {"error": "wrong login or password"}, None + + if path == "/api/logout" and method == "POST": + return 200, {"ok": True}, _cookie("", clear=True) + + # everything below requires a session + if not uid: + return 401, {"error": "not logged in"}, None + + if path == "/api/feedback" and method == "POST": + info = user_info(uid, db) or {} + ok = submit_feedback(uid, info.get("username"), payload.get("text"), db) + return (200, {"ok": True}, None) if ok else (400, {"error": "empty feedback"}, None) + + if path.startswith("/api/admin/"): + if not _is_admin(uid, db): + return 403, {"error": "admins only"}, None + if path == "/api/admin/stats" and method == "GET": + return 200, admin_stats(db), None + if path == "/api/admin/invite" and method == "POST": + return 200, {"code": mint_invite(db)}, None + return 404, {"error": "not found"}, None + + if path == "/api/presets" and method == "GET": + return 200, {"presets": list_presets(uid, db)}, None + + if path.startswith("/api/presets/"): + name = _unquote(path[len("/api/presets/"):]) + if method == "GET": + data = load_preset(uid, name, db) + return (200, {"name": name, "data": data}, None) if data is not None else (404, {"error": "no such preset"}, None) + if method in ("PUT", "POST"): + return (200, {"ok": True}, None) if save_preset(uid, name, payload.get("data"), db) else (400, {"error": "bad preset"}, None) + if method == "DELETE": + return (200, {"ok": True}, None) if delete_preset(uid, name, db) else (404, {"error": "no such preset"}, None) + + return 404, {"error": "not found"}, None + + +def _unquote(s: str) -> str: + from urllib.parse import unquote + return unquote(s) + + +# ---- CLI + self-test -------------------------------------------------------- +def _selftest(): + import tempfile + db = os.path.join(tempfile.mkdtemp(), "t.db") + os.environ["GODSTRUMENT_SECRET"] = "test-secret-not-real" + init_db(db) + code = mint_invite(db) + # signup gates + assert signup("bo", "bad", "password123", code, db)[0] is False, "bad email rejected" + assert signup("bo", "a@b.co", "short", code, db)[0] is False, "short password rejected" + assert signup("no spaces!", "a@b.co", "password123", code, db)[0] is False, "bad username rejected" + ok, _, uid = signup("Nova", "Friend@Example.com", "hunter2pass", code, db) + assert ok and uid, "signup should succeed with a valid invite" + assert signup("Other", "other@x.co", "hunter2pass", code, db)[0] is False, "invite is single-use" + assert signup("Nova2", "friend@example.com", "hunter2pass", mint_invite(db), db)[0] is False, "email unique" + assert signup("nova", "new@x.co", "hunter2pass", mint_invite(db), db)[0] is False, "username unique (case-insensitive)" + # login by email OR username + assert login("friend@example.com", "hunter2pass", db) == uid + assert login("Nova", "hunter2pass", db) == uid, "login by username works" + assert login("friend@example.com", "wrong", db) is None + # session round-trip + tok = make_token(uid, db) + assert read_token(tok, db) == uid + assert read_token(tok + "x", db) is None, "tampered token rejected" + assert read_token("garbage", db) is None + # presets + assert save_preset(uid, "my patch", {"routes": [1, 2, 3], "bpm": 120}, db) + assert load_preset(uid, "my patch", db)["bpm"] == 120 + assert save_preset(uid, "my patch", {"bpm": 96}, db) # overwrite + assert load_preset(uid, "my patch", db)["bpm"] == 96 + assert [p["name"] for p in list_presets(uid, db)] == ["my patch"] + assert load_preset(999, "my patch", db) is None, "presets are per-user" + assert delete_preset(uid, "my patch", db) and not list_presets(uid, db) + # http layer: unauth blocked, signup+login cookie flow works + st, _, _ = handle_api("GET", "/api/presets", b"", "", db) + assert st == 401, "presets require a session" + st, resp, ck = handle_api("POST", "/api/signup", + json.dumps({"username": "Mika", "email": "mika@x.co", + "password": "orbit12345", "invite": mint_invite(db)}).encode(), "", db) + assert st == 200 and resp["username"] == "Mika" and ck and "HttpOnly" in ck and "Secure" in ck + st, resp, ck = handle_api("POST", "/api/login", + json.dumps({"identifier": "Nova", "password": "hunter2pass"}).encode(), "", db) + assert st == 200 and resp["email"] == "friend@example.com" + cookie = ck.split(";")[0] + st, resp, _ = handle_api("GET", "/api/me", b"", cookie, db) + assert st == 200 and resp["username"] == "Nova" + # feedback + admin gating + assert handle_api("POST", "/api/feedback", json.dumps({"text": "love it, one bug tho"}).encode(), cookie, db)[0] == 200 + assert handle_api("GET", "/api/admin/stats", b"", cookie, db)[0] == 403, "non-admin blocked from admin" + assert make_admin("friend@example.com", db), "grant admin" + st, stats, _ = handle_api("GET", "/api/admin/stats", b"", cookie, db) + assert st == 200 and stats["users"] >= 1, "admin can read stats" + assert any("one bug tho" in f["text"] for f in stats["feedback"]), "feedback shows in stats" + assert handle_api("POST", "/api/admin/invite", b"", cookie, db)[1].get("code"), "admin can mint codes" + print("auth self-test: all checks passed.") + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--db", default=DEFAULT_DB) + ap.add_argument("--init", action="store_true") + ap.add_argument("--mint-invite", nargs="?", type=int, const=1, metavar="N") + ap.add_argument("--users", action="store_true") + ap.add_argument("--make-admin", metavar="EMAIL") + ap.add_argument("--feedback", action="store_true", help="print recent feedback") + ap.add_argument("--selftest", action="store_true") + args = ap.parse_args() + if args.selftest: + _selftest(); return + if args.make_admin: + print("granted admin" if make_admin(args.make_admin, args.db) else "no such user"); return + if args.feedback: + for f in admin_stats(args.db)["feedback"]: + print(f" [{f['when']}] @{f['username']}: {f['text']}") + return + if args.init: + init_db(args.db); print(f"initialized {args.db}"); return + if args.mint_invite: + for _ in range(args.mint_invite): + print("invite:", mint_invite(args.db)) + return + if args.users: + con = _con(args.db) + for row in con.execute("SELECT id,username,email,created FROM users ORDER BY id"): + print(f" #{row[0]:3} {row[1]:16} {row[2]:30} joined {time.strftime('%Y-%m-%d', time.localtime(row[3]))}") + con.close(); return + ap.print_help() + + +if __name__ == "__main__": + main() diff --git a/config.json b/config.json index 2f3fd83..d60fb19 100644 --- a/config.json +++ b/config.json @@ -132,7 +132,8 @@ "season.north": {"norm": {"lo": 0, "hi": 1}, "filter": {"min_cutoff": 0.08, "beta": 0.001}, "label": "northern summer"}, "season.south": {"norm": {"lo": 0, "hi": 1}, "filter": {"min_cutoff": 0.08, "beta": 0.001}, "label": "southern summer"}, "fire.count": {"norm": {"lo": 0, "hi": 600}, "filter": {"min_cutoff": 0.2, "beta": 0.002}, "label": "wildfires worldwide"}, - "fire.event": {"type": "event", "attack": 0.01, "decay": 0.5, "label": "new wildfire"} + "fire.event": {"type": "event", "attack": 0.01, "decay": 0.5, "label": "new wildfire"}, + "midi.note.event": {"type": "event", "norm": {"lo": 0, "hi": 127}, "attack": 0.005, "decay": 0.4, "label": "MIDI note velocity"} }, "routes": [ diff --git a/hub.py b/hub.py index 4cf7937..02f4980 100644 --- a/hub.py +++ b/hub.py @@ -69,6 +69,14 @@ class Signal: if "lo" in nrm and "hi" in nrm: fixed = (nrm["lo"], nrm["hi"]) mode = "fixed" + elif not nrm and (name.startswith(("midi.", "hand.", "tof.", "light.")) + or self.type == "event"): + # Control inputs (knobs, hand, sensors) and bare events already + # speak 0..1 — adaptively rescaling a constant stream would pin it + # to 0.5 (a held knob) or halve every event. Pass them through. + # ponytail: prefix allowlist; give any exception an explicit norm. + fixed = (0.0, 1.0) + mode = "fixed" self.norm = AdaptiveNormalizer(nrm.get("window", 600), mode, fixed) flt = spec.get("filter", {}) self.filter = OneEuroFilter(flt.get("min_cutoff", 1.2), @@ -103,10 +111,16 @@ class Hub: self.matrix = ModMatrix.from_config(cfg) self.tweaks = TweakBank(cfg) self.rate = cfg.get("control_rate_hz", 60) + # read-only mode: broadcast to viewers but ignore every incoming command + # (the ws control channel has no auth — safe to expose publicly this way). + # Set via config "readonly": true or env GODSTRUMENT_READONLY=1. + self.readonly = bool(cfg.get("readonly")) or \ + os.environ.get("GODSTRUMENT_READONLY", "") not in ("", "0") self.signals: dict[str, Signal] = {} self._raw_lock = threading.Lock() self._raw_inbox: dict[str, tuple[float, float]] = {} # key -> (val,t) + self._last_raw: dict[str, float] = {} # last value fed to the normalizer self._event_q: "queue.Queue[tuple[str,float,float]]" = queue.Queue() self.out = SimpleUDPClient(cfg.get("osc_out_host", "127.0.0.1"), @@ -184,7 +198,18 @@ class Hub: for key, (raw, rt) in inbox.items(): sig = self._ensure_signal(key) - v = sig.update_continuous(raw, t) + if self._last_raw.get(key) != raw: + # only feed the normalizer/filter when the value actually + # changed. Workers re-emit their last value on a cadence with + # a fresh timestamp each time, so gating on arrival time would + # still pad an adaptive window with duplicates; gating on the + # value means the window holds N distinct samples (not N ticks + # of polling). ponytail: windows now forget by sample, not + # wall-clock — the right call for this instrument. + v = sig.update_continuous(raw, t) + self._last_raw[key] = raw + else: + v = sig.value # unchanged datum; hold the last value sources[key] = v wire_sources[key] = {"raw": round(raw, 4), "norm": round(v, 4), "label": sig.label, @@ -238,23 +263,35 @@ class Hub: # ---- outputs ------------------------------------------------------- def _emit(self, dests: dict[str, float]): midi_map = self.cfg.get("midi", {}).get("map", {}) + notes_live = set() # note dests sounding this frame for dest, val in dests.items(): self.out.send_message(f"/out/{dest}", float(val)) m = midi_map.get(dest) - if m and self.midi: - if m.get("note") and val >= 40: # quantized note number - note = int(val) - if self._last_notes.get(dest) != note: - if self._last_notes.get(dest): - self.midi.send_message([0x80 | m.get("channel", 0), - self._last_notes[dest], 0]) - self.midi.send_message([0x90 | m.get("channel", 0), - note, 100]) - self._last_notes[dest] = note - elif "cc" in m: - cc = int(_clamp01(val) * 127) - self.midi.send_message([0xB0 | m.get("channel", 0), - m["cc"], cc]) + if not (m and self.midi): + continue + if m.get("note"): # note-mapped: val IS a note number + note = int(round(val)) + if not 24 <= note <= 127: # matches the viz Web-MIDI guard + continue + notes_live.add(dest) + if self._last_notes.get(dest) != note: + if self._last_notes.get(dest) is not None: + self.midi.send_message([0x80 | m.get("channel", 0), + self._last_notes[dest], 0]) + self.midi.send_message([0x90 | m.get("channel", 0), + note, 100]) + self._last_notes[dest] = note + elif "cc" in m: + cc = int(_clamp01(val) * 127) + self.midi.send_message([0xB0 | m.get("channel", 0), + m["cc"], cc]) + # release any held note whose source went silent (muted/gated) this frame + if self.midi: + for dest, note in list(self._last_notes.items()): + if note is not None and dest not in notes_live: + ch = midi_map.get(dest, {}).get("channel", 0) + self.midi.send_message([0x80 | ch, note, 0]) + self._last_notes[dest] = None async def _broadcast(self, t, sources, dests, active, macros=None): if not self.clients: @@ -285,9 +322,13 @@ class Hub: try: await ws.send(json.dumps({ "hello": "godstrument", + "readonly": self.readonly, + # full route spec so a logged-in client can compute its own mix "routes": [{"source": r.get("source"), "dest": r.get("dest"), "label": r.get("label", ""), - "amount": round(r.get("amount", 0), 3)} + "amount": round(r.get("amount", 0), 3), + "curve": r.get("curve", "lin"), + "gate": r.get("gate"), "quantize": r.get("quantize")} for r in self.matrix.routes], "groups": self.cfg.get("groups", {}), "midi": self.cfg.get("midi", {}).get("map", {}), @@ -311,7 +352,7 @@ class Hub: def _save_patch(self, name): if not name: return - safe = "".join(c for c in name if c.isalnum() or c in "-_ ").strip() or "patch" + safe = _safe_name(name) or "patch" os.makedirs(self.patches_dir, exist_ok=True) data = {"routes": self.matrix.routes, "groups": self.tweaks.dump_groups(), @@ -323,10 +364,13 @@ class Hub: print(f" saved template '{safe}'") def _load_patch(self, name): + safe = _safe_name(name) # no path traversal / absolute paths + if not safe: + return try: - with open(os.path.join(self.patches_dir, name + ".json")) as f: + with open(os.path.join(self.patches_dir, safe + ".json")) as f: data = json.load(f) - except (OSError, TypeError): + except (OSError, ValueError): return if "routes" in data: self.matrix.routes = [dict(r) for r in data["routes"]] @@ -334,9 +378,11 @@ class Hub: self.tweaks.load_groups(data["groups"]) if "state" in data: self.tweaks.load_state(data["state"]) - print(f" loaded spell '{name}'") + print(f" loaded spell '{safe}'") def _handle_cmd(self, m: dict): + if self.readonly: # public spectator: ignore all control commands + return cmd = m.get("cmd") if cmd == "mute_toggle": self.tweaks.toggle(m.get("target"), "mute") @@ -344,15 +390,15 @@ class Hub: self.tweaks.toggle(m.get("target"), m.get("param", "mute")) elif cmd == "set": self.tweaks.set_param(m.get("target"), m.get("param", "gain"), - float(m.get("value", 0.0))) + _ffloat(m.get("value"), 0.0)) elif cmd == "add_route": self.matrix.add_route(m.get("source"), m.get("dest"), - float(m.get("amount", 0.5))) + _ffloat(m.get("amount"), 0.5)) elif cmd == "remove_route": self.matrix.remove_route(m.get("source"), m.get("dest")) elif cmd == "set_route": self.matrix.set_route_amount(m.get("source"), m.get("dest"), - float(m.get("amount", 0.5))) + _ffloat(m.get("amount"), 0.5)) elif cmd == "make_group": self.tweaks.add_group(m.get("name"), m.get("members", [])) elif cmd == "save_patch": @@ -415,9 +461,76 @@ def _open_midi(midi_cfg: dict): def _serve_http(directory: str, port: int): - handler = partial(SimpleHTTPRequestHandler, directory=directory) - httpd = ThreadingHTTPServer(("127.0.0.1", port), handler) - httpd.RequestHandlerClass.log_message = lambda *a, **k: None + try: + import auth # invite-only accounts + presets, mounted at /api/* + except Exception as e: # noqa + auth = None + print(f" (accounts disabled: {e})") + + class Handler(SimpleHTTPRequestHandler): + def __init__(self, *a, **k): + super().__init__(*a, directory=directory, **k) + + def log_message(self, *a, **k): + pass + + def end_headers(self): + # never let a browser serve a stale console after a deploy + if not self.path.startswith("/api/"): + self.send_header("Cache-Control", "no-cache, must-revalidate") + super().end_headers() + + def _api(self, method): + path = self.path.split("?", 1)[0] + if auth is None: + return self.send_error(503, "accounts unavailable") + length = int(self.headers.get("Content-Length", 0) or 0) + if length > 512 * 1024: + return self.send_error(413, "too large") + body = self.rfile.read(length) if length else b"" + try: + status, resp, set_cookie = auth.handle_api( + method, path, body, self.headers.get("Cookie", "")) + except Exception: # never leak a stack trace to the client + status, resp, set_cookie = 500, {"error": "server error"}, None + payload = json.dumps(resp).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.send_header("Cache-Control", "no-store") + if set_cookie: + self.send_header("Set-Cookie", set_cookie) + self.end_headers() + self.wfile.write(payload) + + def _force_https(self): + # visitors on http can't keep a Secure session cookie -> send them to + # https (Cloudflare sets X-Forwarded-Proto to the visitor's scheme). + if self.headers.get("X-Forwarded-Proto", "").lower() == "http": + host = self.headers.get("Host", "") + if host: + self.send_response(301) + self.send_header("Location", f"https://{host}{self.path}") + self.send_header("Content-Length", "0") + self.end_headers() + return True + return False + + def do_GET(self): + if self._force_https(): + return + self._api("GET") if self.path.startswith("/api/") else super().do_GET() + + def do_POST(self): + self._api("POST") if self.path.startswith("/api/") else self.send_error(501) + + def do_PUT(self): + self._api("PUT") if self.path.startswith("/api/") else self.send_error(501) + + def do_DELETE(self): + self._api("DELETE") if self.path.startswith("/api/") else self.send_error(501) + + httpd = ThreadingHTTPServer(("127.0.0.1", port), Handler) threading.Thread(target=httpd.serve_forever, daemon=True).start() @@ -425,6 +538,23 @@ def _clamp01(x: float) -> float: return 0.0 if x < 0.0 else 1.0 if x > 1.0 else x +def _safe_name(name) -> str: + """A patch name safe to use as a filename: basename only, alnum/-/_/space. + Defeats path traversal ('../x') and absolute paths from the ws command.""" + base = os.path.basename(str(name or "")) + return "".join(c for c in base if c.isalnum() or c in "-_ ").strip() + + +def _ffloat(x, default: float = 0.0) -> float: + """float() that rejects NaN/Infinity (json.loads accepts them) so a crafted + command can't poison the matrix or blind the console.""" + try: + v = float(x) + except (TypeError, ValueError): + return default + return v if math.isfinite(v) else default + + def main(): ap = argparse.ArgumentParser() ap.add_argument("--config", default=os.path.join(HERE, "config.json")) diff --git a/normalize.py b/normalize.py index dd05889..e4e058b 100644 --- a/normalize.py +++ b/normalize.py @@ -145,7 +145,12 @@ class ImpulseEnvelope: self._amount = 0.0 def trigger(self, amount: float = 1.0, t: float | None = None): - self._amount = _clamp01(amount) + amt = _clamp01(amount) + # peak-hold: a smaller event arriving mid-decay must not chop a bigger + # one that's still ringing (nor zero it out on a negative magnitude). + if t is not None and self._t_trigger is not None and amt <= self.value(t): + return + self._amount = amt self._t_trigger = t self._peak = 0.0 diff --git a/run.py b/run.py index 630d443..3432675 100644 --- a/run.py +++ b/run.py @@ -18,6 +18,7 @@ from __future__ import annotations import argparse import json import os +import signal import subprocess import sys import time @@ -145,22 +146,25 @@ def main(): procs.append(hub) time.sleep(2.0) # let the hub bind its ports - for path in PROFILES[args.profile]: - p = launch(path, worker_args(path, loc)) + workers: list[dict] = [] # supervised: relaunched if they die mid-set + + def start_worker(path, extra=None): + p = launch(path, extra) if p: procs.append(p) + workers.append({"path": path, "extra": extra, "proc": p, "next_ok": 0.0}) + return p + + for path in PROFILES[args.profile]: + start_worker(path, worker_args(path, loc)) time.sleep(0.15) if args.dealgod: - d = launch("workers/world_dealgod.py") - if d: - procs.append(d) + if start_worker("workers/world_dealgod.py"): print(" dealgod market worker -> reading the warehouse") if args.record: - r = launch("recorder.py") - if r: - procs.append(r) + if start_worker("recorder.py"): print(" recording -> godstrument.db") url = f"http://localhost:{http_port}" @@ -172,11 +176,29 @@ def main(): except Exception: pass + # so `kill` / systemd stop (SIGTERM) runs the finally block instead of + # orphaning the hub + ~20 workers with their ports still bound + def _on_sigterm(*_): + raise KeyboardInterrupt + signal.signal(signal.SIGTERM, _on_sigterm) + try: while True: if hub.poll() is not None: print(" hub exited; shutting down.") break + now = time.time() + for w in workers: # revive any worker that died + if w["proc"].poll() is not None and now >= w["next_ok"]: + print(f" ({os.path.basename(w['path'])} died; restarting)") + np = launch(w["path"], w["extra"]) + if np: + try: # swap the dead handle out, + procs[procs.index(w["proc"])] = np # don't leak it + except ValueError: + procs.append(np) + w["proc"] = np + w["next_ok"] = now + 10.0 # rate-limit restart storms time.sleep(0.5) except KeyboardInterrupt: print("\n godstrument sleeps.") diff --git a/test_fixes.py b/test_fixes.py new file mode 100644 index 0000000..5c2e647 --- /dev/null +++ b/test_fixes.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +"""Self-check for the review fixes. Run: python3 test_fixes.py +Asserts the behaviours that were broken now hold. No framework, no fixtures.""" +import math + +import json +import os + +from normalize import AdaptiveNormalizer, ImpulseEnvelope +from transform import Tweak, TweakBank +from hub import Signal, Hub, _safe_name, _ffloat + + +def test_control_signals_pass_through(): + # a knob/hand signal (midi./tof./hand./light.) with no norm spec must NOT be + # adaptively rescaled (that pinned held knobs to 0.5 and froze the planet). + s = Signal("midi.cc.20", {}) + assert s.norm.mode == "fixed", "control input should be fixed 0..1" + assert abs(s.norm(1.0) - 1.0) < 1e-9 and abs(s.norm(0.0)) < 1e-9 + # a bare event (fire.event) must trigger at full magnitude, not 0.5 + e = Signal("fire.event", {"type": "event"}) + assert e.norm.mode == "fixed" + assert abs(e.norm(1.0) - 1.0) < 1e-9, "constant-magnitude event must reach 1.0" + # a world signal with no spec keeps adaptive behaviour + w = Signal("crypto.vel", {}) + assert w.norm.mode == "minmax" + + +def test_normalizer_collapses_on_repeats(): + # identical values collapse to 0.5 (hi==lo) — this is WHY the hub must not + # re-feed the window with a worker's re-emitted duplicates. + n = AdaptiveNormalizer(100, "minmax") + for _ in range(50): + out = n(0.7) + assert abs(out - 0.5) < 1e-9 + + +def test_hub_value_dedup_gate(): + # Replicates the hub's inbox gate: workers re-emit their last value on a + # cadence (fresh timestamp each time), so gating on arrival time would still + # pad the window with duplicates. Gating on the VALUE keeps only real changes. + n = AdaptiveNormalizer(600, "minmax") + last = object() + # 10 datagrams, each with a distinct arrival tick, but only 2 distinct values + stream = [(100.0, i) for i in range(5)] + [(200.0, i) for i in range(5, 10)] + for raw, _rt in stream: + if last != raw: # the gate hub.control_loop now uses + n(raw) + last = raw + assert len(n.buf) == 2, f"re-emits leaked into the window: {len(n.buf)}" + + +def test_group_freeze_is_per_member(): + # one group Tweak applied to 3 members must hold EACH at its own value. + g = Tweak({"freeze": 1.0}) + vals = {"a": 0.8, "b": 0.2, "c": 0.55} + held = {k: g.apply(v, k) for k, v in vals.items()} + assert held == vals, f"freeze must hold each member; got {held}" + # and a second frame keeps holding the same per-member values + held2 = {k: g.apply(0.0, k) for k in vals} + assert held2 == vals, f"freeze must persist per member; got {held2}" + + +def test_group_smooth_is_per_member(): + g = Tweak({"smooth": 0.9}) + # push two members apart; each must chase its own target, not a shared blend + for _ in range(200): + a = g.apply(1.0, "a") + b = g.apply(0.0, "b") + assert a > 0.9 and b < 0.1, f"per-member smooth failed: a={a} b={b}" + + +def test_envelope_peak_hold(): + env = ImpulseEnvelope(attack=0.01, decay=1.0) + env.trigger(1.0, t=0.0) + big = env.value(0.1) # a large event ringing + env.trigger(0.1, t=0.1) # a small event arrives mid-decay + after = env.value(0.1) + assert after >= big - 1e-9, f"small event chopped the big one: {big}->{after}" + # a negative magnitude must never zero a ringing envelope + env.trigger(-5.0, t=0.11) + assert env.value(0.11) > 0.5, "negative-mag event zeroed the swell" + # but when nothing is ringing, a fresh event still triggers + env2 = ImpulseEnvelope(0.01, 1.0) + env2.trigger(1.0, t=0.0) + assert env2.value(0.02) > 0.9 + + +def test_safe_name_blocks_traversal(): + assert _safe_name("../../etc/passwd") == "passwd" + assert _safe_name("/absolute/secret") == "secret" + assert _safe_name("my spell-2") == "my spell-2" + assert _safe_name("..") == "" + assert _safe_name(None) == "" + + +def test_ffloat_rejects_non_finite(): + assert _ffloat(float("nan"), 0.5) == 0.5 + assert _ffloat(float("inf"), 0.5) == 0.5 + assert _ffloat("1.25") == 1.25 + assert _ffloat(None, 0.3) == 0.3 + + +def test_spell_saves_muted_sources(): + # a source muted by clicking its sphere must be considered "active" so + # dump_state persists it into a spell. + cfg = {"groups": {}, "tweaks": {}, "controls": []} + tb = TweakBank(cfg) + tb.toggle("weather.temp", "mute") # mute a single source + state = tb.dump_state() + assert "weather.temp" in state["sources"], "muted source dropped from spell" + + +def test_regroup_drops_ghost_membership(): + cfg = {"groups": {"x": {"members": ["a", "b", "c", "d"]}}, + "tweaks": {}, "controls": []} + tb = TweakBank(cfg) + tb.groups["x"].set("mute", 1.0) # mute the group + tb.add_group("x", ["a", "b"]) # redefine with fewer members + # c and d must no longer be silenced by group x + assert not tb.is_muted("c"), "ghost membership kept muting a removed source" + assert tb.is_muted("a"), "remaining member should still be muted" + + +def test_readonly_ignores_commands(): + # a public (read-only) hub must ignore every incoming ws command + cfg = json.load(open(os.path.join(os.path.dirname(__file__), "config.json"))) + os.environ["GODSTRUMENT_READONLY"] = "1" + try: + h = Hub(cfg) + finally: + os.environ.pop("GODSTRUMENT_READONLY", None) + assert h.readonly + before = [dict(r) for r in h.matrix.routes] + h._handle_cmd({"cmd": "add_route", "source": "x", "dest": "glitch", "amount": 1}) + h._handle_cmd({"cmd": "set", "target": "@planet", "param": "mute", "value": 1}) + h._handle_cmd({"cmd": "load_patch", "name": "../../etc/passwd"}) + assert [dict(r) for r in h.matrix.routes] == before, "readonly leaked a route change" + assert not h.tweaks.groups["planet"].p["mute"], "readonly leaked a mute" + + +def main(): + tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + for t in tests: + t() + print(f" ok {t.__name__}") + print(f"\n{len(tests)} checks passed.") + + +if __name__ == "__main__": + main() diff --git a/transform.py b/transform.py index 68ca296..572759e 100644 --- a/transform.py +++ b/transform.py @@ -38,8 +38,8 @@ class Tweak: for k, v in spec.items(): if k in self.p: self.p[k] = float(v) if not isinstance(v, bool) else float(v) - self._held: float | None = None - self._lag: float | None = None # exponential-smoothing state + self._held: dict[str, float] = {} # freeze holds, keyed by signal name + self._lag: dict[str, float] = {} # smoothing state, keyed by signal name self.orbit_lfo = 1.0 # external planetary-orbit gain (hub-set) def set(self, param: str, value: float): @@ -50,19 +50,23 @@ class Tweak: def active(self) -> bool: p = self.p return (p["gain"] != 1.0 or p["offset"] != 0.0 or p["invert"] >= 0.5 - or p["freeze"] >= 0.5 or p["smooth"] > 0.0) + or p["freeze"] >= 0.5 or p["smooth"] > 0.0 or p["mute"] >= 0.5) - def apply(self, x: float) -> float: + def apply(self, x: float, key: str = "") -> float: + # `key` is the signal name: a group Tweak is applied to every member, so + # freeze/smooth state must be kept per-signal, not once for the group. p = self.p # mute: silence this source/group entirely if p["mute"] >= 0.5: return 0.0 - # freeze: hold the last emitted value (a CDJ pause on the data) + # freeze: hold each signal's own last value (a CDJ pause on the data) if p["freeze"] >= 0.5: - if self._held is None: - self._held = x if self._lag is None else self._lag - return self._held - self._held = None + h = self._held.get(key) + if h is None: + h = self._lag.get(key, x) + self._held[key] = h + return h + self._held.pop(key, None) v = (1.0 - x) if p["invert"] >= 0.5 else x v = v * p["gain"] * self.orbit_lfo + p["offset"] @@ -71,10 +75,11 @@ class Tweak: s = p["smooth"] if s > 0.0: alpha = 1.0 - min(0.985, s * 0.985) # smooth=0 -> 1 (pass), 1 -> ~0.015 - self._lag = v if self._lag is None else self._lag + alpha * (v - self._lag) - v = self._lag + prev = self._lag.get(key) + self._lag[key] = v if prev is None else prev + alpha * (v - prev) + v = self._lag[key] else: - self._lag = v + self._lag[key] = v return _clamp01(v) @@ -142,10 +147,10 @@ class TweakBank: for name, val in sources.items(): v = val for g in self.member_groups.get(name, ()): # stack every concept-group - v = self.groups[g].apply(v) + v = self.groups[g].apply(v, name) st = self.source_tweaks.get(name) if st: - v = st.apply(v) + v = st.apply(v, name) out[name] = v return out @@ -181,6 +186,12 @@ class TweakBank: """Create/replace a group at runtime (browser multi-select -> group).""" if not name or not members: return + # drop stale memberships from a previous definition so removed sources + # don't keep getting muted/tweaked by this group forever + for m in self.group_members.get(name, []): + gl = self.member_groups.get(m) + if gl and name in gl: + gl.remove(name) self.group_members[name] = list(members) self.groups.setdefault(name, Tweak()) self.group_label[name] = label or name diff --git a/workers/world_bank.py b/workers/world_bank.py index 21d7bff..3c80c6b 100644 --- a/workers/world_bank.py +++ b/workers/world_bank.py @@ -57,19 +57,26 @@ def main(): f"(re-query every {args.poll:g}s)") values: dict[str, float] = {} - last_poll = 0.0 + next_poll, retry = 0.0, 60.0 while True: - if time.time() - last_poll > args.poll or not values: + if time.time() >= next_poll: + got = False for ind, (addr, name) in INDICATORS.items(): try: v, year = fetch(ind) if v is not None: values[addr] = v + got = True print(f" 🌍 {name}: {v} ({year})") - time.sleep(1.0) # be gentle with the API except Exception as e: # noqa print(f"[worldbank] {ind} warn: {e}") - last_poll = time.time() + time.sleep(1.0) # be gentle with the API + if got: + next_poll, retry = time.time() + args.poll, 60.0 + else: # nothing came back — back off + print(f"[worldbank] no values; retry in {retry:.0f}s") + next_poll = time.time() + retry + retry = min(retry * 2, args.poll) for addr, v in values.items(): client.send_message("/gs/" + addr, v) time.sleep(10.0) diff --git a/workers/world_crypto.py b/workers/world_crypto.py index a1234c2..af0e2da 100644 --- a/workers/world_crypto.py +++ b/workers/world_crypto.py @@ -74,32 +74,25 @@ async def main(): print(f"[{NAME}] emitting /gs/crypto/price,/gs/crypto/vel every <1s (ws) / {POLL_S:g}s (rest)") backoff = 1.0 - rest_only = False while True: - if rest_only: - # We could not open the socket at all; stay on REST but keep - # occasionally retrying the socket by resetting the flag on error. - try: - await poll_rest(client) - except Exception as e: - print(f"[{NAME}] REST loop error: {e}; retrying") - await asyncio.sleep(POLL_S) - continue - try: await run_websocket(client) # Clean disconnect (stream ended): reconnect quickly. print(f"[{NAME}] websocket closed; reconnecting") backoff = 1.0 + await asyncio.sleep(1.0) # avoid a hot loop on an instant re-close except Exception as e: - print(f"[{NAME}] websocket error: {e}; backoff {backoff:g}s") - await asyncio.sleep(backoff) + # Bridge the gap on REST for the backoff window, then always retry + # the socket — never latch onto REST for the rest of the session. + print(f"[{NAME}] websocket error: {e}; REST bridge {backoff:g}s, then retry") + try: + await asyncio.wait_for(poll_rest(client), timeout=backoff) + except asyncio.TimeoutError: + pass + except Exception as e2: + print(f"[{NAME}] REST bridge error: {e2}") + await asyncio.sleep(min(backoff, POLL_S)) backoff = min(backoff * 2, 30.0) - # If we have never managed to connect, drop to REST fallback, - # but keep trying the socket periodically via a bounded retry. - if backoff >= 8.0 and not rest_only: - print(f"[{NAME}] websocket unreachable; falling back to REST polling") - rest_only = True if __name__ == "__main__": diff --git a/workers/world_dealgod.py b/workers/world_dealgod.py index 4577314..653204a 100644 --- a/workers/world_dealgod.py +++ b/workers/world_dealgod.py @@ -90,7 +90,11 @@ def main(): state = {"vel": None, "turn": None} def load(): - rows, err = fetch(args.ssh, args.days, args.binmin) + try: + rows, err = fetch(args.ssh, args.days, args.binmin) + except Exception as e: # SSH timeout, missing ssh binary, tailnet stall + print(f"[dealgod] query failed ({e}); retrying") + return if len(rows) >= 2: vel, turn = build(rows, args.into) state["vel"], state["turn"] = vel, turn diff --git a/workers/world_debt.py b/workers/world_debt.py index 372942f..b94d8ac 100644 --- a/workers/world_debt.py +++ b/workers/world_debt.py @@ -45,19 +45,21 @@ def main(): f"(re-query every {args.poll:g}s)") total, vel, last = 0.0, 0.0, None - last_poll = 0.0 + next_poll, retry = 0.0, 60.0 while True: - try: - if time.time() - last_poll > args.poll or total == 0.0: + if time.time() >= next_poll: + try: total, vel, date = fetch() - last_poll = time.time() if date != last: print(f" 💰 debt ${total/1e12:.3f}T (daily churn ${vel/1e9:.1f}B)") last = date - client.send_message("/gs/debt/total", total) - client.send_message("/gs/debt/vel", vel) - except Exception as e: # noqa - print(f"[debt] warn: {e}") + next_poll, retry = time.time() + args.poll, 60.0 + except Exception as e: # noqa + print(f"[debt] warn: {e}; retry in {retry:.0f}s") + next_poll = time.time() + retry + retry = min(retry * 2, args.poll) + client.send_message("/gs/debt/total", total) + client.send_message("/gs/debt/vel", vel) time.sleep(5.0) diff --git a/workers/world_econ.py b/workers/world_econ.py index 02c6929..89cdfdf 100644 --- a/workers/world_econ.py +++ b/workers/world_econ.py @@ -53,19 +53,23 @@ def main(): f"(re-query every {args.poll:g}s)") yoy, mom, last = 0.0, 0.0, None - last_poll = 0.0 + next_poll, retry = 0.0, 60.0 while True: - try: - if time.time() - last_poll > args.poll or last is None: + if time.time() >= next_poll: + try: yoy, mom, label = fetch() - last_poll = time.time() if label != last: print(f" 📈 inflation {yoy:.2f}% YoY ({label}, {mom:+.2f}% MoM)") last = label - client.send_message("/gs/econ/inflation", yoy) - client.send_message("/gs/econ/cpi_mom", mom) - except Exception as e: # noqa - print(f"[econ] warn: {e}") + next_poll, retry = time.time() + args.poll, 60.0 + except Exception as e: # noqa + # back off on failure: BLS allows ~25 req/day unregistered, so a + # 10s retry-storm would exhaust the quota and lock us out all day + print(f"[econ] warn: {e}; retry in {retry:.0f}s") + next_poll = time.time() + retry + retry = min(retry * 2, args.poll) + client.send_message("/gs/econ/inflation", yoy) + client.send_message("/gs/econ/cpi_mom", mom) time.sleep(10.0) diff --git a/workers/world_fire.py b/workers/world_fire.py index cf78ed2..9dae399 100644 --- a/workers/world_fire.py +++ b/workers/world_fire.py @@ -44,10 +44,10 @@ def main(): seen: set[str] = set() count = 0 first = True - last_poll = 0.0 + next_poll, retry = 0.0, 60.0 while True: - try: - if time.time() - last_poll > args.poll or first: + if time.time() >= next_poll: + try: events = fetch() count = len(events) for e in events: @@ -59,10 +59,12 @@ def main(): print(f" 🔥 {e.get('title', 'wildfire')}") print(f" {count} wildfires burning worldwide") first = False - last_poll = time.time() - client.send_message("/gs/fire/count", float(count)) - except Exception as e: # noqa - print(f"[fire] warn: {e}") + next_poll, retry = time.time() + args.poll, 60.0 + except Exception as e: # noqa + print(f"[fire] warn: {e}; retry in {retry:.0f}s") + next_poll = time.time() + retry + retry = min(retry * 2, args.poll) + client.send_message("/gs/fire/count", float(count)) time.sleep(10.0) diff --git a/workers/world_planes.py b/workers/world_planes.py index 1d780f0..66a1918 100644 --- a/workers/world_planes.py +++ b/workers/world_planes.py @@ -16,8 +16,11 @@ NAME = "world_planes" API = "https://opensky-network.org/api/states/all" UA = "Godstrument/1.0 (live instrument; contact monsterrobotparty@gmail.com)" -POLL_OK = 30.0 # normal poll interval (s) -POLL_429 = 120.0 # backoff poll interval after HTTP 429 +# OpenSky anonymous budget is ~400 credits/day; a bbox query costs 2-4 credits, +# so <200 requests/day -> poll no faster than ~1 per 450s or the feed 429s out +# and freezes for the rest of the day. (Add OAuth2 creds for the 4000 tier.) +POLL_OK = 450.0 # normal poll interval (s) +POLL_429 = 600.0 # backoff poll interval after HTTP 429 REEMIT = 10.0 # re-emit last values every 10s @@ -100,8 +103,12 @@ def main(): poll = POLL_OK except urllib.error.HTTPError as ex: if ex.code == 429: - poll = POLL_429 - print(f"[{NAME}] HTTP 429 rate limited; backing off to {POLL_429:.0f}s") + retry_after = ex.headers.get("X-Rate-Limit-Retry-After-Seconds") + try: + poll = max(POLL_429, float(retry_after)) + except (TypeError, ValueError): + poll = POLL_429 + print(f"[{NAME}] HTTP 429 rate limited; backing off to {poll:.0f}s") else: print(f"[{NAME}] HTTP {ex.code}; retrying in {poll:.0f}s") except (urllib.error.URLError, TimeoutError, OSError,