fix(mcp): efficiency and data race improvements

cmd/internal/browse.go

@@ -170,18 +170,37 @@ func listExecutables(ctx *context.Context, cmd *cobra.Command, _ []string) {
 		}
 	}
 
+	applyFilters := func(execs executable.ExecutableList) executable.ExecutableList {
+		return execs.
+			FilterByWorkspaceWithVisibility(wsFilter, visibilityFilter).
+			FilterByNamespace(nsFilter).
+			FilterByVerb(executable.Verb(verbFilter)).
+			FilterByTags(tagsFilter).
+			FilterByAnnotations(annotationFilter).
+			FilterBySubstring(substr)
+	}
+
 	allExecs, err := ctx.ExecutableCache.GetExecutableList()
 	if err != nil {
 		errhandler.HandleFatal(ctx, cmd, err)
 	}
-	filteredExec := allExecs
-	filteredExec = filteredExec.
-		FilterByWorkspaceWithVisibility(wsFilter, visibilityFilter).
-		FilterByNamespace(nsFilter).
-		FilterByVerb(executable.Verb(verbFilter)).
-		FilterByTags(tagsFilter).
-		FilterByAnnotations(annotationFilter).
-		FilterBySubstring(substr)
+	filteredExec := applyFilters(allExecs)
+
+	// The persisted cache has no TTL or file-watcher (see viewExecutable's identical retry for a
+	// single ref), so it can predate executables added since it was last synced. An empty
+	// filtered result is the only signal available here; force one rescan and retry before
+	// concluding the filter is genuinely empty.
+	if len(filteredExec) == 0 {
+		logger.Log().Debugf("no executables matched filters, syncing cache and retrying")
+		if err := ctx.ExecutableCache.Update(); err != nil {
+			errhandler.HandleFatal(ctx, cmd, err)
+		}
+		allExecs, err = ctx.ExecutableCache.GetExecutableList()
+		if err != nil {
+			errhandler.HandleFatal(ctx, cmd, err)
+		}
+		filteredExec = applyFilters(allExecs)
+	}
 
 	if TUIEnabled(ctx, cmd) {
 		runFunc := func(ref string) error { return runByRef(ctx, cmd, ref) }

internal/mcp/output_types.go

@@ -106,6 +106,9 @@ type ExecutableListOutput struct {
 // ExecutionOutput is the output of the execute tool.
 type ExecutionOutput struct {
 	Output string `json:"output"`
+	// Truncated is true when Output was cut to protect the context window. The run itself
+	// completed in full; the untruncated output remains available via get_execution_logs.
+	Truncated bool `json:"truncated,omitempty"`
 }
 
 // LogEntry represents a single execution log record.
@@ -154,6 +157,10 @@ type WriteFlowFileOutput struct {
 	Path        string   `json:"path"`
 	Executables []string `json:"executables"`
 	Overwritten bool     `json:"overwritten"`
+	// SyncFailed is true when the file was written successfully but the follow-up executable
+	// cache refresh failed. The write itself is unaffected; a subsequent list_executables/
+	// get_executable call may still show stale results until sync_executables is run manually.
+	SyncFailed bool `json:"syncFailed,omitempty"`
 }
 
 // WorkspaceConfigOutput is the output of the get_workspace_config tool.

internal/mcp/server_test.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/mark3labs/mcp-go/client"
@@ -320,6 +321,26 @@ var _ = Describe("MCP Server", func() {
 				Expect(err).ToNot(HaveOccurred())
 				Expect(getTextContent(result)).To(ContainSubstring("execution result with no args"))
 			})
+
+			It("should cap oversized output and report truncation", func() {
+				huge := strings.Repeat("x", 300_000)
+				mockExecutor.EXPECT().
+					ExecuteContext(gomock.Any(), "test", "test:test-flow").
+					Return(huge, nil)
+
+				result, err := mcpClient.CallTool(ctx, newCallToolRequest("execute", map[string]interface{}{
+					"executable_verb": "test",
+					"executable_id":   "test:test-flow",
+				}))
+				Expect(err).ToNot(HaveOccurred())
+
+				var out flowMcp.ExecutionOutput
+				Expect(json.Unmarshal([]byte(getTextContent(result)), &out)).To(Succeed())
+				Expect(out.Truncated).To(BeTrue())
+				Expect(len(out.Output)).To(BeNumerically("<=", 200_000))
+				// The tail is kept, since errors typically surface at the end of a run's output.
+				Expect(out.Output).To(HaveSuffix(strings.Repeat("x", 100)))
+			})
 		})
 
 		Context("run_command tool", func() {
@@ -357,6 +378,23 @@ var _ = Describe("MCP Server", func() {
 				Expect(err).ToNot(HaveOccurred())
 				Expect(result.IsError).To(BeTrue())
 			})
+
+			It("should cap oversized output and report truncation", func() {
+				huge := strings.Repeat("y", 300_000)
+				mockExecutor.EXPECT().
+					ExecuteContext(gomock.Any(), "exec", "--cmd", "echo hi").
+					Return(huge, nil)
+
+				result, err := mcpClient.CallTool(ctx, newCallToolRequest("run_command", map[string]interface{}{
+					"command": "echo hi",
+				}))
+				Expect(err).ToNot(HaveOccurred())
+
+				var out flowMcp.ExecutionOutput
+				Expect(json.Unmarshal([]byte(getTextContent(result)), &out)).To(Succeed())
+				Expect(out.Truncated).To(BeTrue())
+				Expect(len(out.Output)).To(BeNumerically("<=", 200_000))
+			})
 		})
 
 		Context("run_executable tool", func() {
@@ -511,6 +549,8 @@ executables:
     exec:
       cmd: echo greet
 `
+				mockExecutor.EXPECT().ExecuteContext(gomock.Any(), "sync").Return("synced", nil)
+
 				result, err := mcpClient.CallTool(ctx, newCallToolRequest("write_flowfile", map[string]interface{}{
 					"path":    flowPath,
 					"content": validYAML,
@@ -523,12 +563,35 @@ executables:
 				Expect(json.Unmarshal([]byte(text), &out)).To(Succeed())
 				Expect(out.Path).To(Equal(flowPath))
 				Expect(out.Executables).To(ContainElements("hello", "greet"))
+				Expect(out.SyncFailed).To(BeFalse())
 
 				// Verify file was actually written
 				_, statErr := os.Stat(flowPath)
 				Expect(statErr).ToNot(HaveOccurred())
 			})
 
+			It("should report SyncFailed when the post-write cache refresh fails, without failing the write", func() {
+				tmpDir := GinkgoTB().TempDir()
+				flowPath := filepath.Join(tmpDir, "sync-fail.flow")
+
+				mockExecutor.EXPECT().ExecuteContext(gomock.Any(), "sync").Return("", errors.New("sync exploded"))
+
+				result, err := mcpClient.CallTool(ctx, newCallToolRequest("write_flowfile", map[string]interface{}{
+					"path":    flowPath,
+					"content": "executables: []",
+				}))
+				Expect(err).ToNot(HaveOccurred())
+				Expect(result.IsError).To(BeFalse())
+
+				var out flowMcp.WriteFlowFileOutput
+				Expect(json.Unmarshal([]byte(getTextContent(result)), &out)).To(Succeed())
+				Expect(out.SyncFailed).To(BeTrue())
+
+				// The write itself still succeeded despite the sync failure.
+				_, statErr := os.Stat(flowPath)
+				Expect(statErr).ToNot(HaveOccurred())
+			})
+
 			It("should reject invalid file extension", func() {
 				result, err := mcpClient.CallTool(ctx, newCallToolRequest("write_flowfile", map[string]interface{}{
 					"path":    "/tmp/bad.txt",
@@ -571,6 +634,8 @@ executables:
 				flowPath := filepath.Join(tmpDir, "existing.flow")
 				Expect(os.WriteFile(flowPath, []byte("executables: []\n"), 0600)).To(Succeed())
 
+				mockExecutor.EXPECT().ExecuteContext(gomock.Any(), "sync").Return("synced", nil)
+
 				result, err := mcpClient.CallTool(ctx, newCallToolRequest("write_flowfile", map[string]interface{}{
 					"path":      flowPath,
 					"content":   "executables: []",

internal/mcp/tools.go

@@ -16,6 +16,22 @@ func boolPtr(b bool) *bool {
 	return &b
 }
 
+// maxExecutionOutputBytes caps captured subprocess output returned by execute/run_command/
+// run_executable so a large run (e.g. `validate`, which chains generate+lint+test+e2e) can't
+// return a single unbounded JSON-RPC message. Mirrors the cap get_execution_logs already
+// enforces (see maxLogContentMaxBytes) so no MCP tool response is unbounded.
+const maxExecutionOutputBytes = 200_000
+
+// capOutput truncates output to at most maxExecutionOutputBytes, keeping the tail since that's
+// where errors and failures surface. Returns the (possibly truncated) output and whether
+// truncation happened; the full, untruncated output remains available via get_execution_logs.
+func capOutput(output string) (string, bool) {
+	if len(output) <= maxExecutionOutputBytes {
+		return output, false
+	}
+	return output[len(output)-maxExecutionOutputBytes:], true
+}
+
 // sendProgress sends a progress notification to the client if a progress token was provided.
 // It silently ignores errors (e.g., no active session in test contexts).
 func sendProgress(srv *server.MCPServer, ctx context.Context, token any, progress, total float64, message string) {

internal/mcp/tools_executable.go

@@ -95,7 +95,8 @@ func addExecutableTools(srv *server.MCPServer, executor CommandExecutor) {
 	writeFlowfile := mcp.NewTool("write_flowfile",
 		mcp.WithDescription("Create or update a .flow workflow file. Use when the user wants to add or modify "+
 			"automation — builds, tests, deploys, scripts. Validates the YAML against the schema before "+
-			"writing. Prefer this over writing YAML files directly."),
+			"writing, then refreshes flow's executable cache so the change is immediately visible to "+
+			"list_executables/get_executable. Prefer this over writing YAML files directly."),
 		mcp.WithString("path", mcp.Required(),
 			mcp.Description("Absolute or workspace-relative path for the flowfile (must end in .flow or .flow.yaml)")),
 		mcp.WithString("content", mcp.Required(),
@@ -108,7 +109,7 @@ func addExecutableTools(srv *server.MCPServer, executor CommandExecutor) {
 		DestructiveHint: boolPtr(true), ReadOnlyHint: boolPtr(false),
 		IdempotentHint: boolPtr(false), OpenWorldHint: boolPtr(false),
 	}
-	srv.AddTool(writeFlowfile, writeFlowfileHandler(srv))
+	srv.AddTool(writeFlowfile, writeFlowfileHandler(srv, executor))
 }
 
 func getExecutableHandler(executor CommandExecutor) server.ToolHandlerFunc {
@@ -227,12 +228,14 @@ func executeFlowHandler(srv *server.MCPServer, executor CommandExecutor) server.
 
 		if err != nil {
 			ref := strings.Join([]string{executableVerb, executableID}, " ")
-			return toolError(ErrCodeExecutionFailed, fmt.Sprintf("%s execution failed: %s", ref, output)), nil
+			capped, _ := capOutput(output)
+			return toolError(ErrCodeExecutionFailed, fmt.Sprintf("%s execution failed: %s", ref, capped)), nil
 		}
 
 		sendProgress(srv, ctx, progressToken, 2, 2, "Complete")
 
-		result := ExecutionOutput{Output: output}
+		capped, truncated := capOutput(output)
+		result := ExecutionOutput{Output: capped, Truncated: truncated}
 		jsonData, _ := json.Marshal(result)
 		return mcp.NewToolResultStructured(result, string(jsonData)), nil
 	}
@@ -381,18 +384,20 @@ func runTransientTool(
 	sendProgress(srv, ctx, progressToken, 1, 2, "Processing result")
 
 	if err != nil {
-		return toolError(ErrCodeExecutionFailed, fmt.Sprintf("%s: %s", failMsg, output)), nil
+		capped, _ := capOutput(output)
+		return toolError(ErrCodeExecutionFailed, fmt.Sprintf("%s: %s", failMsg, capped)), nil
 	}
 
 	sendProgress(srv, ctx, progressToken, 2, 2, "Complete")
 
-	result := ExecutionOutput{Output: output}
+	capped, truncated := capOutput(output)
+	result := ExecutionOutput{Output: capped, Truncated: truncated}
 	jsonData, _ := json.Marshal(result)
 	return mcp.NewToolResultStructured(result, string(jsonData)), nil
 }
 
-func writeFlowfileHandler(srv *server.MCPServer) server.ToolHandlerFunc {
-	return func(_ context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
+func writeFlowfileHandler(srv *server.MCPServer, executor CommandExecutor) server.ToolHandlerFunc {
+	return func(ctx context.Context, request mcp.CallToolRequest) (*mcp.CallToolResult, error) {
 		path, err := request.RequireString("path")
 		if err != nil {
 			return toolError(ErrCodeInvalidInput, "path is required"), nil
@@ -436,12 +441,23 @@ func writeFlowfileHandler(srv *server.MCPServer) server.ToolHandlerFunc {
 			execNames = append(execNames, exec.Name)
 		}
 
+		// write_flowfile is the one executable-mutating tool that writes directly rather than
+		// shelling to the CLI, so nothing else refreshes the persisted executable cache — without
+		// this, a list_executables/get_executable call right after would still see stale state.
+		// Best-effort: the file is already written and valid, so a sync failure here doesn't
+		// invalidate the write itself.
+		syncFailed := false
+		if _, err := executor.ExecuteContext(ctx, "sync"); err != nil {
+			syncFailed = true
+		}
+
 		srv.SendNotificationToAllClients("notifications/resources/list_changed", nil)
 
 		output := WriteFlowFileOutput{
 			Path:        absPath,
 			Executables: execNames,
 			Overwritten: overwrite,
+			SyncFailed:  syncFailed,
 		}
 		jsonData, _ := json.Marshal(output)
 		return mcp.NewToolResultStructured(output, string(jsonData)), nil

main.go

@@ -27,7 +27,7 @@ func main() {
 
 	archiveDir, archiveID := initLogArchive()
 	loggerOpts := logger.InitOptions{
-		StdOut:           io.Stdout,
+		StdOut:           loggerStdOut(),
 		LogMode:          cfg.DefaultLogMode,
 		Theme:            logger.Theme(cfg.Theme.String()),
 		ArchiveDirectory: archiveDir,
@@ -60,6 +60,16 @@ func main() {
 	}
 }
 
+// loggerStdOut returns stderr for `flow mcp`: that command's stdout carries JSON-RPC framing for
+// the whole session, and any log line interleaved into it would corrupt the stream and sever the
+// client's connection. stderr is safe — MCP hosts capture it separately as server diagnostics.
+func loggerStdOut() *os.File {
+	if len(os.Args) > 1 && os.Args[1] == "mcp" {
+		return os.Stderr
+	}
+	return io.Stdout
+}
+
 func initLogArchive() (dir, id string) {
 	if args := os.Args; len(args) > 1 && slices.Contains(executable.ValidVerbs(), executable.Verb(args[1])) {
 		dir = filesystem.LogsDir()