geniusrun/backend/internal/garmin/pyscript/wrapper.py

"""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."""
    global _client, _auth_state
    email = os.environ.get("GARMIN_EMAIL")
    password = os.environ.get("GARMIN_PASSWORD")
    if not email or not password:
        return
    try:
        _client = Garmin(email, password)
        _client.login(tokenstore=TOKENSTORE)
        _auth_state = "authenticated"
    except Exception as exc:
        _debug(f"startup tokenstore login failed: {exc}")
        _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()