mirror of
https://github.com/Ed1s0nZ/CyberStrikeAI.git
synced 2026-10-18 21:47:17 +02:00
fix: snapshot MCP configuration and make tool deadline failures consistent
This commit is contained in:
1 parent
e8a5392b37
commit
b841296f0e
7 files changed
+83
-8
No files matched your search
@@ -1,7 +1,9 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"maps"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -52,6 +54,11 @@ func expandEnvVar(s string) string {
|
||||
// ExpandConfigEnv 展开 ExternalMCPServerConfig 中所有支持环境变量的字段。
|
||||
// 展开范围:Command、Args、Env values、URL、Headers values。
|
||||
func ExpandConfigEnv(cfg *ExternalMCPServerConfig) {
|
||||
// Config structs are passed by value to background MCP connections. Their
|
||||
// slices and maps still share storage; expansion must not mutate that storage.
|
||||
cfg.Args = slices.Clone(cfg.Args)
|
||||
cfg.Env = maps.Clone(cfg.Env)
|
||||
cfg.Headers = maps.Clone(cfg.Headers)
|
||||
cfg.Command = expandEnvVar(cfg.Command)
|
||||
for i, arg := range cfg.Args {
|
||||
cfg.Args[i] = expandEnvVar(arg)
|
||||
|
||||
@@ -79,3 +79,20 @@ func TestExpandConfigEnv(t *testing.T) {
|
||||
t.Errorf("Headers[Authorization] = %q, want %q", cfg.Headers["Authorization"], "Bearer secret123")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExpandConfigEnvDoesNotMutateSharedConfig(t *testing.T) {
|
||||
t.Setenv("CSAI_TEST_EXPAND", "expanded")
|
||||
original := ExternalMCPServerConfig{
|
||||
Args: []string{"${CSAI_TEST_EXPAND}"},
|
||||
Env: map[string]string{"VALUE": "${CSAI_TEST_EXPAND}"},
|
||||
Headers: map[string]string{"X-Value": "${CSAI_TEST_EXPAND}"},
|
||||
}
|
||||
copy := original
|
||||
ExpandConfigEnv(©)
|
||||
if original.Args[0] != "${CSAI_TEST_EXPAND}" || original.Env["VALUE"] != "${CSAI_TEST_EXPAND}" || original.Headers["X-Value"] != "${CSAI_TEST_EXPAND}" {
|
||||
t.Fatal("expansion mutated a config shared with another goroutine")
|
||||
}
|
||||
if copy.Args[0] != "expanded" || copy.Env["VALUE"] != "expanded" || copy.Headers["X-Value"] != "expanded" {
|
||||
t.Fatal("copied config was not expanded")
|
||||
}
|
||||
}
|
||||
@@ -172,13 +172,6 @@ func (h *ExternalMCPHandler) AddOrUpdateExternalMCP(c *gin.Context) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
|
||||
// 添加或更新配置
|
||||
if err := h.manager.AddOrUpdateConfig(name, req.Config); err != nil {
|
||||
h.logger.Error("添加或更新外部MCP配置失败", zap.Error(err))
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "添加或更新配置失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
// 更新内存中的配置
|
||||
if h.config.ExternalMCP.Servers == nil {
|
||||
h.config.ExternalMCP.Servers = make(map[string]config.ExternalMCPServerConfig)
|
||||
@@ -197,6 +190,13 @@ func (h *ExternalMCPHandler) AddOrUpdateExternalMCP(c *gin.Context) {
|
||||
// 展开 ${VAR} 环境变量
|
||||
config.ExpandConfigEnv(&cfg)
|
||||
|
||||
// 添加或更新配置
|
||||
if err := h.manager.AddOrUpdateConfig(name, cfg); err != nil {
|
||||
h.logger.Error("添加或更新外部MCP配置失败", zap.Error(err))
|
||||
c.JSON(http.StatusInternalServerError, gin.H{"error": "添加或更新配置失败: " + err.Error()})
|
||||
return
|
||||
}
|
||||
|
||||
h.config.ExternalMCP.Servers[name] = cfg
|
||||
|
||||
// 保存到配置文件
|
||||
|
||||
@@ -56,6 +56,34 @@ func cleanupTestConfig(configPath string) {
|
||||
os.Remove(configPath + ".backup")
|
||||
}
|
||||
|
||||
func TestExternalMCPConnectionReceivesExpandedSnapshot(t *testing.T) {
|
||||
t.Setenv("CSAI_TEST_MCP_COMMAND", "not-executed")
|
||||
t.Setenv("CSAI_TEST_MCP_VALUE", "expanded-test-value")
|
||||
router, h, configPath := setupTestRouter()
|
||||
defer cleanupTestConfig(configPath)
|
||||
body, err := json.Marshal(AddOrUpdateExternalMCPRequest{Config: config.ExternalMCPServerConfig{
|
||||
Command: "${CSAI_TEST_MCP_COMMAND}", Args: []string{"${CSAI_TEST_MCP_VALUE}"}, Disabled: true,
|
||||
Env: map[string]string{"VALUE": "${CSAI_TEST_MCP_VALUE}"}, Headers: map[string]string{"X-Value": "${CSAI_TEST_MCP_VALUE}"},
|
||||
}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req := httptest.NewRequest(http.MethodPut, "/api/external-mcp/test-env", bytes.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
w := httptest.NewRecorder()
|
||||
router.ServeHTTP(w, req)
|
||||
if w.Code != http.StatusOK {
|
||||
t.Fatalf("save: %d %s", w.Code, w.Body)
|
||||
}
|
||||
cfg := h.manager.GetConfigs()["test-env"]
|
||||
if cfg.Command != "not-executed" || len(cfg.Args) != 1 || cfg.Args[0] != "expanded-test-value" || cfg.Env["VALUE"] != "expanded-test-value" || cfg.Headers["X-Value"] != "expanded-test-value" {
|
||||
t.Fatalf("connection received unexpanded snapshot: %#v", cfg)
|
||||
}
|
||||
if !cfg.Disabled || cfg.ExternalMCPEnable {
|
||||
t.Fatal("disabled connection was activated")
|
||||
}
|
||||
}
|
||||
|
||||
func TestExternalMCPHandler_AddOrUpdateExternalMCP_Stdio(t *testing.T) {
|
||||
router, _, configPath := setupTestRouter()
|
||||
defer cleanupTestConfig(configPath)
|
||||
|
||||
@@ -86,6 +86,14 @@ func runToolNode(ctx context.Context, args RunArgs, node graphNode, state *Workf
|
||||
if err == nil && toolCtx.Err() != nil {
|
||||
err = toolCtx.Err()
|
||||
}
|
||||
// The MCP worker and caller have timers for the same absolute deadline.
|
||||
// The worker may finish first while the caller's Done notification is still
|
||||
// waiting to run; the deadline itself is authoritative in either ordering.
|
||||
if err == nil {
|
||||
if deadline, ok := toolCtx.Deadline(); ok && !time.Now().Before(deadline) {
|
||||
err = context.DeadlineExceeded
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
errText := err.Error()
|
||||
return outputMap(envelope("tool", node.ID, node.Type, "failed", ""), map[string]any{"tool_name": toolName, "arguments": toolArgs, "error": errText}), false, "failed", errText
|
||||
|
||||
@@ -107,8 +107,12 @@ func endOutputMap(node graphNode, value any) map[string]any {
|
||||
}
|
||||
|
||||
func toolOutputMap(node graphNode, output string, toolName string, args map[string]any, executionID string, isError bool) map[string]any {
|
||||
status := "completed"
|
||||
if isError {
|
||||
status = "failed"
|
||||
}
|
||||
typed := ToolOutput{
|
||||
NodeOutputEnvelope: envelope("tool", node.ID, node.Type, "completed", output),
|
||||
NodeOutputEnvelope: envelope("tool", node.ID, node.Type, status, output),
|
||||
ToolName: toolName,
|
||||
Arguments: args,
|
||||
ExecutionID: executionID,
|
||||
|
||||
@@ -88,3 +88,14 @@ func TestRunToolNodeAppliesTimeoutToActualMCPExecution(t *testing.T) {
|
||||
t.Fatal("actual MCP handler was not executed")
|
||||
}
|
||||
}
|
||||
|
||||
func TestFailedToolOutputDoesNotClaimCompletedStatus(t *testing.T) {
|
||||
out := toolOutputMap(graphNode{ID: "tool-1", Type: "tool"}, "tool failed", "lookup", nil, "exec-1", true)
|
||||
if out["status"] != "failed" {
|
||||
t.Fatalf("failed output claimed success: %#v", out)
|
||||
}
|
||||
typed, ok := out["typed"].(ToolOutput)
|
||||
if !ok || typed.Status != "failed" || !typed.IsError {
|
||||
t.Fatalf("typed output disagrees with failure: %#v", out["typed"])
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user