You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
265 lines
12 KiB
265 lines
12 KiB
package controlplane
|
|
|
|
import (
|
|
"database/sql"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
// AgentDirectoryFilter is a bounded, server-side filter for the Agent directory.
|
|
// Empty values mean "all"; it deliberately has no free-form SQL fields.
|
|
type AgentDirectoryFilter struct {
|
|
Page int `json:"page"`
|
|
PageSize int `json:"pageSize"`
|
|
Name string `json:"name"`
|
|
ConnectionStatus string `json:"connectionStatus"`
|
|
MCPStatus string `json:"mcpStatus"`
|
|
LLMStatus string `json:"llmStatus"`
|
|
SkillStatus string `json:"skillStatus"`
|
|
}
|
|
|
|
// AgentDirectoryItem is the safe, one-row operational summary used in a paged list.
|
|
type AgentDirectoryItem struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
Status string `json:"status"`
|
|
LastSeenAt string `json:"lastSeenAt"`
|
|
LLMProfile string `json:"llmProfile"`
|
|
LLMStatus string `json:"llmStatus"`
|
|
LLMFailure string `json:"llmFailure"`
|
|
SkillTotal int `json:"skillTotal"`
|
|
SkillApplied int `json:"skillApplied"`
|
|
SkillStatus string `json:"skillStatus"`
|
|
SkillFailure string `json:"skillFailure"`
|
|
MCPStatus string `json:"mcpStatus"`
|
|
MCPFailure string `json:"mcpFailure"`
|
|
LastDeploymentAt string `json:"lastDeploymentAt"`
|
|
RestartStatus string `json:"restartStatus"`
|
|
RestartFailure string `json:"restartFailure"`
|
|
}
|
|
|
|
type AgentDirectoryPage struct {
|
|
Items []AgentDirectoryItem `json:"items"`
|
|
Page int `json:"page"`
|
|
PageSize int `json:"pageSize"`
|
|
TotalItems int `json:"totalItems"`
|
|
TotalPages int `json:"totalPages"`
|
|
}
|
|
|
|
type DeploymentFailure struct {
|
|
Resource string `json:"resource"`
|
|
Reason string `json:"reason"`
|
|
UpdatedAt string `json:"updatedAt"`
|
|
}
|
|
|
|
// AgentOperationalDetail keeps the detail view intentionally operational: recent
|
|
// tasks, safe audit/deployment trail and current deployment failures only.
|
|
type AgentOperationalDetail struct {
|
|
Agent Agent `json:"agent"`
|
|
Deployment AgentDirectoryItem `json:"deployment"`
|
|
RecentTasks []Task `json:"recentTasks"`
|
|
RecentAudit []AuditEvent `json:"recentAudit"`
|
|
DeploymentHistory []AuditEvent `json:"deploymentHistory"`
|
|
Failures []DeploymentFailure `json:"failures"`
|
|
}
|
|
|
|
var agentDirectoryBaseSQL = `WITH skill_summary AS (
|
|
SELECT agent_id, COUNT(*) AS total,
|
|
SUM(CASE WHEN apply_status = 'applied' THEN 1 ELSE 0 END) AS applied,
|
|
SUM(CASE WHEN apply_status = 'failed' THEN 1 ELSE 0 END) AS failed,
|
|
SUM(CASE WHEN apply_status = 'pending' THEN 1 ELSE 0 END) AS pending,
|
|
SUM(CASE WHEN apply_status = 'sent' THEN 1 ELSE 0 END) AS sent,
|
|
MAX(updated_at) AS updated_at,
|
|
MAX(CASE WHEN apply_status = 'failed' THEN last_error ELSE '' END) AS failure
|
|
FROM agent_skill_assignments GROUP BY agent_id
|
|
), directory AS (
|
|
SELECT a.agent_id, a.name, a.status, a.last_seen_at,
|
|
COALESCE(p.name, '') AS llm_profile, COALESCE(l.apply_status, '未部署') AS llm_status,
|
|
COALESCE(l.last_error, '') AS llm_failure,
|
|
COALESCE(s.total, 0) AS skill_total, COALESCE(s.applied, 0) AS skill_applied,
|
|
CASE WHEN COALESCE(s.total, 0) = 0 THEN '未选择'
|
|
WHEN COALESCE(s.failed, 0) > 0 THEN 'failed'
|
|
WHEN COALESCE(s.pending, 0) > 0 THEN 'pending'
|
|
WHEN COALESCE(s.sent, 0) > 0 THEN 'sent'
|
|
ELSE 'applied' END AS skill_status,
|
|
COALESCE(s.failure, '') AS skill_failure,
|
|
` + mcpStatusExpression("a.agent_id") + ` AS mcp_status,
|
|
` + mcpErrorExpression("a.agent_id") + ` AS mcp_failure,
|
|
COALESCE((SELECT MAX(updated_at) FROM agent_mcp_assignments m WHERE m.agent_id = a.agent_id), '') AS mcp_updated_at,
|
|
MAX(COALESCE(l.updated_at, ''), COALESCE((SELECT MAX(updated_at) FROM agent_mcp_assignments m WHERE m.agent_id = a.agent_id), ''), COALESCE(s.updated_at, '')) AS last_deployment_at,
|
|
COALESCE(r.status, '') AS restart_status, COALESCE(r.failure, '') AS restart_failure
|
|
FROM managed_agents a
|
|
LEFT JOIN agent_llm_assignments l ON l.agent_id = a.agent_id
|
|
LEFT JOIN llm_profiles p ON p.profile_id = l.profile_id
|
|
LEFT JOIN skill_summary s ON s.agent_id = a.agent_id
|
|
LEFT JOIN agent_restart_operations r ON r.agent_id = a.agent_id
|
|
)
|
|
`
|
|
|
|
func normaliseAgentDirectoryFilter(filter AgentDirectoryFilter) (AgentDirectoryFilter, error) {
|
|
filter.Name = strings.TrimSpace(filter.Name)
|
|
filter.ConnectionStatus = strings.TrimSpace(filter.ConnectionStatus)
|
|
filter.MCPStatus = strings.TrimSpace(filter.MCPStatus)
|
|
filter.LLMStatus = strings.TrimSpace(filter.LLMStatus)
|
|
filter.SkillStatus = strings.TrimSpace(filter.SkillStatus)
|
|
if len([]rune(filter.Name)) > 80 {
|
|
return filter, errors.New("Agent 名称搜索不能超过 80 个字符")
|
|
}
|
|
if filter.Page < 1 {
|
|
filter.Page = 1
|
|
}
|
|
if filter.PageSize < 1 {
|
|
filter.PageSize = 12
|
|
}
|
|
if filter.PageSize > 50 {
|
|
filter.PageSize = 50
|
|
}
|
|
if !oneOf(filter.ConnectionStatus, "", "online", "offline") ||
|
|
!oneOf(filter.MCPStatus, "", "未启用", "pending", "sent", "applied", "failed") ||
|
|
!oneOf(filter.LLMStatus, "", "未部署", "pending", "sent", "applied", "failed") ||
|
|
!oneOf(filter.SkillStatus, "", "未选择", "pending", "sent", "applied", "failed") {
|
|
return filter, errors.New("Agent 筛选条件无效")
|
|
}
|
|
return filter, nil
|
|
}
|
|
|
|
func oneOf(value string, allowed ...string) bool {
|
|
for _, candidate := range allowed {
|
|
if value == candidate {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func agentDirectoryWhere(filter AgentDirectoryFilter) (string, []any) {
|
|
clauses, args := make([]string, 0, 5), make([]any, 0, 5)
|
|
if filter.Name != "" {
|
|
escaped := strings.NewReplacer("\\", "\\\\", "%", "\\%", "_", "\\_").Replace(filter.Name)
|
|
clauses, args = append(clauses, "name LIKE ? ESCAPE '\\'"), append(args, "%"+escaped+"%")
|
|
}
|
|
for _, criterion := range []struct{ column, value string }{{"status", filter.ConnectionStatus}, {"mcp_status", filter.MCPStatus}, {"llm_status", filter.LLMStatus}, {"skill_status", filter.SkillStatus}} {
|
|
if criterion.value != "" {
|
|
clauses, args = append(clauses, criterion.column+" = ?"), append(args, criterion.value)
|
|
}
|
|
}
|
|
if len(clauses) == 0 {
|
|
return "", args
|
|
}
|
|
return " WHERE " + strings.Join(clauses, " AND "), args
|
|
}
|
|
|
|
// AgentDirectory provides database pagination and all Agent filters in one query.
|
|
func (c *ControlPlane) AgentDirectory(filter AgentDirectoryFilter) (AgentDirectoryPage, error) {
|
|
c.expireStaleAgents()
|
|
filter, err := normaliseAgentDirectoryFilter(filter)
|
|
if err != nil {
|
|
return AgentDirectoryPage{}, err
|
|
}
|
|
where, args := agentDirectoryWhere(filter)
|
|
page := AgentDirectoryPage{Items: make([]AgentDirectoryItem, 0), Page: filter.Page, PageSize: filter.PageSize}
|
|
if err := c.db.QueryRow(agentDirectoryBaseSQL+"SELECT COUNT(*) FROM directory"+where, args...).Scan(&page.TotalItems); err != nil {
|
|
return page, fmt.Errorf("无法统计 Agent 列表: %w", err)
|
|
}
|
|
page.TotalPages = (page.TotalItems + filter.PageSize - 1) / filter.PageSize
|
|
if page.TotalPages > 0 && page.Page > page.TotalPages {
|
|
page.Page = page.TotalPages
|
|
}
|
|
queryArgs := append(append([]any{}, args...), filter.PageSize, (page.Page-1)*filter.PageSize)
|
|
rows, err := c.db.Query(agentDirectoryBaseSQL+`SELECT agent_id, name, status, last_seen_at, llm_profile, llm_status, llm_failure,
|
|
skill_total, skill_applied, skill_status, skill_failure, mcp_status, mcp_failure, last_deployment_at, restart_status, restart_failure
|
|
FROM directory`+where+" ORDER BY CASE status WHEN 'online' THEN 0 ELSE 1 END, name COLLATE NOCASE, agent_id LIMIT ? OFFSET ?", queryArgs...)
|
|
if err != nil {
|
|
return page, fmt.Errorf("无法读取 Agent 列表: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
for rows.Next() {
|
|
var item AgentDirectoryItem
|
|
if err := rows.Scan(&item.ID, &item.Name, &item.Status, &item.LastSeenAt, &item.LLMProfile, &item.LLMStatus, &item.LLMFailure, &item.SkillTotal, &item.SkillApplied, &item.SkillStatus, &item.SkillFailure, &item.MCPStatus, &item.MCPFailure, &item.LastDeploymentAt, &item.RestartStatus, &item.RestartFailure); err != nil {
|
|
return page, err
|
|
}
|
|
item.LLMFailure, item.SkillFailure, item.MCPFailure = sanitizeDisplayText(item.LLMFailure), sanitizeDisplayText(item.SkillFailure), sanitizeDisplayText(item.MCPFailure)
|
|
page.Items = append(page.Items, item)
|
|
}
|
|
return page, rows.Err()
|
|
}
|
|
|
|
// AgentOperationalDetails returns bounded, safe operational context for one Agent.
|
|
func (c *ControlPlane) AgentOperationalDetails(agentID string) (AgentOperationalDetail, error) {
|
|
if strings.TrimSpace(agentID) == "" {
|
|
return AgentOperationalDetail{}, errors.New("Agent 标识无效")
|
|
}
|
|
// Fetch by ID directly so duplicate display names cannot select another Agent.
|
|
var item AgentDirectoryItem
|
|
err := c.db.QueryRow(agentDirectoryBaseSQL+`SELECT agent_id, name, status, last_seen_at, llm_profile, llm_status, llm_failure,
|
|
skill_total, skill_applied, skill_status, skill_failure, mcp_status, mcp_failure, last_deployment_at, restart_status, restart_failure FROM directory WHERE agent_id = ?`, agentID).Scan(
|
|
&item.ID, &item.Name, &item.Status, &item.LastSeenAt, &item.LLMProfile, &item.LLMStatus, &item.LLMFailure, &item.SkillTotal, &item.SkillApplied, &item.SkillStatus, &item.SkillFailure, &item.MCPStatus, &item.MCPFailure, &item.LastDeploymentAt, &item.RestartStatus, &item.RestartFailure,
|
|
)
|
|
if errors.Is(err, sql.ErrNoRows) {
|
|
return AgentOperationalDetail{}, errors.New("未找到 Agent")
|
|
}
|
|
if err != nil {
|
|
return AgentOperationalDetail{}, fmt.Errorf("无法读取 Agent 详情: %w", err)
|
|
}
|
|
detail := AgentOperationalDetail{Agent: Agent{ID: item.ID, Name: item.Name, Status: item.Status, LastSeenAt: item.LastSeenAt, RestartStatus: item.RestartStatus, RestartFailure: item.RestartFailure}, Deployment: item, RecentTasks: make([]Task, 0), RecentAudit: make([]AuditEvent, 0), DeploymentHistory: make([]AuditEvent, 0), Failures: make([]DeploymentFailure, 0)}
|
|
detail.RecentTasks, err = c.recentTasksForAgent(agentID, 8)
|
|
if err != nil {
|
|
return detail, err
|
|
}
|
|
detail.RecentAudit, err = c.agentAuditEvents(agentID, false, 12)
|
|
if err != nil {
|
|
return detail, err
|
|
}
|
|
detail.DeploymentHistory, err = c.agentAuditEvents(agentID, true, 12)
|
|
if err != nil {
|
|
return detail, err
|
|
}
|
|
for _, failure := range []DeploymentFailure{{"LLM", item.LLMFailure, item.LastDeploymentAt}, {"Skill", item.SkillFailure, item.LastDeploymentAt}, {"MCP", item.MCPFailure, item.LastDeploymentAt}} {
|
|
if strings.TrimSpace(failure.Reason) != "" {
|
|
failure.Reason = sanitizeDisplayText(failure.Reason)
|
|
detail.Failures = append(detail.Failures, failure)
|
|
}
|
|
}
|
|
return detail, nil
|
|
}
|
|
|
|
func (c *ControlPlane) recentTasksForAgent(agentID string, maximum int) ([]Task, error) {
|
|
rows, err := c.db.Query("SELECT task_id, agent_id, COALESCE(NULLIF(agent_name, ''), '未知 Agent'), instruction, status, created_at, COALESCE(started_at, ''), COALESCE(finished_at, ''), COALESCE(deadline_at, ''), COALESCE(attempt, 1), COALESCE(retry_of, ''), COALESCE(output, ''), COALESCE(error, ''), COALESCE(execution_source, ''), COALESCE(session_id, '') FROM task_runs WHERE agent_id = ? ORDER BY created_at DESC LIMIT ?", agentID, maximum)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("无法读取 Agent 最近任务: %w", 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.SessionID); err != nil {
|
|
return nil, err
|
|
}
|
|
tasks = append(tasks, task)
|
|
}
|
|
return tasks, rows.Err()
|
|
}
|
|
|
|
func (c *ControlPlane) agentAuditEvents(agentID string, deploymentOnly bool, maximum int) ([]AuditEvent, error) {
|
|
where := "(actor = ? OR target LIKE ?)"
|
|
args := []any{"agent:" + agentID, "agent:" + agentID + "%"}
|
|
if deploymentOnly {
|
|
where += " AND (action LIKE 'llm.%' OR action LIKE 'skill.%' OR action LIKE 'mcp.%')"
|
|
}
|
|
rows, err := c.db.Query("SELECT event_id, action, actor, target, outcome, detail, created_at FROM audit_events WHERE "+where+" ORDER BY event_id DESC LIMIT ?", append(args, maximum)...)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("无法读取 Agent 审计记录: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
events := make([]AuditEvent, 0)
|
|
for rows.Next() {
|
|
var event AuditEvent
|
|
if err := rows.Scan(&event.ID, &event.Action, &event.Actor, &event.Target, &event.Outcome, &event.Detail, &event.CreatedAt); err != nil {
|
|
return nil, err
|
|
}
|
|
event.Detail = sanitizeDisplayText(event.Detail)
|
|
events = append(events, event)
|
|
}
|
|
return events, rows.Err()
|
|
}
|
|
|