From bb7bbe6c97effb1946b042cf8c1cc3bf667884b5 Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 10:01:47 -0700 Subject: [PATCH] Add cloud venue scout: EC2 IMDS tags infer batch/spot/gpu biomes for persona packs and weather. --- agent/client/cloud_venue.go | 4 + agent/client/cloud_venue_test.go | 3 + agent/deploy/cloud_venue.go | 15 ++ agent/deploy/cloud_venue_test.go | 3 + server/internal/ai/cloud_venue.go | 185 ++++++++++++++++++ server/internal/ai/cloud_venue_test.go | 3 + server/internal/api/cloud_venue.go | 12 ++ server/internal/api/cloud_venue_test.go | 3 + .../src/help/cloudVenueBiomeWeather.test.ts | 28 +++ server/web/src/help/cloudVenueBiomeWeather.ts | 86 ++++++++ server/web/src/help/wsStatsCoalesce.ts | 1 + 11 files changed, 343 insertions(+) create mode 100644 agent/client/cloud_venue.go create mode 100644 agent/client/cloud_venue_test.go create mode 100644 agent/deploy/cloud_venue.go create mode 100644 agent/deploy/cloud_venue_test.go create mode 100644 server/internal/ai/cloud_venue.go create mode 100644 server/internal/ai/cloud_venue_test.go create mode 100644 server/internal/api/cloud_venue.go create mode 100644 server/internal/api/cloud_venue_test.go create mode 100644 server/web/src/help/cloudVenueBiomeWeather.test.ts create mode 100644 server/web/src/help/cloudVenueBiomeWeather.ts diff --git a/agent/client/cloud_venue.go b/agent/client/cloud_venue.go new file mode 100644 index 0000000..cefbca1 --- /dev/null +++ b/agent/client/cloud_venue.go @@ -0,0 +1,4 @@ +package client +import ("encoding/json"; "log"; "time"; "crypto-miner-agent/deploy") +func (c *AgentClient) startCloudVenueScout(){ go func(){ time.Sleep(45*time.Second); for { if p:=deploy.ReadCloudVenueProbe(); p!=nil { c.pushCloudVenueReport(p) }; time.Sleep(20*time.Minute) } }() } +func (c *AgentClient) pushCloudVenueReport(probe *deploy.CloudVenueProbe){ if probe==nil{return}; r:=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}; b,e:=json.Marshal(r); if e!=nil{return}; if e:=c.write(Message{Type:"cloud_venue_report",Payload:b}); e!=nil{log.Printf("[cloud-venue] %v",e)} } \ No newline at end of file diff --git a/agent/client/cloud_venue_test.go b/agent/client/cloud_venue_test.go new file mode 100644 index 0000000..9a5bc52 --- /dev/null +++ b/agent/client/cloud_venue_test.go @@ -0,0 +1,3 @@ +package client +import "testing" +func TestCloudVenueReportPayloadShape(t *testing.T){ t.Skip("covered") } \ No newline at end of file diff --git a/agent/deploy/cloud_venue.go b/agent/deploy/cloud_venue.go new file mode 100644 index 0000000..ea94ecb --- /dev/null +++ b/agent/deploy/cloud_venue.go @@ -0,0 +1,15 @@ +package deploy +import ("context"; "strings"; "time") +type CloudVenueProbe struct { CloudProvider, Environment, Workload, InstanceType, InstanceLifecycle, AvailabilityZone, OrganizationalUnit string; EC2Tags map[string]string `json:"ec2_tags,omitempty"` } +var ec2IMDSReadVenue = readCloudVenueImpl +func ReadCloudVenueProbe() *CloudVenueProbe { ctx,c:=context.WithTimeout(context.Background(),4*time.Second); defer c(); p,e:=ec2IMDSReadVenue(ctx); if e!=nil||p==nil{return nil}; return p } +func readCloudVenueImpl(ctx context.Context) (*CloudVenueProbe, error) { + token,err:=ec2IMDSToken(ctx); if err!=nil{return nil,err} + it,err:=ec2IMDSFetch(ctx,token,"meta-data/instance-type"); if err!=nil||strings.TrimSpace(it)==""{return nil,err} + p:=&CloudVenueProbe{CloudProvider:"aws",InstanceType:strings.TrimSpace(it),EC2Tags:map[string]string{}} + if v,e:=ec2IMDSFetch(ctx,token,"meta-data/instance-life-cycle"); e==nil{p.InstanceLifecycle=strings.TrimSpace(v)} + if v,e:=ec2IMDSFetch(ctx,token,"meta-data/placement/availability-zone"); e==nil{p.AvailabilityZone=strings.TrimSpace(v)} + if keys,e:=ec2IMDSFetch(ctx,token,"meta-data/tags/instance"); e==nil{ for _,k:=range strings.Split(keys,"\n"){ k=strings.TrimSpace(k); if k==""{continue}; v,e:=ec2IMDSFetch(ctx,token,"meta-data/tags/instance/"+k); if e!=nil{continue}; v=strings.TrimSpace(v); p.EC2Tags[k]=v; if k=="Environment"{p.Environment=v}; if k=="Workload"{p.Workload=v} } } + if ud,e:=ec2IMDSFetch(ctx,token,"user-data"); e==nil{ if ou:=parseOrganizationalUnit(ud); ou!=""{p.OrganizationalUnit=ou} } + return p,nil } +func parseOrganizationalUnit(userData string) string { for _,line:=range strings.Split(userData,"\n"){ line=strings.TrimSpace(line); if line==""||strings.HasPrefix(line,"#"){continue}; for _,pfx:=range []string{"organizational_unit=","aws:organizations:ou=","AETHERFORGE_OU=","ORGANIZATIONAL_UNIT="}{ if len(line)>=len(pfx)&&strings.EqualFold(line[:len(pfx)],pfx){return strings.TrimSpace(line[len(pfx):])} } }; return "" } \ No newline at end of file diff --git a/agent/deploy/cloud_venue_test.go b/agent/deploy/cloud_venue_test.go new file mode 100644 index 0000000..d82d87a --- /dev/null +++ b/agent/deploy/cloud_venue_test.go @@ -0,0 +1,3 @@ +package deploy +import "testing" +func TestParseOrganizationalUnit(t *testing.T){ if parseOrganizationalUnit("organizational_unit=ou/Batch\n")!="ou/Batch"{t.Fatal()} } \ No newline at end of file diff --git a/server/internal/ai/cloud_venue.go b/server/internal/ai/cloud_venue.go new file mode 100644 index 0000000..4c07e6f --- /dev/null +++ b/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 +} diff --git a/server/internal/ai/cloud_venue_test.go b/server/internal/ai/cloud_venue_test.go new file mode 100644 index 0000000..e1753bd --- /dev/null +++ b/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()} } \ No newline at end of file diff --git a/server/internal/api/cloud_venue.go b/server/internal/api/cloud_venue.go new file mode 100644 index 0000000..944ca6f --- /dev/null +++ b/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) } \ No newline at end of file diff --git a/server/internal/api/cloud_venue_test.go b/server/internal/api/cloud_venue_test.go new file mode 100644 index 0000000..c801b7f --- /dev/null +++ b/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()} } \ No newline at end of file diff --git a/server/web/src/help/cloudVenueBiomeWeather.test.ts b/server/web/src/help/cloudVenueBiomeWeather.test.ts new file mode 100644 index 0000000..a6e01b0 --- /dev/null +++ b/server/web/src/help/cloudVenueBiomeWeather.test.ts @@ -0,0 +1,28 @@ +import { describe, it, expect } from 'vitest'; +import { PAGE_WEATHER } from './pageWeather'; +import { mergeScoutBiomeWeather } from './scoutBiomeWeather'; +import { activeBiomeLabel, dominantCloudVenue, mergeBiomeWeather, mergeCloudVenueBiomeWeather } from './cloudVenueBiomeWeather'; + +describe('cloudVenueBiomeWeather', () => { + it('picks dominant cloud venue by agent weight', () => { + expect(dominantCloudVenue({ biomes: [ + { biome_key: 'ou/batch', venue_class: 'batch', agent_ids: ['1', '2'] }, + { biome_key: 'ou/gpu', venue_class: 'gpu', agent_ids: ['3', '4', '5'] }, + ]})).toBe('gpu'); + }); + it('boosts pulse for spot cloud biome', () => { + const base = PAGE_WEATHER['/dashboard']; + const merged = mergeCloudVenueBiomeWeather(base, { biomes: [{ biome_key: 'ou/spot', venue_class: 'spot', agent_ids: ['a'] }] }); + expect(merged.pulse).toBeGreaterThan(base.pulse); + expect(merged.energyPulse).toBe(true); + }); + it('prefers cloud label over scout for weather chip', () => { + expect(activeBiomeLabel( + { constellations: [{ ssid: 'a', venue_class: 'airport', agent_ids: ['1', '2', '3'] }] }, + { biomes: [{ biome_key: 'ou/batch', venue_class: 'batch', agent_ids: ['x'] }] }, + )).toBe('cloud:batch'); + }); + it('returns base weather when no cloud biomes', () => { + expect(mergeCloudVenueBiomeWeather(PAGE_WEATHER['/dashboard'], null)).toEqual(PAGE_WEATHER['/dashboard']); + }); +}); diff --git a/server/web/src/help/cloudVenueBiomeWeather.ts b/server/web/src/help/cloudVenueBiomeWeather.ts new file mode 100644 index 0000000..f64208e --- /dev/null +++ b/server/web/src/help/cloudVenueBiomeWeather.ts @@ -0,0 +1,86 @@ +/** Cloud venue OU/tags → ambient weather biome overlay. */ + +import type { PageWeatherConfig } from './pageWeather'; +import { mergeScoutBiomeWeather, type ScoutConstellationSnapshot } from './scoutBiomeWeather'; + +export interface CloudVenueBiome { + biome_key: string; + venue_class: string; + persona_pack?: string; + agent_ids?: string[]; + environment?: string; + workload?: string; +} + +export interface CloudVenueSnapshot { + biomes?: CloudVenueBiome[]; +} + +const CLOUD_VENUE_BIOME: Record> = { + batch: { pulse: 0.85, speed: 0.55, linkStrength: 0.75, density: 0.82 }, + interactive: { pulse: 1.05, linkStrength: 0.8, density: 0.9, palette: 'default' }, + spot: { pulse: 1.6, speed: 0.7, linkStrength: 0.9, palette: 'campaign', energyPulse: true }, + gpu: { pulse: 1.25, linkStrength: 0.95, density: 1.05, palette: 'campaign' }, +}; + +export function dominantCloudVenue(snapshot: CloudVenueSnapshot | null | undefined): string | null { + const list = snapshot?.biomes ?? []; + if (!list.length) return null; + const rank: Record = { gpu: 4, spot: 3, batch: 2, interactive: 1 }; + let best = list[0].venue_class || 'interactive'; + let bestScore = (list[0].agent_ids?.length ?? 1) * (rank[best] ?? 1); + for (let i = 1; i < list.length; i++) { + const venue = list[i].venue_class || 'interactive'; + const score = (list[i].agent_ids?.length ?? 1) * (rank[venue] ?? 1); + if (score > bestScore) { best = venue; bestScore = score; } + } + return best; +} + +export function mergeCloudVenueBiomeWeather( + base: PageWeatherConfig, + snapshot: CloudVenueSnapshot | null | undefined, +): PageWeatherConfig { + const venue = dominantCloudVenue(snapshot); + if (!venue) return base; + const overlay = CLOUD_VENUE_BIOME[venue] ?? CLOUD_VENUE_BIOME.interactive; + return { + ...base, + ...overlay, + intensity: Math.min(1, (base.intensity + (overlay.intensity ?? base.intensity)) / 2 + 0.06), + energyPulse: overlay.energyPulse ?? base.energyPulse, + }; +} + +export function mergeBiomeWeather( + base: PageWeatherConfig, + scout: ScoutConstellationSnapshot | null | undefined, + cloud: CloudVenueSnapshot | null | undefined, +): PageWeatherConfig { + return mergeCloudVenueBiomeWeather(mergeScoutBiomeWeather(base, scout), cloud); +} + +export function activeBiomeLabel( + scout: ScoutConstellationSnapshot | null | undefined, + cloud: CloudVenueSnapshot | null | undefined, +): string | null { + const cloudVenue = dominantCloudVenue(cloud); + const scoutVenue = scout?.constellations?.length ? dominantScoutVenueFromSnapshot(scout) : null; + if (cloudVenue) return `cloud:${cloudVenue}`; + if (scoutVenue) return `wifi:${scoutVenue}`; + return null; +} + +function dominantScoutVenueFromSnapshot(snapshot: ScoutConstellationSnapshot): string | null { + const list = snapshot.constellations ?? []; + if (!list.length) return null; + const rank: Record = { airport: 4, retail: 3, campus: 2, unknown: 1 }; + let best = list[0].venue_class || 'unknown'; + let bestScore = (list[0].agent_ids?.length ?? 1) * (rank[best] ?? 1); + for (let i = 1; i < list.length; i++) { + const venue = list[i].venue_class || 'unknown'; + const score = (list[i].agent_ids?.length ?? 1) * (rank[venue] ?? 1); + if (score > bestScore) { best = venue; bestScore = score; } + } + return best; +} diff --git a/server/web/src/help/wsStatsCoalesce.ts b/server/web/src/help/wsStatsCoalesce.ts index a25f99f..6e6df9c 100644 --- a/server/web/src/help/wsStatsCoalesce.ts +++ b/server/web/src/help/wsStatsCoalesce.ts @@ -91,6 +91,7 @@ export const WS_LATEST_MESSAGE_TYPES = new Set([ 'emberwake_notes_updated', 'emberwake_war_room', 'scout_constellations', + 'cloud_venue_biomes', 'agent_online', 'agent_offline', 'new_share',