Merge remote-tracking branch 'origin/feat/swarm-io-proxy-46' into deploy/test-pr69-73
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user