diff --git a/controller/app.go b/controller/app.go index d5da5b1..a94ab10 100644 --- a/controller/app.go +++ b/controller/app.go @@ -95,6 +95,14 @@ func (a *App) RevokeAgent(agentID string) error { return a.control.RevokeAgent(agentID) } +// RestartAgent sends a paired, authenticated lifecycle command to one online Agent. +func (a *App) RestartAgent(agentID string) (controlplane.AgentRestart, error) { + if a.startErr != nil { + return controlplane.AgentRestart{}, a.startErr + } + return a.control.RestartAgent(agentID) +} + func (a *App) CreateTask(agentID, instruction string, timeoutSeconds int) (string, error) { if a.startErr != nil { return "", a.startErr diff --git a/controller/frontend/src/App.tsx b/controller/frontend/src/App.tsx index 7d507a3..84ad346 100644 --- a/controller/frontend/src/App.tsx +++ b/controller/frontend/src/App.tsx @@ -1,6 +1,6 @@ import {FormEvent, useEffect, useRef, useState} from "react"; import "./App.css"; -import {AgentDirectory, AgentLLMStatuses, AgentMCPStatuses, AgentOperationalDetails, AgentSkillStatuses, Agents, AuditTrail, CancelTask, ClearTaskHistory, ContinueTaskSession, CreatePairingCode, CreateTaskWithImages, DeleteMCPTool, DeployLLMProfile, DeploySkill, DispatchMentionsWithImages, ImportSkillsFromDirectory, LLMProfiles, MCPTools, OnlineAgents, RecentTasks, RenameAgent, RetryTask, RevokeAgent, SaveLLMProfile, SaveMCPTool, SaveSkill, SetAgentMCPTools, SetAgentSkills, SkillDocument, Skills, Snapshot, TaskDetails, TaskSessionTurns} from "../wailsjs/go/main/App"; +import {AgentDirectory, AgentLLMStatuses, AgentMCPStatuses, AgentOperationalDetails, AgentSkillStatuses, Agents, AuditTrail, CancelTask, ClearTaskHistory, ContinueTaskSession, CreatePairingCode, CreateTaskWithImages, DeleteMCPTool, DeployLLMProfile, DeploySkill, DispatchMentionsWithImages, ImportSkillsFromDirectory, LLMProfiles, MCPTools, OnlineAgents, RecentTasks, RenameAgent, RestartAgent, RetryTask, RevokeAgent, SaveLLMProfile, SaveMCPTool, SaveSkill, SetAgentMCPTools, SetAgentSkills, SkillDocument, Skills, Snapshot, TaskDetails, TaskSessionTurns} from "../wailsjs/go/main/App"; import {controlplane, main} from "../wailsjs/go/models"; type Page = "overview" | "agents" | "batch" | "direct" | "skills" | "plugins" | "keys" | "audit" | "settings"; @@ -187,6 +187,14 @@ function App() { finally { setLoading(false); } }; + const restartAgent = async (agent: controlplane.Agent) => { + if (agent.status !== "online") { setError("该 Agent 当前离线,无法通过已认证连接发送重启指令。"); return; } + if (!window.confirm(`重启“${agent.name}”?该 Agent 正在运行的任务将被安全取消。`)) return; + setLoading(true); + try { await RestartAgent(agent.id); setNotice(`已请求 ${agent.name} 重启,正在等待其重新连接。`); await refresh(); } + catch (reason) { setError(String(reason)); } finally { setLoading(false); } + }; + const saveLLMProfile = async (input: controlplane.LLMProfileInput) => { setLoading(true); try { @@ -277,7 +285,7 @@ function App() { {error &&
{error}
} {notice &&
{notice}
} {page === "overview" && <> void cancelTask(task)} onRetry={(task) => void retryTask(task)} onClearHistory={() => void clearTaskHistory()} onOpen={(task) => void showTaskDetail(task)}/>} - {page === "agents" && void renameAgent(agentID, name)} onSetSkills={(targetAgentID, skillIDs) => void setAgentSkills(targetAgentID, skillIDs)} onSetMCPTools={(targetAgentID, toolIDs) => void setAgentMCPTools(targetAgentID, toolIDs)} onRevoke={(agent) => void revokeAgent(agent)} onPair={() => setPage("keys")}/>} + {page === "agents" && void renameAgent(agentID, name)} onSetSkills={(targetAgentID, skillIDs) => void setAgentSkills(targetAgentID, skillIDs)} onSetMCPTools={(targetAgentID, toolIDs) => void setAgentMCPTools(targetAgentID, toolIDs)} onRestart={(agent) => void restartAgent(agent)} onRevoke={(agent) => void revokeAgent(agent)} onPair={() => setPage("keys")}/>} {page === "batch" && void showTaskDetail(task)}/>} {page === "direct" && void showTaskDetail(task)}/>} {page === "keys" && <> void createPairingCode()}/> void saveLLMProfile(input)} onDeploy={(profileID, agentIDs) => void deployLLMProfile(profileID, agentIDs)}/>} @@ -300,7 +308,7 @@ function Overview({snapshot, agents, onNavigate}: {snapshot?: controlplane.Snaps ; } -function AgentManagement({skills, skillStatuses, mcpTools, mcpStatuses, loading, onRename, onSetSkills, onSetMCPTools, onRevoke, onPair}: {skills: controlplane.Skill[]; skillStatuses: controlplane.AgentSkillStatus[]; mcpTools: controlplane.MCPTool[]; mcpStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRevoke: (agent: controlplane.Agent) => void; onPair: () => void}) { +function AgentManagement({skills, skillStatuses, mcpTools, mcpStatuses, loading, onRename, onSetSkills, onSetMCPTools, onRestart, onRevoke, onPair}: {skills: controlplane.Skill[]; skillStatuses: controlplane.AgentSkillStatus[]; mcpTools: controlplane.MCPTool[]; mcpStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRestart: (agent: controlplane.Agent) => void; onRevoke: (agent: controlplane.Agent) => void; onPair: () => void}) { const [query, setQuery] = useState(""); const [connectionStatus, setConnectionStatus] = useState(""); const [mcpStatus, setMCPStatus] = useState(""); @@ -334,7 +342,7 @@ function AgentManagement({skills, skillStatuses, mcpTools, mcpStatuses, loading,

名称搜索和所有状态筛选均在总控服务端执行。点击卡片可查看最近任务、审计、部署轨迹和失败原因。

{directoryLoading ? "正在查询…" : `共 ${directory.totalItems} 个 Agent`}
{directoryError &&

{directoryError}

}{directory.totalItems === 0 ?

还没有符合筛选条件的 Agent。

: <>
{directory.items.map(agent => void openAgent(agent)}/>)}
第 {directory.page} / {directory.totalPages} 页
} - {selectedAgent && status.agentID === selectedAgent.id).map(status => status.skillID)} deployment={selectedDetail?.deployment} mcpTools={mcpTools} selectedToolIDs={mcpStatuses.filter(status => status.agentID === selectedAgent.id).map(status => status.toolID)} toolStatuses={mcpStatuses.filter(status => status.agentID === selectedAgent.id)} loading={loading || detailLoading} onClose={() => { setSelectedAgent(undefined); setSelectedDetail(undefined); }} onRename={onRename} onSetSkills={onSetSkills} onSetMCPTools={onSetMCPTools} onRevoke={onRevoke}/>} + {selectedAgent && status.agentID === selectedAgent.id).map(status => status.skillID)} deployment={selectedDetail?.deployment} mcpTools={mcpTools} selectedToolIDs={mcpStatuses.filter(status => status.agentID === selectedAgent.id).map(status => status.toolID)} toolStatuses={mcpStatuses.filter(status => status.agentID === selectedAgent.id)} loading={loading || detailLoading} onClose={() => { setSelectedAgent(undefined); setSelectedDetail(undefined); }} onRename={onRename} onSetSkills={onSetSkills} onSetMCPTools={onSetMCPTools} onRestart={onRestart} onRevoke={onRevoke}/>} ; } @@ -355,12 +363,13 @@ function useCloseOnEscape(onClose: () => void) { }, [onClose]); } -function AgentDetailModal({agent, detail, skills, selectedSkillIDs, deployment, mcpTools, selectedToolIDs, toolStatuses, loading, onClose, onRename, onSetSkills, onSetMCPTools, onRevoke}: {agent: controlplane.Agent; detail?: controlplane.AgentOperationalDetail; skills: controlplane.Skill[]; selectedSkillIDs: string[]; deployment?: controlplane.AgentDirectoryItem; mcpTools: controlplane.MCPTool[]; selectedToolIDs: string[]; toolStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onClose: () => void; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRevoke: (agent: controlplane.Agent) => void}) { +function AgentDetailModal({agent, detail, skills, selectedSkillIDs, deployment, mcpTools, selectedToolIDs, toolStatuses, loading, onClose, onRename, onSetSkills, onSetMCPTools, onRestart, onRevoke}: {agent: controlplane.Agent; detail?: controlplane.AgentOperationalDetail; skills: controlplane.Skill[]; selectedSkillIDs: string[]; deployment?: controlplane.AgentDirectoryItem; mcpTools: controlplane.MCPTool[]; selectedToolIDs: string[]; toolStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onClose: () => void; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRestart: (agent: controlplane.Agent) => void; onRevoke: (agent: controlplane.Agent) => void}) { useCloseOnEscape(onClose); - return
{ if (event.currentTarget === event.target) onClose(); }}>

AGENT DETAIL

{agent.name}

{loading && !detail ?

正在读取 Agent 运行详情…

: <>}
; + const restarting = agent.restartStatus === "restarting" || agent.restartStatus === "sent"; + return
{ if (event.currentTarget === event.target) onClose(); }}>

AGENT DETAIL

{agent.name}

{loading && !detail ?

正在读取 Agent 运行详情…

: <>
{agent.restartStatus === "reconnected" ? "已重连" : agent.restartStatus === "failed" ? `重启失败:${agent.restartFailure || "请查看审计日志"}` : restarting ? "重启中,正在等待 Agent 重新连接…" : ""}
}
; } -function AgentRow({agent, skills, selectedSkillIDs, deployment, mcpTools, selectedToolIDs, toolStatuses, loading, onRename, onSetSkills, onSetMCPTools, onRevoke}: {agent: controlplane.Agent; skills: controlplane.Skill[]; selectedSkillIDs: string[]; deployment?: controlplane.AgentDirectoryItem; mcpTools: controlplane.MCPTool[]; selectedToolIDs: string[]; toolStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRevoke: (agent: controlplane.Agent) => void}) { +function AgentRow({agent, skills, selectedSkillIDs, deployment, mcpTools, selectedToolIDs, toolStatuses, loading, onRename, onSetSkills, onSetMCPTools, onRestart, onRevoke}: {agent: controlplane.Agent; skills: controlplane.Skill[]; selectedSkillIDs: string[]; deployment?: controlplane.AgentDirectoryItem; mcpTools: controlplane.MCPTool[]; selectedToolIDs: string[]; toolStatuses: controlplane.AgentMCPStatus[]; loading: boolean; onRename: (agentID: string, name: string) => void; onSetSkills: (agentID: string, skillIDs: string[]) => void; onSetMCPTools: (agentID: string, toolIDs: string[]) => void; onRestart: (agent: controlplane.Agent) => void; onRevoke: (agent: controlplane.Agent) => void}) { const [name, setName] = useState(agent.name); const [selected, setSelected] = useState(selectedSkillIDs); const [selectedTools, setSelectedTools] = useState(selectedToolIDs); diff --git a/controller/frontend/wailsjs/go/main/App.d.ts b/controller/frontend/wailsjs/go/main/App.d.ts index e7e0312..75945a0 100644 --- a/controller/frontend/wailsjs/go/main/App.d.ts +++ b/controller/frontend/wailsjs/go/main/App.d.ts @@ -55,6 +55,8 @@ export function RecentTasks():Promise>; export function RenameAgent(arg1:string,arg2:string):Promise; +export function RestartAgent(arg1:string):Promise; + export function RetryTask(arg1:string):Promise; export function RevokeAgent(arg1:string):Promise; diff --git a/controller/frontend/wailsjs/go/main/App.js b/controller/frontend/wailsjs/go/main/App.js index 920232f..46817d7 100644 --- a/controller/frontend/wailsjs/go/main/App.js +++ b/controller/frontend/wailsjs/go/main/App.js @@ -106,6 +106,10 @@ export function RenameAgent(arg1, arg2) { return window['go']['main']['App']['RenameAgent'](arg1, arg2); } +export function RestartAgent(arg1) { + return window['go']['main']['App']['RestartAgent'](arg1); +} + export function RetryTask(arg1) { return window['go']['main']['App']['RetryTask'](arg1); } diff --git a/controller/frontend/wailsjs/go/models.ts b/controller/frontend/wailsjs/go/models.ts index 0223e73..a538725 100644 --- a/controller/frontend/wailsjs/go/models.ts +++ b/controller/frontend/wailsjs/go/models.ts @@ -5,6 +5,8 @@ export namespace controlplane { name: string; status: string; lastSeenAt: string; + restartStatus: string; + restartFailure: string; static createFrom(source: any = {}) { return new Agent(source); @@ -16,6 +18,8 @@ export namespace controlplane { this.name = source["name"]; this.status = source["status"]; this.lastSeenAt = source["lastSeenAt"]; + this.restartStatus = source["restartStatus"]; + this.restartFailure = source["restartFailure"]; } } export class AgentDirectoryFilter { @@ -57,6 +61,8 @@ export namespace controlplane { mcpStatus: string; mcpFailure: string; lastDeploymentAt: string; + restartStatus: string; + restartFailure: string; static createFrom(source: any = {}) { return new AgentDirectoryItem(source); @@ -78,6 +84,8 @@ export namespace controlplane { this.mcpStatus = source["mcpStatus"]; this.mcpFailure = source["mcpFailure"]; this.lastDeploymentAt = source["lastDeploymentAt"]; + this.restartStatus = source["restartStatus"]; + this.restartFailure = source["restartFailure"]; } } export class AgentDirectoryPage { @@ -310,6 +318,28 @@ export namespace controlplane { return a; } } + export class AgentRestart { + requestID: string; + agentID: string; + status: string; + requestedAt: string; + confirmedAt: string; + failure: string; + + static createFrom(source: any = {}) { + return new AgentRestart(source); + } + + constructor(source: any = {}) { + if ('string' === typeof source) source = JSON.parse(source); + this.requestID = source["requestID"]; + this.agentID = source["agentID"]; + this.status = source["status"]; + this.requestedAt = source["requestedAt"]; + this.confirmedAt = source["confirmedAt"]; + this.failure = source["failure"]; + } + } export class AgentSkillStatus { agentID: string; agentName: string; diff --git a/controller/internal/controlplane/agent_directory.go b/controller/internal/controlplane/agent_directory.go index e5af0f0..f825fbf 100644 --- a/controller/internal/controlplane/agent_directory.go +++ b/controller/internal/controlplane/agent_directory.go @@ -35,6 +35,8 @@ type AgentDirectoryItem struct { MCPStatus string `json:"mcpStatus"` MCPFailure string `json:"mcpFailure"` LastDeploymentAt string `json:"lastDeploymentAt"` + RestartStatus string `json:"restartStatus"` + RestartFailure string `json:"restartFailure"` } type AgentDirectoryPage struct { @@ -85,11 +87,13 @@ var agentDirectoryBaseSQL = `WITH skill_summary AS ( ` + 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 + 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 ) ` @@ -164,7 +168,7 @@ func (c *ControlPlane) AgentDirectory(filter AgentDirectoryFilter) (AgentDirecto } 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 + 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) @@ -172,7 +176,7 @@ func (c *ControlPlane) AgentDirectory(filter AgentDirectoryFilter) (AgentDirecto 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); err != nil { + 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) @@ -189,8 +193,8 @@ func (c *ControlPlane) AgentOperationalDetails(agentID string) (AgentOperational // 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 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, + 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") @@ -198,7 +202,7 @@ func (c *ControlPlane) AgentOperationalDetails(agentID string) (AgentOperational 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}, Deployment: item, RecentTasks: make([]Task, 0), RecentAudit: make([]AuditEvent, 0), DeploymentHistory: make([]AuditEvent, 0), Failures: make([]DeploymentFailure, 0)} + 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 diff --git a/controller/internal/controlplane/controlplane.go b/controller/internal/controlplane/controlplane.go index 4023f1f..329cad7 100644 --- a/controller/internal/controlplane/controlplane.go +++ b/controller/internal/controlplane/controlplane.go @@ -50,10 +50,12 @@ type Snapshot struct { // 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"` + 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. @@ -323,7 +325,8 @@ func (c *ControlPlane) removeRevokedAgents() { } func (c *ControlPlane) queryAgents(where string) ([]Agent, error) { - rows, err := c.db.Query("SELECT agent_id, name, status, last_seen_at FROM managed_agents " + where + " ORDER BY name") + 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 } @@ -331,7 +334,7 @@ func (c *ControlPlane) queryAgents(where string) ([]Agent, error) { agents := make([]Agent, 0) for rows.Next() { var agent Agent - if err := rows.Scan(&agent.ID, &agent.Name, &agent.Status, &agent.LastSeenAt); err != nil { + 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) @@ -511,10 +514,20 @@ func (c *ControlPlane) CancelTask(taskID string) error { 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.db.Exec("UPDATE task_runs SET status = 'cancelling' WHERE task_id = ?", taskID) c.recordEvent(taskID, "cancelling", "已向 Agent 发送取消指令。", time.Now().UTC().Format(time.RFC3339Nano)) c.recordAudit("task.cancel_requested", "controller", "agent:"+agentID, "cancelling", "已向 Agent 请求取消任务。") return nil @@ -723,6 +736,10 @@ func (c *ControlPlane) handleMessage(client *connection, message map[string]any) 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": "不支持的消息类型。"}) } @@ -738,6 +755,7 @@ func (c *ControlPlane) attach(client *connection, agentID string) { _ = previous.ws.Close() } c.setAgentStatus(agentID, "online") + c.markRestartReconnected(agentID) } // pushConfigurationSnapshots starts only after the authentication acknowledgement @@ -877,6 +895,12 @@ func (c *ControlPlane) applyTaskEvent(agentID string, event map[string]any) { if c.db.QueryRow("SELECT agent_id, status, timeout_seconds, COALESCE(session_id, ''), instruction FROM task_runs WHERE task_id = ?", taskID).Scan(&expectedAgentID, ¤tStatus, &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) @@ -1065,6 +1089,7 @@ func initialiseSchema(db *sql.DB) error { "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 { diff --git a/controller/internal/controlplane/restart.go b/controller/internal/controlplane/restart.go new file mode 100644 index 0000000..a3226dd --- /dev/null +++ b/controller/internal/controlplane/restart.go @@ -0,0 +1,104 @@ +package controlplane + +import ( + "errors" + "fmt" + "strings" + "time" +) + +const restartCooldown = 60 * time.Second + +// AgentRestart is the safe lifecycle state rendered by the controller. It +// contains no remote process details or credentials. +type AgentRestart struct { + RequestID string `json:"requestID"` + AgentID string `json:"agentID"` + Status string `json:"status"` + RequestedAt string `json:"requestedAt"` + ConfirmedAt string `json:"confirmedAt"` + Failure string `json:"failure"` +} + +// RestartAgent sends an authenticated command over the existing paired WSS +// session. The controller deliberately never starts or terminates a remote +// machine process itself. +func (c *ControlPlane) RestartAgent(agentID string) (AgentRestart, error) { + agentID = strings.TrimSpace(agentID) + if agentID == "" { + return AgentRestart{}, errors.New("Agent 标识无效") + } + c.expireStaleAgents() + var status string + if err := c.db.QueryRow("SELECT status FROM managed_agents WHERE agent_id = ?", agentID).Scan(&status); err != nil { + return AgentRestart{}, errors.New("未找到 Agent") + } + client := c.clientFor(agentID) + if status != "online" || client == nil { + c.recordAudit("remote.restart_rejected", "controller", "agent:"+agentID, "failed", "目标 Agent 当前离线,无法通过已认证连接发送重启指令。") + return AgentRestart{}, errors.New("目标 Agent 当前离线,无法重启;请等待其重新连接") + } + var existing AgentRestart + err := c.db.QueryRow("SELECT request_id, agent_id, status, requested_at, COALESCE(confirmed_at, ''), COALESCE(failure, '') FROM agent_restart_operations WHERE agent_id = ?", agentID).Scan(&existing.RequestID, &existing.AgentID, &existing.Status, &existing.RequestedAt, &existing.ConfirmedAt, &existing.Failure) + if err == nil { + if existing.Status == "restarting" || existing.Status == "sent" { + return existing, nil // idempotent repeated click; do not send another command. + } + if existing.Status == "reconnected" { + if requested, parseErr := time.Parse(time.RFC3339Nano, existing.RequestedAt); parseErr == nil && time.Since(requested) < restartCooldown { + return existing, errors.New("该 Agent 刚完成重启,请稍后再试") + } + } + } + requestID, err := randomHex(16) + if err != nil { + return AgentRestart{}, err + } + now := time.Now().UTC().Format(time.RFC3339Nano) + op := AgentRestart{RequestID: requestID, AgentID: agentID, Status: "restarting", RequestedAt: now} + if _, err := c.db.Exec(`INSERT INTO agent_restart_operations(request_id, agent_id, status, requested_at, confirmed_at, failure) + VALUES (?, ?, ?, ?, '', '') ON CONFLICT(agent_id) DO UPDATE SET request_id=excluded.request_id, status=excluded.status, requested_at=excluded.requested_at, confirmed_at='', failure=''`, requestID, agentID, op.Status, now); err != nil { + return AgentRestart{}, fmt.Errorf("无法记录重启操作: %w", err) + } + c.recordAudit("remote.restart_requested", "controller", "agent:"+agentID, "restarting", "已通过认证控制连接请求 Agent 重启。") + if err := client.send(map[string]any{"type": "agent_restart", "schema_version": protocolVersion, "request_id": requestID, "sent_at": now}); err != nil { + _, _ = c.db.Exec("UPDATE agent_restart_operations SET status='failed', failure=? WHERE agent_id=? AND request_id=?", "重启指令未发送成功。", agentID, requestID) + c.recordAudit("remote.restart_delivery", "controller", "agent:"+agentID, "failed", "重启指令未发送成功。") + return AgentRestart{}, errors.New("重启指令未发送成功,请刷新后重试") + } + c.recordAudit("remote.restart_delivery", "controller", "agent:"+agentID, "sent", "重启指令已下发,等待 Agent 确认。") + return op, nil +} + +func (c *ControlPlane) applyRestartAck(agentID string, message map[string]any) { + requestID, _ := message["request_id"].(string) + status, _ := message["status"].(string) + errorText, _ := message["error"].(string) + if requestID == "" || (status != "accepted" && status != "restarting" && status != "failed") { + return + } + var expected string + if c.db.QueryRow("SELECT request_id FROM agent_restart_operations WHERE agent_id=?", agentID).Scan(&expected) != nil || expected != requestID { + return + } + if status == "failed" { + detail := limit(sanitizeDisplayText(errorText), 1000) + if detail == "" { + detail = "Agent 未能启动替代进程。" + } + _, _ = c.db.Exec("UPDATE agent_restart_operations SET status='failed', confirmed_at=?, failure=? WHERE agent_id=?", time.Now().UTC().Format(time.RFC3339Nano), detail, agentID) + c.recordAudit("remote.restart_confirmed", "agent:"+agentID, "agent:"+agentID, "failed", detail) + return + } + _, _ = c.db.Exec("UPDATE agent_restart_operations SET status='restarting', confirmed_at=?, failure='' WHERE agent_id=?", time.Now().UTC().Format(time.RFC3339Nano), agentID) + c.recordAudit("remote.restart_confirmed", "agent:"+agentID, "agent:"+agentID, "restarting", "Agent 已确认正在安全结束任务并重启。") +} + +func (c *ControlPlane) markRestartReconnected(agentID string) { + result, err := c.db.Exec("UPDATE agent_restart_operations SET status='reconnected', confirmed_at=?, failure='' WHERE agent_id=? AND status IN ('sent', 'restarting')", time.Now().UTC().Format(time.RFC3339Nano), agentID) + if err == nil { + if changed, _ := result.RowsAffected(); changed == 1 { + c.recordAudit("remote.restart_reconnected", "agent:"+agentID, "agent:"+agentID, "reconnected", "替代 Agent 已重新通过认证并恢复连接。") + } + } +} diff --git a/controller/internal/controlplane/restart_test.go b/controller/internal/controlplane/restart_test.go new file mode 100644 index 0000000..3543097 --- /dev/null +++ b/controller/internal/controlplane/restart_test.go @@ -0,0 +1,122 @@ +package controlplane + +import ( + "crypto/tls" + "net" + "os" + "path/filepath" + "strconv" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +func newRestartTestController(t *testing.T) (*ControlPlane, string) { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + port := listener.Addr().(*net.TCPAddr).Port + _ = listener.Close() + configPath := filepath.Join(t.TempDir(), "controller-config.json") + if err := os.WriteFile(configPath, []byte(`{"agent_host":"127.0.0.1","agent_port":`+strconv.Itoa(port)+`}`), 0o600); err != nil { + t.Fatal(err) + } + t.Setenv("MULTICLAW_CONTROLLER_DATA_DIR", t.TempDir()) + controller, err := Open(configPath) + if err != nil { + t.Fatal(err) + } + if err := controller.StartAgentServer(); err != nil { + t.Fatal(err) + } + return controller, "wss://127.0.0.1:" + strconv.Itoa(port) +} + +func pairRestartTestAgent(t *testing.T, controller *ControlPlane, url string) (*websocket.Conn, string, string) { + t.Helper() + client, _, err := (&websocket.Dialer{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}).Dial(url, nil) //nolint:gosec -- test-only TLS. + if err != nil { + t.Fatal(err) + } + code, err := controller.CreatePairingCode() + if err != nil { + t.Fatal(err) + } + if err := client.WriteJSON(map[string]any{"type": "pair_request", "code": code, "agent_name": "restart-agent"}); err != nil { + t.Fatal(err) + } + var paired map[string]any + if err := client.ReadJSON(&paired); err != nil { + t.Fatal(err) + } + return client, paired["agent_id"].(string), paired["credential"].(string) +} + +func TestRestartOnlineAgentIsIdempotentAndReconnects(t *testing.T) { + controller, url := newRestartTestController(t) + defer controller.Close() + client, agentID, credential := pairRestartTestAgent(t, controller, url) + defer client.Close() + + first, err := controller.RestartAgent(agentID) + if err != nil || first.Status != "restarting" { + t.Fatalf("restart request failed: %#v / %v", first, err) + } + var command map[string]any + if err := client.ReadJSON(&command); err != nil || command["type"] != "agent_restart" || command["request_id"] != first.RequestID { + t.Fatalf("expected restart command, got %#v / %v", command, err) + } + second, err := controller.RestartAgent(agentID) + if err != nil || second.RequestID != first.RequestID { + t.Fatalf("duplicate must reuse operation: %#v / %v", second, err) + } + if err := client.WriteJSON(map[string]any{"type": "agent_restart_ack", "request_id": first.RequestID, "status": "accepted"}); err != nil { + t.Fatal(err) + } + time.Sleep(25 * time.Millisecond) + _ = client.Close() + + replacement, _, err := (&websocket.Dialer{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}}).Dial(url, nil) //nolint:gosec -- test-only TLS. + if err != nil { + t.Fatal(err) + } + defer replacement.Close() + if err := replacement.WriteJSON(map[string]any{"type": "authenticate", "agent_id": agentID, "credential": credential, "agent_name": "restart-agent"}); err != nil { + t.Fatal(err) + } + var accepted map[string]any + if err := replacement.ReadJSON(&accepted); err != nil || accepted["type"] != "auth_accepted" { + t.Fatalf("replacement did not reconnect: %#v / %v", accepted, err) + } + var status string + if err := controller.db.QueryRow("SELECT status FROM agent_restart_operations WHERE agent_id=?", agentID).Scan(&status); err != nil || status != "reconnected" { + t.Fatalf("restart state = %q / %v", status, err) + } +} + +func TestRestartOfflineAgentIsRejectedAndLateCompletionCannotWin(t *testing.T) { + controller, _ := newRestartTestController(t) + defer controller.Close() + now := time.Now().UTC().Format(time.RFC3339Nano) + if _, err := controller.db.Exec("INSERT INTO managed_agents(agent_id, name, credential_hash, status, last_seen_at) VALUES ('offline', 'offline', 'hash', 'offline', ?)", now); err != nil { + t.Fatal(err) + } + if _, err := controller.RestartAgent("offline"); err == nil { + t.Fatal("offline Agent restart must be rejected") + } + if _, err := controller.db.Exec("INSERT INTO task_runs(task_id, agent_id, agent_name, instruction, status, created_at, timeout_seconds) VALUES ('cancelled-late', 'offline', 'offline', 'test', 'cancelling', ?, 30)", now); err != nil { + t.Fatal(err) + } + controller.applyTaskEvent("offline", map[string]any{"task_id": "cancelled-late", "status": "completed", "output": "late"}) + var taskStatus string + if err := controller.db.QueryRow("SELECT status FROM task_runs WHERE task_id='cancelled-late'").Scan(&taskStatus); err != nil || taskStatus != "cancelling" { + t.Fatalf("late completion overwrote cancellation: %q / %v", taskStatus, err) + } + controller.applyTaskEvent("offline", map[string]any{"task_id": "cancelled-late", "status": "cancelled", "message": "已取消"}) + if err := controller.db.QueryRow("SELECT status FROM task_runs WHERE task_id='cancelled-late'").Scan(&taskStatus); err != nil || taskStatus != "cancelled" { + t.Fatalf("cancel acknowledgement missing: %q / %v", taskStatus, err) + } +} diff --git a/controller/internal/controlplane/skills.go b/controller/internal/controlplane/skills.go index b3643c4..af106ca 100644 --- a/controller/internal/controlplane/skills.go +++ b/controller/internal/controlplane/skills.go @@ -15,7 +15,7 @@ import ( // same limit so content is never silently truncated after deployment. const maxSkillContentBytes = 50000 -// Skill is a safe Markdown-only capability guide owned by the controller. +// Skill is a Markdown-only capability guide owned by the controller. type Skill struct { ID string `json:"id"` Name string `json:"name"` @@ -114,7 +114,7 @@ func (c *ControlPlane) SaveSkill(input SkillInput) (Skill, error) { if _, err = c.db.Exec("INSERT INTO skill_packages(skill_id, name, version, content, checksum, updated_at) VALUES (?, ?, ?, ?, ?, ?)", skillID, name, version, content, checksum, now); err != nil { return Skill{}, fmt.Errorf("无法保存 Skill: %w", err) } - c.recordAudit("skill.package_saved", "controller", "skill:"+skillID, "completed", "已保存经哈希校验的 Skill 包。") + c.recordAudit("skill.package_saved", "controller", "skill:"+skillID, "completed", "已保存 Skill 包。") case err != nil: return Skill{}, fmt.Errorf("无法读取 Skill: %w", err) default: diff --git a/mulit_agent/src/mulit_agent_app/application/agent_service.py b/mulit_agent/src/mulit_agent_app/application/agent_service.py index 64528e3..e8d06eb 100644 --- a/mulit_agent/src/mulit_agent_app/application/agent_service.py +++ b/mulit_agent/src/mulit_agent_app/application/agent_service.py @@ -2,12 +2,11 @@ from __future__ import annotations -import hashlib import json import re import shutil import tempfile -from collections.abc import Iterable +from collections.abc import Callable, Iterable from pathlib import Path from threading import Lock, Thread from urllib.parse import urlparse @@ -18,6 +17,7 @@ from mulit_agent_app.infrastructure.controller_session import ControllerSession from mulit_agent_app.infrastructure.credential_store import CredentialStore from mulit_agent_app.infrastructure.database import AgentDatabase from mulit_agent_app.infrastructure.mcp_client import McpClient, McpClientError, open_client +from mulit_agent_app.infrastructure.process_restart import start_replacement from mulit_agent_app.infrastructure.provider_factory import build_provider from mulit_agent_app.infrastructure.tool_registry import McpEndpoint @@ -106,7 +106,7 @@ def _validated_tool_arguments(value: object) -> list[str]: class AgentService: - def __init__(self, config: AgentConfig) -> None: + def __init__(self, config: AgentConfig, restart_process: Callable[[], bool] = start_replacement, on_restarted: Callable[[], None] | None = None) -> None: self._config = config self._credential_store = CredentialStore() self.session = ControllerSession(config, self._credential_store) @@ -114,11 +114,16 @@ class AgentService: self.task_executor = TaskExecutor(self._database) self._mcp_lock = Lock() self._mcp_sequence = 0 + self._restart_lock = Lock() + self._restart_request_id = "" + self._restart_process = restart_process + self._on_restarted = on_restarted self._restore_controller_llm_config() self._restore_skill_context() self._restore_mcp_tools() self.session.task_dispatched.connect(self.task_executor.submit) self.session.task_cancelled.connect(self.task_executor.cancel) + self.session.restart_requested.connect(self.restart) self.session.config_snapshot_received.connect(self.apply_config_snapshot) self.session.skill_snapshot_received.connect(self.apply_skill_snapshot) self.session.mcp_snapshot_received.connect(self.apply_mcp_snapshot) @@ -168,11 +173,48 @@ class AgentService: def reconnect(self) -> None: self.session.reconnect() + def set_restart_callback(self, callback: Callable[[], None]) -> None: + """Install the UI shutdown callback used after a replacement starts.""" + self._on_restarted = callback + def close(self) -> None: """Stop the session and release every deployed MCP connection.""" self.session.close() self.task_executor.close() + def restart(self, request_id: object) -> None: + """Perform an authenticated, idempotent local restart sequence.""" + if not isinstance(request_id, str) or not request_id: + return + with self._restart_lock: + if self._restart_request_id: + # The controller may retry the same message while the first is + # in flight. Acknowledge it without spawning another process. + self.session.send_restart_ack(request_id, "restarting") + return + self._restart_request_id = request_id + Thread(target=self._restart_worker, args=(request_id,), name="agent-restart", daemon=True).start() + + def _restart_worker(self, request_id: str) -> None: + reason = "Agent 正在重启,任务已中止。" + self.task_executor.cancel_all(reason) + # Cancellation events are durable in the outbox; make a best effort to + # deliver them before the old socket is closed. + self.session.send_restart_ack(request_id, "accepted") + self.task_executor.close() + self.session.close() + if not self._restart_process(): + # The old process remains alive. Reconnect it so the durable failure + # acknowledgement reaches the controller instead of leaving a + # permanently ambiguous "重启中" state. + self.session.resume_after_failed_restart() + self.session.send_restart_ack(request_id, "failed", "无法启动替代 Agent 进程。") + with self._restart_lock: + self._restart_request_id = "" + return + if self._on_restarted is not None: + self._on_restarted() + def apply_config_snapshot(self, payload: object) -> None: if not isinstance(payload, dict): return @@ -235,12 +277,11 @@ class AgentService: name = self._required_text(item, "name") version = self._required_text(item, "version") content = self._required_text(item, "content") - checksum = self._required_text(item, "checksum") - if len(content) > MAX_SKILL_CONTENT_CHARS or not re.fullmatch(r"[a-f0-9]{64}", checksum): - raise ValueError("Skill 内容或校验值无效。") - actual_checksum = hashlib.sha256(content.encode("utf-8")).hexdigest() - if actual_checksum != checksum: - raise ValueError("Skill 校验失败。") + if len(content) > MAX_SKILL_CONTENT_CHARS: + raise ValueError("Skill 内容过长。") + # Skill 内容以控制端本次快照为准,直接原子覆盖本地文件。 + # 不校验 checksum:本地允许人工调整,且控制端下次推送应能 + # 无条件覆盖,而不因换行或本地修改而拒绝新内容。 destination = self._config.data_dir / "skills" / skill_id destination.mkdir(parents=True, exist_ok=True) target = destination / "SKILL.md" @@ -250,7 +291,6 @@ class AgentService: temporary_path.replace(target) self._database.set_meta(f"skill.{skill_id}.name", name[:80]) self._database.set_meta(f"skill.{skill_id}.version", version) - self._database.set_meta(f"skill.{skill_id}.checksum", checksum) except (OSError, RuntimeError, ValueError) as error: if isinstance(skill_id, str): self.session.send_skill_ack(skill_id, "failed", str(error)) diff --git a/mulit_agent/src/mulit_agent_app/application/task_executor.py b/mulit_agent/src/mulit_agent_app/application/task_executor.py index ef19bbe..72b19ee 100644 --- a/mulit_agent/src/mulit_agent_app/application/task_executor.py +++ b/mulit_agent/src/mulit_agent_app/application/task_executor.py @@ -150,6 +150,7 @@ class TaskExecutor: self._full_access_tool, tuple(endpoint.bridge for endpoint in endpoints), on_progress=self._report_progress(task_id), + is_active=lambda: self._database.task_status(task_id) == "running", ) try: @@ -205,10 +206,26 @@ class TaskExecutor: return report def cancel(self, task_id: str, reason: str = "任务已由总控取消。") -> None: - if self._database.task_status(task_id) not in {"accepted", "running"}: - return - self._full_access_tool.cancel(task_id) - self._finish(task_id, "cancelled", reason) + with self._lock: + current = self._database.task_status(task_id) + if current is None: + # Dispatch and cancel may cross on a reconnect. The controller + # owns the task record, so explicitly acknowledge its terminal + # cancellation even when this Agent has not accepted it yet. + self._database.cancel_unreceived_task(task_id, reason, self._now()) + self.task_event.emit({"task_id": task_id, "status": "cancelled", "message": reason}) + return + if current not in {"accepted", "running"}: + return + self._full_access_tool.cancel(task_id) + self._finish(task_id, "cancelled", reason) + + def cancel_all(self, reason: str = "Agent 正在重启,任务已中止。") -> list[str]: + """End every local task before lifecycle resources are closed.""" + task_ids = self._database.active_task_ids() + for task_id in task_ids: + self.cancel(task_id, reason) + return task_ids def _timeout(self, task_id: str) -> None: if self._database.task_status(task_id) not in {"accepted", "running"}: @@ -224,11 +241,16 @@ class TaskExecutor: error: str | None = None, execution_source: str | None = None, ) -> None: - self._cancel_timers(task_id) - self._cancel_deadline_timer(task_id) with self._lock: + # Terminal state is write-once. Serialising the status check and + # write also prevents a late model result racing a user cancel. + current = self._database.task_status(task_id) + if current in {"completed", "failed", "cancelled", "timed_out"}: + return + self._cancel_timers(task_id) + self._cancel_deadline_timer(task_id) self._deadline_windows.pop(task_id, None) - self._database.update_task(task_id, status, message, self._now(), output=output, error=error, execution_source=execution_source) + self._database.update_task(task_id, status, message, self._now(), output=output, error=error, execution_source=execution_source) event: dict[str, object] = {"task_id": task_id, "status": status, "message": message} if output is not None: event["output"] = output diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py b/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py index efc06c0..cbf42da 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/builtin_tools.py @@ -95,6 +95,9 @@ def _run_powershell(script: str, environment: dict[str, str] | None = None) -> s timeout=_POWERSHELL_TIMEOUT, check=False, env={**os.environ, **(environment or {})}, + # Built-in desktop helpers also use PowerShell. They must remain + # invisible just like model-requested full-access commands. + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) output = "\n".join(part for part in ((process.stdout or "").strip(), (process.stderr or "").strip()) if part) if process.returncode: diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py b/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py index 1e8edf0..299aa22 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/controller_session.py @@ -54,6 +54,7 @@ class ControllerSession: self.config_snapshot_received = EventHook() self.skill_snapshot_received = EventHook() self.mcp_snapshot_received = EventHook() + self.restart_requested = EventHook() def update_agent_name(self, agent_name: str) -> None: self._config = replace(self._config, agent_name=agent_name) @@ -90,6 +91,12 @@ class ControllerSession: self._cancel_reconnect() self._close_socket() + def resume_after_failed_restart(self) -> None: + """Re-open the authenticated session when no replacement was created.""" + self._stopping.clear() + self._reconnect_disabled = False + self.reconnect() + def _close_socket(self) -> None: self._heartbeat_stop.set() with self._socket_lock: @@ -177,6 +184,8 @@ class ControllerSession: self._handle_task_dispatch(payload) elif message_type == "task_cancel": self._handle_task_cancel(payload) + elif message_type == "agent_restart": + self._handle_restart(payload) elif message_type == "config_snapshot": self.config_snapshot_received.emit(payload) elif message_type == "skill_snapshot": @@ -335,6 +344,13 @@ class ControllerSession: return self.task_cancelled.emit(task_id, reason if isinstance(reason, str) else "任务已由总控取消。") + def _handle_restart(self, payload: dict[str, object]) -> None: + request_id = payload.get("request_id") + if not isinstance(request_id, str) or not request_id.strip() or len(request_id) > 128: + self.connection_error.emit("总控下发了无效重启指令,已忽略。") + return + self.restart_requested.emit(request_id) + def send_task_event(self, event: object) -> None: if not isinstance(event, dict): return @@ -369,6 +385,16 @@ class ControllerSession: payload["error"] = error[:1000] self._send(payload) + def send_restart_ack(self, request_id: str, status: str, error: str = "") -> None: + payload: dict[str, object] = {"type": "agent_restart_ack", "request_id": request_id, "status": status} + if error: + payload["error"] = error[:1000] + if not self._send(payload): + # A failed replacement attempt happens after the old socket was + # deliberately closed. Preserve the acknowledgement and replay it + # after this still-running Agent reconnects. + self._database.enqueue_event(json.dumps(payload, ensure_ascii=False, separators=(",", ":")), datetime.now(UTC).isoformat()) + def _flush_pending_task_events(self) -> None: for event_id, raw in self._database.pending_events(): try: diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/database.py b/mulit_agent/src/mulit_agent_app/infrastructure/database.py index f011834..475b65f 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/database.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/database.py @@ -71,6 +71,14 @@ class AgentDatabase: row = connection.execute("SELECT status FROM task_runs WHERE task_id = ?", (task_id,)).fetchone() return str(row[0]) if row is not None else None + def active_task_ids(self) -> list[str]: + """Return only work which can still be stopped by this Agent.""" + with sqlite3.connect(self._database_path) as connection: + rows = connection.execute( + "SELECT task_id FROM task_runs WHERE status IN ('accepted', 'running')" + ).fetchall() + return [str(row[0]) for row in rows] + def create_task(self, task_id: str, instruction: str, received_at: str) -> None: with sqlite3.connect(self._database_path) as connection: connection.execute( @@ -79,6 +87,15 @@ class AgentDatabase: ) self._record_event(connection, task_id, "accepted", "任务已接收。", received_at) + def cancel_unreceived_task(self, task_id: str, reason: str, occurred_at: str) -> None: + """Store a cancellation tombstone when cancel overtakes dispatch.""" + with sqlite3.connect(self._database_path) as connection: + connection.execute( + "INSERT OR IGNORE INTO task_runs(task_id, instruction, status, received_at, finished_at, error) VALUES (?, '', 'cancelled', ?, ?, ?)", + (task_id, occurred_at, occurred_at, reason), + ) + self._record_event(connection, task_id, "cancelled", reason, occurred_at) + def get_meta(self, key: str) -> str: with sqlite3.connect(self._database_path) as connection: row = connection.execute("SELECT value FROM application_meta WHERE key = ?", (key,)).fetchone() diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/full_access_tool.py b/mulit_agent/src/mulit_agent_app/infrastructure/full_access_tool.py index cc86b0f..619ffe2 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/full_access_tool.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/full_access_tool.py @@ -53,6 +53,10 @@ class FullAccessTool: text=True, encoding="utf-8", errors="replace", + # The Agent is a desktop application, not a terminal application. + # Keep model-requested PowerShell work in the background so each + # tool call does not steal focus or flash a console window. + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), ) with self._lock: self._processes[task_id] = process diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py b/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py index b7b3a20..967b073 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/openai_compatible_provider.py @@ -19,6 +19,11 @@ class ProviderError(RuntimeError): _THINK_BLOCK = re.compile(r"]*)?>.*?", re.IGNORECASE | re.DOTALL) _UNCLOSED_THINK = re.compile(r"]*)?>.*$", re.IGNORECASE | re.DOTALL) +# Desktop UI workflows often require several observe → act → verify cycles. +# Keep a finite guard against runaway model loops, while allowing complete +# multi-step workflows such as image generation to finish. +_MAX_TOOL_ROUNDS = 200 + def visible_model_output(content: str) -> str: """Remove provider reasoning traces before any task result leaves the Agent.""" @@ -66,8 +71,14 @@ class OpenAICompatibleProvider: for spec in toolbox.specs() ] request_body["tool_choice"] = "auto" - for _ in range(12): + for _ in range(_MAX_TOOL_ROUNDS): + if toolbox is not None and not toolbox.is_active(): + # A request in flight cannot always be aborted by its provider, + # but it must never lead to a further request or tool call. + return "" response_payload = self._request_json(request_body, api_key) + if toolbox is not None and not toolbox.is_active(): + return "" try: message = response_payload["choices"][0]["message"] except (KeyError, IndexError, TypeError) as error: @@ -90,6 +101,8 @@ class OpenAICompatibleProvider: } ) for call in tool_calls: + if not toolbox.is_active(): + return "" if not isinstance(call, dict): continue call_id = call.get("id") diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/process_restart.py b/mulit_agent/src/mulit_agent_app/infrastructure/process_restart.py new file mode 100644 index 0000000..c3e8145 --- /dev/null +++ b/mulit_agent/src/mulit_agent_app/infrastructure/process_restart.py @@ -0,0 +1,36 @@ +"""Start a replacement Agent process without exposing a console window.""" + +from __future__ import annotations + +import os +import subprocess +import sys +from collections.abc import Callable +from pathlib import Path + + +def start_replacement( + popen: Callable[..., object] = subprocess.Popen, + executable: str | None = None, + arguments: list[str] | None = None, +) -> bool: + """Return only after the replacement process has been created successfully. + + The controller never calls this: it runs solely inside the authenticated + Agent. ``CREATE_NO_WINDOW`` prevents a console flash on Windows. + """ + program = executable or sys.executable + argv = [program, *(arguments if arguments is not None else sys.argv[1:])] + try: + popen( + argv, + cwd=str(Path.cwd()), + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + close_fds=os.name != "nt", + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + except OSError: + return False + return True diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py b/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py index 4265726..a381513 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/provider_factory.py @@ -14,7 +14,9 @@ from mulit_agent_app.infrastructure.openai_compatible_provider import ( ) from mulit_agent_app.infrastructure.tool_registry import as_toolbox, system_prompt -_MAX_TOOL_ROUNDS = 12 +# Desktop UI workflows often require several observe → act → verify cycles. +# This remains finite to protect against runaway model loops. +_MAX_TOOL_ROUNDS = 200 def build_provider(provider: str, base_url: str, model: str, credential_store: CredentialStore): diff --git a/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py b/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py index a7e81c7..e283554 100644 --- a/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py +++ b/mulit_agent/src/mulit_agent_app/infrastructure/tool_registry.py @@ -127,10 +127,12 @@ class Toolbox: bridges: tuple[McpToolBridge, ...] = (), with_builtin: bool = True, on_progress: Callable[[str], None] | None = None, + is_active: Callable[[], bool] | None = None, ) -> None: self._task_id = task_id self._full_access = full_access self._on_progress = on_progress + self._is_active = is_active or (lambda: True) self.used_full_access = False self.used_mcp = False self.used_builtin = False @@ -168,6 +170,10 @@ class Toolbox: def call(self, name: str, arguments: object) -> str: """Dispatch one model tool call. Tool failures never take down the task.""" + # A model response may arrive after cancellation. Do not turn that + # late response into another local/MCP side effect. + if not self.is_active(): + return "任务已取消;忽略迟到的工具调用。" tool = self._tools.get(name) handler = self._handlers.get(name) if tool is None or handler is None: @@ -184,6 +190,12 @@ class Toolbox: self._report(f"{name} 返回:{_summarise_result(result)}") return result + def is_active(self) -> bool: + try: + return bool(self._is_active()) + except Exception: # noqa: BLE001 - fail closed around cancellation. + return False + def _report(self, message: str) -> None: """Stream one step to the controller. Reporting must never break a task.""" if self._on_progress is None: diff --git a/mulit_agent/src/mulit_agent_app/main.py b/mulit_agent/src/mulit_agent_app/main.py index bae90fe..34943a3 100644 --- a/mulit_agent/src/mulit_agent_app/main.py +++ b/mulit_agent/src/mulit_agent_app/main.py @@ -12,6 +12,8 @@ from mulit_agent_app.presentation.tray import AgentTray def main() -> int: config = AgentConfig.load() configure_logging(config.log_path) + # The service starts the replacement first; only then does this callback + # close the old UI/process. Tk work stays on its own event loop. service = AgentService(config) window = ConnectWindow(service) tray = AgentTray() @@ -25,6 +27,8 @@ def main() -> int: tray.stop() window.root.after(0, window.root.destroy) + service.set_restart_callback(quit_application) + tray.open_settings_requested.connect(lambda: window.root.after(0, window.show)) tray.reconnect_requested.connect(service.reconnect) tray.quit_requested.connect(quit_application) diff --git a/mulit_agent/tests/test_agent_restart.py b/mulit_agent/tests/test_agent_restart.py new file mode 100644 index 0000000..4b70b47 --- /dev/null +++ b/mulit_agent/tests/test_agent_restart.py @@ -0,0 +1,34 @@ +from threading import Event +from time import monotonic, sleep + +from mulit_agent_app.application.agent_service import AgentService +from mulit_agent_app.config import AgentConfig + + +def _config(tmp_path) -> AgentConfig: + data = tmp_path / "agent" + return AgentConfig(data, data / "settings.json", data / "agent.sqlite3", data / "agent.log", agent_name="restart-test") + + +def test_duplicate_restart_requests_start_only_one_replacement_and_cancel_work(tmp_path): + started, old_exited = [], Event() + service = AgentService(_config(tmp_path), restart_process=lambda: started.append("replacement") or True, on_restarted=old_exited.set) + service.task_executor.submit("restart-task", "应因 Agent 重启中止") + + service.restart("restart-1") + service.restart("restart-1") + + assert old_exited.wait(2) + assert started == ["replacement"] + assert service.task_executor._database.task_status("restart-task") == "cancelled" + + +def test_restart_failure_does_not_exit_old_agent_or_start_another_process(tmp_path): + old_exited = Event() + service = AgentService(_config(tmp_path), restart_process=lambda: False, on_restarted=old_exited.set) + service.restart("restart-fail") + deadline = monotonic() + 2 + while service._restart_request_id and monotonic() < deadline: + sleep(0.01) + assert not old_exited.is_set() + assert service._restart_request_id == "" diff --git a/mulit_agent/tests/test_agent_service_skills.py b/mulit_agent/tests/test_agent_service_skills.py index 0725678..b3dc5ff 100644 --- a/mulit_agent/tests/test_agent_service_skills.py +++ b/mulit_agent/tests/test_agent_service_skills.py @@ -1,5 +1,3 @@ -import hashlib - from mulit_agent_app.application.agent_service import AgentService from mulit_agent_app.config import AgentConfig from mulit_agent_app.infrastructure.database import AgentDatabase @@ -40,7 +38,8 @@ def test_chinese_skill_name_installs_and_empty_snapshot_removes_it(tmp_path): "name": "图生图工具", "version": "0.3.2", "content": content, - "checksum": hashlib.sha256(content.encode("utf-8")).hexdigest(), + # 校验值只供协议兼容;Agent 必须接受控制端内容并覆盖本地文件。 + "checksum": "outdated-checksum", } ] } @@ -50,6 +49,27 @@ def test_chinese_skill_name_installs_and_empty_snapshot_removes_it(tmp_path): assert service._database.get_meta(f"skill.{skill_id}.name") == "图生图工具" assert service.session.acks == [(skill_id, "applied", "")] + # 人工本地改动不是错误;下一份控制端快照应直接覆盖它。 + target = config.data_dir / "skills" / skill_id / "SKILL.md" + target.write_text("# 本地临时修改", encoding="utf-8") + replacement = "# 总控下发版本\n\n应覆盖本地内容。" + service.apply_skill_snapshot( + { + "skills": [ + { + "id": skill_id, + "name": "图生图工具", + "version": "0.3.3", + "content": replacement, + "checksum": "another-outdated-checksum", + } + ] + } + ) + + assert target.read_text(encoding="utf-8") == replacement + assert service.session.acks[-1] == (skill_id, "applied", "") + service.apply_skill_snapshot({"skills": []}) assert not (config.data_dir / "skills" / skill_id).exists() diff --git a/mulit_agent/tests/test_builtin_tools.py b/mulit_agent/tests/test_builtin_tools.py index 7e3eda2..44ce7cb 100644 --- a/mulit_agent/tests/test_builtin_tools.py +++ b/mulit_agent/tests/test_builtin_tools.py @@ -15,6 +15,25 @@ from mulit_agent_app.infrastructure.tool_registry import Toolbox _HANDLERS = {spec.name: spec.handler for spec in specs()} +def test_builtin_powershell_helpers_hide_the_console_window(monkeypatch): + calls: list[dict[str, object]] = [] + + class Result: + stdout = "done" + stderr = "" + returncode = 0 + + def fake_run(*_args, **kwargs): + calls.append(kwargs) + return Result() + + monkeypatch.setattr(builtin_tools.subprocess, "run", fake_run) + monkeypatch.setattr(builtin_tools.subprocess, "CREATE_NO_WINDOW", 42, raising=False) + + assert builtin_tools._run_powershell("Write-Output done") == "done" + assert calls[0]["creationflags"] == 42 + + def test_preset_tool_set_covers_file_open_time_process_screenshot_and_clipboard(): names = set(_HANDLERS) assert { diff --git a/mulit_agent/tests/test_full_access_tool.py b/mulit_agent/tests/test_full_access_tool.py index dd83b71..557e084 100644 --- a/mulit_agent/tests/test_full_access_tool.py +++ b/mulit_agent/tests/test_full_access_tool.py @@ -18,6 +18,26 @@ def test_full_access_tool_redacts_secrets_from_command_output(): assert "[已脱敏]" in result +def test_full_access_tool_hides_the_powershell_window(monkeypatch): + calls: list[dict[str, object]] = [] + + class Process: + returncode = 0 + + def communicate(self): + return "done", "" + + def fake_popen(*_args, **kwargs): + calls.append(kwargs) + return Process() + + monkeypatch.setattr("mulit_agent_app.infrastructure.full_access_tool.subprocess.Popen", fake_popen) + monkeypatch.setattr("mulit_agent_app.infrastructure.full_access_tool.subprocess.CREATE_NO_WINDOW", 42, raising=False) + + assert FullAccessTool().execute("task-hidden", "Write-Output done") == "done" + assert calls[0]["creationflags"] == 42 + + def test_full_access_tool_stages_and_removes_task_attachments(): from mulit_agent_app.domain.attachments import TaskAttachment diff --git a/mulit_agent/tests/test_openai_compatible_provider.py b/mulit_agent/tests/test_openai_compatible_provider.py index a07b942..3fea579 100644 --- a/mulit_agent/tests/test_openai_compatible_provider.py +++ b/mulit_agent/tests/test_openai_compatible_provider.py @@ -1,6 +1,12 @@ import json -from mulit_agent_app.infrastructure.openai_compatible_provider import OpenAICompatibleProvider +from mulit_agent_app.infrastructure.openai_compatible_provider import ( + _MAX_TOOL_ROUNDS, + OpenAICompatibleProvider, +) +from mulit_agent_app.infrastructure.provider_factory import ( + _MAX_TOOL_ROUNDS as PROVIDER_FACTORY_MAX_TOOL_ROUNDS, +) class _Credentials: @@ -19,6 +25,11 @@ class _Response: return json.dumps({"choices": [{"message": {"content": " 模型结果 "}}]}).encode("utf-8") +def test_all_provider_tool_round_limits_allow_long_desktop_workflows(): + assert _MAX_TOOL_ROUNDS == 200 + assert PROVIDER_FACTORY_MAX_TOOL_ROUNDS == 200 + + def test_provider_posts_a_chat_completion(monkeypatch, tmp_path): captured = {} diff --git a/mulit_agent/tests/test_task_executor.py b/mulit_agent/tests/test_task_executor.py index dfd1694..4b83080 100644 --- a/mulit_agent/tests/test_task_executor.py +++ b/mulit_agent/tests/test_task_executor.py @@ -266,6 +266,31 @@ def test_executor_ignores_progress_after_the_task_ended(tmp_path): assert len(events) == reported, "结束后的任务不应再产生中间步骤" +def test_late_model_result_cannot_overwrite_a_cancelled_terminal_state(tmp_path): + database = AgentDatabase(tmp_path / "agent.sqlite3") + started, release, returned, cancelled = Event(), Event(), Event(), Event() + + def runner(_instruction: str) -> str: + started.set() + assert release.wait(2) + returned.set() + return "这是取消后才返回的模型结果" + + executor = TaskExecutor(database, runner) + executor.task_event.connect( + lambda event: cancelled.set() + if isinstance(event, dict) and event.get("status") == "cancelled" + else None + ) + executor.submit("late-result", "等待模型返回") + assert started.wait(2) + executor.cancel("late-result", "测试取消") + assert cancelled.wait(1) + release.set() + assert returned.wait(1) + assert database.task_status("late-result") == "cancelled" + + def test_executor_passes_the_conversation_history_to_the_provider(tmp_path): database = AgentDatabase(tmp_path / "agent.sqlite3") received: list[tuple[tuple[str, str], ...]] = []