diff --git a/heicode/controller/agent_runtime_client.go b/heicode/controller/agent_runtime_client.go index 2650bc65..60101c7d 100644 --- a/heicode/controller/agent_runtime_client.go +++ b/heicode/controller/agent_runtime_client.go @@ -526,6 +526,29 @@ func agentRuntimeStopPath(cfg agentRuntimeConfig, runtimeDeploymentID string) st return strings.ReplaceAll(path, "{deployment_id}", url.PathEscape(runtimeDeploymentID)) } +// agentRuntimeSwarmInputPath / agentRuntimeSwarmResultPath 构造 swarm 运行时的追加输入 / 结果 +// 端点(agent_swarm PR#41 runtime-contract:POST/GET …/{deployment_id}/{input,result})。默认沿用 +// create-path 的 /api/agent/swarm/deployments 基路径,可经 SWARM_RUNTIME_{INPUT,RESULT}_PATH 覆盖。 +func agentRuntimeSwarmInputPath(cfg agentRuntimeConfig, runtimeDeploymentID string) string { + return swarmRuntimeDeploymentSubPath(cfg, runtimeDeploymentID, "input", "SWARM_RUNTIME_INPUT_PATH") +} + +func agentRuntimeSwarmResultPath(cfg agentRuntimeConfig, runtimeDeploymentID string) string { + return swarmRuntimeDeploymentSubPath(cfg, runtimeDeploymentID, "result", "SWARM_RUNTIME_RESULT_PATH") +} + +func swarmRuntimeDeploymentSubPath(cfg agentRuntimeConfig, runtimeDeploymentID, sub, envKey string) string { + path := strings.TrimSpace(common.GetEnvOrDefaultString(envKey, "")) + if path == "" { + base := strings.TrimRight(strings.TrimSpace(cfg.CreatePath), "/") + if base == "" { + base = "/api/agent/swarm/deployments" + } + path = base + "/{deployment_id}/" + sub + } + return strings.ReplaceAll(path, "{deployment_id}", url.PathEscape(strings.TrimSpace(runtimeDeploymentID))) +} + func agentRuntimeStopPathForRecord(cfg agentRuntimeConfig, record agentDeploymentRecord) string { // {deployment_id} falls back to the swarm id so sub-agile-style paths still // resolve for records that only persisted a runtime swarm id. diff --git a/heicode/controller/agent_swarm_query.go b/heicode/controller/agent_swarm_query.go index 6337217f..cceb6da3 100644 --- a/heicode/controller/agent_swarm_query.go +++ b/heicode/controller/agent_swarm_query.go @@ -501,6 +501,167 @@ func callSwarmRuntimeStop(ctx context.Context, cfg agentRuntimeConfig, dep model }, nil } +// callSwarmRuntimeJSON 向 Swarm 运行时发起一次 JSON 请求并归一化 envelope(复用 stop 的解析 helper)。 +// 返回 data(envelope.data 或顶层)、HTTP 状态、错误。错误信息仅含 HTTP 状态 + 运行时响应体, +// 绝不含调用方请求体(如 /input 的指令文本)。 +func callSwarmRuntimeJSON(ctx context.Context, cfg agentRuntimeConfig, method, endpoint string, body []byte, dep model.AgentDeployment) (map[string]any, int, error) { + var reader io.Reader + if len(body) > 0 { + reader = bytes.NewReader(body) + } + req, err := http.NewRequestWithContext(ctx, method, endpoint, reader) + if err != nil { + return nil, 0, err + } + if len(body) > 0 { + req.Header.Set("Content-Type", "application/json") + } + req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(cfg.Token)) + if cid := strings.TrimSpace(dep.CorrelationID); cid != "" { + req.Header.Set("X-Correlation-ID", cid) + } + client := &http.Client{Timeout: cfg.Timeout} + resp, err := client.Do(req) + if err != nil { + return nil, 0, err + } + defer resp.Body.Close() + respBody, readErr := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if readErr != nil { + return nil, resp.StatusCode, readErr + } + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, resp.StatusCode, fmt.Errorf("runtime returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(respBody))) + } + var envelope map[string]any + if len(respBody) > 0 { + if err := common.Unmarshal(respBody, &envelope); err != nil { + return nil, resp.StatusCode, err + } + } + if message := agentRuntimeEnvelopeError(envelope); message != "" { + return nil, resp.StatusCode, errors.New(message) + } + return extractAgentRuntimeData(envelope), resp.StatusCode, nil +} + +// HeicodeAppendSwarmInput: POST /api/heicode/swarms/:id/input — 追加用户输入(写路,代理到 Swarm)。 +// agent_swarm PR#41 runtime-contract:POST …/{deployment_id}/input,注入 source=user_append 任务 +// (终态 run 自动 reopen 为 running;stopped 拒绝)。**指令文本仅转发,绝不落 HM 日志/审计/事件** +// (#40「原文不进事件流」+ #46「勿落日志」)。与 stop 一样需运行时开关 + base_url + 服务令牌,缺一即拒, +// 绝不伪造受理。 +func HeicodeAppendSwarmInput(c *gin.Context) { + dep, ok := findUserSwarmDeployment(c) + if !ok { + return + } + cfg := agentRuntimeClientConfigForMode(agentRuntimeModeSwarm) + if !cfg.Enabled || strings.TrimSpace(cfg.BaseURL) == "" || strings.TrimSpace(cfg.Token) == "" { + agentError(c, "POLICY_REJECTED", + "swarm input not enabled: requires SWARM_RUNTIME_ENABLED=true with SWARM_RUNTIME_BASE_URL and SWARM_RUNTIME_SERVICE_TOKEN") + return + } + var body struct { + Instruction string `json:"instruction"` + } + if err := c.ShouldBindJSON(&body); err != nil { + agentError(c, "POLICY_REJECTED", err.Error()) + return + } + if strings.TrimSpace(body.Instruction) == "" { + agentError(c, "POLICY_REJECTED", "instruction is required") + return + } + runtimeID := firstNonEmpty(strings.TrimSpace(dep.RuntimeDeploymentID), strings.TrimSpace(dep.RuntimeSwarmID)) + if runtimeID == "" { + agentError(c, "POLICY_REJECTED", "swarm run has no runtime deployment id yet") + return + } + endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeSwarmInputPath(cfg, runtimeID)) + if err != nil { + agentError(c, "RUNTIME_UNAVAILABLE", err.Error()) + return + } + payload, err := common.Marshal(gin.H{ + "instruction": body.Instruction, + "manager_deployment_id": dep.DeploymentID, + }) + if err != nil { + agentError(c, "POLICY_REJECTED", err.Error()) + return + } + data, _, err := callSwarmRuntimeJSON(c.Request.Context(), cfg, http.MethodPost, endpoint, payload, dep) + if err != nil { + // err 仅含 HTTP 状态 + 运行时响应体,不含指令文本。 + common.SysLog("HeicodeAppendSwarmInput: " + err.Error()) + agentError(c, "RUNTIME_UNAVAILABLE", "failed to append input at runtime: "+err.Error()) + return + } + // 审计:只记发生过一次 append,**不记指令文本**。 + recordAgentAuditEvent(agentEvent{ + EventID: "evt_" + common.GetUUID()[:12], + Event: "swarm.input_appended", + SchemaVersion: 1, + UserID: dep.UserID, + ChannelID: dep.ChannelID, + BindingScope: dep.BindingScope, + DeploymentID: dep.DeploymentID, + CorrelationID: dep.CorrelationID, + OccurredAt: agentNow(), + }, "agent_swarm_query", dep.DeploymentID, agentRequestID(c), "ok") + common.ApiSuccess(c, gin.H{ + "deployment_id": dep.DeploymentID, + "swarm_id": dep.RuntimeSwarmID, + "accepted": true, + "task_id": stringFromMap(data, "task_id"), + "runtime_status": stringFromMap(data, "status", "runtime_status"), + "note": "input appended; original instruction is not echoed into the event stream", + }) +} + +// HeicodeGetSwarmResult: GET /api/heicode/swarms/:id/result — 用户面结果(读路,代理到 Swarm)。 +// agent_swarm PR#41 runtime-contract:GET …/{deployment_id}/result → +// {summary, deliverable, artifacts[], termination_reason, status};产物内容按各 artifact 的 uri +// (git/runtime)取,非内联。透出前递归剔除敏感键(stripSensitiveKeys)兜底,绝不含 secret_ref 等。 +// 读路:需 base_url + 服务令牌(运行时可达),不强制 SWARM_RUNTIME_ENABLED(与派发开关解耦)。 +func HeicodeGetSwarmResult(c *gin.Context) { + dep, ok := findUserSwarmDeployment(c) + if !ok { + return + } + cfg := agentRuntimeClientConfigForMode(agentRuntimeModeSwarm) + if strings.TrimSpace(cfg.BaseURL) == "" || strings.TrimSpace(cfg.Token) == "" { + agentError(c, "POLICY_REJECTED", + "swarm result not available: requires SWARM_RUNTIME_BASE_URL and SWARM_RUNTIME_SERVICE_TOKEN") + return + } + runtimeID := firstNonEmpty(strings.TrimSpace(dep.RuntimeDeploymentID), strings.TrimSpace(dep.RuntimeSwarmID)) + if runtimeID == "" { + agentError(c, "POLICY_REJECTED", "swarm run has no runtime deployment id yet") + return + } + endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeSwarmResultPath(cfg, runtimeID)) + if err != nil { + agentError(c, "RUNTIME_UNAVAILABLE", err.Error()) + return + } + data, _, err := callSwarmRuntimeJSON(c.Request.Context(), cfg, http.MethodGet, endpoint, nil, dep) + if err != nil { + common.SysLog("HeicodeGetSwarmResult: " + err.Error()) + agentError(c, "RUNTIME_UNAVAILABLE", "failed to fetch swarm result at runtime: "+err.Error()) + return + } + common.ApiSuccess(c, gin.H{ + "deployment_id": dep.DeploymentID, + "swarm_id": dep.RuntimeSwarmID, + "status": stringFromMap(data, "status"), + "summary": stringFromMap(data, "summary"), + "termination_reason": stringFromMap(data, "termination_reason"), + "deliverable": stripSensitiveKeys(data["deliverable"]), + "artifacts": stripSensitiveKeys(data["artifacts"]), + }) +} + // swarmTerminalEventTypes 是 agent_swarm event-schema v1(#15)的运行终态事件。命中即可结束 SSE 流。 var swarmTerminalEventTypes = map[string]bool{ "swarm.completed": true, diff --git a/heicode/controller/swarm_io_proxy_test.go b/heicode/controller/swarm_io_proxy_test.go new file mode 100644 index 00000000..2f7152e1 --- /dev/null +++ b/heicode/controller/swarm_io_proxy_test.go @@ -0,0 +1,114 @@ +package controller + +import ( + "context" + "io" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + + "github.com/heicode/manager/common" + "github.com/heicode/manager/model" +) + +// #46/#28:input/result 代理端点的运行时路径遵循 PR#41 契约 …/{deployment_id}/{input,result}, +// 默认沿用 create-path 基路径,可经 env 覆盖。 +func TestAgentRuntimeSwarmSubPaths(t *testing.T) { + cfg := agentRuntimeConfig{CreatePath: "/api/agent/swarm/deployments"} + require.Equal(t, "/api/agent/swarm/deployments/rt-9/input", agentRuntimeSwarmInputPath(cfg, "rt-9")) + require.Equal(t, "/api/agent/swarm/deployments/rt-9/result", agentRuntimeSwarmResultPath(cfg, "rt-9")) + + // 空 CreatePath -> 回退到 canonical 默认基路径。 + require.Equal(t, "/api/agent/swarm/deployments/x/input", agentRuntimeSwarmInputPath(agentRuntimeConfig{}, "x")) + + // env 覆盖。 + t.Setenv("SWARM_RUNTIME_RESULT_PATH", "/api/swarms/{deployment_id}/result") + require.Equal(t, "/api/swarms/rt-9/result", agentRuntimeSwarmResultPath(cfg, "rt-9")) +} + +// input 转发:命中契约路径/方法/鉴权,body 携带 instruction + manager_deployment_id,解析 envelope.data。 +func TestCallSwarmRuntimeJSON_InputForwards(t *testing.T) { + var gotMethod, gotPath, gotAuth, gotCorr, gotBody string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + gotCorr = r.Header.Get("X-Correlation-ID") + b, _ := io.ReadAll(r.Body) + gotBody = string(b) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"task_id":"task-append-1","status":"running"}}`)) + })) + defer srv.Close() + + cfg := agentRuntimeConfig{Enabled: true, BaseURL: srv.URL, Token: "svc", CreatePath: "/api/agent/swarm/deployments", Timeout: 5 * time.Second} + dep := model.AgentDeployment{DeploymentID: "dep-1", RuntimeDeploymentID: "rt-9", CorrelationID: "cor-3"} + endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeSwarmInputPath(cfg, dep.RuntimeDeploymentID)) + require.NoError(t, err) + payload, err := common.Marshal(gin.H{"instruction": "refine the report", "manager_deployment_id": dep.DeploymentID}) + require.NoError(t, err) + + data, status, err := callSwarmRuntimeJSON(context.Background(), cfg, http.MethodPost, endpoint, payload, dep) + require.NoError(t, err) + require.Equal(t, http.StatusOK, status) + require.Equal(t, "task-append-1", stringFromMap(data, "task_id")) + require.Equal(t, http.MethodPost, gotMethod) + require.Equal(t, "/api/agent/swarm/deployments/rt-9/input", gotPath) + require.Equal(t, "Bearer svc", gotAuth) + require.Equal(t, "cor-3", gotCorr) + require.Contains(t, gotBody, "refine the report") + require.Contains(t, gotBody, "dep-1") +} + +// result 解析 + 透出脱敏:deliverable/artifacts 内若混入 secret_ref/token 等敏感键,视图须递归剔除。 +func TestCallSwarmRuntimeJSON_ResultParsesAndSanitizes(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + require.Equal(t, http.MethodGet, r.Method) + require.Equal(t, "/api/agent/swarm/deployments/rt-9/result", r.URL.Path) + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"success":true,"data":{"status":"completed","summary":"done","termination_reason":"completed","deliverable":{"text":"the answer","secret_ref":"azkv://heicode-kv/x"},"artifacts":[{"uri":"git://repo/a.patch","token":"leak"}]}}`)) + })) + defer srv.Close() + + cfg := agentRuntimeConfig{BaseURL: srv.URL, Token: "svc", CreatePath: "/api/agent/swarm/deployments", Timeout: 5 * time.Second} + dep := model.AgentDeployment{DeploymentID: "dep-1", RuntimeDeploymentID: "rt-9"} + endpoint, err := agentRuntimeURL(cfg.BaseURL, agentRuntimeSwarmResultPath(cfg, dep.RuntimeDeploymentID)) + require.NoError(t, err) + + data, _, err := callSwarmRuntimeJSON(context.Background(), cfg, http.MethodGet, endpoint, nil, dep) + require.NoError(t, err) + require.Equal(t, "completed", stringFromMap(data, "status")) + require.Equal(t, "done", stringFromMap(data, "summary")) + + deliverable, ok := stripSensitiveKeys(data["deliverable"]).(map[string]any) + require.True(t, ok) + require.Equal(t, "the answer", deliverable["text"]) + _, hasSecret := deliverable["secret_ref"] + require.False(t, hasSecret, "secret_ref 必须从 deliverable 剔除") + + artifacts, ok := stripSensitiveKeys(data["artifacts"]).([]any) + require.True(t, ok) + require.Len(t, artifacts, 1) + first := artifacts[0].(map[string]any) + require.Equal(t, "git://repo/a.patch", first["uri"]) + _, hasToken := first["token"] + require.False(t, hasToken, "token 必须从 artifact 剔除") +} + +// 运行时非 2xx → 返回错误(handler 据此回 RUNTIME_UNAVAILABLE,不伪造受理)。 +func TestCallSwarmRuntimeJSON_RuntimeError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`upstream down`)) + })) + defer srv.Close() + cfg := agentRuntimeConfig{BaseURL: srv.URL, Token: "t", CreatePath: "/api/agent/swarm/deployments", Timeout: 5 * time.Second} + endpoint, _ := agentRuntimeURL(cfg.BaseURL, agentRuntimeSwarmResultPath(cfg, "rt")) + _, status, err := callSwarmRuntimeJSON(context.Background(), cfg, http.MethodGet, endpoint, nil, model.AgentDeployment{DeploymentID: "d", RuntimeDeploymentID: "rt"}) + require.Error(t, err) + require.Equal(t, http.StatusBadGateway, status) +} diff --git a/heicode/router/api-router.go b/heicode/router/api-router.go index 2eb9d5ab..1949bff3 100644 --- a/heicode/router/api-router.go +++ b/heicode/router/api-router.go @@ -560,6 +560,10 @@ func SetApiRouter(router *gin.Engine) { heicodeAgentRoute.GET("/swarms/:id/events/stream", controller.HeicodeStreamSwarmEvents) heicodeAgentRoute.GET("/swarms/:id/artifacts", controller.HeicodeListSwarmArtifacts) heicodeAgentRoute.POST("/swarms/:id/stop", controller.HeicodeStopSwarm) + // I/O 闭环(#46/#28):追加输入(写,代理到 Swarm POST …/input)+ 结果交付(读, + // 代理到 Swarm GET …/result)。契约见 agent_swarm PR#41 / CLIENT_GUIDE §9。 + heicodeAgentRoute.POST("/swarms/:id/input", controller.HeicodeAppendSwarmInput) + heicodeAgentRoute.GET("/swarms/:id/result", controller.HeicodeGetSwarmResult) } // Client↔agent access control (HM-provided, AM-optional). Called by the