// Package controlplane owns the Wails controller's local state and Agent protocol.
package controlplane

import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/sha256"
	"crypto/x509"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"encoding/pem"
	"errors"
	"fmt"
	"math/big"
	"net"
	"net/http"
	"os"
	"path/filepath"
	"regexp"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/gorilla/websocket"
	_ "modernc.org/sqlite"
)

const protocolVersion = 1

const agentHeartbeatTimeout = 30 * time.Second

// Config contains only portable, non-sensitive controller settings.
type Config struct {
	ConfigVersion int    `json:"config_version"`
	AgentHost     string `json:"agent_host"`
	AgentPort     int    `json:"agent_port"`
}

// Snapshot is the summary rendered by the Wails frontend.
type Snapshot struct {
	OnlineAgents int    `json:"onlineAgents"`
	RunningTasks int    `json:"runningTasks"`
	Listener     string `json:"listener"`
	LocalIP      string `json:"localIP"`
	AgentURL     string `json:"agentURL"`
}

// Agent is a safe summary of a registered Agent.
type Agent struct {
	ID             string `json:"id"`
	Name           string `json:"name"`
	Status         string `json:"status"`
	LastSeenAt     string `json:"lastSeenAt"`
	RestartStatus  string `json:"restartStatus"`
	RestartFailure string `json:"restartFailure"`
}

// Task is a persisted controller task summary. Credentials never appear here.
type Task struct {
	ID              string `json:"id"`
	AgentID         string `json:"agentID"`
	AgentName       string `json:"agentName"`
	Instruction     string `json:"instruction"`
	Status          string `json:"status"`
	CreatedAt       string `json:"createdAt"`
	StartedAt       string `json:"startedAt"`
	FinishedAt      string `json:"finishedAt"`
	DeadlineAt      string `json:"deadlineAt"`
	Attempt         int    `json:"attempt"`
	RetryOf         string `json:"retryOf"`
	Output          string `json:"output"`
	Error           string `json:"error"`
	Execution       string `json:"execution"`
	AttachmentCount int    `json:"attachmentCount"`
	// SessionID groups the turns of one multi-turn conversation; empty for a
	// one-shot task, which is what 单独指派 sends.
	SessionID string `json:"sessionID"`
}

// TaskEvent is one safe, persisted point in a task's execution timeline.
// Its message is written by the controller or the paired Agent and never carries
// controller credentials.
type TaskEvent struct {
	Type      string `json:"type"`
	Message   string `json:"message"`
	CreatedAt string `json:"createdAt"`
}

// TaskDetail contains the complete task result and its durable event timeline.
type TaskDetail struct {
	Task        Task             `json:"task"`
	Events      []TaskEvent      `json:"events"`
	Attachments []TaskAttachment `json:"attachments"`
}

// TaskRequest is one independently dispatched item from the multi-Agent task board.
// It intentionally contains only the same safe fields accepted by CreateTask.
type TaskRequest struct {
	AgentID        string `json:"agentID"`
	Instruction    string `json:"instruction"`
	TimeoutSeconds int    `json:"timeoutSeconds"`
}

// TaskDispatchResult gives each submitted row its own outcome. A failed row never
// prevents the controller from dispatching other valid rows in the same batch.
type TaskDispatchResult struct {
	AgentID string `json:"agentID"`
	TaskID  string `json:"taskID"`
	Error   string `json:"error"`
}

type connection struct {
	agentID string
	ws      *websocket.Conn
	mu      sync.Mutex
}

// ControlPlane is independent of Wails and is safe to expose only through narrow use cases.
type ControlPlane struct {
	db              *sql.DB
	config          Config
	certificatePath string
	privateKeyPath  string
	fingerprint     string
	mu              sync.Mutex
	connections     map[string]*connection
	server          *http.Server
	listening       bool
}

// Open prepares the isolated data directory, schema and self-signed TLS identity.
func Open(configPath string) (*ControlPlane, error) {
	config, err := loadConfig(configPath)
	if err != nil {
		return nil, err
	}
	dataDir := os.Getenv("MULTICLAW_CONTROLLER_DATA_DIR")
	if dataDir == "" {
		dataRoot, err := os.UserConfigDir()
		if err != nil {
			return nil, fmt.Errorf("无法定位本机数据目录: %w", err)
		}
		dataDir = filepath.Join(dataRoot, "MultiClawControllerWails")
	}
	if err := os.MkdirAll(filepath.Join(dataDir, "tls"), 0o700); err != nil {
		return nil, fmt.Errorf("无法创建本机数据目录: %w", err)
	}
	db, err := sql.Open("sqlite", filepath.Join(dataDir, "controller.sqlite3"))
	if err != nil {
		return nil, fmt.Errorf("无法打开本机数据库: %w", err)
	}
	// The controller has a small local SQLite store but several WSS callbacks can
	// arrive together. Serialising its single writer avoids transient SQLITE_BUSY
	// failures while preserving durable deployment and audit transitions.
	db.SetMaxOpenConns(1)
	db.SetMaxIdleConns(1)
	if _, err := db.Exec("PRAGMA busy_timeout = 5000"); err != nil {
		db.Close()
		return nil, fmt.Errorf("无法配置本机数据库: %w", err)
	}
	if err := initialiseSchema(db); err != nil {
		db.Close()
		return nil, err
	}
	certPath := filepath.Join(dataDir, "tls", "cert.pem")
	keyPath := filepath.Join(dataDir, "tls", "key.pem")
	fingerprint, err := ensureCertificate(certPath, keyPath)
	if err != nil {
		db.Close()
		return nil, err
	}
	return &ControlPlane{
		db:              db,
		config:          config,
		certificatePath: certPath,
		privateKeyPath:  keyPath,
		fingerprint:     fingerprint,
		connections:     make(map[string]*connection),
	}, nil
}

// StartAgentServer starts the mutually authenticated WSS control plane.
func (c *ControlPlane) StartAgentServer() error {
	c.mu.Lock()
	defer c.mu.Unlock()
	if c.listening {
		return nil
	}
	mux := http.NewServeMux()
	mux.HandleFunc("/", c.handleWebSocket)
	c.server = &http.Server{Addr: net.JoinHostPort(c.config.AgentHost, fmt.Sprint(c.config.AgentPort)), Handler: mux}
	listener, err := net.Listen("tcp", c.server.Addr)
	if err != nil {
		return fmt.Errorf("无法启动 WSS 服务: %w", err)
	}
	c.listening = true
	go func() {
		_ = c.server.ServeTLS(listener, c.certificatePath, c.privateKeyPath)
		c.mu.Lock()
		c.listening = false
		c.mu.Unlock()
	}()
	return nil
}

// Close stops networking and releases local resources.
func (c *ControlPlane) Close() {
	c.mu.Lock()
	for _, client := range c.connections {
		_ = client.ws.Close()
	}
	c.connections = make(map[string]*connection)
	server := c.server
	c.mu.Unlock()
	if server != nil {
		_ = server.Close()
	}
	_ = c.db.Close()
}

// CreatePairingCode creates a one-use, ten-minute code carrying the TLS pin.
func (c *ControlPlane) CreatePairingCode() (string, error) {
	pairingID, err := randomHex(16)
	if err != nil {
		return "", err
	}
	secret, err := randomHex(24)
	if err != nil {
		return "", err
	}
	_, err = c.db.Exec(
		"INSERT INTO pairing_tokens(pairing_id, secret_hash, certificate_fingerprint, expires_at) VALUES (?, ?, ?, ?)",
		pairingID, hash(secret), c.fingerprint, time.Now().UTC().Add(10*time.Minute).Format(time.RFC3339Nano),
	)
	if err != nil {
		return "", fmt.Errorf("无法保存配对码: %w", err)
	}
	c.recordAudit("pairing.code_created", "controller", "pairing-token", "created", "已生成一个十分钟有效的一次性配对码。")
	return fmt.Sprintf("v%d.%s.%s.%s", protocolVersion, pairingID, secret, c.fingerprint), nil
}

// Snapshot returns current WSS and task status without exposing sensitive state.
func (c *ControlPlane) Snapshot() (Snapshot, error) {
	c.expireTasks()
	c.expireStaleAgents()
	var online, running int
	if err := c.db.QueryRow("SELECT COUNT(*) FROM managed_agents WHERE status = 'online'").Scan(&online); err != nil {
		return Snapshot{}, err
	}
	if err := c.db.QueryRow("SELECT COUNT(*) FROM task_runs WHERE status IN ('queued', 'accepted', 'running', 'cancelling')").Scan(&running); err != nil {
		return Snapshot{}, err
	}
	c.mu.Lock()
	listening := c.listening
	c.mu.Unlock()
	listener := "未启动"
	if listening {
		listener = "运行中（WSS）"
	}
	localIP := bestLocalAddress()
	return Snapshot{OnlineAgents: online, RunningTasks: running, Listener: listener, LocalIP: localIP, AgentURL: c.agentURLFor(localIP)}, nil
}

// OnlineAgents lists only currently usable dispatch targets.
func (c *ControlPlane) OnlineAgents() ([]Agent, error) {
	c.expireStaleAgents()
	return c.queryAgents("WHERE status = 'online'")
}

// Agents lists all currently enrolled Agents. Revoked pairing records are removed.
func (c *ControlPlane) Agents() ([]Agent, error) {
	c.expireStaleAgents()
	c.removeRevokedAgents()
	return c.queryAgents("")
}

// RenameAgent changes only the display name; the immutable Agent identity stays unchanged.
func (c *ControlPlane) RenameAgent(agentID, name string) error {
	name = strings.TrimSpace(name)
	if agentID == "" || name == "" {
		return errors.New("Agent 名称不能为空")
	}
	name = limit(name, 80)
	result, err := c.db.Exec("UPDATE managed_agents SET name = ? WHERE agent_id = ?", name, agentID)
	if err != nil {
		return fmt.Errorf("无法更新 Agent 名称: %w", err)
	}
	changed, err := result.RowsAffected()
	if err != nil || changed != 1 {
		return errors.New("未找到要更新的 Agent")
	}
	c.recordAudit("agent.renamed", "controller", "agent:"+agentID, "completed", "已更新 Agent 显示名称。")
	return nil
}

// RevokeAgent ends the active connection and removes its pairing record.
func (c *ControlPlane) RevokeAgent(agentID string) error {
	if agentID == "" {
		return errors.New("Agent ID 不能为空")
	}
	c.mu.Lock()
	client := c.connections[agentID]
	delete(c.connections, agentID)
	c.mu.Unlock()
	if client != nil {
		_ = client.ws.Close()
	}
	result, err := c.db.Exec("DELETE FROM managed_agents WHERE agent_id = ?", agentID)
	if err != nil {
		return fmt.Errorf("无法撤销 Agent: %w", err)
	}
	changed, err := result.RowsAffected()
	if err != nil || changed != 1 {
		return errors.New("未找到要撤销的 Agent")
	}
	c.recordAudit("pairing.revoked", "controller", "agent:"+agentID, "completed", "已撤销配对并使旧凭据失效。")
	return nil
}

func (c *ControlPlane) removeRevokedAgents() {
	_, _ = c.db.Exec("DELETE FROM managed_agents WHERE status = 'revoked'")
}

func (c *ControlPlane) queryAgents(where string) ([]Agent, error) {
	where = strings.ReplaceAll(where, "status", "a.status")
	rows, err := c.db.Query("SELECT a.agent_id, a.name, a.status, a.last_seen_at, COALESCE(r.status, ''), COALESCE(r.failure, '') FROM managed_agents a LEFT JOIN agent_restart_operations r ON r.agent_id = a.agent_id " + where + " ORDER BY a.name")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	agents := make([]Agent, 0)
	for rows.Next() {
		var agent Agent
		if err := rows.Scan(&agent.ID, &agent.Name, &agent.Status, &agent.LastSeenAt, &agent.RestartStatus, &agent.RestartFailure); err != nil {
			return nil, err
		}
		agents = append(agents, agent)
	}
	return agents, rows.Err()
}

// CreateTaskSessionResult tells the UI which conversation a task joined.
type CreateTaskSessionResult struct {
	TaskID    string `json:"taskID"`
	SessionID string `json:"sessionID"`
	Turns     int    `json:"turns"`
}

// CreateTask persists and sends a text-only task to an authenticated Agent.
func (c *ControlPlane) CreateTask(agentID, instruction string, timeoutSeconds int) (string, error) {
	return c.createTask(agentID, instruction, timeoutSeconds, 1, "")
}

// CreateTaskWithImages creates one explicitly-addressed multimodal task.
func (c *ControlPlane) CreateTaskWithImages(agentID, instruction string, timeoutSeconds int, inputs []TaskImageInput) (string, error) {
	images, err := normalizeTaskImages(inputs)
	if err != nil {
		return "", err
	}
	return c.createTaskWithImages(agentID, instruction, timeoutSeconds, 1, "", images, "")
}

// CreateTasks dispatches independently-addressed tasks for the task board. This is
// deliberately not a broadcast: every row has its own target, timeout and result.
func (c *ControlPlane) CreateTasks(requests []TaskRequest) ([]TaskDispatchResult, error) {
	if len(requests) == 0 {
		return nil, errors.New("请至少添加一条任务")
	}
	if len(requests) > 20 {
		return nil, errors.New("一次最多下发 20 条任务")
	}
	results := make([]TaskDispatchResult, 0, len(requests))
	for _, request := range requests {
		result := TaskDispatchResult{AgentID: request.AgentID}
		taskID, err := c.CreateTask(request.AgentID, request.Instruction, request.TimeoutSeconds)
		if err != nil {
			result.Error = sanitizeDisplayText(err.Error())
		} else {
			result.TaskID = taskID
		}
		results = append(results, result)
	}
	return results, nil
}

func (c *ControlPlane) createTask(agentID, instruction string, timeoutSeconds, attempt int, retryOf string) (string, error) {
	return c.createTaskWithImages(agentID, instruction, timeoutSeconds, attempt, retryOf, nil, "")
}

func (c *ControlPlane) createTaskWithImages(agentID, instruction string, timeoutSeconds, attempt int, retryOf string, images []taskImage, sessionID string) (string, error) {
	if err := validateSafeTaskText(instruction); err != nil {
		return "", err
	}
	return c.createTaskWithInstructionsAndImages(agentID, instruction, instruction, timeoutSeconds, attempt, retryOf, images, sessionID)
}

// createTaskWithInstructions stores the concise user-visible instruction while
// optionally dispatching a bounded private context to the target Agent. The
// protocol and lifecycle stay exactly the same for both kinds of task.
func (c *ControlPlane) createTaskWithInstructions(agentID, instruction, dispatchInstruction string, timeoutSeconds, attempt int, retryOf string) (string, error) {
	return c.createTaskWithInstructionsAndImages(agentID, instruction, dispatchInstruction, timeoutSeconds, attempt, retryOf, nil, "")
}

func (c *ControlPlane) createTaskWithInstructionsAndImages(agentID, instruction, dispatchInstruction string, timeoutSeconds, attempt int, retryOf string, images []taskImage, sessionID string) (string, error) {
	c.expireStaleAgents()
	instruction = strings.TrimSpace(instruction)
	dispatchInstruction = strings.TrimSpace(dispatchInstruction)
	if agentID == "" || instruction == "" {
		return "", errors.New("请选择在线 Agent 并输入任务内容")
	}
	if timeoutSeconds < 1 || timeoutSeconds > 3600 {
		return "", errors.New("任务超时时间必须在 1 到 3600 秒之间")
	}
	var agentName string
	if err := c.db.QueryRow("SELECT name FROM managed_agents WHERE agent_id = ?", agentID).Scan(&agentName); err != nil {
		return "", errors.New("目标 Agent 不存在")
	}
	taskID, err := randomHex(16)
	if err != nil {
		return "", err
	}
	now := time.Now().UTC().Format(time.RFC3339Nano)
	deadline := time.Now().UTC().Add(time.Duration(timeoutSeconds) * time.Second).Format(time.RFC3339Nano)
	if _, err := c.db.Exec(
		"INSERT INTO task_runs(task_id, agent_id, agent_name, instruction, status, created_at, deadline_at, timeout_seconds, attempt, retry_of, session_id) VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?)",
		taskID, agentID, agentName, instruction, now, deadline, timeoutSeconds, attempt, retryOf, sessionID,
	); err != nil {
		return "", fmt.Errorf("无法保存任务: %w", err)
	}
	for position, image := range images {
		if _, err := c.db.Exec("INSERT INTO task_images(task_id, position, name, media_type, data_url) VALUES (?, ?, ?, ?, ?)", taskID, position+1, image.Name, image.MediaType, image.DataURL); err != nil {
			return "", fmt.Errorf("无法保存任务图片: %w", err)
		}
	}
	c.recordEvent(taskID, "queued", "任务已进入下发队列。", now)
	c.recordAudit("task.created", "controller", "agent:"+agentID, "queued", fmt.Sprintf("已创建任务；超时为 %d 秒。", timeoutSeconds))
	if client := c.clientFor(agentID); client != nil {
		if err := c.sendQueuedTask(client, taskID); err != nil {
			return "", errors.New("任务已安全排队，等待 Agent 重新连接后下发")
		}
	}
	return taskID, nil
}

// dispatchQueuedTasks replays durable queued work only after an authenticated
// Agent session is attached. Repeated delivery is safe: the Agent task ID is
// idempotent and it returns its existing lifecycle state.
func (c *ControlPlane) dispatchQueuedTasks(agentID string) {
	client := c.clientFor(agentID)
	if client == nil {
		return
	}
	rows, err := c.db.Query("SELECT task_id FROM task_runs WHERE agent_id = ? AND status = 'queued' ORDER BY created_at", agentID)
	if err != nil {
		return
	}
	defer rows.Close()
	for rows.Next() {
		var taskID string
		if rows.Scan(&taskID) == nil {
			_ = c.sendQueuedTask(client, taskID)
		}
	}
}

func (c *ControlPlane) sendQueuedTask(client *connection, taskID string) error {
	var instruction, deadline, sessionID string
	var attempt int
	if err := c.db.QueryRow("SELECT instruction, deadline_at, attempt, COALESCE(session_id, '') FROM task_runs WHERE task_id = ? AND status = 'queued'", taskID).Scan(&instruction, &deadline, &attempt, &sessionID); err != nil {
		return err
	}
	images, err := c.taskImages(taskID)
	if err != nil {
		return err
	}
	payload := map[string]any{"type": "task_dispatch", "schema_version": protocolVersion, "task_id": taskID, "instruction": instruction, "attachments": images, "sent_at": time.Now().UTC().Format(time.RFC3339Nano), "deadline_at": deadline, "attempt": attempt}
	if sessionID != "" {
		history, historyErr := c.sessionHistory(sessionID, c.sessionTurnLimit(sessionID))
		if historyErr != nil {
			return historyErr
		}
		payload["session_id"] = sessionID
		if len(history) > 0 {
			// A follow-up turn carries the earlier plain user/assistant text so the
			// model keeps the thread without shipping tool transcripts.
			payload["history"] = history
		}
	}
	return client.send(payload)
}

// sessionTurnLimit reads the configured conversation length, defaulting safely.
func (c *ControlPlane) sessionTurnLimit(sessionID string) int {
	var turnLimit int
	if err := c.db.QueryRow("SELECT turn_limit FROM task_sessions WHERE session_id = ?", sessionID).Scan(&turnLimit); err != nil {
		return defaultSessionTurnLimit
	}
	return normaliseTurnLimit(turnLimit)
}

// CancelTask asks the target Agent to stop an active task. The Agent owns final acknowledgement.
func (c *ControlPlane) CancelTask(taskID string) error {
	var agentID, status string
	if err := c.db.QueryRow("SELECT agent_id, status FROM task_runs WHERE task_id = ?", taskID).Scan(&agentID, &status); err != nil {
		return errors.New("未找到要取消的任务")
	}
	if !isActiveTask(status) {
		return errors.New("该任务已结束，不能取消")
	}
	client := c.clientFor(agentID)
	if client == nil {
		return errors.New("目标 Agent 当前不在线，暂不能取消")
	}
	// Persist the cancellation intent before sending.  An Agent can respond on
	// another goroutine immediately; recording it first makes every late
	// completed/failed event ineligible to overwrite the requested outcome.
	result, err := c.db.Exec("UPDATE task_runs SET status = 'cancelling' WHERE task_id = ? AND status = ?", taskID, status)
	if err != nil {
		return errors.New("无法记录取消状态，请刷新后重试")
	}
	if changed, _ := result.RowsAffected(); changed != 1 {
		return errors.New("任务状态已变化，请刷新后重试")
	}
	if err := client.send(map[string]any{"type": "task_cancel", "schema_version": protocolVersion, "task_id": taskID, "reason": "由总控取消"}); err != nil {
		_, _ = c.db.Exec("UPDATE task_runs SET status = ? WHERE task_id = ? AND status = 'cancelling'", status, taskID)
		return errors.New("取消指令未发送成功，请刷新后重试")
	}
	c.recordEvent(taskID, "cancelling", "已向 Agent 发送取消指令。", time.Now().UTC().Format(time.RFC3339Nano))
	c.recordAudit("task.cancel_requested", "controller", "agent:"+agentID, "cancelling", "已向 Agent 请求取消任务。")
	return nil
}

// RetryTask creates a distinct retry attempt for a non-successful task without overwriting history.
func (c *ControlPlane) RetryTask(taskID string) (string, error) {
	var agentID, instruction, status, sessionID string
	var timeoutSeconds, attempt int
	if err := c.db.QueryRow("SELECT agent_id, instruction, status, timeout_seconds, attempt, COALESCE(session_id, '') FROM task_runs WHERE task_id = ?", taskID).Scan(&agentID, &instruction, &status, &timeoutSeconds, &attempt, &sessionID); err != nil {
		return "", errors.New("未找到要重试的任务")
	}
	if status != "failed" && status != "cancelled" && status != "timed_out" {
		return "", errors.New("只能重试失败、已取消或已超时的任务")
	}
	images, imageErr := c.taskImages(taskID)
	if imageErr != nil {
		return "", imageErr
	}
	// A retry repeats the same turn, so it stays in the same conversation.
	retryID, err := c.createTaskWithImages(agentID, instruction, timeoutSeconds, attempt+1, taskID, images, sessionID)
	if err == nil {
		c.recordAudit("task.retry_created", "controller", "agent:"+agentID, "queued", "已创建失败任务的重试尝试。")
	}
	return retryID, err
}

// RecentTasks provides durable task summaries for the dashboard.
func (c *ControlPlane) RecentTasks() ([]Task, error) {
	c.expireTasks()
	rows, err := c.db.Query("SELECT r.task_id, r.agent_id, COALESCE(NULLIF(r.agent_name, ''), '未知 Agent'), r.instruction, r.status, r.created_at, COALESCE(r.started_at, ''), COALESCE(r.finished_at, ''), COALESCE(r.deadline_at, ''), COALESCE(r.attempt, 1), COALESCE(r.retry_of, ''), COALESCE(r.output, ''), COALESCE(r.error, ''), COALESCE(r.execution_source, ''), COUNT(i.position), COALESCE(r.session_id, '') FROM task_runs r LEFT JOIN task_images i ON i.task_id = r.task_id GROUP BY r.task_id ORDER BY r.created_at DESC LIMIT 24")
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	tasks := make([]Task, 0)
	for rows.Next() {
		var task Task
		if err := rows.Scan(&task.ID, &task.AgentID, &task.AgentName, &task.Instruction, &task.Status, &task.CreatedAt, &task.StartedAt, &task.FinishedAt, &task.DeadlineAt, &task.Attempt, &task.RetryOf, &task.Output, &task.Error, &task.Execution, &task.AttachmentCount, &task.SessionID); err != nil {
			return nil, err
		}
		tasks = append(tasks, task)
	}
	return tasks, rows.Err()
}

// TaskDetails returns one task and the controller events recorded for it.
func (c *ControlPlane) TaskDetails(taskID string) (TaskDetail, error) {
	c.expireTasks()
	var detail TaskDetail
	err := c.db.QueryRow("SELECT r.task_id, r.agent_id, COALESCE(NULLIF(r.agent_name, ''), '未知 Agent'), r.instruction, r.status, r.created_at, COALESCE(r.started_at, ''), COALESCE(r.finished_at, ''), COALESCE(r.deadline_at, ''), COALESCE(r.attempt, 1), COALESCE(r.retry_of, ''), COALESCE(r.output, ''), COALESCE(r.error, ''), COALESCE(r.execution_source, ''), COUNT(i.position), COALESCE(r.session_id, '') FROM task_runs r LEFT JOIN task_images i ON i.task_id = r.task_id WHERE r.task_id = ? GROUP BY r.task_id", taskID).Scan(
		&detail.Task.ID, &detail.Task.AgentID, &detail.Task.AgentName, &detail.Task.Instruction, &detail.Task.Status, &detail.Task.CreatedAt, &detail.Task.StartedAt, &detail.Task.FinishedAt, &detail.Task.DeadlineAt, &detail.Task.Attempt, &detail.Task.RetryOf, &detail.Task.Output, &detail.Task.Error, &detail.Task.Execution, &detail.Task.AttachmentCount, &detail.Task.SessionID,
	)
	if errors.Is(err, sql.ErrNoRows) {
		return TaskDetail{}, errors.New("未找到任务")
	}
	if err != nil {
		return TaskDetail{}, err
	}
	rows, err := c.db.Query("SELECT event_type, message, created_at FROM task_events WHERE task_id = ? ORDER BY event_id ASC", taskID)
	if err != nil {
		return TaskDetail{}, err
	}
	defer rows.Close()
	detail.Events = make([]TaskEvent, 0)
	for rows.Next() {
		var event TaskEvent
		if err := rows.Scan(&event.Type, &event.Message, &event.CreatedAt); err != nil {
			return TaskDetail{}, err
		}
		detail.Events = append(detail.Events, event)
	}
	if err := rows.Err(); err != nil {
		return TaskDetail{}, err
	}
	attachments, err := c.taskAttachmentPreviews(taskID)
	if err != nil {
		return TaskDetail{}, err
	}
	detail.Attachments = attachments
	return detail, nil
}

// ClearTaskHistory removes only terminal task records and their event timelines.
// Active work remains cancellable and auditable; audit records themselves are never removed.
func (c *ControlPlane) ClearTaskHistory() (int, error) {
	transaction, err := c.db.Begin()
	if err != nil {
		return 0, fmt.Errorf("无法开始清除任务历史: %w", err)
	}
	defer transaction.Rollback()
	const terminal = "'completed', 'failed', 'cancelled', 'timed_out'"
	if _, err := transaction.Exec("DELETE FROM task_images WHERE task_id IN (SELECT task_id FROM task_runs WHERE status IN (" + terminal + "))"); err != nil {
		return 0, fmt.Errorf("无法清除任务图片: %w", err)
	}
	if _, err := transaction.Exec("DELETE FROM task_events WHERE task_id IN (SELECT task_id FROM task_runs WHERE status IN (" + terminal + "))"); err != nil {
		return 0, fmt.Errorf("无法清除任务事件: %w", err)
	}
	result, err := transaction.Exec("DELETE FROM task_runs WHERE status IN (" + terminal + ")")
	if err != nil {
		return 0, fmt.Errorf("无法清除任务历史: %w", err)
	}
	if err := transaction.Commit(); err != nil {
		return 0, fmt.Errorf("无法完成清除任务历史: %w", err)
	}
	deleted, _ := result.RowsAffected()
	c.recordAudit("task.history_cleared", "controller", "task-history", "completed", fmt.Sprintf("已清除 %d 条已结束任务记录。", deleted))
	return int(deleted), nil
}

func (c *ControlPlane) handleWebSocket(writer http.ResponseWriter, request *http.Request) {
	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
	ws, err := upgrader.Upgrade(writer, request, nil)
	if err != nil {
		return
	}
	client := &connection{ws: ws}
	defer func() {
		c.disconnect(client)
		_ = ws.Close()
	}()
	for {
		_, raw, err := ws.ReadMessage()
		if err != nil {
			return
		}
		var message map[string]any
		if json.Unmarshal(raw, &message) != nil {
			_ = client.send(map[string]any{"type": "error", "message": "消息不是有效 JSON。"})
			continue
		}
		c.handleMessage(client, message)
	}
}

func (c *ControlPlane) handleMessage(client *connection, message map[string]any) {
	messageType, _ := message["type"].(string)
	switch messageType {
	case "pair_request":
		code, _ := message["code"].(string)
		if !c.consumePairingCode(code) {
			_ = client.send(map[string]any{"type": "pair_rejected", "message": "配对码无效、已使用或已过期。"})
			return
		}
		name, _ := message["agent_name"].(string)
		agentID, credential, err := c.registerAgent(name)
		if err != nil {
			_ = client.send(map[string]any{"type": "error", "message": "无法注册 Agent。"})
			return
		}
		c.attach(client, agentID)
		_ = client.send(map[string]any{"type": "pair_accepted", "agent_id": agentID, "credential": credential})
		c.pushConfigurationSnapshots(agentID)
		go c.dispatchQueuedTasks(agentID)
	case "authenticate":
		agentID, _ := message["agent_id"].(string)
		credential, _ := message["credential"].(string)
		if !c.authenticate(agentID, credential) {
			_ = client.send(map[string]any{"type": "auth_rejected", "message": "Agent 凭据无效。"})
			return
		}
		c.attach(client, agentID)
		if name, _ := message["agent_name"].(string); strings.TrimSpace(name) != "" {
			_ = c.RenameAgent(agentID, name)
		}
		_ = client.send(map[string]any{"type": "auth_accepted"})
		c.pushConfigurationSnapshots(agentID)
		go c.dispatchQueuedTasks(agentID)
	case "agent_rename":
		if client.agentID == "" {
			_ = client.send(map[string]any{"type": "error", "message": "请先完成认证。"})
			return
		}
		name, _ := message["name"].(string)
		if err := c.RenameAgent(client.agentID, name); err != nil {
			_ = client.send(map[string]any{"type": "error", "message": err.Error()})
			return
		}
		_ = client.send(map[string]any{"type": "agent_rename_ack"})
	case "heartbeat":
		if client.agentID == "" {
			_ = client.send(map[string]any{"type": "error", "message": "请先完成认证。"})
			return
		}
		c.setAgentStatus(client.agentID, "online")
		_ = client.send(map[string]any{"type": "heartbeat_ack"})
	case "task_event":
		if client.agentID == "" {
			_ = client.send(map[string]any{"type": "error", "message": "请先完成认证。"})
			return
		}
		c.applyTaskEvent(client.agentID, message)
	case "config_ack":
		if client.agentID == "" {
			_ = client.send(map[string]any{"type": "error", "message": "请先完成认证。"})
			return
		}
		c.applyLLMConfigAck(client.agentID, message)
	case "skill_ack":
		if client.agentID == "" {
			_ = client.send(map[string]any{"type": "error", "message": "请先完成认证。"})
			return
		}
		c.applySkillAck(client.agentID, message)
	case "mcp_ack":
		if client.agentID != "" {
			c.applyMCPAck(client.agentID, message)
		}
	case "agent_restart_ack":
		if client.agentID != "" {
			c.applyRestartAck(client.agentID, message)
		}
	default:
		_ = client.send(map[string]any{"type": "error", "message": "不支持的消息类型。"})
	}
}

func (c *ControlPlane) attach(client *connection, agentID string) {
	client.agentID = agentID
	c.mu.Lock()
	previous := c.connections[agentID]
	c.connections[agentID] = client
	c.mu.Unlock()
	if previous != nil && previous != client {
		_ = previous.ws.Close()
	}
	c.setAgentStatus(agentID, "online")
	c.markRestartReconnected(agentID)
}

// pushConfigurationSnapshots starts only after the authentication acknowledgement
// has been written, so a reconnect never receives a deployment before auth_accepted.
func (c *ControlPlane) pushConfigurationSnapshots(agentID string) {
	go c.pushLLMConfig(agentID)
	go c.pushSkills(agentID)
	go c.pushMCP(agentID, false)
}

func (c *ControlPlane) disconnect(client *connection) {
	if client.agentID == "" {
		return
	}
	c.mu.Lock()
	if c.connections[client.agentID] == client {
		delete(c.connections, client.agentID)
		c.mu.Unlock()
		c.setAgentStatus(client.agentID, "offline")
		return
	}
	c.mu.Unlock()
}

func (c *ControlPlane) clientFor(agentID string) *connection {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.connections[agentID]
}

func (c *ControlPlane) consumePairingCode(code string) bool {
	parts := strings.Split(code, ".")
	if len(parts) != 4 || parts[0] != fmt.Sprintf("v%d", protocolVersion) || parts[3] != c.fingerprint {
		return false
	}
	transaction, err := c.db.Begin()
	if err != nil {
		return false
	}
	defer transaction.Rollback()
	var secretHash, fingerprint, expiresAt string
	var usedAt sql.NullString
	err = transaction.QueryRow("SELECT secret_hash, certificate_fingerprint, expires_at, used_at FROM pairing_tokens WHERE pairing_id = ?", parts[1]).Scan(&secretHash, &fingerprint, &expiresAt, &usedAt)
	if err != nil || usedAt.Valid || secretHash != hash(parts[2]) || fingerprint != c.fingerprint {
		return false
	}
	expires, err := time.Parse(time.RFC3339Nano, expiresAt)
	if err != nil || !expires.After(time.Now().UTC()) {
		return false
	}
	if _, err := transaction.Exec("UPDATE pairing_tokens SET used_at = ? WHERE pairing_id = ? AND used_at IS NULL", time.Now().UTC().Format(time.RFC3339Nano), parts[1]); err != nil {
		return false
	}
	return transaction.Commit() == nil
}

func (c *ControlPlane) registerAgent(name string) (string, string, error) {
	agentID, err := randomHex(16)
	if err != nil {
		return "", "", err
	}
	credential, err := randomHex(32)
	if err != nil {
		return "", "", err
	}
	name = strings.TrimSpace(name)
	if name == "" {
		name = "未命名 Agent"
	}
	name = string([]rune(name)[:min(len([]rune(name)), 80)])
	_, err = c.db.Exec("INSERT INTO managed_agents(agent_id, name, credential_hash, status, last_seen_at) VALUES (?, ?, ?, 'online', ?)", agentID, name, hash(credential), time.Now().UTC().Format(time.RFC3339Nano))
	if err == nil {
		c.recordAudit("pairing.completed", "controller", "agent:"+agentID, "completed", "新的 Agent 已完成安全配对。")
	}
	return agentID, credential, err
}

func (c *ControlPlane) authenticate(agentID, credential string) bool {
	var storedHash, status string
	if c.db.QueryRow("SELECT credential_hash, status FROM managed_agents WHERE agent_id = ?", agentID).Scan(&storedHash, &status) != nil {
		return false
	}
	return status != "revoked" && storedHash == hash(credential)
}

func (c *ControlPlane) setAgentStatus(agentID, status string) {
	_, _ = c.db.Exec("UPDATE managed_agents SET status = ?, last_seen_at = ? WHERE agent_id = ?", status, time.Now().UTC().Format(time.RFC3339Nano), agentID)
}

// expireStaleAgents prevents a dead network connection from remaining dispatchable forever.
func (c *ControlPlane) expireStaleAgents() {
	cutoff := time.Now().UTC().Add(-agentHeartbeatTimeout).Format(time.RFC3339Nano)
	rows, err := c.db.Query("SELECT agent_id FROM managed_agents WHERE status = 'online' AND last_seen_at < ?", cutoff)
	if err != nil {
		return
	}
	staleIDs := make([]string, 0)
	for rows.Next() {
		var agentID string
		if rows.Scan(&agentID) == nil {
			staleIDs = append(staleIDs, agentID)
		}
	}
	rows.Close()
	if len(staleIDs) == 0 {
		return
	}
	for _, agentID := range staleIDs {
		_, _ = c.db.Exec("UPDATE managed_agents SET status = 'offline' WHERE agent_id = ? AND status = 'online'", agentID)
	}
	c.mu.Lock()
	staleConnections := make([]*connection, 0, len(staleIDs))
	for _, agentID := range staleIDs {
		if client := c.connections[agentID]; client != nil {
			delete(c.connections, agentID)
			staleConnections = append(staleConnections, client)
		}
	}
	c.mu.Unlock()
	for _, client := range staleConnections {
		_ = client.ws.Close()
	}
}

func (c *ControlPlane) applyTaskEvent(agentID string, event map[string]any) {
	taskID, _ := event["task_id"].(string)
	status, _ := event["status"].(string)
	if status == "progress" {
		c.applyTaskProgress(agentID, taskID, event)
		return
	}
	if taskID == "" || !isTaskStatus(status) {
		return
	}
	var expectedAgentID, currentStatus, sessionID, instruction string
	var timeoutSeconds int
	if c.db.QueryRow("SELECT agent_id, status, timeout_seconds, COALESCE(session_id, ''), instruction FROM task_runs WHERE task_id = ?", taskID).Scan(&expectedAgentID, &currentStatus, &timeoutSeconds, &sessionID, &instruction) != nil || expectedAgentID != agentID || isTerminal(currentStatus) {
		return
	}
	// Cancellation is a write-once intent.  An already-running model or MCP
	// request may return late, but it must not change the requested terminal
	// outcome to completed/failed or resurrect the task as running.
	if currentStatus == "cancelling" && status != "cancelled" {
		return
	}
	message, _ := event["message"].(string)
	output, _ := event["output"].(string)
	errText, _ := event["error"].(string)
	execution, _ := event["execution"].(string)
	if execution != "desktop-filesystem" && execution != "full-access" && execution != "builtin" && execution != "mcp" && execution != "llm" && execution != "demo" {
		execution = ""
	}
	message = sanitizeDisplayText(message)
	output = sanitizeDisplayText(output)
	errText = sanitizeDisplayText(errText)
	now := time.Now().UTC().Format(time.RFC3339Nano)
	if status == "running" {
		deadline := time.Now().UTC().Add(time.Duration(timeoutSeconds) * time.Second).Format(time.RFC3339Nano)
		_, _ = c.db.Exec("UPDATE task_runs SET status = ?, started_at = COALESCE(started_at, ?), deadline_at = ? WHERE task_id = ?", status, now, deadline, taskID)
	} else if isTerminal(status) {
		_, _ = c.db.Exec("UPDATE task_runs SET status = ?, finished_at = ?, output = ?, error = ?, execution_source = COALESCE(NULLIF(?, ''), execution_source) WHERE task_id = ?", status, now, limit(output, 12000), limit(errText, 2000), execution, taskID)
	} else {
		deadline := time.Now().UTC().Add(time.Duration(timeoutSeconds) * time.Second).Format(time.RFC3339Nano)
		_, _ = c.db.Exec("UPDATE task_runs SET status = ?, deadline_at = ? WHERE task_id = ?", status, deadline, taskID)
	}
	if message == "" {
		message = "Agent 上报状态：" + status
	}
	c.recordEvent(taskID, status, limit(message, 1000), now)
	// Only a successful turn joins the conversation, so a failed attempt cannot
	// poison the context of the next follow-up instruction.
	if status == "completed" && sessionID != "" {
		c.appendSessionTurn(sessionID, instruction, output)
	}
	c.recordAudit("task.status_changed", "agent:"+agentID, "agent:"+agentID, status, "Agent 已上报任务状态变更。")
	if execution == "desktop-filesystem" && isTerminal(status) {
		detail := output
		if detail == "" {
			detail = errText
		}
		c.recordAudit("mcp.tool_execution", "agent:"+agentID, "task:"+taskID+":desktop-filesystem", status, detail)
	}
	if execution == "full-access" && isTerminal(status) {
		detail := output
		if detail == "" {
			detail = errText
		}
		c.recordAudit("task.full_access_execution", "agent:"+agentID, "task:"+taskID+":full-access", status, detail)
	}
	if execution == "builtin" && isTerminal(status) {
		detail := output
		if detail == "" {
			detail = errText
		}
		c.recordAudit("task.builtin_tool_execution", "agent:"+agentID, "task:"+taskID+":builtin", status, detail)
	}
	if execution == "mcp" && isTerminal(status) {
		detail := output
		if detail == "" {
			detail = errText
		}
		c.recordAudit("mcp.tool_execution", "agent:"+agentID, "task:"+taskID+":mcp", status, detail)
	}
}

// applyTaskProgress records one intermediate step of a running task without
// changing its lifecycle state. A running tool call is also real progress, so
// the no-progress deadline is renewed instead of expiring a busy task.
func (c *ControlPlane) applyTaskProgress(agentID, taskID string, event map[string]any) {
	if taskID == "" {
		return
	}
	var expectedAgentID, currentStatus string
	var timeoutSeconds int
	if c.db.QueryRow("SELECT agent_id, status, timeout_seconds FROM task_runs WHERE task_id = ?", taskID).Scan(&expectedAgentID, &currentStatus, &timeoutSeconds) != nil ||
		expectedAgentID != agentID || isTerminal(currentStatus) {
		return
	}
	message, _ := event["message"].(string)
	message = limit(sanitizeDisplayText(message), 500)
	if message == "" {
		return
	}
	now := time.Now().UTC()
	if isActiveTask(currentStatus) {
		deadline := now.Add(time.Duration(timeoutSeconds) * time.Second).Format(time.RFC3339Nano)
		_, _ = c.db.Exec("UPDATE task_runs SET deadline_at = ? WHERE task_id = ?", deadline, taskID)
	}
	c.recordEvent(taskID, "progress", message, now.Format(time.RFC3339Nano))
}

func (c *ControlPlane) expireTasks() {
	now := time.Now().UTC()
	rows, err := c.db.Query("SELECT task_id, agent_id, deadline_at FROM task_runs WHERE status IN ('queued', 'accepted', 'running', 'cancelling') AND deadline_at IS NOT NULL AND deadline_at != ''")
	if err != nil {
		return
	}
	type expiredTask struct{ id, agentID string }
	expired := make([]expiredTask, 0)
	for rows.Next() {
		var taskID, agentID, deadlineAt string
		if rows.Scan(&taskID, &agentID, &deadlineAt) != nil {
			continue
		}
		deadline, parseErr := time.Parse(time.RFC3339Nano, deadlineAt)
		if parseErr != nil || deadline.After(now) {
			continue
		}
		expired = append(expired, expiredTask{id: taskID, agentID: agentID})
	}
	if rows.Close() != nil {
		return
	}
	for _, task := range expired {
		result, updateErr := c.db.Exec("UPDATE task_runs SET status = 'timed_out', error = ?, finished_at = ? WHERE task_id = ? AND status IN ('queued', 'accepted', 'running', 'cancelling')", "任务超过截止时间。", now.Format(time.RFC3339Nano), task.id)
		if updateErr != nil {
			continue
		}
		changed, _ := result.RowsAffected()
		if changed != 1 {
			continue
		}
		c.recordEvent(task.id, "timed_out", "任务超过截止时间，总控已停止等待。", now.Format(time.RFC3339Nano))
		if client := c.clientFor(task.agentID); client != nil {
			_ = client.send(map[string]any{"type": "task_cancel", "schema_version": protocolVersion, "task_id": task.id, "reason": "任务已超时"})
		}
	}
}

func (c *ControlPlane) markTaskFailed(taskID, message string) {
	now := time.Now().UTC().Format(time.RFC3339Nano)
	_, _ = c.db.Exec("UPDATE task_runs SET status = 'failed', error = ?, finished_at = ? WHERE task_id = ?", message, now, taskID)
	c.recordEvent(taskID, "failed", message, now)
}

func (c *ControlPlane) recordEvent(taskID, eventType, message, createdAt string) {
	_, _ = c.db.Exec("INSERT INTO task_events(task_id, event_type, message, created_at) VALUES (?, ?, ?, ?)", taskID, eventType, message, createdAt)
}

func (c *connection) send(value any) error {
	encoded, err := json.Marshal(value)
	if err != nil {
		return err
	}
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.ws.WriteMessage(websocket.TextMessage, encoded)
}

func (c *ControlPlane) agentURL() string {
	return c.agentURLFor(bestLocalAddress())
}

func (c *ControlPlane) agentURLFor(host string) string {
	return fmt.Sprintf("wss://%s:%d", host, c.config.AgentPort)
}

func loadConfig(configPath string) (Config, error) {
	config := Config{ConfigVersion: 1, AgentHost: "0.0.0.0", AgentPort: 8443}
	raw, err := os.ReadFile(configPath)
	if err != nil && !os.IsNotExist(err) {
		return config, err
	}
	if err == nil && json.Unmarshal(raw, &config) != nil {
		return config, errors.New("controller-config.json 格式无效")
	}
	if config.AgentHost == "" {
		config.AgentHost = "0.0.0.0"
	}
	if config.AgentPort < 1 || config.AgentPort > 65535 {
		return config, errors.New("Agent 服务端口必须在 1 到 65535 之间")
	}
	return config, nil
}

func initialiseSchema(db *sql.DB) error {
	statements := []string{
		"CREATE TABLE IF NOT EXISTS pairing_tokens (pairing_id TEXT PRIMARY KEY, secret_hash TEXT NOT NULL, certificate_fingerprint TEXT NOT NULL, expires_at TEXT NOT NULL, used_at TEXT)",
		"CREATE TABLE IF NOT EXISTS managed_agents (agent_id TEXT PRIMARY KEY, name TEXT NOT NULL, credential_hash TEXT NOT NULL, status TEXT NOT NULL, last_seen_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS task_runs (task_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, agent_name TEXT NOT NULL DEFAULT '', instruction TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL, deadline_at TEXT, timeout_seconds INTEGER NOT NULL DEFAULT 300, attempt INTEGER NOT NULL DEFAULT 1, retry_of TEXT, started_at TEXT, finished_at TEXT, output TEXT, error TEXT, execution_source TEXT)",
		"CREATE TABLE IF NOT EXISTS task_events (event_id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, event_type TEXT NOT NULL, message TEXT NOT NULL, created_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS task_images (task_id TEXT NOT NULL, position INTEGER NOT NULL, name TEXT NOT NULL, media_type TEXT NOT NULL, data_url TEXT NOT NULL, PRIMARY KEY(task_id, position))",
		"CREATE TABLE IF NOT EXISTS llm_secrets (secret_id TEXT PRIMARY KEY, encrypted_value BLOB NOT NULL, updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS llm_profiles (profile_id TEXT PRIMARY KEY, name TEXT NOT NULL, provider TEXT NOT NULL, base_url TEXT NOT NULL, model TEXT NOT NULL, supports_images BOOLEAN NOT NULL DEFAULT FALSE, secret_id TEXT NOT NULL, updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS agent_llm_assignments (agent_id TEXT PRIMARY KEY, profile_id TEXT NOT NULL, config_version INTEGER NOT NULL, apply_status TEXT NOT NULL, last_error TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS skill_packages (skill_id TEXT PRIMARY KEY, name TEXT NOT NULL, version TEXT NOT NULL, content TEXT NOT NULL, checksum TEXT NOT NULL, updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS agent_skill_assignments (agent_id TEXT NOT NULL, skill_id TEXT NOT NULL, apply_status TEXT NOT NULL, last_error TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL, PRIMARY KEY(agent_id, skill_id))",
		"CREATE TABLE IF NOT EXISTS agent_skill_selections (agent_id TEXT PRIMARY KEY, updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS mcp_tools (tool_id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT NOT NULL DEFAULT '', kind TEXT NOT NULL, endpoint TEXT NOT NULL DEFAULT '', command TEXT NOT NULL DEFAULT '', args TEXT NOT NULL DEFAULT '[]', version TEXT NOT NULL DEFAULT '1.0.0', updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS agent_mcp_assignments (agent_id TEXT NOT NULL, tool_id TEXT NOT NULL, deployment_version INTEGER NOT NULL DEFAULT 1, apply_status TEXT NOT NULL, last_error TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL, PRIMARY KEY(agent_id, tool_id))",
		"CREATE TABLE IF NOT EXISTS task_sessions (session_id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, turn_limit INTEGER NOT NULL DEFAULT 50, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)",
		"CREATE TABLE IF NOT EXISTS task_session_messages (session_id TEXT NOT NULL, position INTEGER NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL, PRIMARY KEY(session_id, position))",
		"CREATE TABLE IF NOT EXISTS audit_events (event_id INTEGER PRIMARY KEY AUTOINCREMENT, action TEXT NOT NULL, actor TEXT NOT NULL, target TEXT NOT NULL, outcome TEXT NOT NULL, detail TEXT NOT NULL, created_at TEXT NOT NULL, operator TEXT NOT NULL DEFAULT '', target_agent_id TEXT NOT NULL DEFAULT '', resource_type TEXT NOT NULL DEFAULT 'system', result TEXT NOT NULL DEFAULT '', security_detail TEXT NOT NULL DEFAULT '')",
		"CREATE TABLE IF NOT EXISTS agent_restart_operations (agent_id TEXT PRIMARY KEY, request_id TEXT NOT NULL, status TEXT NOT NULL, requested_at TEXT NOT NULL, confirmed_at TEXT, failure TEXT NOT NULL DEFAULT '')",
	}
	for _, statement := range statements {
		if _, err := db.Exec(statement); err != nil {
			return fmt.Errorf("无法初始化控制端数据库: %w", err)
		}
	}
	// Existing preview databases get additive columns without losing task history.
	for _, statement := range []string{
		"ALTER TABLE task_runs ADD COLUMN deadline_at TEXT",
		"ALTER TABLE task_runs ADD COLUMN timeout_seconds INTEGER NOT NULL DEFAULT 300",
		"ALTER TABLE task_runs ADD COLUMN attempt INTEGER NOT NULL DEFAULT 1",
		"ALTER TABLE task_runs ADD COLUMN retry_of TEXT",
		"ALTER TABLE task_runs ADD COLUMN execution_source TEXT",
		"ALTER TABLE task_runs ADD COLUMN agent_name TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE task_runs ADD COLUMN session_id TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE llm_profiles ADD COLUMN supports_images BOOLEAN NOT NULL DEFAULT FALSE",
		"ALTER TABLE agent_mcp_assignments ADD COLUMN deployment_version INTEGER NOT NULL DEFAULT 1",
		"ALTER TABLE mcp_tools ADD COLUMN command TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE mcp_tools ADD COLUMN args TEXT NOT NULL DEFAULT '[]'",
		"ALTER TABLE audit_events ADD COLUMN operator TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE audit_events ADD COLUMN target_agent_id TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE audit_events ADD COLUMN resource_type TEXT NOT NULL DEFAULT 'system'",
		"ALTER TABLE audit_events ADD COLUMN result TEXT NOT NULL DEFAULT ''",
		"ALTER TABLE audit_events ADD COLUMN security_detail TEXT NOT NULL DEFAULT ''",
	} {
		if _, err := db.Exec(statement); err != nil && !strings.Contains(err.Error(), "duplicate column name") {
			return fmt.Errorf("无法升级控制端数据库: %w", err)
		}
	}
	// Backfill records created before task ownership was persisted. New records
	// keep this display name even if the Agent is later revoked.
	if _, err := db.Exec("UPDATE task_runs SET agent_name = COALESCE((SELECT name FROM managed_agents WHERE managed_agents.agent_id = task_runs.agent_id), '') WHERE COALESCE(agent_name, '') = ''"); err != nil {
		return fmt.Errorf("无法补全任务 Agent 名称: %w", err)
	}
	if err := seedBuiltinMCPTools(db); err != nil {
		return err
	}
	if err := backfillAuditEventFields(db); err != nil {
		return err
	}
	return nil
}

func ensureCertificate(certPath, keyPath string) (string, error) {
	if raw, err := os.ReadFile(certPath); err == nil {
		block, _ := pem.Decode(raw)
		if block == nil {
			return "", errors.New("TLS 证书文件无效")
		}
		return fingerprint(block.Bytes), nil
	}
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	if err != nil {
		return "", err
	}
	serial, err := rand.Int(rand.Reader, new(big.Int).Lsh(big.NewInt(1), 128))
	if err != nil {
		return "", err
	}
	template := x509.Certificate{SerialNumber: serial, NotBefore: time.Now().Add(-time.Minute), NotAfter: time.Now().AddDate(5, 0, 0), KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, DNSNames: []string{"multiclaw-controller"}}
	der, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
	if err != nil {
		return "", err
	}
	keyDER, err := x509.MarshalECPrivateKey(key)
	if err != nil {
		return "", err
	}
	if err := os.WriteFile(certPath, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), 0o600); err != nil {
		return "", err
	}
	if err := os.WriteFile(keyPath, pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), 0o600); err != nil {
		return "", err
	}
	return fingerprint(der), nil
}

func bestLocalAddress() string {
	connection, err := net.Dial("udp", "192.0.2.1:9")
	if err == nil {
		defer connection.Close()
		host, _, splitErr := net.SplitHostPort(connection.LocalAddr().String())
		if splitErr == nil && isUsableLANIPv4(net.ParseIP(host)) {
			return host
		}
	}

	// VPN/proxy tunnel adapters can become the system default route. Prefer an
	// active private IPv4 address on a normal LAN interface when that happens.
	candidates := make([]string, 0)
	interfaces, err := net.Interfaces()
	if err == nil {
		for _, networkInterface := range interfaces {
			if networkInterface.Flags&net.FlagUp == 0 || networkInterface.Flags&net.FlagLoopback != 0 || networkInterface.Flags&net.FlagPointToPoint != 0 {
				continue
			}
			addresses, addressErr := networkInterface.Addrs()
			if addressErr != nil {
				continue
			}
			for _, address := range addresses {
				ip, _, parseErr := net.ParseCIDR(address.String())
				if parseErr == nil && isUsableLANIPv4(ip) {
					candidates = append(candidates, ip.String())
				}
			}
		}
	}
	if len(candidates) == 0 {
		return "127.0.0.1"
	}
	sort.Strings(candidates)
	return candidates[0]
}

func isUsableLANIPv4(ip net.IP) bool {
	if ip == nil || ip.To4() == nil || !ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() {
		return false
	}
	return !ip.Equal(net.ParseIP("198.18.0.1"))
}

func randomHex(length int) (string, error) {
	bytes := make([]byte, length)
	if _, err := rand.Read(bytes); err != nil {
		return "", err
	}
	return hex.EncodeToString(bytes), nil
}

func hash(value string) string { return fingerprint([]byte(value)) }
func fingerprint(value []byte) string {
	digest := sha256.Sum256(value)
	return hex.EncodeToString(digest[:])
}
func isTaskStatus(status string) bool {
	return status == "accepted" || status == "running" || isTerminal(status)
}
func isTerminal(status string) bool {
	return status == "completed" || status == "failed" || status == "cancelled" || status == "timed_out"
}
func isActiveTask(status string) bool {
	return status == "queued" || status == "accepted" || status == "running" || status == "cancelling"
}
func limit(value string, maximum int) string {
	runes := []rune(value)
	if len(runes) > maximum {
		return string(runes[:maximum])
	}
	return value
}

var thinkBlockPattern = regexp.MustCompile(`(?is)<think(?:\s[^>]*)?>.*?</think\s*>`)
var unclosedThinkPattern = regexp.MustCompile(`(?is)<think(?:\s[^>]*)?>.*$`)
var displaySecretPattern = regexp.MustCompile(`(?i)(\b(?:api[_ -]?key|password|passwd|secret|token|credential|authorization)|用户名|账号|密码|密钥)\s*[:=：]\s*[^\s,;；]+`)
var pairingCredentialPattern = regexp.MustCompile(`(?i)\bv1\.[0-9a-f]{32}\.[0-9a-f]{48}\.[0-9a-f]{64}\b`)
var rawAPIKeyPattern = regexp.MustCompile(`\b(?:sk-[A-Za-z0-9_-]{16,}|AIza[A-Za-z0-9_-]{20,})\b`)

func sanitizeDisplayText(value string) string {
	value = unclosedThinkPattern.ReplaceAllString(thinkBlockPattern.ReplaceAllString(value, ""), "")
	value = pairingCredentialPattern.ReplaceAllString(value, "[配对凭据已脱敏]")
	value = rawAPIKeyPattern.ReplaceAllString(value, "[API Key 已脱敏]")
	return strings.TrimSpace(displaySecretPattern.ReplaceAllString(value, "$1=[已脱敏]"))
}