"""Subprocess wrapper around garminconnect, spoken to over newline-delimited JSON on stdin/stdout by geniusrund's internal/garmin package. See docs/superpowers/specs/2026-07-25-garmin-direct-wrapper-design.md.""" import json import os import queue import sys import threading import traceback from garminconnect import Garmin TOKENSTORE = os.path.expanduser(os.environ.get("GARMIN_TOKENSTORE", "~/.garth")) _client = None _auth_state = "unauthenticated" _mfa_input_queue = queue.Queue() _login_result_queue = queue.Queue() def _debug(msg): print(f"[garmin-wrapper debug] {msg}", file=sys.stderr, flush=True) def _prompt_mfa(): _debug("garminconnect invoked prompt_mfa() -- this is a REAL MFA challenge from Garmin") code = _mfa_input_queue.get(timeout=300) _debug(f"prompt_mfa() handing code of length {len(code)} back to garminconnect") return code def _startup_login(): """Silently resume a cached tokenstore session at process start, so a freshly (re)spawned subprocess is already authenticated for background syncs that never call the explicit authenticate command. Bounded to the same 10s timeout as _handle_authenticate: the actual login runs on a background daemon thread, and this function waits up to 10s for it before returning either way. This runs before main() enters its stdin dispatch loop, so a slow/rate-limited Garmin login must never block indefinitely here -- if it times out, the thread keeps running and will update _auth_state whenever it eventually finishes (success or failure), same as today, just without wedging the whole subprocess unresponsive in the meantime. Deliberately uses its own private, function-local result queue rather than the module-level _login_result_queue that _handle_authenticate and _handle_complete_mfa share -- those two are legitimately two halves of one explicit, MFA-capable login flow and must share a queue for the MFA handoff to work, but this is a plain background tokenstore resume with no MFA involved. Sharing the queue here would let this thread's stale result (arriving after the 10s timeout below) be dequeued by an unrelated, later authenticate/complete_mfa call instead of that call's own fresh result.""" global _client, _auth_state email = os.environ.get("GARMIN_EMAIL") password = os.environ.get("GARMIN_PASSWORD") if not email or not password: return _client = Garmin(email, password) result_queue = queue.Queue() # private to this call, never shared with authenticate/complete_mfa def _do_login(): try: _client.login(tokenstore=TOKENSTORE) result_queue.put(("success", None)) except Exception as exc: result_queue.put(("error", str(exc))) threading.Thread(target=_do_login, daemon=True).start() try: status, err = result_queue.get(timeout=10) if status == "success": _auth_state = "authenticated" else: _debug(f"startup tokenstore login failed: {err}") _auth_state = "unauthenticated" except queue.Empty: _debug( "startup tokenstore login hit the 10s timeout -- still running in the " "background and will update auth state whenever it finishes" ) _auth_state = "unauthenticated" def _handle_authenticate(_params): global _client, _auth_state email = os.environ.get("GARMIN_EMAIL", "") password = os.environ.get("GARMIN_PASSWORD", "") if not email or not password: return { "status": "failed", "message": "GARMIN_EMAIL and GARMIN_PASSWORD environment variables are required.", } def _do_login(): _debug(f"background login thread starting _client.login(tokenstore={TOKENSTORE})") try: _client.login(tokenstore=TOKENSTORE) _debug("_client.login() returned successfully") _login_result_queue.put(("success", None)) except Exception as exc: _debug(f"_client.login() raised {type(exc).__name__}: {exc}") _debug(traceback.format_exc()) _login_result_queue.put(("error", str(exc))) _client = Garmin(email, password) _client.prompt_mfa = _prompt_mfa threading.Thread(target=_do_login, daemon=True).start() try: status, err = _login_result_queue.get(timeout=10) _debug(f"authenticate got result within 10s timeout: status={status}") if status == "success": _auth_state = "authenticated" return {"status": "success", "message": "Authenticated successfully."} return {"status": "failed", "message": f"Authentication failed: {err}"} except queue.Empty: _debug( "authenticate hit the 10s timeout with no result yet -- reporting mfa_required, " "but this does NOT necessarily mean prompt_mfa() was actually invoked; check " "whether the 'REAL MFA challenge' debug line above appears to tell real MFA " "apart from a merely slow login." ) _auth_state = "mfa_pending" return { "status": "mfa_required", "message": "MFA required. Garmin has sent a verification code to your registered email or phone.", } def _handle_complete_mfa(params): global _auth_state if _auth_state != "mfa_pending": return {"status": "failed", "message": "No MFA in progress. Call authenticate first."} code = params["code"] _debug(f"complete_mfa received a code of length {len(code)}, pushing to mfa queue") _mfa_input_queue.put(code) try: status, err = _login_result_queue.get(timeout=30) _debug(f"complete_mfa got result: status={status} err={err}") if status == "success": _auth_state = "authenticated" return {"status": "success", "message": "MFA accepted. Authenticated successfully."} return {"status": "failed", "message": f"Authentication failed after MFA: {err}"} except queue.Empty: _auth_state = "unauthenticated" return { "status": "failed", "message": "Timed out waiting for authentication to complete. Call authenticate again.", } def _handle_call(params): if _auth_state != "authenticated": raise RuntimeError("Not authenticated. Call authenticate first.") method = params["method"] args = params.get("args") or {} fn = getattr(_client, method) return fn(**args) _HANDLERS = { "authenticate": _handle_authenticate, "complete_mfa": _handle_complete_mfa, "call": _handle_call, } def dispatch(req): handler = _HANDLERS.get(req.get("cmd")) if handler is None: return {"id": req.get("id"), "error": f"unknown cmd {req.get('cmd')!r}"} try: result = handler(req.get("params") or {}) return {"id": req["id"], "result": result} except Exception as exc: _debug(f"{req.get('cmd')} raised {type(exc).__name__}: {exc}") _debug(traceback.format_exc()) return {"id": req.get("id"), "error": str(exc)} def main(): _startup_login() for line in sys.stdin: line = line.strip() if not line: continue req = json.loads(line) resp = dispatch(req) print(json.dumps(resp), flush=True) if __name__ == "__main__": main()