Add adversarial L4 court chamber with Seer court_debate feed.
Some checks failed
CI Docker Mining Proof / Linux agent hashrate proof (push) Has been cancelled

Prosecutor and Public Defender use real fleet telemetry only; Judge dispatches L4 commands and emits full transcripts via seer_events and emberwake_court_debate when ai_control_enabled.
This commit is contained in:
AetherForge
2026-06-07 09:21:32 -07:00
parent e1cecd8aef
commit 5853f80c48
18 changed files with 2152 additions and 9 deletions

View File

@@ -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()
}

View File

@@ -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)
}
}

View File

@@ -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, ",")
}

View File

@@ -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 ReedSolomon 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
}

View File

@@ -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)
}
}

View File

@@ -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,
}),
})
}

View File

@@ -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)
}
}

View File

@@ -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)
}

View File

@@ -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)
}
}

View File

@@ -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)})
}

View File

@@ -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)
}
}

View File

@@ -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)
}

View File

@@ -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")
}
}

View File

@@ -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))

View File

@@ -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:]
}

View File

@@ -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)
}
}

View File

@@ -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()
}

View File

@@ -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)
}
}