kriss/geniusrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(garmin): all-JSON wrapper protocol and log stream

Browse Source 
Error responses from wrapper.py now carry error_type and traceback in
the protocol itself, logged as attrs on the Go side's per-call record.
wrapper.py's free-text stderr debug prints become JSON log lines
({level,msg,...attrs}); client.go parses that stream and re-emits it
through the application logger (level-mapped, source=garmin-wrapper),
wrapping any non-JSON line instead of passing it through raw. The
wrapper also survives malformed request lines and unserializable
results with JSON responses instead of crashing with a raw traceback.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Kriss Vila
2026-08-04 16:42:25 +02:00
parent 431315bac3
commit dcbb1d8bb0
6 changed files with 200 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
backend/internal/garmin/client.go
View File

@@ -77,11 +77,16 @@ type wireRequest struct {
}
// wireResponse is one line read from the wrapper subprocess's stdout.
// ErrorType/Traceback carry the Python-side failure detail in the protocol
// itself (see wrapper.py's dispatch), so the Go side logs one complete
// structured record per failed call instead of correlating stderr noise.
type wireResponse struct {
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error string `json:"error,omitempty"`
NotFound bool `json:"not_found,omitempty"`
ID int `json:"id"`
Result json.RawMessage `json:"result,omitempty"`
Error string `json:"error,omitempty"`
ErrorType string `json:"error_type,omitempty"`
Traceback string `json:"traceback,omitempty"`
NotFound bool `json:"not_found,omitempty"`
}
// ErrNotFound wraps any error a Client method returns when the wrapper
@@ -186,13 +191,16 @@ func (c *subprocessClient) ensureStarted(ctx context.Context) error {
if err := cmd.Start(); err != nil {
return fmt.Errorf("spawn garmin wrapper subprocess: %w", err)
}
// wrapper.py logs auth/rate-limit diagnostics to stderr. Per os/exec's
// StderrPipe docs, it's incorrect to call Wait before all reads from the
// pipe have completed, so close() waits on stderrDone before Wait-ing.
// wrapper.py logs auth/rate-limit diagnostics to stderr as JSON lines
// (see its _log helper); forwardWrapperStderr re-emits them through the
// application logger so the backend's output stays one JSON stream. Per
// os/exec's StderrPipe docs, it's incorrect to call Wait before all
// reads from the pipe have completed, so close() waits on stderrDone
// before Wait-ing.
c.stderrDone = make(chan struct{})
stderrDone := c.stderrDone
go func() {
io.Copy(os.Stderr, stderr)
forwardWrapperStderr(stderr)
close(stderrDone)
}()
@@ -217,6 +225,7 @@ func (c *subprocessClient) ensureStarted(ctx context.Context) error {
// these wire params.
func (c *subprocessClient) roundTrip(ctx context.Context, cmdName string, params any) (result json.RawMessage, err error) {
start := time.Now()
var errorType, traceback string // Python-side failure detail, set from the error response
defer func() {
attrs := []slog.Attr{
slog.String("cmd", cmdName),
@@ -230,6 +239,12 @@ func (c *subprocessClient) roundTrip(ctx context.Context, cmdName string, params
if err != nil {
level = slog.LevelError
attrs = append(attrs, slog.String("error", err.Error()))
if errorType != "" {
attrs = append(attrs, slog.String("error_type", errorType))
}
if traceback != "" {
attrs = append(attrs, slog.String("traceback", traceback))
}
}
applog.FromContext(ctx).LogAttrs(context.Background(), level, "Garmin wrapper call", attrs...)
}()
@@ -261,6 +276,7 @@ func (c *subprocessClient) roundTrip(ctx context.Context, cmdName string, params
return nil, err
}
if resp.Error != "" {
errorType, traceback = resp.ErrorType, resp.Traceback
if resp.NotFound {
err = fmt.Errorf("%s: %s: %w", cmdName, resp.Error, ErrNotFound)
} else {
@@ -352,6 +368,52 @@ func (c *subprocessClient) close() error {
return err
}
// forwardWrapperStderr re-emits the wrapper subprocess's stderr through
// the application logger. wrapper.py writes JSON lines ({"level","msg",
// ...attrs} -- see its _log helper), which are decoded and re-logged at
// the corresponding level with their attrs preserved. Anything that isn't
// such a line (a Python startup crash before _log exists, a chatty
// third-party library printing directly) is wrapped as a warn-level
// "garmin wrapper stderr" record rather than passed through raw, so the
// backend's combined output is JSON no matter what the subprocess does.
func forwardWrapperStderr(r io.Reader) {
scanner := bufio.NewScanner(r)
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) // tracebacks can exceed the default token size
for scanner.Scan() {
line := scanner.Text()
var entry map[string]any
if err := json.Unmarshal([]byte(line), &entry); err != nil || entry["msg"] == nil {
slog.Warn("garmin wrapper stderr", "line", line)
continue
}
msg, _ := entry["msg"].(string)
levelName, _ := entry["level"].(string)
delete(entry, "msg")
delete(entry, "level")
attrs := make([]slog.Attr, 0, len(entry)+1)
attrs = append(attrs, slog.String("source", "garmin-wrapper"))
for k, v := range entry {
attrs = append(attrs, slog.Any(k, v))
}
slog.LogAttrs(context.Background(), wrapperLogLevel(levelName), msg, attrs...)
}
}
// wrapperLogLevel maps wrapper.py's level strings onto slog levels,
// defaulting to info for anything unrecognized.
func wrapperLogLevel(level string) slog.Level {
switch level {
case "debug":
return slog.LevelDebug
case "warn", "warning":
return slog.LevelWarn
case "error":
return slog.LevelError
default:
return slog.LevelInfo
}
}
func truncate(s string, n int) string {
if len(s) <= n {
return s

40
backend/internal/garmin/client_test.go
View File

@@ -486,3 +486,43 @@ func TestSubprocessClient_RoundTrip_LogsFailedCallAtErrorLevel(t *testing.T) {
t.Errorf("error field = %q, want it to mention \"boom\"", errMsg)
}
}
func TestForwardWrapperStderr_ReemitsJSONAndWrapsFreeText(t *testing.T) {
var buf bytes.Buffer
prev := slog.Default()
slog.SetDefault(slog.New(slog.NewJSONHandler(&buf, &slog.HandlerOptions{Level: slog.LevelDebug})))
defer slog.SetDefault(prev)
stderr := strings.NewReader(
`{"level":"warn","msg":"startup tokenstore login failed","error":"boom"}` + "\n" +
"Traceback (most recent call last): free text\n",
)
forwardWrapperStderr(stderr)
lines := strings.Split(strings.TrimSpace(buf.String()), "\n")
if len(lines) != 2 {
t.Fatalf("expected 2 log lines, got %d: %q", len(lines), buf.String())
}
var first map[string]any
if err := json.Unmarshal([]byte(lines[0]), &first); err != nil {
t.Fatalf("first line not JSON: %v", err)
}
if first["level"] != "WARN" || first["msg"] != "startup tokenstore login failed" {
t.Errorf("first = %v, want the wrapper's level and msg preserved", first)
}
if first["error"] != "boom" || first["source"] != "garmin-wrapper" {
t.Errorf("first = %v, want error attr preserved and source=garmin-wrapper", first)
}
var second map[string]any
if err := json.Unmarshal([]byte(lines[1]), &second); err != nil {
t.Fatalf("second line not JSON: %v", err)
}
if second["msg"] != "garmin wrapper stderr" || second["level"] != "WARN" {
t.Errorf("second = %v, want free text wrapped as a warn record", second)
}
if line, _ := second["line"].(string); !strings.Contains(line, "Traceback") {
t.Errorf("second line attr = %q, want the raw text preserved", line)
}
}

17
backend/internal/garmin/wrapper/tests/test_wrapper.py
View File

@@ -129,7 +129,13 @@ def test_call_propagates_garminconnect_exception_as_error():
resp = wrapper.dispatch({
"id": 9, "cmd": "call", "params": {"method": "get_activity_splits", "args": {"activity_id": "999"}},
})
assert resp == {"id": 9, "error": "not found"}
assert resp["id"] == 9
assert resp["error"] == "not found"
# Failure detail travels in the protocol so the Go side can log one
# complete structured record (no stderr correlation needed).
assert resp["error_type"] == "Exception"
assert "Exception: not found" in resp["traceback"]
assert "not_found" not in resp
def test_dispatch_unknown_cmd_is_error():
@@ -280,7 +286,10 @@ def test_call_marks_not_found_error_specifically():
resp = wrapper.dispatch({
"id": 11, "cmd": "call", "params": {"method": "get_workout_by_id", "args": {"workout_id": "999"}},
})
assert resp == {"id": 11, "error": "API Error 404", "not_found": True}
assert resp["id"] == 11
assert resp["error"] == "API Error 404"
assert resp["not_found"] is True
assert resp["error_type"] == "GarminConnectNotFoundError"
def test_call_does_not_mark_other_errors_as_not_found():
@@ -290,4 +299,6 @@ def test_call_does_not_mark_other_errors_as_not_found():
resp = wrapper.dispatch({
"id": 12, "cmd": "call", "params": {"method": "get_workout_by_id", "args": {"workout_id": "999"}},
})
assert resp == {"id": 12, "error": "rate limited"}
assert resp["id"] == 12
assert resp["error"] == "rate limited"
assert "not_found" not in resp

96
backend/internal/garmin/wrapper/wrapper.py
View File

@@ -24,14 +24,21 @@ _login_result_queue = queue.Queue()
_startup_login_attempted = False
def _debug(msg):
print(f"[garmin-wrapper debug] {msg}", file=sys.stderr, flush=True)
def _log(level, msg, **attrs):
"""Emit one JSON log line on stderr. internal/garmin/client.go reads
stderr line by line and re-emits these through the Go application
logger, so the backend's combined output stays a single JSON stream --
never print free text to stderr or stdout from this process (stdout is
reserved for the request/response protocol)."""
entry = {"level": level, "msg": msg}
entry.update(attrs)
print(json.dumps(entry), file=sys.stderr, flush=True)
def _prompt_mfa():
_debug("garminconnect invoked prompt_mfa() -- this is a REAL MFA challenge from Garmin")
_log("info", "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")
_log("debug", "prompt_mfa() handing code back to garminconnect", code_length=len(code))
return code
@@ -91,12 +98,13 @@ def _startup_login():
if status == "success":
_auth_state = "authenticated"
else:
_debug(f"startup tokenstore login failed: {err}")
_log("warn", "startup tokenstore login failed", error=err)
_auth_state = "unauthenticated"
except queue.Empty:
_debug(
_log(
"warn",
"startup tokenstore login hit the 10s timeout -- still running in the "
"background and will update auth state whenever it finishes"
"background and will update auth state whenever it finishes",
)
_auth_state = "unauthenticated"
@@ -118,14 +126,19 @@ def _handle_authenticate(_params):
}
def _do_login():
_debug(f"background login thread starting _client.login(tokenstore={TOKENSTORE})")
_log("debug", "background login thread starting _client.login()", tokenstore=TOKENSTORE)
try:
_client.login(tokenstore=TOKENSTORE)
_debug("_client.login() returned successfully")
_log("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())
_log(
"error",
"_client.login() raised",
error=str(exc),
error_type=type(exc).__name__,
traceback=traceback.format_exc(),
)
_login_result_queue.put(("error", str(exc)))
_client = Garmin(email, password)
@@ -134,17 +147,18 @@ def _handle_authenticate(_params):
try:
status, err = _login_result_queue.get(timeout=10)
_debug(f"authenticate got result within 10s timeout: status={status}")
_log("debug", "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(
_log(
"warn",
"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."
"whether the 'REAL MFA challenge' log line above appears to tell real MFA "
"apart from a merely slow login.",
)
_auth_state = "mfa_pending"
return {
@@ -160,12 +174,12 @@ def _handle_complete_mfa(params):
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")
_log("debug", "complete_mfa received a code, pushing to mfa queue", code_length=len(code))
_mfa_input_queue.put(code)
try:
status, err = _login_result_queue.get(timeout=30)
_debug(f"complete_mfa got result: status={status} err={err}")
_log("debug", "complete_mfa got result", status=status, error=err)
if status == "success":
_auth_state = "authenticated"
return {"status": "success", "message": "MFA accepted. Authenticated successfully."}
@@ -206,9 +220,15 @@ def dispatch(req):
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())
resp = {"id": req.get("id"), "error": str(exc)}
# The full failure detail travels IN the response -- error text,
# exception class, and traceback -- so the Go side can log one
# complete structured record instead of correlating stderr noise.
resp = {
"id": req.get("id"),
"error": str(exc),
"error_type": type(exc).__name__,
"traceback": traceback.format_exc(),
}
# A 404 (e.g. get_workout_by_id for a workout deleted on Garmin's
# side after being linked to an activity) is definitive, not a
# transient failure worth retrying forever -- marked specifically so
@@ -224,10 +244,40 @@ def main():
line = line.strip()
if not line:
continue
req = json.loads(line)
try:
req = json.loads(line)
except json.JSONDecodeError as exc:
# A malformed request has no id to answer to -- log and keep
# serving rather than crashing the subprocess.
_log("error", "malformed request line", error=str(exc))
continue
resp = dispatch(req)
print(json.dumps(resp), flush=True)
try:
print(json.dumps(resp), flush=True)
except (TypeError, ValueError) as exc:
# A non-JSON-serializable handler result must still produce a
# protocol response, or the Go side would block on a reply.
print(
json.dumps(
{
"id": req.get("id"),
"error": f"unserializable result: {exc}",
"error_type": type(exc).__name__,
}
),
flush=True,
)
if __name__ == "__main__":
main()
try:
main()
except Exception as exc: # last resort: die loudly, but still in JSON
_log(
"error",
"wrapper crashed",
error=str(exc),
error_type=type(exc).__name__,
traceback=traceback.format_exc(),
)
raise SystemExit(1)