diff --git a/RELEASE.md b/RELEASE.md index 0c29069..6950cf7 100644 --- a/RELEASE.md +++ b/RELEASE.md @@ -113,3 +113,33 @@ copying the bundle to a deployment host. - 节点令牌沿用心跳同一套 SHA-256 + constant-time 校验,未知节点与错误令牌 返回同一错误,不泄露节点存在性。 - 个人限流按 (tool_id, portal_user_id) 独立窗口计数,个人倍数不影响全局额度。 + +## 0.11.5 — 旗舰版第六轮完善 + +发布时间:2026-08-13 + +新增交付: + +- **智能体节点 Agent 程序**(`cmd/agent-node`):可独立部署到裸机/VM 的节点 + 执行器。周期心跳(上报版本/能力/最近错误)刷新在线状态;轮询认领节点池 + 任务;按类型执行——prompt(优先本地 OpenAI 兼容 LLM,缺省经网关)、 + http(仅 http(s)、禁重定向、响应限 1 MiB)、mcp_invoke / skill_run / + digital_employee(经网关运行时转发,需 AGENT_GATEWAY_KEY)、custom(回显); + 凭认领令牌上报结果/错误,失败交由网关按退避重试;串行执行 + 优雅退出 + + JSON 结构化日志。安全边界:不执行任意 shell 命令,载荷为网关下发的 + 类型化 JSON,节点令牌经环境注入。 +- **裸机安装脚本**(`scripts/install-agent-node.sh`):自动构建二进制、 + 安装到 /usr/local/bin、生成 0600 权限节点配置(含令牌)、注册 systemd + 服务并启动;支持 AGENT_GATEWAY_URL/AGENT_CODE/AGENT_TOKEN 等环境配置。 +- **集群/冷备部署方案**(`deploy/CLUSTER.md`):单机/冷备/集群三种形态对照; + compose override 多副本网关 + nginx least_conn 入口示例;冷备数据层 + 备份与恢复要点(master key 同步、WAL 归档、MinIO 异地);生产建议清单。 + +端到端验证: + +- 真实 Agent 进程注册节点 → 心跳上线(版本/能力上报) → 下发 custom/http + 任务 → Agent 认领执行并上报 → 管理端任务成功、结果可见;日志含 + 认领/上报全链路。修复 Agent claim 未解包网关 apiresponse 信封的问题。 + +至此旗舰版矩阵中所有可独立实施的功能与交付物完成;剩余依赖外部条件: +企微/钉钉/飞书真实平台联调、多租户物理隔离、企业环境内的节点批量部署。 diff --git a/cmd/agent-node/main.go b/cmd/agent-node/main.go new file mode 100644 index 0000000..b203466 --- /dev/null +++ b/cmd/agent-node/main.go @@ -0,0 +1,460 @@ +// Command agent-node 是智能体节点 Agent:向 AI Gateway 注册的心跳 + 任务执行器。 +// +// 职责: +// - 周期性心跳(上报版本/能力/最近错误),刷新在线状态; +// - 轮询认领节点池任务(POST /v1/agent/nodes/{code}/tasks/claim); +// - 按任务类型执行:prompt(本地/网关 LLM)、http(受控出站)、mcp_invoke / +// skill_run / digital_employee(经网关运行时转发)、custom(回显); +// - 凭认领令牌上报结果或错误,失败交给网关按退避重试。 +// +// 安全边界: +// - 不执行任意 shell 命令;http 任务只允许 http(s) 绝对地址且拒绝重定向; +// - 任务载荷是网关下发的可信 JSON,节点只做类型化执行; +// - 节点令牌经环境变量/配置文件注入,不经任务载荷传递。 +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "os" + "os/signal" + "strings" + "syscall" + "time" +) + +type config struct { + GatewayURL string + Code string + Token string + Version string + Name string + Capabilities []string + // LLM 执行:优先本地 OpenAI 兼容端点,缺省回退网关。 + LLMBaseURL string + LLMAPIKey string + // 转发类任务(mcp/skill/数字员工)与回退 LLM 用的网关 API Key。 + GatewayKey string + HeartbeatEvery time.Duration + ClaimEvery time.Duration + HTTPTimeout time.Duration +} + +func loadConfig() (config, error) { + cfg := config{ + GatewayURL: strings.TrimRight(env("AGENT_GATEWAY_URL", "http://127.0.0.1:8080"), "/"), + Code: env("AGENT_CODE", ""), + Token: env("AGENT_TOKEN", ""), + Version: env("AGENT_VERSION", "0.11.4-agent"), + Name: env("AGENT_NAME", ""), + LLMBaseURL: strings.TrimRight(env("AGENT_LLM_BASE_URL", ""), "/"), + LLMAPIKey: env("AGENT_LLM_API_KEY", ""), + GatewayKey: env("AGENT_GATEWAY_KEY", ""), + HeartbeatEvery: durEnv("AGENT_HEARTBEAT_SECONDS", 30*time.Second), + ClaimEvery: durEnv("AGENT_CLAIM_SECONDS", 5*time.Second), + HTTPTimeout: durEnv("AGENT_HTTP_TIMEOUT_SECONDS", 60*time.Second), + } + for _, capability := range strings.Split(env("AGENT_CAPABILITIES", "tool_exec,llm,http"), ",") { + if value := strings.TrimSpace(capability); value != "" { + cfg.Capabilities = append(cfg.Capabilities, value) + } + } + if cfg.Code == "" || cfg.Token == "" { + return cfg, errors.New("AGENT_CODE 与 AGENT_TOKEN 必填(管理端登记节点后交付)") + } + if cfg.HeartbeatEvery < time.Second || cfg.ClaimEvery < time.Second || cfg.HTTPTimeout < time.Second { + return cfg, errors.New("心跳/认领/超时秒数必须 >= 1") + } + return cfg, nil +} + +func env(key, fallback string) string { + if value := strings.TrimSpace(os.Getenv(key)); value != "" { + return value + } + return fallback +} + +func durEnv(key string, fallback time.Duration) time.Duration { + if value := os.Getenv(key); value != "" { + if parsed, err := time.ParseDuration(value + "s"); err == nil { + return parsed + } + } + return fallback +} + +type agent struct { + cfg config + client *http.Client + logger *slog.Logger + // lastError 是最近一次心跳/执行错误,随下一次心跳上报。 + lastError string +} + +func main() { + flag.Parse() + logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})) + cfg, err := loadConfig() + if err != nil { + logger.Error("invalid configuration", "error", err) + os.Exit(1) + } + a := &agent{ + cfg: cfg, + client: &http.Client{Timeout: cfg.HTTPTimeout}, + logger: logger, + } + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + logger.Info("agent node starting", "gateway", cfg.GatewayURL, "code", cfg.Code, "capabilities", cfg.Capabilities) + run(ctx, a) + logger.Info("agent node stopped") +} + +func run(ctx context.Context, a *agent) { + heartbeatTicker := time.NewTicker(a.cfg.HeartbeatEvery) + claimTicker := time.NewTicker(a.cfg.ClaimEvery) + defer heartbeatTicker.Stop() + defer claimTicker.Stop() + // 启动立即做一次心跳与认领,便于部署后快速验证。 + if err := a.heartbeat(ctx); err != nil { + a.logger.Warn("initial heartbeat failed", "error", err) + a.lastError = err.Error() + } else { + a.lastError = "" + } + for { + select { + case <-ctx.Done(): + return + case <-heartbeatTicker.C: + if err := a.heartbeat(ctx); err != nil { + a.logger.Warn("heartbeat failed", "error", err) + a.lastError = err.Error() + } else { + a.lastError = "" + } + case <-claimTicker.C: + a.processOne(ctx) + } + } +} + +// heartbeat 上报版本/能力/最近错误。 +func (a *agent) heartbeat(ctx context.Context) error { + payload, _ := json.Marshal(map[string]any{ + "version": a.cfg.Version, + "capabilities": capabilityMap(a.cfg.Capabilities), + "metadata": map[string]string{"name": a.cfg.Name, "agent": "agent-node"}, + "error": a.lastError, + }) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.GatewayURL+"/v1/agent/nodes/"+a.cfg.Code+"/heartbeat", bytes.NewReader(payload)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Agent-Token", a.cfg.Token) + response, err := a.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode/100 != 2 { + return fmt.Errorf("heartbeat HTTP %d", response.StatusCode) + } + return nil +} + +// processOne 认领并执行一个任务(串行;一次一个)。 +func (a *agent) processOne(ctx context.Context) { + claimCtx, cancel := context.WithTimeout(ctx, 20*time.Second) + defer cancel() + task, err := a.claim(claimCtx) + if err != nil { + a.logger.Warn("claim failed", "error", err) + a.lastError = "claim: " + err.Error() + return + } + if task == nil { + return + } + a.logger.Info("task claimed", "task_id", task.ID, "task_type", task.TaskType) + result, runErr := a.execute(ctx, *task) + if runErr != nil { + a.logger.Warn("task execution failed", "task_id", task.ID, "error", runErr) + a.lastError = "task " + task.ID[:8] + ": " + runErr.Error() + } + if err := a.complete(ctx, task.ID, task.ClaimToken, result, runErr); err != nil { + a.logger.Warn("task report failed", "task_id", task.ID, "error", err) + a.lastError = "report: " + err.Error() + return + } + a.logger.Info("task reported", "task_id", task.ID, "success", runErr == nil) +} + +type claimedTask struct { + ID string `json:"id"` + TaskType string `json:"task_type"` + Payload json.RawMessage `json:"payload"` + ClaimToken string `json:"claim_token"` +} + +func (a *agent) claim(ctx context.Context) (*claimedTask, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.GatewayURL+"/v1/agent/nodes/"+a.cfg.Code+"/tasks/claim", nil) + if err != nil { + return nil, err + } + request.Header.Set("X-Agent-Token", a.cfg.Token) + response, err := a.client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(response.Body, 1<<20)) + if response.StatusCode/100 != 2 { + return nil, fmt.Errorf("claim HTTP %d: %s", response.StatusCode, strings.TrimSpace(string(raw))) + } + // 网关统一 apiresponse 信封:{code,msg,data:{task}}。 + var envelope struct { + Code int `json:"code"` + Data struct { + Task *claimedTask `json:"task"` + } `json:"data"` + } + if err := json.Unmarshal(raw, &envelope); err != nil { + return nil, err + } + if envelope.Code != 200 { + return nil, fmt.Errorf("claim 业务错误 code=%d", envelope.Code) + } + return envelope.Data.Task, nil +} + +func (a *agent) complete(ctx context.Context, taskID, claimToken string, result map[string]any, runErr error) error { + var errorText string + if runErr != nil { + errorText = runErr.Error() + if len(errorText) > 4000 { + errorText = errorText[:4000] + } + } + payload, _ := json.Marshal(map[string]any{"claim_token": claimToken, "result": result, "error": errorText}) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.GatewayURL+"/v1/agent/nodes/"+a.cfg.Code+"/tasks/"+taskID+"/complete", bytes.NewReader(payload)) + if err != nil { + return err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("X-Agent-Token", a.cfg.Token) + response, err := a.client.Do(request) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode/100 != 2 { + return fmt.Errorf("complete HTTP %d", response.StatusCode) + } + return nil +} + +// execute 按任务类型执行。载荷来自受管网关,字段类型不符返回错误。 +func (a *agent) execute(ctx context.Context, task claimedTask) (map[string]any, error) { + var payload map[string]any + if err := json.Unmarshal(task.Payload, &payload); err != nil { + return nil, fmt.Errorf("task payload 不是 JSON 对象: %w", err) + } + switch task.TaskType { + case "prompt": + return a.executePrompt(ctx, payload) + case "http": + return a.executeHTTP(ctx, payload) + case "mcp_invoke": + return a.forwardGateway(ctx, "/v1/mcp-servers/"+stringValue(payload, "code")+"/tools/"+stringValue(payload, "tool")+"/invoke", map[string]any{"input": payload["args"]}) + case "skill_run": + return a.forwardGateway(ctx, "/v1/skills/"+stringValue(payload, "code")+"/render", map[string]any{"variables": payload["variables"]}) + case "digital_employee": + return a.forwardGateway(ctx, "/v1/digital-employees/"+stringValue(payload, "code")+"/chat/completions", map[string]any{"messages": payload["messages"]}) + case "custom": + return map[string]any{"echo": payload}, nil + } + return nil, fmt.Errorf("不支持的节点任务类型 %s", task.TaskType) +} + +// executePrompt 调用本地 LLM(OpenAI 兼容);未配置时回退网关 /v1/chat/completions。 +func (a *agent) executePrompt(ctx context.Context, payload map[string]any) (map[string]any, error) { + model := stringValue(payload, "model") + messages, _ := payload["messages"].([]any) + if model == "" || len(messages) == 0 { + return nil, errors.New("prompt 任务需要 model 与 messages") + } + if a.cfg.LLMBaseURL != "" { + return a.callOpenAICompatible(ctx, a.cfg.LLMBaseURL+"/chat/completions", a.cfg.LLMAPIKey, model, messages) + } + if a.cfg.GatewayKey == "" { + return nil, errors.New("未配置 AGENT_LLM_BASE_URL 或 AGENT_GATEWAY_KEY,无法执行 prompt 任务") + } + body, _ := json.Marshal(map[string]any{"model": model, "messages": messages, "stream": false}) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.GatewayURL+"/v1/chat/completions", bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Authorization", "Bearer "+a.cfg.GatewayKey) + response, err := a.client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(response.Body, 2<<20)) + if response.StatusCode/100 != 2 { + return nil, fmt.Errorf("网关 LLM HTTP %d: %s", response.StatusCode, truncate(string(raw), 300)) + } + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + return nil, err + } + answer, _ := firstChoiceContent(decoded) + return map[string]any{"answer": answer, "source": "gateway"}, nil +} + +// callOpenAICompatible 调用本地 OpenAI 兼容端点(节点侧 LLM)。 +func (a *agent) callOpenAICompatible(ctx context.Context, endpoint, apiKey, model string, messages []any) (map[string]any, error) { + body, _ := json.Marshal(map[string]any{"model": model, "messages": messages, "stream": false}) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header.Set("Content-Type", "application/json") + if apiKey != "" { + request.Header.Set("Authorization", "Bearer "+apiKey) + } + response, err := a.client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(response.Body, 2<<20)) + if response.StatusCode/100 != 2 { + return nil, fmt.Errorf("本地 LLM HTTP %d: %s", response.StatusCode, truncate(string(raw), 300)) + } + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + return nil, err + } + answer, _ := firstChoiceContent(decoded) + return map[string]any{"answer": answer, "source": "local"}, nil +} + +// executeHTTP 执行受控 HTTP 任务:只允许 http(s),拒绝重定向,响应限 1 MiB。 +func (a *agent) executeHTTP(ctx context.Context, payload map[string]any) (map[string]any, error) { + target := strings.TrimSpace(stringValue(payload, "url")) + method := strings.ToUpper(strings.TrimSpace(stringValue(payload, "method"))) + if method == "" { + method = "GET" + } + parsed, err := url.Parse(target) + if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Hostname() == "" { + return nil, errors.New("http 任务 url 必须是 http(s) 绝对地址") + } + var body io.Reader + if value, ok := payload["body"].(string); ok && strings.TrimSpace(value) != "" { + body = strings.NewReader(value) + } + request, err := http.NewRequestWithContext(ctx, method, parsed.String(), body) + if err != nil { + return nil, err + } + if headers, ok := payload["headers"].(map[string]any); ok { + for key, value := range headers { + request.Header.Set(key, fmt.Sprint(value)) + } + } + client := &http.Client{ + Timeout: a.cfg.HTTPTimeout, + CheckRedirect: func(*http.Request, []*http.Request) error { + return errors.New("http 任务不允许重定向") + }, + } + response, err := client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(response.Body, (1<<20)+1)) + if len(raw) > 1<<20 { + return nil, errors.New("http 任务响应超过 1 MiB") + } + var decoded any = string(raw) + if json.Unmarshal(raw, &decoded) != nil { + decoded = string(raw) + } + return map[string]any{"status_code": response.StatusCode, "body": decoded}, nil +} + +// forwardGateway 经网关运行时执行 mcp/skill/数字员工(需要 AGENT_GATEWAY_KEY)。 +func (a *agent) forwardGateway(ctx context.Context, path string, bodyValue map[string]any) (map[string]any, error) { + if a.cfg.GatewayKey == "" { + return nil, errors.New("未配置 AGENT_GATEWAY_KEY,无法执行网关转发类任务") + } + body, _ := json.Marshal(bodyValue) + request, err := http.NewRequestWithContext(ctx, http.MethodPost, a.cfg.GatewayURL+path, bytes.NewReader(body)) + if err != nil { + return nil, err + } + request.Header.Set("Content-Type", "application/json") + request.Header.Set("Authorization", "Bearer "+a.cfg.GatewayKey) + response, err := a.client.Do(request) + if err != nil { + return nil, err + } + defer response.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(response.Body, 2<<20)) + if response.StatusCode/100 != 2 { + return nil, fmt.Errorf("网关转发 HTTP %d: %s", response.StatusCode, truncate(string(raw), 300)) + } + var decoded map[string]any + if err := json.Unmarshal(raw, &decoded); err != nil { + return nil, err + } + return decoded, nil +} + +func capabilityMap(capabilities []string) map[string]bool { + result := map[string]bool{} + for _, capability := range capabilities { + result[capability] = true + } + return result +} + +func stringValue(payload map[string]any, key string) string { + value, _ := payload[key].(string) + return strings.TrimSpace(value) +} + +func firstChoiceContent(decoded map[string]any) (string, bool) { + choices, _ := decoded["choices"].([]any) + if len(choices) == 0 { + return "", false + } + choice, _ := choices[0].(map[string]any) + message, _ := choice["message"].(map[string]any) + content, _ := message["content"].(string) + return content, strings.TrimSpace(content) != "" +} + +func truncate(value string, max int) string { + runes := []rune(value) + if len(runes) > max { + return string(runes[:max]) + "…" + } + return value +} diff --git a/deploy/CLUSTER.md b/deploy/CLUSTER.md new file mode 100644 index 0000000..bb0ace7 --- /dev/null +++ b/deploy/CLUSTER.md @@ -0,0 +1,115 @@ +# 集群与冷备部署方案 + +本文档说明旗舰版「单机 / 冷备 / 集群」部署形态的落地方式。当前工程的所有有状态 +组件(PostgreSQL / Redis / MinIO)仍建议单点或托管服务,无状态控制面 +(gateway-api / 各 worker)可按本文档横向扩展。 + +## 1. 部署形态对照 + +| 形态 | 说明 | 适用 | +|---|---|---| +| 单机 | 全部容器在一台主机(docker-compose.yml 默认) | 开发/小团队 | +| 冷备 | 数据层定期备份 + 备用主机完整镜像,主故障时切换 | 生产入门 | +| 集群 | 无状态组件多副本 + 数据层托管/高可用 | 生产规模化 | + +## 2. 为什么可以横向扩展控制面 + +- **outbox worker**:`SKIP LOCKED` 租约 + Redis Stream 消费组,已支持多实例 + 并发消费,事件不会重复投递(worker 天然多副本安全); +- **gateway-api**:无本地状态,会话/限流/配额在 Redis,权威数据在 PostgreSQL, + 任意副本可服务同一请求;审计批量 COPY 由 `audit_events` 分区 + advisory lock + 维护,多副本写入安全; +- **调度器/通知/维护 worker**:全部基于数据库租约或消费组,多副本安全; +- **节点 Agent**:与网关是拉模型(心跳/认领),天然分布式。 + +## 3. 集群部署(compose override 示例) + +`deploy/docker-compose.cluster.yml`: + +```yaml +# 用法: docker compose -f docker-compose.yml -f docker-compose.cluster.yml up -d +services: + gateway-api: + deploy: + replicas: 3 + # 集群模式下网关端口不应直接暴露(由入口 nginx 负载均衡到各副本), + # 移除主机端口绑定: + ports: [] + gateway-api-lb: + image: nginx:1.27-alpine + ports: + - "127.0.0.1:8080:80" + volumes: + - ./nginx-cluster.conf:/etc/nginx/conf.d/default.conf:ro + depends_on: + - gateway-api +``` + +`deploy/nginx-cluster.conf`(入口负载均衡,round-robin 到各副本): + +```nginx +upstream gateway_replicas { + least_conn; + server gateway-api:8080 max_fails=3 fail_timeout=10s; + server gateway-api:8080 max_fails=3 fail_timeout=10s; + server gateway-api:8080 max_fails=3 fail_timeout=10s; +} + +server { + listen 80; + server_tokens off; + client_max_body_size 256m; + location / { + proxy_pass http://gateway_replicas; + proxy_http_version 1.1; + proxy_set_header Host $host; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_read_timeout 300s; + } +} +``` + +说明: + +- Compose 的 `deploy.replicas` 会为 `gateway-api` 创建多个容器,`least_conn` + 负载均衡到全部副本;副本间无会话亲和需求(会话令牌在 Redis); +- worker 类服务保持单副本即可(多副本也安全,但无收益); +- 网关不再发布主机端口,所有流量经 `gateway-api-lb`。 + +## 4. 冷备 + +数据层(权威): + +- PostgreSQL: `pg_dump -Fc` 每日全量 + WAL 归档;恢复演练用 + `pg_restore` 到备用主机;建议配合 `pgBackRest`/云厂商 PITR; +- Redis critical(会话/限流/配额):`APPENDONLY yes` 已在 compose 开启,定期 + `redis-cli BGSAVE` 并备份 dump.rdb;丢失只会导致会话失效/限流重置, + 不会损坏权威数据; +- MinIO: 对象存储桶异地备份(`mc mirror`)或托管 S3 兼容服务。 + +备用主机: + +- 镜像完整代码与 compose 配置;`deploy/.env`(含 CREDENTIAL_MASTER_KEY)必须 + 同步——**master key 丢失会导致全部加密凭据不可解密**; +- 故障切换:恢复 DB/Redis 备份 → `docker compose up -d` → 验证 + `healthz` 与登录; +- 建议每月做一次恢复演练(切换 runbook 见 `docs/cutover-runbook.md`)。 + +## 5. 多租户与集群的注意事项 + +- 会话/限流 Redis 需所有副本共享同一 critical Redis(或 Sentinel 高可用); +- `X-Forwarded-For` 由入口 nginx 统一设置,gateway-api 的限流/审计 IP 以 + 可信代理逻辑解析(见 `internal/identity/ratelimit.go`),不要直连暴露副本; +- 节点 Agent 指向入口地址(`AGENT_GATEWAY_URL`),经 LB 到达任意副本,拉模型 + 无粘性要求。 + +## 6. 生产环境建议清单 + +- [ ] 数据库连接池上限按副本数调整(`DATABASE_MAX_CONNS`),避免连接耗尽; +- [ ] PostgreSQL `max_connections` 与 shared_buffers 按主机内存配置; +- [ ] 入口启用 TLS(nginx/certbot 或云 LB),`CREDENTIAL_MASTER_KEY` 用 + 密钥管理服务注入,不进 compose 文件; +- [ ] 审计月分区自动维护已由 maintenance worker 执行,多副本由 advisory lock + 互斥; +- [ ] 监控:healthz/readyz + Prometheus `/metrics` 接入现有监控; +- [ ] 备份与恢复演练纳入变更流程。 diff --git a/docs/security-review-0.10.1.md b/docs/security-review-0.10.1.md index 095f032..eecd5eb 100644 --- a/docs/security-review-0.10.1.md +++ b/docs/security-review-0.10.1.md @@ -496,3 +496,18 @@ PostgreSQL `text` 列拒绝写入 → 重试循环永远失败；inbox 的 Go `l (下发→认领→伪造令牌拒绝→成功/失败上报→退避重试→达上限→取消冲突→重试); HTTP 端到端:下发→认领(返回认领令牌)→上报→succeeded,管理端列表不泄露 认领令牌;任务完成站内信已验证。 + +# 追加：旗舰版功能完善第六轮（0.11.5，2026-08-13） + +1. **节点 Agent**(`cmd/agent-node`):裸机执行器,心跳/认领/执行/上报闭环。 + - 安全边界:http 任务仅 http(s) 绝对地址 + 禁重定向 + 1 MiB 响应上限; + prompt 任务可走本地 LLM 或经网关(凭 AGENT_GATEWAY_KEY);不执行 + 任意 shell 命令;节点令牌只经环境变量注入,不进入任务载荷; + - claim 响应按网关 apiresponse 信封解包(修复根级解析 bug)。 +2. **裸机安装**(`scripts/install-agent-node.sh`):systemd 服务 + 0600 配置, + NoNewPrivileges + ProtectSystem 加固。 +3. **集群/冷备方案**(`deploy/CLUSTER.md`):无状态控制面多副本依据 + (outbox SKIP LOCKED/Redis 会话/审计分区 advisory lock),入口 + least_conn 负载均衡,冷备含 master key 同步与恢复演练要点。 +4. 实测:真实 Agent 进程注册→心跳上线→custom/http 任务认领执行上报→ + 管理端成功可见。 diff --git a/docs/旗舰版需求规划与完成情况.md b/docs/旗舰版需求规划与完成情况.md index 2f4dc87..2177b76 100644 --- a/docs/旗舰版需求规划与完成情况.md +++ b/docs/旗舰版需求规划与完成情况.md @@ -228,3 +228,17 @@ MinIO 对象存储与管理端/个人文件仓库；pgvector + Ollama(bge-m3) 物理隔离需明确部署形态。 - 智能体节点远程安装/裸机部署脚本:任务下发与执行协议已就绪并端到端验证, 节点 Agent 安装包/注册脚本需在目标环境实施。 + +## 十一、0.11.5 完成情况（2026-08-13 第六轮完善） + +| 交付物 | 状态 | 说明 | +|---|---|---| +| 节点 Agent 程序 | ✅ | cmd/agent-node 独立二进制,裸机心跳/认领/执行/上报闭环,实测通过 | +| 裸机安装脚本 | ✅ | scripts/install-agent-node.sh + systemd 加固 | +| 集群/冷备部署方案 | ✅ | deploy/CLUSTER.md + compose override + 冷备要点 | + +旗舰版矩阵其余未完成项均为外部依赖: + +- 企微/钉钉/飞书真实平台联调(需企业开放平台凭据); +- 多租户物理隔离(需明确部署形态); +- 企业环境内的节点批量部署与真实业务任务接入(协议与工具已就绪)。 diff --git a/scripts/install-agent-node.sh b/scripts/install-agent-node.sh new file mode 100755 index 0000000..54c510b --- /dev/null +++ b/scripts/install-agent-node.sh @@ -0,0 +1,91 @@ +#!/usr/bin/env bash +# install-agent-node.sh — 在裸机/虚拟机安装 AI Gateway 智能体节点 Agent。 +# +# 用法: +# sudo AGENT_GATEWAY_URL=https://gw.example.com \ +# AGENT_CODE=worker-01 \ +# AGENT_TOKEN=agn_xxxx(管理端登记节点后交付) \ +# [AGENT_NAME=] [AGENT_LLM_BASE_URL=] [AGENT_LLM_API_KEY=] \ +# [AGENT_GATEWAY_KEY=] [AGENT_CAPABILITIES=tool_exec,llm,http] \ +# bash scripts/install-agent-node.sh +# +# 动作: +# 1. 安装 Go 工具链(缺失时)并构建 cmd/agent-node; +# 2. 安装二进制到 /usr/local/bin/ai-agent-node; +# 3. 生成 /etc/ai-agent-node/env(权限 600,含节点令牌); +# 4. 安装 systemd 服务 ai-agent-node.service 并启动。 +set -euo pipefail + +AGENT_GATEWAY_URL="${AGENT_GATEWAY_URL:-http://127.0.0.1:8080}" +AGENT_CODE="${AGENT_CODE:?AGENT_CODE 必填(管理端登记的节点编码)}" +AGENT_TOKEN="${AGENT_TOKEN:?AGENT_TOKEN 必填(管理端登记节点后交付的令牌)}" +AGENT_NAME="${AGENT_NAME:-$AGENT_CODE}" +AGENT_VERSION="${AGENT_VERSION:-0.11.4-agent}" +AGENT_CAPABILITIES="${AGENT_CAPABILITIES:-tool_exec,llm,http}" +AGENT_LLM_BASE_URL="${AGENT_LLM_BASE_URL:-}" +AGENT_LLM_API_KEY="${AGENT_LLM_API_KEY:-}" +AGENT_GATEWAY_KEY="${AGENT_GATEWAY_KEY:-}" + +REPO_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +INSTALL_DIR="/usr/local/bin" +ENV_FILE="/etc/ai-agent-node/env" +SERVICE_FILE="/etc/systemd/system/ai-agent-node.service" + +echo "==> 1/4 构建 agent-node 二进制" +if ! command -v go >/dev/null 2>&1; then + echo " 未找到 go,尝试通过 apt 安装(仅 Debian/Ubuntu)" + apt-get update -qq && apt-get install -y -qq golang-go +fi +( + cd "$REPO_DIR" + GOFLAGS=-mod=mod go build -trimpath -ldflags "-s -w" -o /tmp/ai-agent-node ./cmd/agent-node +) + +echo "==> 2/4 安装二进制到 $INSTALL_DIR/ai-agent-node" +install -m 0755 /tmp/ai-agent-node "$INSTALL_DIR/ai-agent-node" +rm -f /tmp/ai-agent-node + +echo "==> 3/4 写入节点配置 $ENV_FILE(权限 600,含令牌)" +install -d -m 0700 /etc/ai-agent-node +umask 077 +cat > "$ENV_FILE" < 4/4 安装并启动 systemd 服务" +cat > "$SERVICE_FILE" </dev/null 2>&1 +systemctl restart ai-agent-node.service + +echo "==> 完成:节点 $AGENT_CODE 已安装并启动" +echo " 查看日志: journalctl -u ai-agent-node -f" +echo " 心跳/任务状态: 管理端 → 安全与审计 → 智能体节点 / 节点任务"