diff --git a/internal/config/envexpand.go b/internal/config/envexpand.go index 0ffc1784..26faa0c1 100644 --- a/internal/config/envexpand.go +++ b/internal/config/envexpand.go @@ -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) diff --git a/internal/config/envexpand_test.go b/internal/config/envexpand_test.go index a17c4514..60d91944 100644 --- a/internal/config/envexpand_test.go +++ b/internal/config/envexpand_test.go @@ -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") + } +} diff --git a/internal/handler/external_mcp.go b/internal/handler/external_mcp.go index e042046f..0da18210 100644 --- a/internal/handler/external_mcp.go +++ b/internal/handler/external_mcp.go @@ -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 // 保存到配置文件 diff --git a/internal/handler/external_mcp_test.go b/internal/handler/external_mcp_test.go index e4cf3c1f..2c859f0a 100644 --- a/internal/handler/external_mcp_test.go +++ b/internal/handler/external_mcp_test.go @@ -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) diff --git a/internal/workflow/nodes.go b/internal/workflow/nodes.go index 8351f9cd..ebad82a8 100644 --- a/internal/workflow/nodes.go +++ b/internal/workflow/nodes.go @@ -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 diff --git a/internal/workflow/structured_outputs.go b/internal/workflow/structured_outputs.go index 1c3200a5..f4adb640 100644 --- a/internal/workflow/structured_outputs.go +++ b/internal/workflow/structured_outputs.go @@ -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, diff --git a/internal/workflow/tool_timeout_test.go b/internal/workflow/tool_timeout_test.go index c98c3828..74a37d40 100644 --- a/internal/workflow/tool_timeout_test.go +++ b/internal/workflow/tool_timeout_test.go @@ -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"]) + } +}