From d534865b33794ecaec83f3963999def1c2ce8ae5 Mon Sep 17 00:00:00 2001 From: LLMGuardX Dev Date: Thu, 13 Aug 2026 14:30:37 +0800 Subject: [PATCH] =?UTF-8?q?0.11.4:=20=E6=97=97=E8=88=B0=E7=89=88=E7=AC=AC?= =?UTF-8?q?=E4=BA=94=E8=BD=AE=E5=AE=8C=E5=96=84(=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E4=BD=93=E8=8A=82=E7=82=B9=E4=BB=BB=E5=8A=A1=E4=B8=8B=E5=8F=91?= =?UTF-8?q?/=E4=B8=AA=E4=BA=BA=E6=99=BA=E8=83=BD=E4=BD=93=E5=AE=89?= =?UTF-8?q?=E5=85=A8=E7=AD=96=E7=95=A5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 节点任务:管理端向节点池下发(Prompt/HTTP/MCP/Skill/数字员工/自定义), 指定节点或池路由,认领 SKIP LOCKED + 15 分钟租约,认领令牌防重放上报, 失败 30s×次数退避重入队,达上限 failed,支持取消/重试,完成与失败站内信。 - 个人智能体安全策略:auto_approve_tools 跳过个人调用审批门; rate_limit_multiplier 按 (tool,user) 独立窗口放宽个人限流(全局额度不受影响)。 - 修复存量缺陷:/v1/agent/nodes/ 未挂 publicMux,节点心跳/认领端点在部署 拓扑下不可达。 - 迁移 000046;任务全链路集成测试连真实库通过,HTTP 端到端验证 (下发→认领→伪造令牌拒绝→上报→succeeded,列表不泄露认领令牌); 25 包测试通过,前后端构建通过。 --- RELEASE.md | 28 ++ cmd/gateway-api/main.go | 5 + docs/security-review-0.10.1.md | 18 + docs/旗舰版需求规划与完成情况.md | 16 + internal/agentnode/http.go | 117 +++++- internal/agentnode/tasks.go | 336 ++++++++++++++++++ internal/agentnode/tasks_integration_test.go | 117 ++++++ internal/identity/http.go | 1 + internal/workbench/agentpolicy.go | 122 +++++++ internal/workbench/inbox.go | 4 + internal/workbench/tools.go | 64 +++- migrations/000046_agent_tasks.sql | 46 +++ web/apps/admin/src/api/agent-nodes.ts | 13 + .../src/views/gateway/agent-tasks/index.vue | 162 +++++++++ web/apps/portal/src/api/portal.ts | 5 + .../src/views/portal/security/index.vue | 43 ++- 16 files changed, 1082 insertions(+), 15 deletions(-) create mode 100644 internal/agentnode/tasks.go create mode 100644 internal/agentnode/tasks_integration_test.go create mode 100644 internal/workbench/agentpolicy.go create mode 100644 migrations/000046_agent_tasks.sql create mode 100644 web/apps/admin/src/views/gateway/agent-tasks/index.vue diff --git a/RELEASE.md b/RELEASE.md index b9b98bc..0c29069 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -85,3 +85,31 @@ copying the bundle to a deployment host. - 个人渠道令牌只存 SHA-256 摘要、仅创建/轮换时显示一次,校验走 constant-time 比较;渠道代码全局唯一。 - 平台环境变量与个人变量同用 AES-256-GCM 加密,管理端仅系统管理员可写。 + +## 0.11.4 — 旗舰版第五轮完善 + +发布时间:2026-08-13 + +新增功能: + +- **智能体节点任务下发**(迁移 000046):管理端「安全与审计 → 节点任务」向节点池 + 下发任务(Prompt/HTTP/MCP/Skill/数字员工/自定义),支持指定节点或池路由、 + 最大尝试次数;节点经令牌认证后 `POST /v1/agent/nodes/{code}/tasks/claim` + 按 FIFO 认领(SKIP LOCKED 防多节点重复认领),执行后凭认领令牌上报结果 + (`/tasks/{id}/complete`,防重放/防串扰);失败按 30s×次数退避自动重入队, + 达上限标记 failed;支持取消排队任务与失败重试;完成/失败经站内信通知管理员。 +- **个人智能体安全策略**:门户「账号安全」新增智能体安全策略——个人工具 + 自动审批(个人凭据调用需审批工具时不再拦截)与个人限流倍数(个人调用 + 按 (tool,user) 独立窗口放宽限流,默认倍数 1);运行时治理即时生效。 +- 修复: `/v1/agent/nodes/` 从未挂载 publicMux(心跳端点在部署拓扑下不可达, + 0.11.0 存量缺陷)。 + +迁移:000046_agent_tasks(共 46 个迁移)。 + +安全要点: + +- 任务认领 SKIP LOCKED + 15 分钟认领租约;上报必须携带认领令牌,伪造/复用 + 一律拒绝;认领令牌仅认领响应返回,管理端列表/详情不回传。 +- 节点令牌沿用心跳同一套 SHA-256 + constant-time 校验,未知节点与错误令牌 + 返回同一错误,不泄露节点存在性。 +- 个人限流按 (tool_id, portal_user_id) 独立窗口计数,个人倍数不影响全局额度。 diff --git a/cmd/gateway-api/main.go b/cmd/gateway-api/main.go index b03c4be..e31bb9c 100644 --- a/cmd/gateway-api/main.go +++ b/cmd/gateway-api/main.go @@ -323,6 +323,7 @@ func main() { traceHandler := trace.NewAdminHTTPHandler(traceStore, identityService) agentNodeStore := agentnode.NewStore(db) agentNodeHandler := agentnode.NewHTTPHandler(agentNodeStore, identityService) + agentPolicyHandler := workbench.NewAgentPolicyHTTPHandler(workbench.NewAgentPolicyService(db), identityService) applicationKeyCipher, err := cryptox.NewKeyring( cfg.Credentials.MasterKey, cfg.Credentials.KEKVersion, cfg.Credentials.KEKKeyring, "application-runtime-key", ) @@ -444,6 +445,8 @@ func main() { controlMux.Handle("/api/v1/admin/agent-sessions", traceHandler) controlMux.Handle("/api/v1/admin/agent-nodes", agentNodeHandler) controlMux.Handle("/api/v1/admin/agent-nodes/", agentNodeHandler) + controlMux.Handle("/api/v1/admin/agent-tasks", agentNodeHandler) + controlMux.Handle("/api/v1/admin/agent-tasks/", agentNodeHandler) controlMux.Handle("/api/v1/agent/nodes/", agentNodeHandler) controlMux.Handle("/api/v1/admin/channels", channelHandler) controlMux.Handle("/api/v1/admin/channels/", channelHandler) @@ -483,6 +486,7 @@ func main() { controlMux.Handle("/api/v1/portal/stats", portalHandler) controlMux.Handle("/api/v1/portal/tools", portalHandler) controlMux.Handle("/api/v1/portal/chat/", portalHandler) + controlMux.Handle("/api/v1/portal/agent-policy", agentPolicyHandler) controlMux.Handle("/api/v1/portal/personal-channels", portalHandler) controlMux.Handle("/api/v1/portal/personal-channels/", portalHandler) controlMux.Handle("/api/v1/portal/digital-employees", portalHandler) @@ -506,6 +510,7 @@ func main() { publicMux.Handle("/v1/mcp-servers/", workbenchRuntime) publicMux.Handle("/v1/channels/", channelInboundHandler) publicMux.Handle("/v1/personal-channels/", portalHandler) + publicMux.Handle("/v1/agent/nodes/", agentNodeHandler) publicMux.Handle("/v1/digital-employees/", workbenchRuntime) publicMux.Handle("/v1/", governedGateway) server := httpserver.New(httpserver.Dependencies{ diff --git a/docs/security-review-0.10.1.md b/docs/security-review-0.10.1.md index b79f162..095f032 100644 --- a/docs/security-review-0.10.1.md +++ b/docs/security-review-0.10.1.md @@ -478,3 +478,21 @@ PostgreSQL `text` 列拒绝写入 → 重试循环永远失败；inbox 的 Go `l 校验;租户概览展示配额与当月用量。 8. 修复:渠道 Save 空 API Key 时 `encrypted_api_key` NOT NULL 违约(空 bytea); 安全报表 inet 列二进制扫描失败(`ip::text`)。 + +# 追加：旗舰版功能完善第五轮（0.11.4，2026-08-13） + +1. **智能体节点任务下发**(迁移 000046 + `internal/agentnode/tasks.go`)： + 任务表 `agent_tasks`(类型/载荷/池路由/重试/认领令牌/结果);认领用 + `FOR UPDATE SKIP LOCKED` 保证多节点并发互斥,15 分钟租约;上报凭认领令牌 + 校验(伪造/重放拒绝),失败按 30s×attempts 退避重入队,达 max_attempts 标记 + failed;取消仅限排队中,失败任务可重试;事件经 outbox → 站内信。 +2. **个人智能体安全策略**(`portal_agent_policies`):auto_approve_tools 跳过 + 个人调用的审批门;rate_limit_multiplier 按 (tool,user) 独立窗口放宽个人 + 限流;`tool_rate_usage` 主键扩展为 (tool_id, portal_user_id, window_start), + 全零 UUID 为全局窗口哨兵。 +3. **修复存量缺陷**:`/v1/agent/nodes/` 未挂 publicMux,节点心跳/认领端点 + 在部署拓扑下全部不可达;已挂载并端到端验证。 +4. 验证:任务全链路集成测试 `TestAgentTaskPostgreSQLLifecycle` 连真实库通过 + (下发→认领→伪造令牌拒绝→成功/失败上报→退避重试→达上限→取消冲突→重试); + HTTP 端到端:下发→认领(返回认领令牌)→上报→succeeded,管理端列表不泄露 + 认领令牌;任务完成站内信已验证。 diff --git a/docs/旗舰版需求规划与完成情况.md b/docs/旗舰版需求规划与完成情况.md index 7712424..2f4dc87 100644 --- a/docs/旗舰版需求规划与完成情况.md +++ b/docs/旗舰版需求规划与完成情况.md @@ -212,3 +212,19 @@ MinIO 对象存储与管理端/个人文件仓库；pgvector + Ollama(bge-m3) 逻辑隔离 + 配额管控已完成,物理隔离需明确部署形态后实施。 - 智能体节点远程安装/任务下发:节点登记/心跳/路由预览已完成,远程安装 与真实节点执行需部署 Agent 环境后验收。 + +## 十、0.11.4 完成情况（2026-08-13 第五轮完善） + +| 功能 | 状态 | 说明 | +|---|---|---| +| 智能体节点任务下发 | ✅ | 池路由认领(SKIP LOCKED)/认领令牌上报/退避重试/取消/失败重试/管理端页面/站内信 | +| 个人智能体安全策略 | ✅ | 个人工具自动审批 + 个人限流倍数,运行时即时生效 | + +至此旗舰版矩阵中可独立实施的功能已全部完成。剩余依赖外部条件项: + +- 企微/钉钉/飞书真实平台联调(扫码登录与渠道):协议实现 + 单测覆盖, + 需企业开放平台应用凭据冒烟。 +- 多租户物理隔离(独立 schema/库):逻辑隔离(tenant_id)与配额管控已完成, + 物理隔离需明确部署形态。 +- 智能体节点远程安装/裸机部署脚本:任务下发与执行协议已就绪并端到端验证, + 节点 Agent 安装包/注册脚本需在目标环境实施。 diff --git a/internal/agentnode/http.go b/internal/agentnode/http.go index 1692cfb..8cd3c1d 100644 --- a/internal/agentnode/http.go +++ b/internal/agentnode/http.go @@ -43,7 +43,13 @@ func NewHTTPHandler(store *Store, identityService *identity.Service) *HTTPHandle h.mux.HandleFunc("PUT /api/v1/admin/agent-nodes/{id}", h.update) h.mux.HandleFunc("DELETE /api/v1/admin/agent-nodes/{id}", h.delete) h.mux.HandleFunc("POST /api/v1/admin/agent-nodes/{id}/rotate-token", h.rotateToken) - h.mux.HandleFunc("POST /api/v1/agent/nodes/{code}/heartbeat", h.heartbeat) + h.mux.HandleFunc("GET /api/v1/admin/agent-tasks", h.listTasks) + h.mux.HandleFunc("POST /api/v1/admin/agent-tasks", h.createTask) + h.mux.HandleFunc("POST /api/v1/admin/agent-tasks/{id}/cancel", h.cancelTask) + h.mux.HandleFunc("POST /api/v1/admin/agent-tasks/{id}/retry", h.retryTask) + h.mux.HandleFunc("POST /v1/agent/nodes/{code}/heartbeat", h.heartbeat) + h.mux.HandleFunc("POST /v1/agent/nodes/{code}/tasks/claim", h.claimTask) + h.mux.HandleFunc("POST /v1/agent/nodes/{code}/tasks/{id}/complete", h.completeTask) return h } @@ -216,3 +222,112 @@ func writeHeartbeatError(w http.ResponseWriter, err error) { apiresponse.Error(w, http.StatusInternalServerError, "节点心跳处理失败") } } + +// --- 节点任务下发/认领/上报 --- + +func (h *HTTPHandler) listTasks(w http.ResponseWriter, r *http.Request) { + if _, ok := h.require(w, r, identity.PermissionAgentNodeRead); !ok { + return + } + items, err := h.store.ListTasks(r.Context(), r.URL.Query().Get("status")) + if err != nil { + writeError(w, err) + return + } + apiresponse.OK(w, items) +} + +func (h *HTTPHandler) createTask(w http.ResponseWriter, r *http.Request) { + actor, ok := h.require(w, r, identity.PermissionAgentNodeManage) + if !ok { + return + } + var input struct { + TaskType string `json:"task_type"` + Payload json.RawMessage `json:"payload"` + PoolType string `json:"pool_type"` + PoolCode string `json:"pool_code"` + NodeID *string `json:"node_id"` + MaxAttempts int `json:"max_attempts"` + } + if !decodeJSON(w, r, &input) { + return + } + item, err := h.store.CreateTask(r.Context(), TaskInput{TaskType: input.TaskType, Payload: input.Payload, PoolType: input.PoolType, PoolCode: input.PoolCode, NodeID: input.NodeID, MaxAttempts: input.MaxAttempts}, actor.ID) + if err != nil { + writeTaskError(w, err) + return + } + apiresponse.OK(w, item) +} + +func (h *HTTPHandler) cancelTask(w http.ResponseWriter, r *http.Request) { + if _, ok := h.require(w, r, identity.PermissionAgentNodeManage); !ok { + return + } + if err := h.store.CancelTask(r.Context(), r.PathValue("id")); err != nil { + writeTaskError(w, err) + return + } + apiresponse.OK(w, map[string]bool{"cancelled": true}) +} + +func (h *HTTPHandler) retryTask(w http.ResponseWriter, r *http.Request) { + if _, ok := h.require(w, r, identity.PermissionAgentNodeManage); !ok { + return + } + if err := h.store.RetryTask(r.Context(), r.PathValue("id")); err != nil { + writeTaskError(w, err) + return + } + apiresponse.OK(w, map[string]bool{"retried": true}) +} + +// claimTask 节点拉取下一个可执行任务(池路由 + SKIP LOCKED 认领)。 +func (h *HTTPHandler) claimTask(w http.ResponseWriter, r *http.Request) { + item, err := h.store.ClaimTask(r.Context(), r.PathValue("code"), r.Header.Get("X-Agent-Token")) + if err != nil { + writeTaskError(w, err) + return + } + if item.ID == "" { + apiresponse.OK(w, map[string]any{"task": nil}) + return + } + apiresponse.OK(w, map[string]any{"task": item}) +} + +// completeTask 节点上报任务结果;claim_token 匹配才生效。 +func (h *HTTPHandler) completeTask(w http.ResponseWriter, r *http.Request) { + var input struct { + ClaimToken string `json:"claim_token"` + Result json.RawMessage `json:"result"` + Error string `json:"error"` + } + if !decodeJSON(w, r, &input) { + return + } + item, err := h.store.CompleteTask(r.Context(), r.PathValue("code"), r.Header.Get("X-Agent-Token"), r.PathValue("id"), input.ClaimToken, input.Result, input.Error) + if err != nil { + writeTaskError(w, err) + return + } + apiresponse.OK(w, item) +} + +func writeTaskError(w http.ResponseWriter, err error) { + switch { + case errors.Is(err, ErrTaskUnauthorized), errors.Is(err, ErrInvalidToken): + apiresponse.Error(w, http.StatusUnauthorized, "节点令牌无效或节点已停用") + case errors.Is(err, ErrTaskNotFound), errors.Is(err, ErrNotFound): + apiresponse.Error(w, http.StatusNotFound, "任务不存在") + case errors.Is(err, ErrTaskConflict): + apiresponse.Error(w, http.StatusConflict, "任务状态不允许该操作(可能已被认领或已结束)") + case errors.Is(err, ErrTaskInvalid), errors.Is(err, ErrInvalidInput): + apiresponse.Error(w, http.StatusBadRequest, err.Error()) + case errors.Is(err, ErrStore): + apiresponse.Error(w, http.StatusServiceUnavailable, "任务服务暂不可用") + default: + apiresponse.Error(w, http.StatusInternalServerError, "任务处理失败") + } +} diff --git a/internal/agentnode/tasks.go b/internal/agentnode/tasks.go new file mode 100644 index 0000000..5f353f9 --- /dev/null +++ b/internal/agentnode/tasks.go @@ -0,0 +1,336 @@ +package agentnode + +import ( + "context" + "crypto/sha256" + "crypto/subtle" + "encoding/json" + "errors" + "fmt" + "strings" + "time" + + platformid "aigateway.local/core/internal/platform/id" + "github.com/jackc/pgx/v5" +) + +// Task 是一条下发给节点池的任务。 +type Task struct { + ID string `json:"id"` + TaskType string `json:"task_type"` + Payload json.RawMessage `json:"payload"` + PoolType string `json:"pool_type"` + PoolCode string `json:"pool_code"` + NodeID *string `json:"node_id,omitempty"` + NodeCode string `json:"node_code,omitempty"` + Status string `json:"status"` + Attempts int `json:"attempts"` + MaxAttempts int `json:"max_attempts"` + Result json.RawMessage `json:"result,omitempty"` + Error string `json:"error,omitempty"` + // ClaimToken 是认领时签发的上报凭证,仅认领响应返回,列表/详情不回传。 + ClaimToken string `json:"claim_token,omitempty"` + CreatedAt time.Time `json:"created_at"` + ClaimedAt *time.Time `json:"claimed_at,omitempty"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + AvailableAt time.Time `json:"available_at"` +} + +// TaskInput 是创建任务的输入。 +type TaskInput struct { + TaskType string + Payload json.RawMessage + PoolType string + PoolCode string + NodeID *string + MaxAttempts int +} + +var ( + ErrTaskNotFound = errors.New("agent task not found") + ErrTaskConflict = errors.New("agent task state conflict") + ErrTaskInvalid = errors.New("agent task input invalid") + ErrTaskUnauthorized = errors.New("agent node token invalid") +) + +var taskTypes = map[string]bool{"prompt": true, "http": true, "mcp_invoke": true, "skill_run": true, "digital_employee": true, "custom": true} + +const taskSelect = `SELECT t.id::text,t.task_type,t.payload,t.pool_type,t.pool_code,t.node_id::text,coalesce(n.code,''),t.status,t.attempts,t.max_attempts,t.result,t.error,t.created_at,t.claimed_at,t.finished_at,t.available_at +FROM gateway.agent_tasks t LEFT JOIN gateway.agent_nodes n ON n.id=t.node_id` + +func scanTask(row pgx.Row) (Task, error) { + var item Task + var nodeID *string + err := row.Scan(&item.ID, &item.TaskType, &item.Payload, &item.PoolType, &item.PoolCode, &nodeID, &item.NodeCode, &item.Status, &item.Attempts, &item.MaxAttempts, &item.Result, &item.Error, &item.CreatedAt, &item.ClaimedAt, &item.FinishedAt, &item.AvailableAt) + if errors.Is(err, pgx.ErrNoRows) { + return Task{}, ErrTaskNotFound + } + if err != nil { + return Task{}, err + } + item.NodeID = nodeID + item.Payload = normalizeJSON(item.Payload) + return item, nil +} + +// CreateTask 下发任务到节点池(queued)。 +func (s *Store) CreateTask(ctx context.Context, input TaskInput, actorID string) (Task, error) { + if s == nil || s.pool == nil { + return Task{}, ErrStore + } + input.TaskType = strings.TrimSpace(input.TaskType) + input.PoolType = strings.ToLower(strings.TrimSpace(input.PoolType)) + input.PoolCode = strings.TrimSpace(input.PoolCode) + if !taskTypes[input.TaskType] { + return Task{}, fmt.Errorf("%w: task type must be one of prompt/http/mcp_invoke/skill_run/digital_employee/custom", ErrTaskInvalid) + } + if len(input.Payload) == 0 || !json.Valid(input.Payload) { + return Task{}, fmt.Errorf("%w: payload must be a JSON object", ErrTaskInvalid) + } + var object map[string]any + if json.Unmarshal(input.Payload, &object) != nil || object == nil { + return Task{}, fmt.Errorf("%w: payload must be a JSON object", ErrTaskInvalid) + } + if input.PoolType == "" { + input.PoolType = "private" + } + if input.PoolType != "public" && input.PoolType != "private" { + return Task{}, fmt.Errorf("%w: pool type is invalid", ErrTaskInvalid) + } + if input.PoolCode == "" { + input.PoolCode = "default" + } + if len(input.PoolCode) > 64 { + return Task{}, fmt.Errorf("%w: pool code is invalid", ErrTaskInvalid) + } + if input.MaxAttempts < 1 || input.MaxAttempts > 10 { + input.MaxAttempts = 3 + } + if input.NodeID != nil && strings.TrimSpace(*input.NodeID) != "" { + var exists bool + if err := s.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.agent_nodes WHERE id=$1)`, *input.NodeID).Scan(&exists); err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + if !exists { + return Task{}, fmt.Errorf("%w: target node does not exist", ErrTaskInvalid) + } + } else { + input.NodeID = nil + } + id, err := platformid.NewUUID() + if err != nil { + return Task{}, err + } + eventID, err := platformid.NewUUID() + if err != nil { + return Task{}, err + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + defer func() { _ = tx.Rollback(ctx) }() + _, err = tx.Exec(ctx, `INSERT INTO gateway.agent_tasks(id,task_type,payload,pool_type,pool_code,node_id,max_attempts,created_by) VALUES($1,$2,$3,$4,$5,nullif($6,'')::uuid,$7,nullif($8,'')::uuid)`, id, input.TaskType, input.Payload, input.PoolType, input.PoolCode, input.NodeID, input.MaxAttempts, actorID) + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + payload, _ := json.Marshal(map[string]any{"task_id": id, "task_type": input.TaskType, "pool_type": input.PoolType, "pool_code": input.PoolCode}) + if _, err = tx.Exec(ctx, `INSERT INTO gateway.outbox_events(event_id,event_type,event_version,aggregate_type,aggregate_id,payload) VALUES($1,'agent_task.created',1,'agent_task',$2,$3)`, eventID, id, payload); err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + if err = tx.Commit(ctx); err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + return s.GetTask(ctx, id) +} + +// ListTasks 返回任务列表,可按状态过滤。 +func (s *Store) ListTasks(ctx context.Context, status string) ([]Task, error) { + if s == nil || s.pool == nil { + return nil, ErrStore + } + where, args := " WHERE true", []any{} + if status != "" { + args = append(args, status) + where = " WHERE t.status=$1" + } + rows, err := s.pool.Query(ctx, taskSelect+where+` ORDER BY t.created_at DESC LIMIT 200`, args...) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrStore, err) + } + defer rows.Close() + items := []Task{} + for rows.Next() { + item, err := scanTask(rows) + if err != nil { + return nil, err + } + items = append(items, item) + } + return items, rows.Err() +} + +// GetTask 按 ID 返回任务。 +func (s *Store) GetTask(ctx context.Context, id string) (Task, error) { + if s == nil || s.pool == nil { + return Task{}, ErrStore + } + return scanTask(s.pool.QueryRow(ctx, taskSelect+` WHERE t.id=$1`, strings.TrimSpace(id))) +} + +// CancelTask 取消排队中的任务(已认领/运行中不可取消)。 +func (s *Store) CancelTask(ctx context.Context, id string) error { + if s == nil || s.pool == nil { + return ErrStore + } + tag, err := s.pool.Exec(ctx, `UPDATE gateway.agent_tasks SET status='cancelled',finished_at=clock_timestamp() WHERE id=$1 AND status='queued'`, strings.TrimSpace(id)) + if err != nil { + return fmt.Errorf("%w: %v", ErrStore, err) + } + if tag.RowsAffected() == 0 { + return ErrTaskConflict + } + return nil +} + +// RetryTask 把失败任务重新入队(重置失败状态与退避)。 +func (s *Store) RetryTask(ctx context.Context, id string) error { + if s == nil || s.pool == nil { + return ErrStore + } + tag, err := s.pool.Exec(ctx, `UPDATE gateway.agent_tasks SET status='queued',error='',available_at=clock_timestamp(),finished_at=NULL,claimed_at=NULL WHERE id=$1 AND status='failed'`, strings.TrimSpace(id)) + if err != nil { + return fmt.Errorf("%w: %v", ErrStore, err) + } + if tag.RowsAffected() == 0 { + return ErrTaskConflict + } + return nil +} + +// authenticateNode 校验节点令牌(与心跳同一套恒定时间比较),返回节点行。 +func (s *Store) authenticateNode(ctx context.Context, code, token string) (Node, error) { + code = strings.ToLower(strings.TrimSpace(code)) + token = strings.TrimSpace(token) + if !nodeCodePattern.MatchString(code) || token == "" || len(token) > 512 { + return Node{}, ErrTaskUnauthorized + } + var storedHash []byte + err := s.pool.QueryRow(ctx, `SELECT token_hash FROM gateway.agent_nodes WHERE code=$1 AND enabled`, code).Scan(&storedHash) + if errors.Is(err, pgx.ErrNoRows) { + return Node{}, ErrTaskUnauthorized + } + if err != nil { + return Node{}, fmt.Errorf("%w: %v", ErrStore, err) + } + digest := sha256.Sum256([]byte(token)) + if subtle.ConstantTimeCompare(digest[:], storedHash) != 1 { + return Node{}, ErrTaskUnauthorized + } + var id string + if err := s.pool.QueryRow(ctx, `SELECT id::text FROM gateway.agent_nodes WHERE code=$1 AND enabled`, code).Scan(&id); err != nil { + return Node{}, fmt.Errorf("%w: %v", ErrStore, err) + } + return s.Get(ctx, id) +} + +// ClaimTask 节点认领下一个可用任务:目标节点(或所在池)中最早排队且到期的任务, +// SKIP LOCKED 保证多节点并发不重复认领。返回空 Task 表示当前无任务。 +func (s *Store) ClaimTask(ctx context.Context, code, token string) (Task, error) { + if s == nil || s.pool == nil { + return Task{}, ErrStore + } + node, err := s.authenticateNode(ctx, code, token) + if err != nil { + return Task{}, err + } + claim, err := platformid.NewUUID() + if err != nil { + return Task{}, err + } + var item Task + err = s.pool.QueryRow(ctx, `UPDATE gateway.agent_tasks SET status='claimed',claim_token=$1,claimed_at=clock_timestamp(),available_at=clock_timestamp()+interval '15 minutes' + WHERE id = ( + SELECT t.id FROM gateway.agent_tasks t + WHERE t.status='queued' AND t.available_at<=clock_timestamp() + AND (t.node_id IS NULL OR t.node_id=$2) + AND (t.node_id IS NOT NULL OR (t.pool_type=$3 AND t.pool_code=$4)) + ORDER BY t.created_at LIMIT 1 FOR UPDATE SKIP LOCKED + ) + RETURNING id::text,task_type,payload,pool_type,pool_code,node_id::text,status,attempts,max_attempts,result,error,created_at,claimed_at,finished_at,available_at,claim_token`, claim, node.ID, node.PoolType, node.PoolCode).Scan( + &item.ID, &item.TaskType, &item.Payload, &item.PoolType, &item.PoolCode, &item.NodeID, &item.Status, &item.Attempts, &item.MaxAttempts, &item.Result, &item.Error, &item.CreatedAt, &item.ClaimedAt, &item.FinishedAt, &item.AvailableAt, &item.ClaimToken) + if errors.Is(err, pgx.ErrNoRows) { + return Task{}, nil + } + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + item.Payload = normalizeJSON(item.Payload) + return item, nil +} + +// CompleteTask 节点上报任务结果;claim_token 匹配才生效(防重放/防串扰)。 +// 失败且未达最大重试次数时按退避重新入队。 +func (s *Store) CompleteTask(ctx context.Context, code, token, taskID, claimToken string, result json.RawMessage, taskError string) (Task, error) { + if s == nil || s.pool == nil { + return Task{}, ErrStore + } + if _, err := s.authenticateNode(ctx, code, token); err != nil { + return Task{}, err + } + taskID = strings.TrimSpace(taskID) + claimToken = strings.TrimSpace(claimToken) + if taskID == "" || claimToken == "" || len(taskError) > 4000 { + return Task{}, ErrTaskInvalid + } + if len(result) > 0 && !json.Valid(result) { + return Task{}, fmt.Errorf("%w: result must be valid JSON", ErrTaskInvalid) + } + tx, err := s.pool.Begin(ctx) + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + defer func() { _ = tx.Rollback(ctx) }() + var status, currentError string + var attempts, maxAttempts int + err = tx.QueryRow(ctx, `SELECT status,error,attempts,max_attempts FROM gateway.agent_tasks WHERE id=$1 AND claim_token=$2`, taskID, claimToken).Scan(&status, ¤tError, &attempts, &maxAttempts) + if errors.Is(err, pgx.ErrNoRows) { + return Task{}, ErrTaskConflict + } + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + if status != "claimed" && status != "running" { + return Task{}, ErrTaskConflict + } + attempts++ + nextStatus := "succeeded" + if taskError != "" { + nextStatus = "failed" + if attempts < maxAttempts { + nextStatus = "queued" + } + } + // 失败退避:30s * 已尝试次数。 + backoff := 30 * time.Second * time.Duration(attempts) + _, err = tx.Exec(ctx, `UPDATE gateway.agent_tasks SET status=$2::text,attempts=$3::int,result=$4::jsonb,error=$5::text,claim_token=NULL,finished_at=CASE WHEN $2::text IN ('succeeded','failed','cancelled') THEN clock_timestamp() ELSE NULL END,available_at=CASE WHEN $2::text='queued' THEN clock_timestamp()+$6::interval ELSE available_at END WHERE id=$1::uuid`, + taskID, nextStatus, attempts, normalizeJSON(result), taskError, backoff) + if err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + eventID, _ := platformid.NewUUID() + eventType := "agent_task.completed" + if nextStatus == "failed" { + eventType = "agent_task.failed" + } else if nextStatus == "queued" { + eventType = "agent_task.retry" + } + payload, _ := json.Marshal(map[string]any{"task_id": taskID, "status": nextStatus, "attempts": attempts, "error": taskError}) + if _, err = tx.Exec(ctx, `INSERT INTO gateway.outbox_events(event_id,event_type,event_version,aggregate_type,aggregate_id,payload) VALUES($1,$2,1,'agent_task',$3,$4)`, eventID, eventType, taskID, payload); err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + if err = tx.Commit(ctx); err != nil { + return Task{}, fmt.Errorf("%w: %v", ErrStore, err) + } + return s.GetTask(ctx, taskID) +} diff --git a/internal/agentnode/tasks_integration_test.go b/internal/agentnode/tasks_integration_test.go new file mode 100644 index 0000000..f3d3dc1 --- /dev/null +++ b/internal/agentnode/tasks_integration_test.go @@ -0,0 +1,117 @@ +package agentnode + +import ( + "context" + "encoding/json" + "net" + "os" + "testing" + "time" + + "aigateway.local/core/internal/platform/config" + "aigateway.local/core/internal/platform/database" +) + +// TestAgentTaskPostgreSQLLifecycle 模拟一个节点跑通任务全链路: +// 下发 → 心跳上线 → 认领(SKIP LOCKED) → 上报成功 → 状态校验; +// 以及失败重试入队与并发认领互斥。 +func TestAgentTaskPostgreSQLLifecycle(t *testing.T) { + databaseURL := os.Getenv("AGENT_NODE_TEST_DATABASE_URL") + if databaseURL == "" { + t.Skip("AGENT_NODE_TEST_DATABASE_URL is not set") + } + ctx := context.Background() + pool, err := database.Open(ctx, config.Database{URL: databaseURL, MaxConns: 8, MinConns: 0}) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + _, _ = pool.Exec(ctx, `DELETE FROM gateway.agent_nodes WHERE code='task-node-integration'`) + _, _ = pool.Exec(ctx, `DELETE FROM gateway.agent_tasks WHERE payload->>'tag'='integration'`) + defer pool.Exec(ctx, `DELETE FROM gateway.agent_nodes WHERE code='task-node-integration'`) + defer pool.Exec(ctx, `DELETE FROM gateway.agent_tasks WHERE payload->>'tag'='integration'`) + store := NewStore(pool) + + // 1. 登记节点并心跳上线。 + node, token, err := store.Create(ctx, CreateInput{Code: "task-node-integration", Name: "Task Node", PoolType: "private", PoolCode: "test", Enabled: true}, "") + if err != nil { + t.Fatal(err) + } + if _, err := store.Heartbeat(ctx, node.Code, token, net.ParseIP("192.0.2.11"), HeartbeatInput{Version: "test", Capabilities: map[string]any{"tool_exec": true}}); err != nil { + t.Fatal(err) + } + + // 2. 下发两条任务(同池)。 + payload, _ := json.Marshal(map[string]any{"tag": "integration", "seq": 1}) + first, err := store.CreateTask(ctx, TaskInput{TaskType: "prompt", Payload: payload, PoolType: "private", PoolCode: "test", MaxAttempts: 2}, "") + if err != nil || first.Status != "queued" { + t.Fatalf("create task=%+v err=%v", first, err) + } + payload2, _ := json.Marshal(map[string]any{"tag": "integration", "seq": 2}) + second, err := store.CreateTask(ctx, TaskInput{TaskType: "http", Payload: payload2, PoolType: "private", PoolCode: "test", MaxAttempts: 2}, "") + if err != nil { + t.Fatal(err) + } + + // 3. 节点认领:两条任务按创建顺序各被认领一次。 + claimed1, err := store.ClaimTask(ctx, node.Code, token) + if err != nil || claimed1.ID != first.ID || claimed1.Status != "claimed" { + t.Fatalf("claim1 task=%+v err=%v", claimed1, err) + } + claimed2, err := store.ClaimTask(ctx, node.Code, token) + if err != nil || claimed2.ID != second.ID { + t.Fatalf("claim2 task=%+v err=%v", claimed2, err) + } + if empty, err := store.ClaimTask(ctx, node.Code, token); err != nil || empty.ID != "" { + t.Fatalf("third claim should be empty, got %+v err=%v", empty, err) + } + + // 4. 错误 claim_token 上报必须失败(防串扰)。 + badResult, _ := json.Marshal(map[string]string{"ok": "true"}) + if _, err := store.CompleteTask(ctx, node.Code, token, first.ID, "forged-token", badResult, ""); err == nil { + t.Fatal("complete with forged claim token should fail") + } + + // 5. 成功上报第一条。 + done, err := store.CompleteTask(ctx, node.Code, token, first.ID, claimed1.ClaimToken, badResult, "") + if err != nil || done.Status != "succeeded" || done.Attempts != 1 { + t.Fatalf("complete task=%+v err=%v", done, err) + } + + // 6. 第二条上报失败 → attempts 0 { menus = append(menus, map[string]any{"name": "Security", "path": "/security", "component": "/index/index", "meta": map[string]any{"title": "安全与审计", "icon": "ri:shield-check-line"}, "children": securityChildren}) diff --git a/internal/workbench/agentpolicy.go b/internal/workbench/agentpolicy.go new file mode 100644 index 0000000..f8492d8 --- /dev/null +++ b/internal/workbench/agentpolicy.go @@ -0,0 +1,122 @@ +package workbench + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "strings" + + "aigateway.local/core/internal/identity" + "aigateway.local/core/internal/platform/apiresponse" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// AgentPolicy 是个人智能体安全策略。 +type AgentPolicy struct { + AutoApproveTools bool `json:"auto_approve_tools"` + RateLimitMultiplier int `json:"rate_limit_multiplier"` +} + +// AgentPolicyService 管理个人智能体安全策略。 +type AgentPolicyService struct { + pool *pgxpool.Pool +} + +func NewAgentPolicyService(pool *pgxpool.Pool) *AgentPolicyService { + return &AgentPolicyService{pool: pool} +} + +// Get 返回用户的策略(未配置时返回默认值)。 +func (s *AgentPolicyService) Get(ctx context.Context, portalUserID string) (AgentPolicy, error) { + var policy AgentPolicy + if s == nil || s.pool == nil { + return policy, errors.New("安全策略服务不可用") + } + err := s.pool.QueryRow(ctx, `SELECT auto_approve_tools,rate_limit_multiplier FROM gateway.portal_agent_policies WHERE portal_user_id=$1`, portalUserID).Scan(&policy.AutoApproveTools, &policy.RateLimitMultiplier) + if errors.Is(err, pgx.ErrNoRows) { + policy.RateLimitMultiplier = 1 + return policy, nil + } + if err != nil { + return policy, err + } + return policy, nil +} + +// Set 更新用户的策略。 +func (s *AgentPolicyService) Set(ctx context.Context, portalUserID string, policy AgentPolicy) error { + if s == nil || s.pool == nil { + return errors.New("安全策略服务不可用") + } + if policy.RateLimitMultiplier < 1 || policy.RateLimitMultiplier > 100 { + return errors.New("限流倍数必须在 1-100 之间") + } + _, err := s.pool.Exec(ctx, `INSERT INTO gateway.portal_agent_policies(portal_user_id,auto_approve_tools,rate_limit_multiplier) VALUES($1,$2,$3) + ON CONFLICT(portal_user_id) DO UPDATE SET auto_approve_tools=$2,rate_limit_multiplier=$3,updated_at=clock_timestamp()`, + portalUserID, policy.AutoApproveTools, policy.RateLimitMultiplier) + return err +} + +// AgentPolicyHTTPHandler 门户个人智能体安全策略。 +type AgentPolicyHTTPHandler struct { + service *AgentPolicyService + identity *identity.Service + mux *http.ServeMux +} + +func NewAgentPolicyHTTPHandler(service *AgentPolicyService, identityService *identity.Service) *AgentPolicyHTTPHandler { + h := &AgentPolicyHTTPHandler{service: service, identity: identityService, mux: http.NewServeMux()} + h.mux.HandleFunc("GET /api/v1/portal/agent-policy", h.get) + h.mux.HandleFunc("PUT /api/v1/portal/agent-policy", h.put) + return h +} + +func (h *AgentPolicyHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + h.mux.ServeHTTP(w, r) +} + +func (h *AgentPolicyHTTPHandler) account(w http.ResponseWriter, r *http.Request) (identity.Account, bool) { + account, err := h.identity.Authenticate(r.Context(), identity.KindPortal, r.Header.Get("Authorization")) + if err != nil { + apiresponse.Error(w, http.StatusUnauthorized, "登录状态无效或已过期") + return identity.Account{}, false + } + return account, true +} + +func (h *AgentPolicyHTTPHandler) get(w http.ResponseWriter, r *http.Request) { + a, ok := h.account(w, r) + if !ok { + return + } + policy, err := h.service.Get(r.Context(), a.ID) + if err != nil { + apiresponse.Error(w, http.StatusServiceUnavailable, "安全策略查询失败") + return + } + apiresponse.OK(w, policy) +} + +func (h *AgentPolicyHTTPHandler) put(w http.ResponseWriter, r *http.Request) { + a, ok := h.account(w, r) + if !ok { + return + } + var input struct { + AutoApproveTools bool `json:"auto_approve_tools"` + RateLimitMultiplier int `json:"rate_limit_multiplier"` + } + decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<20)) + decoder.DisallowUnknownFields() + if decoder.Decode(&input) != nil { + apiresponse.Error(w, http.StatusBadRequest, "请求格式无效") + return + } + if err := h.service.Set(r.Context(), a.ID, AgentPolicy{AutoApproveTools: input.AutoApproveTools, RateLimitMultiplier: input.RateLimitMultiplier}); err != nil { + apiresponse.Error(w, http.StatusBadRequest, strings.TrimSpace(err.Error())) + return + } + apiresponse.OK(w, map[string]bool{"saved": true}) +} diff --git a/internal/workbench/inbox.go b/internal/workbench/inbox.go index 6c23d71..58bb26e 100644 --- a/internal/workbench/inbox.go +++ b/internal/workbench/inbox.go @@ -132,6 +132,10 @@ func inboxPlan(eventType string, payload json.RawMessage) []inboxDraft { return []inboxDraft{{RecipientKind: "portal", Category: "approval", Title: "资源申请已处理", Body: "您的资源权限申请(" + payloadValue(payload, "resource_type") + " " + payloadValue(payload, "resource_code") + ")已被" + text, Link: "/portal/requests", UserID: payloadValue(payload, "portal_user_id")}} case "tool_approval.requested": return []inboxDraft{{RecipientKind: "admin", Category: "approval", Title: "工具使用待审批", Body: "工具 " + payloadValue(payload, "tool_code") + " 首次被调用,需审批后才能使用", Link: "/system/approvals", AllAdmins: true}} + case "agent_task.completed": + return []inboxDraft{{RecipientKind: "admin", Category: "task_result", Title: "节点任务已完成", Body: "节点任务(" + payloadValue(payload, "task_id") + ")执行成功", Link: "/security/agent-tasks", AllAdmins: true}} + case "agent_task.failed": + return []inboxDraft{{RecipientKind: "admin", Category: "task_result", Title: "节点任务执行失败", Body: "节点任务(" + payloadValue(payload, "task_id") + ")执行失败: " + payloadValue(payload, "error"), Link: "/security/agent-tasks", AllAdmins: true}} } return nil } diff --git a/internal/workbench/tools.go b/internal/workbench/tools.go index 5c6b534..4b53b26 100644 --- a/internal/workbench/tools.go +++ b/internal/workbench/tools.go @@ -217,12 +217,19 @@ var ErrToolRateLimited = errors.New("工具调用频率超限，请稍后重试" // enforceGovernance 在工具执行前做治理校验:审批标记 + 调用频率上限。 // 审批缺失时自动发起一次申请(每工具至多一个待审项);限流用固定窗口原子 -// upsert,多实例共享同一额度。返回 (allowed, err)。 -func (s *ToolService) enforceGovernance(ctx context.Context, tool Tool) (bool, error) { +// upsert,多实例共享同一额度。apiKeyID 对应的门户用户若配置了个人智能体 +// 安全策略:auto_approve_tools 跳过审批门,rate_limit_multiplier 按倍数 +// 放宽个人限流(个人窗口独立计数)。返回 (allowed, err)。 +func (s *ToolService) enforceGovernance(ctx context.Context, tool Tool, apiKeyID string) (bool, error) { if s == nil || s.assets == nil || s.assets.pool == nil { return false, errors.New("工具服务不可用") } - if tool.ApprovalRequired { + // 解析调用者的个人策略(仅门户运行时凭据可能命中)。 + portalUserID, policy, err := s.personalPolicy(ctx, apiKeyID) + if err != nil { + return false, err + } + if tool.ApprovalRequired && !policy.AutoApproveTools { var approved bool if err := s.assets.pool.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM gateway.tool_approval_requests WHERE tool_id=$1 AND status='approved')`, tool.ID).Scan(&approved); err != nil { return false, err @@ -258,22 +265,56 @@ func (s *ToolService) enforceGovernance(ctx context.Context, tool Tool) (bool, e return false, fmt.Errorf("%w: %s（已自动发起审批）", ErrToolApprovalRequired, tool.Name) } } - if tool.RateLimitRPM > 0 { + limit := tool.RateLimitRPM + if portalUserID != "" && policy.RateLimitMultiplier > 1 { + limit = tool.RateLimitRPM * policy.RateLimitMultiplier + } + if limit > 0 { + userKey := portalUserID + if userKey == "" { + userKey = personalPolicySentinel + } var count int64 - err := s.assets.pool.QueryRow(ctx, `INSERT INTO gateway.tool_rate_usage(tool_id,window_start,call_count) - VALUES($1,date_trunc('minute',clock_timestamp()),1) - ON CONFLICT (tool_id,window_start) DO UPDATE SET call_count=gateway.tool_rate_usage.call_count+1 - RETURNING call_count`, tool.ID).Scan(&count) + err := s.assets.pool.QueryRow(ctx, `INSERT INTO gateway.tool_rate_usage(tool_id,portal_user_id,window_start,call_count) + VALUES($1,$2,date_trunc('minute',clock_timestamp()),1) + ON CONFLICT (tool_id,portal_user_id,window_start) DO UPDATE SET call_count=gateway.tool_rate_usage.call_count+1 + RETURNING call_count`, tool.ID, userKey).Scan(&count) if err != nil { return false, err } - if count > int64(tool.RateLimitRPM) { + if count > int64(limit) { return false, ErrToolRateLimited } } return true, nil } +// personalPolicySentinel 是非个人调用的限流窗口占位键(全零 UUID)。 +const personalPolicySentinel = "00000000-0000-0000-0000-000000000000" + +// personalPolicy 解析 API Key 对应的门户用户及其个人智能体安全策略。 +// 非门户 Key 返回空 userID 与默认策略(不跳过审批、倍数 1)。 +func (s *ToolService) personalPolicy(ctx context.Context, apiKeyID string) (string, AgentPolicy, error) { + if strings.TrimSpace(apiKeyID) == "" { + return "", AgentPolicy{RateLimitMultiplier: 1}, nil + } + var portalUserID *string + err := s.assets.pool.QueryRow(ctx, `SELECT portal_user_id::text FROM gateway.api_keys WHERE id=$1`, apiKeyID).Scan(&portalUserID) + if err != nil || portalUserID == nil || *portalUserID == "" { + return "", AgentPolicy{RateLimitMultiplier: 1}, nil + } + var policy AgentPolicy + err = s.assets.pool.QueryRow(ctx, `SELECT auto_approve_tools,rate_limit_multiplier FROM gateway.portal_agent_policies WHERE portal_user_id=$1`, *portalUserID).Scan(&policy.AutoApproveTools, &policy.RateLimitMultiplier) + if errors.Is(err, pgx.ErrNoRows) { + policy.RateLimitMultiplier = 1 + return *portalUserID, policy, nil + } + if err != nil { + return "", AgentPolicy{}, err + } + return *portalUserID, policy, nil +} + // ListApprovalRequests 返回工具审批申请(含工具信息)。 func (s *ToolService) ListApprovalRequests(ctx context.Context, status string) ([]map[string]any, error) { if s == nil || s.assets == nil || s.assets.pool == nil { @@ -351,8 +392,9 @@ func (s *ToolService) Execute(ctx context.Context, tool Tool, input map[string]a if err = validateToolInput(tool.InputSchema, input); err != nil { return nil, err } - // 治理校验:审批标记 + 频率上限。被拒时不算一次成功调用(但会记 tool_runs 失败)。 - if allowed, governanceErr := s.enforceGovernance(ctx, tool); !allowed { + // 治理校验:审批标记 + 频率上限(个人策略可跳过审批/放宽限流)。被拒时 + // 不算一次成功调用(但会记 tool_runs 失败)。 + if allowed, governanceErr := s.enforceGovernance(ctx, tool, apiKeyID); !allowed { return nil, governanceErr } headers, err := s.headers(tool) diff --git a/migrations/000046_agent_tasks.sql b/migrations/000046_agent_tasks.sql new file mode 100644 index 0000000..f5aad22 --- /dev/null +++ b/migrations/000046_agent_tasks.sql @@ -0,0 +1,46 @@ +-- 000046_agent_tasks.sql — 智能体节点任务下发/认领/上报 + 个人智能体安全策略。 + +-- 任务表:管理员下发,节点经令牌认证后按池认领(SKIP LOCKED),执行后上报结果; +-- 失败按 attempts/max_attempts 退避重入队。 +CREATE TABLE IF NOT EXISTS gateway.agent_tasks ( + id uuid PRIMARY KEY, + task_type varchar(32) NOT NULL CHECK (task_type IN ('prompt', 'http', 'mcp_invoke', 'skill_run', 'digital_employee', 'custom')), + payload jsonb NOT NULL DEFAULT '{}'::jsonb, + pool_type varchar(16) NOT NULL DEFAULT 'private' CHECK (pool_type IN ('public', 'private')), + pool_code varchar(64) NOT NULL DEFAULT 'default', + node_id uuid REFERENCES gateway.agent_nodes(id) ON DELETE SET NULL, + status varchar(16) NOT NULL DEFAULT 'queued' CHECK (status IN ('queued', 'claimed', 'running', 'succeeded', 'failed', 'cancelled')), + claim_token varchar(64), + attempts integer NOT NULL DEFAULT 0 CHECK (attempts >= 0), + max_attempts integer NOT NULL DEFAULT 3 CHECK (max_attempts BETWEEN 1 AND 10), + result jsonb, + error text NOT NULL DEFAULT '' CHECK (length(error) <= 4000), + created_by uuid REFERENCES gateway.admin_accounts(id) ON DELETE SET NULL, + available_at timestamptz NOT NULL DEFAULT clock_timestamp(), + created_at timestamptz NOT NULL DEFAULT clock_timestamp(), + claimed_at timestamptz, + finished_at timestamptz +); + +CREATE INDEX IF NOT EXISTS agent_tasks_queue_idx + ON gateway.agent_tasks (status, available_at, created_at) + WHERE status IN ('queued', 'claimed'); +CREATE INDEX IF NOT EXISTS agent_tasks_time_idx + ON gateway.agent_tasks (created_at DESC); + +-- 工具限流窗口支持按用户拆分:portal_user_id 为全零哨兵表示非个人调用(全局窗口), +-- 个人调用按 (tool_id, portal_user_id) 独立计数,便于个人限流倍数。 +ALTER TABLE gateway.tool_rate_usage + DROP CONSTRAINT IF EXISTS tool_rate_usage_pkey; + +ALTER TABLE gateway.tool_rate_usage + ADD COLUMN IF NOT EXISTS portal_user_id uuid NOT NULL DEFAULT '00000000-0000-0000-0000-000000000000', + ADD PRIMARY KEY (tool_id, portal_user_id, window_start); + +-- 个人智能体安全策略:个人调用跳过工具审批标记、按倍数放宽个人限流。 +CREATE TABLE IF NOT EXISTS gateway.portal_agent_policies ( + portal_user_id uuid PRIMARY KEY REFERENCES gateway.portal_users(id) ON DELETE CASCADE, + auto_approve_tools boolean NOT NULL DEFAULT false, + rate_limit_multiplier integer NOT NULL DEFAULT 1 CHECK (rate_limit_multiplier BETWEEN 1 AND 100), + updated_at timestamptz NOT NULL DEFAULT clock_timestamp() +); diff --git a/web/apps/admin/src/api/agent-nodes.ts b/web/apps/admin/src/api/agent-nodes.ts index 314f9d2..a3a9412 100644 --- a/web/apps/admin/src/api/agent-nodes.ts +++ b/web/apps/admin/src/api/agent-nodes.ts @@ -67,3 +67,16 @@ export function rotateAgentNodeToken(id: string) { export function previewAgentNodeRoute(data: AgentNodeRoutePreviewInput) { return request.post({ url: '/api/v1/admin/agent-nodes/route-preview', params: data }) } + +// --- 节点任务下发 --- +export interface AgentTask { + id: string; task_type: string; payload: Record + pool_type: string; pool_code: string; node_id?: string; node_code: string + status: string; attempts: number; max_attempts: number + result?: Record; error: string + created_at: string; claimed_at?: string; finished_at?: string; available_at: string +} +export const fetchAgentTasks=(status='')=>request.get({url:'/api/v1/admin/agent-tasks',params:{status}}) +export const createAgentTask=(params:{task_type:string;payload:Record;pool_type?:string;pool_code?:string;node_id?:string;max_attempts?:number})=>request.post({url:'/api/v1/admin/agent-tasks',params}) +export const cancelAgentTask=(id:string)=>request.post<{cancelled:boolean}>({url:`/api/v1/admin/agent-tasks/${id}/cancel`}) +export const retryAgentTask=(id:string)=>request.post<{retried:boolean}>({url:`/api/v1/admin/agent-tasks/${id}/retry`}) diff --git a/web/apps/admin/src/views/gateway/agent-tasks/index.vue b/web/apps/admin/src/views/gateway/agent-tasks/index.vue new file mode 100644 index 0000000..355ea4f --- /dev/null +++ b/web/apps/admin/src/views/gateway/agent-tasks/index.vue @@ -0,0 +1,162 @@ + + diff --git a/web/apps/portal/src/api/portal.ts b/web/apps/portal/src/api/portal.ts index d81c1c1..95a4f03 100644 --- a/web/apps/portal/src/api/portal.ts +++ b/web/apps/portal/src/api/portal.ts @@ -109,3 +109,8 @@ export interface EmployeeRun { employee_code:string;employee_name:string;status: export const fetchDigitalEmployees=()=>request.get({url:'/api/v1/portal/digital-employees'}) export const runDigitalEmployee=(code:string,message:string)=>request.post>({url:`/api/v1/portal/digital-employees/${code}/chat`,params:{message}}) export const fetchMyEmployeeRuns=(limit=20)=>request.get({url:'/api/v1/portal/digital-employees/runs',params:{limit}}) + +// --- 个人智能体安全策略 --- +export interface AgentPolicy { auto_approve_tools:boolean;rate_limit_multiplier:number } +export const fetchAgentPolicy=()=>request.get({url:'/api/v1/portal/agent-policy'}) +export const setAgentPolicy=(params:AgentPolicy)=>request.put<{saved:boolean}>({url:'/api/v1/portal/agent-policy',params}) diff --git a/web/apps/portal/src/views/portal/security/index.vue b/web/apps/portal/src/views/portal/security/index.vue index 4c609d8..cf61e74 100644 --- a/web/apps/portal/src/views/portal/security/index.vue +++ b/web/apps/portal/src/views/portal/security/index.vue @@ -38,6 +38,26 @@ + + +
+
+
+
个人工具自动审批
+

开启后你通过个人凭据调用需审批的工具时不再要求管理员审批（平台级审批标记仍适用于其他调用）

+
+ +
+
+
+
个人限流倍数
+

你的个人调用按此倍数放宽工具限流（例如工具限流 10 rpm、倍数 2 时个人可调用 20 rpm）

+
+ +
+
+ +