diff --git a/backend/internal/garmin/client.go b/backend/internal/garmin/client.go index 66d62b4..b4667cb 100644 --- a/backend/internal/garmin/client.go +++ b/backend/internal/garmin/client.go @@ -1,23 +1,32 @@ -// Package garmin wraps the mcp-garmin MCP server as a narrow Go client -// interface, so the rest of geniusrun never deals with MCP/JSON-RPC directly. +// Package garmin wraps a direct garminconnect subprocess (see +// pyscript/wrapper.py) as a narrow Go client interface, so the rest of +// geniusrun never deals with the wire protocol directly. package garmin import ( + "bufio" "context" + _ "embed" "encoding/json" "fmt" - "strconv" - "strings" + "io" + "os" + "os/exec" "sync" - - mcpclient "github.com/mark3labs/mcp-go/client" - "github.com/mark3labs/mcp-go/client/transport" - "github.com/mark3labs/mcp-go/mcp" ) +//go:embed pyscript/wrapper.py +var wrapperScript string + +// maxWrapperLineBytes bounds one JSON-line response from the wrapper +// subprocess -- well above the default 64KB bufio.Scanner limit, since +// get_activity_details responses (per-second telemetry) can be several MB. +const maxWrapperLineBytes = 16 * 1024 * 1024 + // Client is the interface the rest of geniusrun depends on. The real -// implementation drives mcp-garmin over stdio; internal/garmin/mock provides -// a fake for tests and frontend-only development. +// implementation drives an embedded Python wrapper subprocess over stdio; +// internal/garmin/mock provides a fake for tests and frontend-only +// development. type Client interface { // Authenticate triggers Garmin login using credentials the subprocess // was started with. Spawns the subprocess on first call. @@ -42,280 +51,207 @@ type Client interface { Close() error } -// Config configures how the mcp-garmin subprocess is spawned. +// Config configures how the Garmin wrapper subprocess is spawned. type Config struct { - PythonPath string // path to mcp-garmin's venv python executable - ServerPath string // path to mcp-garmin's server.py + // PythonPath is the python3 interpreter to run the embedded wrapper + // script with. Empty defaults to "python3" resolved via PATH. + PythonPath string GarminEmail string GarminPassword string - // TokenStorePath, if set, is passed as GARMIN_TOKENSTORE so mcp-garmin + // TokenStorePath, if set, is passed as GARMIN_TOKENSTORE so the wrapper // persists/resumes a Garmin session there instead of the default ~/.garth. TokenStorePath string } -// mcpClient is the real Client implementation, backed by an mcp-garmin -// subprocess spoken to over stdio MCP. -type mcpClient struct { +// wireRequest is one line sent to the wrapper subprocess's stdin. +type wireRequest struct { + ID int `json:"id"` + Cmd string `json:"cmd"` + Params any `json:"params,omitempty"` +} + +// wireResponse is one line read from the wrapper subprocess's stdout. +type wireResponse struct { + ID int `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` +} + +// callParams is the Params payload for a generic "call" request: dispatches +// to any garminconnect.Garmin method by name. +type callParams struct { + Method string `json:"method"` + Args map[string]any `json:"args,omitempty"` +} + +// authResultWire is the Result payload for authenticate/complete_mfa. +type authResultWire struct { + Status string `json:"status"` + Message string `json:"message"` +} + +func mapAuthStatus(s string) AuthStatus { + switch s { + case "success": + return AuthSuccess + case "mfa_required": + return AuthMFARequired + case "failed": + return AuthFailed + default: + return AuthUnknown + } +} + +// subprocessClient is the real Client implementation, backed by a wrapper +// subprocess spoken to over newline-delimited JSON on stdio. +type subprocessClient struct { cfg Config - mu sync.Mutex // serializes calls; mcp-garmin holds a single shared Garmin client - inner *mcpclient.Client - started bool + mu sync.Mutex // serializes calls; the wrapper holds a single shared Garmin client + cmd *exec.Cmd + stdin io.WriteCloser + enc *json.Encoder + scanner *bufio.Scanner + scriptPath string // temp file holding the embedded wrapper.py, written once + started bool + nextID int } -// NewClient builds a Client. The subprocess is not spawned until the first -// call that needs it (Authenticate, or any data call once authenticated). -func NewClient(cfg Config) Client { - return &mcpClient{cfg: cfg} -} - -func (c *mcpClient) ensureStarted(ctx context.Context) error { +// ensureStarted spawns the wrapper subprocess if it isn't already running. +// Callers must hold c.mu. +func (c *subprocessClient) ensureStarted() error { if c.started { return nil } - env := []string{ + if c.scriptPath == "" { + f, err := os.CreateTemp("", "geniusrun-garmin-wrapper-*.py") + if err != nil { + return fmt.Errorf("write embedded wrapper script: %w", err) + } + if _, err := f.WriteString(wrapperScript); err != nil { + f.Close() + return fmt.Errorf("write embedded wrapper script: %w", err) + } + if err := f.Close(); err != nil { + return fmt.Errorf("write embedded wrapper script: %w", err) + } + c.scriptPath = f.Name() + } + + pythonPath := c.cfg.PythonPath + if pythonPath == "" { + pythonPath = "python3" + } + + cmd := exec.Command(pythonPath, c.scriptPath) + extraEnv := []string{ "GARMIN_EMAIL=" + c.cfg.GarminEmail, "GARMIN_PASSWORD=" + c.cfg.GarminPassword, "PYTHONUNBUFFERED=1", } if c.cfg.TokenStorePath != "" { - env = append(env, "GARMIN_TOKENSTORE="+c.cfg.TokenStorePath) + extraEnv = append(extraEnv, "GARMIN_TOKENSTORE="+c.cfg.TokenStorePath) } + cmd.Env = append(os.Environ(), extraEnv...) - inner, err := mcpclient.NewStdioMCPClient(c.cfg.PythonPath, env, c.cfg.ServerPath) + stdin, err := cmd.StdinPipe() if err != nil { - return fmt.Errorf("spawn mcp-garmin subprocess: %w", err) + return fmt.Errorf("open wrapper stdin: %w", err) + } + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("open wrapper stdout: %w", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return fmt.Errorf("open wrapper stderr: %w", err) } - if stdio, ok := inner.GetTransport().(*transport.Stdio); ok { - go drainStderr(stdio) + if err := cmd.Start(); err != nil { + return fmt.Errorf("spawn garmin wrapper subprocess: %w", err) } + go io.Copy(os.Stderr, stderr) // wrapper.py logs auth/rate-limit diagnostics to stderr - initReq := mcp.InitializeRequest{} - initReq.Params.ProtocolVersion = mcp.LATEST_PROTOCOL_VERSION - initReq.Params.ClientInfo = mcp.Implementation{Name: "geniusrund", Version: "0.0.1"} - if _, err := inner.Initialize(ctx, initReq); err != nil { - inner.Close() - return fmt.Errorf("mcp initialize handshake: %w", err) - } - - c.inner = inner + c.cmd = cmd + c.stdin = stdin + c.enc = json.NewEncoder(stdin) + scanner := bufio.NewScanner(stdout) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + c.scanner = scanner c.started = true + c.nextID = 0 return nil } +// roundTrip sends one request and returns its result payload, or an error +// if the wrapper reported one. Callers must hold c.mu and have already +// called ensureStarted. +func (c *subprocessClient) roundTrip(cmdName string, params any) (json.RawMessage, error) { + c.nextID++ + id := c.nextID + + if err := c.enc.Encode(wireRequest{ID: id, Cmd: cmdName, Params: params}); err != nil { + return nil, fmt.Errorf("write %s request: %w", cmdName, err) + } + + if !c.scanner.Scan() { + if err := c.scanner.Err(); err != nil { + return nil, fmt.Errorf("read %s response: %w", cmdName, err) + } + return nil, fmt.Errorf("read %s response: subprocess closed its output", cmdName) + } + + var resp wireResponse + if err := json.Unmarshal(c.scanner.Bytes(), &resp); err != nil { + return nil, fmt.Errorf("parse %s response: %w", cmdName, err) + } + if resp.ID != id { + return nil, fmt.Errorf("%s response id mismatch: got %d, want %d", cmdName, resp.ID, id) + } + if resp.Error != "" { + return nil, fmt.Errorf("%s: %s", cmdName, resp.Error) + } + return resp.Result, nil +} + // UpdateCredentials implements Client. -func (c *mcpClient) UpdateCredentials(email, password string) { +func (c *subprocessClient) UpdateCredentials(email, password string) { c.mu.Lock() defer c.mu.Unlock() c.cfg.GarminEmail = email c.cfg.GarminPassword = password - - if c.started { - if c.inner != nil { - c.inner.Close() - } - c.inner = nil - c.started = false - } + c.close() } -// drainStderr forwards the subprocess's debug/log output so it isn't -// silently dropped (mcp-garmin logs auth/rate-limit diagnostics there). -func drainStderr(stdio *transport.Stdio) { - buf := make([]byte, 4096) - for { - n, err := stdio.Stderr().Read(buf) - if n > 0 { - fmt.Print(string(buf[:n])) - } - if err != nil { - return - } - } -} - -func (c *mcpClient) callTool(ctx context.Context, name string, args map[string]any) (string, error) { - req := mcp.CallToolRequest{} - req.Params.Name = name - req.Params.Arguments = args - - res, err := c.inner.CallTool(ctx, req) - if err != nil { - return "", fmt.Errorf("call tool %s: %w", name, err) - } - var out strings.Builder - for _, content := range res.Content { - if tc, ok := content.(mcp.TextContent); ok { - out.WriteString(tc.Text) - } - } - if res.IsError { - return "", fmt.Errorf("tool %s returned an error result: %s", name, out.String()) - } - return out.String(), nil -} - -func (c *mcpClient) Authenticate(ctx context.Context) (AuthResult, error) { +// Close implements Client. +func (c *subprocessClient) Close() error { c.mu.Lock() defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return AuthResult{}, err - } - msg, err := c.callTool(ctx, "authenticate", nil) - if err != nil { - return AuthResult{}, err - } - return parseAuthResult(msg), nil + return c.close() } -func (c *mcpClient) CompleteMFA(ctx context.Context, code string) (AuthResult, error) { - c.mu.Lock() - defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return AuthResult{}, err - } - msg, err := c.callTool(ctx, "complete_mfa", map[string]any{"code": code}) - if err != nil { - return AuthResult{}, err - } - return parseAuthResult(msg), nil -} - -func parseAuthResult(msg string) AuthResult { - switch { - case strings.Contains(msg, "Authenticated successfully") || strings.Contains(msg, "MFA accepted"): - return AuthResult{Status: AuthSuccess, Message: msg} - case strings.Contains(msg, "MFA required"): - return AuthResult{Status: AuthMFARequired, Message: msg} - default: - return AuthResult{Status: AuthFailed, Message: msg} - } -} - -func (c *mcpClient) GetActivities(ctx context.Context, startDate, endDate string, limit int) ([]Activity, error) { - c.mu.Lock() - defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return nil, err - } - msg, err := c.callTool(ctx, "get_activities", map[string]any{ - "start_date": startDate, - "end_date": endDate, - "limit": limit, - }) - if err != nil { - return nil, err - } - - var rawActivities []json.RawMessage - if err := json.Unmarshal([]byte(msg), &rawActivities); err != nil { - return nil, fmt.Errorf("get_activities did not return a JSON array (%q): %w", truncate(msg, 200), err) - } - - activities := make([]Activity, 0, len(rawActivities)) - for _, raw := range rawActivities { - var a Activity - if err := json.Unmarshal(raw, &a); err != nil { - return nil, fmt.Errorf("parse activity: %w", err) - } - a.Raw = raw - activities = append(activities, a) - } - return activities, nil -} - -func (c *mcpClient) GetActivitySplits(ctx context.Context, activityID int64) (ActivitySplits, error) { - c.mu.Lock() - defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return ActivitySplits{}, err - } - msg, err := c.callTool(ctx, "get_activity_splits", map[string]any{ - "activity_id": strconv.FormatInt(activityID, 10), - }) - if err != nil { - return ActivitySplits{}, err - } - - var envelope struct { - ActivityID int64 `json:"activityId"` - Laps []json.RawMessage `json:"lapDTOs"` - } - if err := json.Unmarshal([]byte(msg), &envelope); err != nil { - return ActivitySplits{}, fmt.Errorf("get_activity_splits did not return valid JSON (%q): %w", truncate(msg, 200), err) - } - - splits := ActivitySplits{ActivityID: envelope.ActivityID, Laps: make([]Lap, 0, len(envelope.Laps))} - for _, raw := range envelope.Laps { - var l Lap - if err := json.Unmarshal(raw, &l); err != nil { - return ActivitySplits{}, fmt.Errorf("parse lap: %w", err) - } - l.Raw = raw - splits.Laps = append(splits.Laps, l) - } - return splits, nil -} - -func (c *mcpClient) GetActivityDetails(ctx context.Context, activityID int64) (ActivityDetails, error) { - c.mu.Lock() - defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return ActivityDetails{}, err - } - msg, err := c.callTool(ctx, "get_activity_details", map[string]any{ - "activity_id": strconv.FormatInt(activityID, 10), - }) - if err != nil { - return ActivityDetails{}, err - } - - var details ActivityDetails - if err := json.Unmarshal([]byte(msg), &details); err != nil { - return ActivityDetails{}, fmt.Errorf("get_activity_details did not return valid JSON (%q): %w", truncate(msg, 200), err) - } - details.Raw = json.RawMessage(msg) - return details, nil -} - -func (c *mcpClient) GetWorkoutByID(ctx context.Context, workoutID int64) (Workout, error) { - c.mu.Lock() - defer c.mu.Unlock() - - if err := c.ensureStarted(ctx); err != nil { - return Workout{}, err - } - msg, err := c.callTool(ctx, "get_workout_by_id", map[string]any{ - "workout_id": strconv.FormatInt(workoutID, 10), - }) - if err != nil { - return Workout{}, err - } - - var workout Workout - if err := json.Unmarshal([]byte(msg), &workout); err != nil { - return Workout{}, fmt.Errorf("get_workout_by_id did not return valid JSON (%q): %w", truncate(msg, 200), err) - } - workout.Raw = json.RawMessage(msg) - return workout, nil -} - -func (c *mcpClient) Close() error { - c.mu.Lock() - defer c.mu.Unlock() - +// close terminates the subprocess, if running. Callers must hold c.mu. +func (c *subprocessClient) close() error { if !c.started { return nil } - return c.inner.Close() + c.started = false + + if c.stdin != nil { + c.stdin.Close() + } + if c.cmd != nil && c.cmd.Process != nil { + c.cmd.Process.Kill() + } + var err error + if c.cmd != nil { + err = c.cmd.Wait() + } + c.cmd, c.stdin, c.enc, c.scanner = nil, nil, nil, nil + return err } func truncate(s string, n int) string { diff --git a/backend/internal/garmin/client_test.go b/backend/internal/garmin/client_test.go index 368d6e6..3d4286a 100644 --- a/backend/internal/garmin/client_test.go +++ b/backend/internal/garmin/client_test.go @@ -1,29 +1,135 @@ package garmin -import "testing" +import ( + "bufio" + "encoding/json" + "io" + "strings" + "testing" +) -func TestParseAuthResult(t *testing.T) { - cases := []struct { - msg string - want AuthStatus - }{ - {"Authenticated successfully.", AuthSuccess}, - {"MFA accepted. Authenticated successfully.", AuthSuccess}, - {"MFA required. Garmin has sent a verification code...", AuthMFARequired}, - {"Authentication failed: 401 Unauthorized (Invalid Username or Password)", AuthFailed}, - {"Authentication failed after MFA: bad code", AuthFailed}, +// wireResponsePayload is what a fake wrapper handler returns for one +// request; the harness fills in the response ID. +type wireResponsePayload struct { + result json.RawMessage + err string +} + +func fakeResult(v any) wireResponsePayload { + b, err := json.Marshal(v) + if err != nil { + panic(err) } - for _, c := range cases { - got := parseAuthResult(c.msg) - if got.Status != c.want { - t.Errorf("parseAuthResult(%q).Status = %v, want %v", c.msg, got.Status, c.want) + return wireResponsePayload{result: b} +} + +func fakeError(msg string) wireResponsePayload { + return wireResponsePayload{err: msg} +} + +// newFakeWrapperClient wires a subprocessClient to an in-process goroutine +// that plays the Python wrapper's role, so protocol-level Go logic can be +// tested without python3/garminconnect installed. +func newFakeWrapperClient(t *testing.T, handle func(cmd string, params json.RawMessage) wireResponsePayload) *subprocessClient { + t.Helper() + + reqR, reqW := io.Pipe() + respR, respW := io.Pipe() + t.Cleanup(func() { + reqW.Close() + respW.Close() + }) + + go func() { + scanner := bufio.NewScanner(reqR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + enc := json.NewEncoder(respW) + for scanner.Scan() { + var req struct { + ID int `json:"id"` + Cmd string `json:"cmd"` + Params json.RawMessage `json:"params"` + } + if err := json.Unmarshal(scanner.Bytes(), &req); err != nil { + continue + } + payload := handle(req.Cmd, req.Params) + resp := wireResponse{ID: req.ID, Result: payload.result, Error: payload.err} + if err := enc.Encode(resp); err != nil { + return + } } + }() + + scanner := bufio.NewScanner(respR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + + return &subprocessClient{ + started: true, + enc: json.NewEncoder(reqW), + scanner: scanner, } } -func TestMcpClient_UpdateCredentials_ResetsStartedState(t *testing.T) { - c := &mcpClient{cfg: Config{GarminEmail: "old@example.com", GarminPassword: "old"}} - c.started = true // simulate an already-spawned subprocess +func TestSubprocessClient_RoundTrip_DetectsIDMismatch(t *testing.T) { + reqR, reqW := io.Pipe() + respR, respW := io.Pipe() + defer reqW.Close() + defer respW.Close() + + go func() { + scanner := bufio.NewScanner(reqR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + scanner.Scan() // read and discard the one request + json.NewEncoder(respW).Encode(wireResponse{ID: 999, Result: json.RawMessage(`{}`)}) + }() + + scanner := bufio.NewScanner(respR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + c := &subprocessClient{started: true, enc: json.NewEncoder(reqW), scanner: scanner} + + _, err := c.roundTrip("authenticate", nil) + if err == nil || !strings.Contains(err.Error(), "id mismatch") { + t.Fatalf("roundTrip error = %v, want an id mismatch error", err) + } +} + +func TestSubprocessClient_RoundTrip_SubprocessClosedIsError(t *testing.T) { + reqR, reqW := io.Pipe() + respR, respW := io.Pipe() + defer reqW.Close() + + go func() { + scanner := bufio.NewScanner(reqR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + scanner.Scan() + respW.Close() // subprocess "exited": stdout closes with no response + }() + + scanner := bufio.NewScanner(respR) + scanner.Buffer(make([]byte, 0, 64*1024), maxWrapperLineBytes) + c := &subprocessClient{started: true, enc: json.NewEncoder(reqW), scanner: scanner} + + _, err := c.roundTrip("authenticate", nil) + if err == nil || !strings.Contains(err.Error(), "closed") { + t.Fatalf("roundTrip error = %v, want a subprocess-closed error", err) + } +} + +func TestSubprocessClient_RoundTrip_WrapperErrorPropagates(t *testing.T) { + c := newFakeWrapperClient(t, func(cmd string, params json.RawMessage) wireResponsePayload { + return fakeError("boom") + }) + + _, err := c.roundTrip("authenticate", nil) + if err == nil || !strings.Contains(err.Error(), "boom") { + t.Fatalf("roundTrip error = %v, want it to mention %q", err, "boom") + } +} + +func TestSubprocessClient_UpdateCredentials_ResetsStartedState(t *testing.T) { + c := &subprocessClient{cfg: Config{GarminEmail: "old@example.com", GarminPassword: "old"}} + c.started = true // simulate an already-spawned subprocess, no real cmd/pipes c.UpdateCredentials("new@example.com", "new") @@ -33,7 +139,11 @@ func TestMcpClient_UpdateCredentials_ResetsStartedState(t *testing.T) { if c.started { t.Error("started should be reset to false so the next call respawns the subprocess") } - if c.inner != nil { - t.Error("inner should be cleared so ensureStarted spawns a fresh client") +} + +func TestSubprocessClient_Close_NoOpWhenNotStarted(t *testing.T) { + c := &subprocessClient{} + if err := c.Close(); err != nil { + t.Errorf("Close on an unstarted client = %v, want nil", err) } }