Merge pull request #110 from smart-mcp-proxy/fix/upstream-stability

fix(upstream): improve SSE stability and OAuth timeout for JetBrains MCP

Author: Dumbris

.github/workflows/unit-tests.yml

@@ -171,7 +171,41 @@ jobs:
         with:
           node-version: '20'
 
-      - name: Install frontend dependencies
+      - name: Install frontend dependencies (Windows)
+        if: matrix.os == 'windows-latest'
+        shell: pwsh
+        run: |
+          cd frontend
+          $maxRetries = 3
+          $retryCount = 0
+
+          while ($retryCount -lt $maxRetries) {
+            $retryCount++
+            Write-Host "Attempting npm ci (attempt $retryCount/$maxRetries)..."
+
+            npm ci --prefer-offline --no-audit 2>&1 | Out-Host
+
+            if ($LASTEXITCODE -eq 0) {
+              Write-Host "npm ci succeeded"
+              break
+            }
+
+            if ($retryCount -lt $maxRetries) {
+              Write-Host "npm ci failed with exit code $LASTEXITCODE, waiting 5 seconds before retry..."
+              Start-Sleep -Seconds 5
+
+              if (Test-Path node_modules) {
+                Write-Host "Cleaning up node_modules..."
+                Remove-Item -Recurse -Force node_modules -ErrorAction SilentlyContinue
+              }
+            } else {
+              Write-Host "npm ci failed after $maxRetries attempts"
+              exit 1
+            }
+          }
+
+      - name: Install frontend dependencies (Unix)
+        if: matrix.os != 'windows-latest'
         run: cd frontend && npm ci
 
       - name: Build frontend

cmd/mcpproxy-tray/main.go

@@ -42,6 +42,7 @@ var (
 	defaultCoreURL   = "http://127.0.0.1:8080"
 	errNoBundledCore = errors.New("no bundled core binary found")
 	trayAPIKey       = "" // API key generated for core communication
+	shutdownComplete = make(chan struct{}) // Signal when shutdown is complete
 )
 
 // getLogDir returns the standard log directory for the current OS.
@@ -152,21 +153,32 @@ func main() {
 
 	// Create tray application early so icon appears
 	shutdownFunc := func() {
+		defer close(shutdownComplete) // Signal when shutdown is done
+
 		logger.Info("Tray shutdown requested")
-		// IMPORTANT: Send shutdown event to trigger state transition
+
+		// Send shutdown event to state machine
 		stateMachine.SendEvent(state.EventShutdown)
-		// IMPORTANT: Call handleShutdown() directly to terminate core process
-		// We can't rely on the state transition goroutine because cancel() will kill it
+
+		// Shutdown launcher (stops SSE, health monitor, kills core)
 		if launcher != nil {
 			launcher.handleShutdown()
 		}
-		// IMPORTANT: Quit the tray UI BEFORE cancelling context
-		// This prevents reconnection attempts from SSE goroutine
-		logger.Info("Quitting system tray")
-		trayApp.Quit()
-		// Now shutdown state machine and cancel context
+
+		// Shutdown state machine
 		stateMachine.Shutdown()
+
+		// CRITICAL: Cancel context LAST, after all cleanup
+		// This prevents the tray.Run() goroutine from quitting prematurely
+		logger.Info("Cancelling context after cleanup complete")
 		cancel()
+
+		// Give tray.Run() goroutine a moment to see cancellation
+		time.Sleep(100 * time.Millisecond)
+
+		// Finally, quit the tray UI
+		logger.Info("Quitting system tray")
+		trayApp.Quit()
 	}
 
 	trayApp = tray.NewWithAPIClient(api.NewServerAdapter(apiClient), apiClient, logger.Sugar(), version, shutdownFunc)
@@ -214,25 +226,25 @@ func main() {
 	go func() {
 		<-sigCh
 		logger.Info("Received shutdown signal")
-		stateMachine.SendEvent(state.EventShutdown)
-		// IMPORTANT: Call handleShutdown() directly to terminate core process
-		// Same as shutdownFunc - we can't rely on state transition goroutine
-		if launcher != nil {
-			launcher.handleShutdown()
-		}
-		// IMPORTANT: Quit the tray UI BEFORE cancelling context
-		logger.Info("Quitting system tray from signal handler")
-		trayApp.Quit()
-		stateMachine.Shutdown()
-		cancel()
+
+		// Use the same shutdown flow as Quit menu item
+		shutdownFunc()
 	}()
 
 	logger.Info("Starting tray event loop")
 	if err := trayApp.Run(ctx); err != nil && err != context.Canceled {
 		logger.Error("Tray application error", zap.Error(err))
 	}
 
-	// Wait for state machine to shut down gracefully
+	// Wait for shutdown to complete (with timeout)
+	select {
+	case <-shutdownComplete:
+		logger.Info("Shutdown completed successfully")
+	case <-time.After(5 * time.Second):
+		logger.Warn("Shutdown timeout - forcing exit")
+	}
+
+	// Final cleanup
 	stateMachine.Shutdown()
 
 	logger.Info("mcpproxy-tray shutdown complete")
@@ -331,33 +343,6 @@ func resolveCoreURL() string {
 	return defaultCoreURL
 }
 
-// isSocketAvailable checks if a socket/pipe endpoint is accessible
-func isSocketAvailable(endpoint string) bool {
-	// Parse endpoint to extract scheme
-	parsed, err := url.Parse(endpoint)
-	if err != nil {
-		return false
-	}
-
-	switch parsed.Scheme {
-	case "unix":
-		// Check if Unix socket file exists
-		socketPath := parsed.Path
-		if socketPath == "" {
-			socketPath = parsed.Opaque
-		}
-		_, err := os.Stat(socketPath)
-		return err == nil
-	case "npipe":
-		// For Windows named pipes, we can't easily check existence
-		// Return true and let the connection attempt fail if needed
-		return true
-	default:
-		// Not a socket endpoint
-		return false
-	}
-}
-
 func shouldSkipCoreLaunch() bool {
 	value := strings.TrimSpace(os.Getenv("MCPPROXY_TRAY_SKIP_CORE"))
 	return value == "1" || strings.EqualFold(value, "true")
@@ -1361,15 +1346,25 @@ func (cpl *CoreProcessLauncher) handleGeneralError() {
 func (cpl *CoreProcessLauncher) handleShutdown() {
 	cpl.logger.Info("Core process launcher shutting down")
 
-	if cpl.processMonitor != nil {
-		cpl.processMonitor.Shutdown()
-	}
+	// CRITICAL: Stop SSE FIRST before killing core
+	// This prevents SSE from detecting disconnection and trying to reconnect
+	cpl.logger.Info("Stopping SSE connection")
+	cpl.apiClient.StopSSE()
 
+	// Give SSE goroutine a moment to see cancellation and exit cleanly
+	time.Sleep(100 * time.Millisecond)
+
+	// Stop health monitor before killing core
 	if cpl.healthMonitor != nil {
+		cpl.logger.Info("Stopping health monitor")
 		cpl.healthMonitor.Stop()
 	}
 
-	cpl.apiClient.StopSSE()
+	// Finally, kill the core process
+	if cpl.processMonitor != nil {
+		cpl.logger.Info("Shutting down core process")
+		cpl.processMonitor.Shutdown()
+	}
 }
 
 // buildCoreEnvironment builds the environment for the core process

cmd/mcpproxy/tools_cmd.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"mcpproxy-go/internal/config"
+	"mcpproxy-go/internal/transport"
 	"mcpproxy-go/internal/upstream/cli"
 
 	"github.com/spf13/cobra"
@@ -29,16 +30,18 @@ This command is primarily used for debugging upstream server connections and too
 Examples:
   mcpproxy tools list --server=github-server --log-level=trace
   mcpproxy tools list --server=weather-api --log-level=debug
-  mcpproxy tools list --server=local-script --log-level=info`,
+  mcpproxy tools list --server=local-script --log-level=info
+  mcpproxy tools list --server=jetbrains-sse --trace-transport  # Enable HTTP/SSE frame tracing`,
 		RunE: runToolsList,
 	}
 
 	// Command flags
-	serverName    string
-	toolsLogLevel string
-	configPath    string
-	timeout       time.Duration
-	outputFormat  string
+	serverName     string
+	toolsLogLevel  string
+	configPath     string
+	timeout        time.Duration
+	outputFormat   string
+	traceTransport bool // Enable HTTP/SSE frame-by-frame tracing
 )
 
 // GetToolsCommand returns the tools command for adding to the root command
@@ -56,6 +59,7 @@ func init() {
 	toolsListCmd.Flags().StringVarP(&configPath, "config", "c", "", "Path to MCP configuration file (default: ~/.mcpproxy/mcp_config.json)")
 	toolsListCmd.Flags().DurationVarP(&timeout, "timeout", "t", 30*time.Second, "Connection timeout")
 	toolsListCmd.Flags().StringVarP(&outputFormat, "output", "o", "table", "Output format (table, json, yaml)")
+	toolsListCmd.Flags().BoolVar(&traceTransport, "trace-transport", false, "Enable detailed HTTP/SSE frame-by-frame tracing (useful for debugging SSE connection issues)")
 
 	// Mark required flags
 	err := toolsListCmd.MarkFlagRequired("server")
@@ -81,6 +85,14 @@ func runToolsList(_ *cobra.Command, _ []string) error {
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 
+	// Enable transport tracing if requested
+	if traceTransport {
+		transport.GlobalTraceEnabled = true
+		fmt.Println("🔍 HTTP/SSE TRANSPORT TRACING ENABLED")
+		fmt.Println("   All HTTP requests/responses and SSE frames will be logged")
+		fmt.Println()
+	}
+
 	// Load configuration
 	globalConfig, err := loadToolsConfig()
 	if err != nil {

internal/runtime/supervisor/actor_pool.go

@@ -122,7 +122,23 @@ func (p *ActorPoolSimple) RemoveServer(name string) error {
 // ConnectServer tells the manager to connect a server.
 func (p *ActorPoolSimple) ConnectServer(ctx context.Context, name string) error {
 	p.logger.Debug("Connecting server via manager", zap.String("name", name))
-	// Manager handles connection automatically via managed clients
+
+	client, exists := p.manager.GetClient(name)
+	if !exists {
+		return fmt.Errorf("server %s not found", name)
+	}
+
+	// Attempt to connect (managed client will handle state checks)
+	// If client is already connecting/connected, Connect() will return an error
+	// which we log but don't treat as fatal (supervisor will retry)
+	if err := client.Connect(ctx); err != nil {
+		p.logger.Debug("Connect returned error (may be already connecting/connected)",
+			zap.String("server", name),
+			zap.Error(err))
+		// Not returning error - this is expected if client is already connecting
+		// Supervisor's reconciliation logic will handle retries if needed
+	}
+
 	return nil
 }
 

internal/runtime/supervisor/exemption_test.go

@@ -0,0 +1,89 @@
+package supervisor
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+)
+
+// TestCircuitBreaker tests the circuit breaker pattern for inspection failures
+func TestCircuitBreaker(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
+
+	// Test failure recording
+	s := &Supervisor{
+		logger:             logger,
+		inspectionFailures: make(map[string]*inspectionFailureInfo),
+	}
+
+	serverName := "test-server"
+
+	// Record 3 failures
+	for i := 0; i < 3; i++ {
+		s.RecordInspectionFailure(serverName)
+	}
+
+	// Check that circuit breaker is now active
+	allowed, reason, cooldown := s.CanInspect(serverName)
+	assert.False(t, allowed, "Should not allow inspection after 3 failures")
+	assert.Contains(t, reason, "Circuit breaker active")
+	assert.Greater(t, cooldown, time.Duration(0), "Should have cooldown remaining")
+
+	// Verify cooldown duration is approximately 5 minutes
+	assert.InDelta(t, inspectionCooldown.Seconds(), cooldown.Seconds(), 1.0, "Cooldown should be ~5 minutes")
+}
+
+// TestCircuitBreakerReset tests that successful inspection resets the failure counter
+func TestCircuitBreakerReset(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
+
+	s := &Supervisor{
+		logger:             logger,
+		inspectionFailures: make(map[string]*inspectionFailureInfo),
+	}
+
+	serverName := "test-server"
+
+	// Record 2 failures
+	s.RecordInspectionFailure(serverName)
+	s.RecordInspectionFailure(serverName)
+
+	// Verify we're still allowed (under threshold)
+	allowed, _, _ := s.CanInspect(serverName)
+	assert.True(t, allowed, "Should still allow inspection with 2 failures")
+
+	// Record success - should reset counter
+	s.RecordInspectionSuccess(serverName)
+
+	// Verify failure counter was reset
+	failures, inCooldown, _ := s.GetInspectionStats(serverName)
+	assert.Equal(t, 0, failures, "Failure counter should be reset after success")
+	assert.False(t, inCooldown, "Should not be in cooldown after success")
+}
+
+// TestExemptionExpiry tests that exemptions expire correctly
+func TestExemptionExpiry(t *testing.T) {
+	logger, _ := zap.NewDevelopment()
+
+	s := &Supervisor{
+		logger:                logger,
+		inspectionExemptions:  make(map[string]time.Time),
+		inspectionFailures:    make(map[string]*inspectionFailureInfo),
+	}
+
+	serverName := "test-server"
+
+	// Grant exemption with short duration
+	s.inspectionExemptions[serverName] = time.Now().Add(100 * time.Millisecond)
+
+	// Should be exempted immediately
+	assert.True(t, s.IsInspectionExempted(serverName), "Should be exempted immediately")
+
+	// Wait for expiry
+	time.Sleep(150 * time.Millisecond)
+
+	// Should no longer be exempted
+	assert.False(t, s.IsInspectionExempted(serverName), "Should not be exempted after expiry")
+}