Add files via upload

internal/mcp/execution_service.go

@@ -76,6 +76,7 @@ type ExecutionService struct {
 	abortUserNotes map[string]string
 	maxInMemory    int
 	resultMaxBytes int
+	spillRootDir   string
 }
 
 func NewExecutionService(storage MonitorStorage, logger *zap.Logger) *ExecutionService {
@@ -101,6 +102,17 @@ func (s *ExecutionService) ConfigureToolResultMaxBytes(maxBytes int) {
 	s.resultMaxBytes = maxBytes
 }
 
+// ConfigureToolResultSpillRoot sets the reduction-compatible root used when
+// oversized tool results are spilled to local files (empty → tmp/reduction).
+func (s *ExecutionService) ConfigureToolResultSpillRoot(rootDir string) {
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.spillRootDir = strings.TrimSpace(rootDir)
+}
+
 func (s *ExecutionService) Submit(ctx context.Context, req ExecutionRequest) (*ExecutionHandle, error) {
 	if s == nil {
 		return nil, fmt.Errorf("execution service is nil")
@@ -163,6 +175,10 @@ func (s *ExecutionService) Submit(ctx context.Context, req ExecutionRequest) (*E
 
 func (s *ExecutionService) runWorker(ctx context.Context, entry *executionEntry, onDone ExecutionDoneFunc) {
 	id := entry.exec.ID
+	ctx = WithMCPExecutionID(ctx, id)
+	if conv := strings.TrimSpace(entry.exec.ConversationID); conv != "" {
+		ctx = WithMCPConversationID(ctx, conv)
+	}
 	var release func()
 	defer func() {
 		if release != nil {
@@ -212,7 +228,20 @@ func (s *ExecutionService) finishEntry(ctx context.Context, entry *executionEntr
 
 	now := time.Now()
 	s.mu.Lock()
-	result = NormalizeToolResultForStorage(result, s.resultMaxBytes)
+	spill := ToolResultSpillConfig{
+		RootDir:        s.spillRootDir,
+		ConversationID: entry.exec.ConversationID,
+		ExecutionID:    id,
+	}
+	if ctx != nil {
+		if pid := MCPProjectIDFromContext(ctx); pid != "" {
+			spill.ProjectID = pid
+		}
+		if conv := MCPConversationIDFromContext(ctx); conv != "" {
+			spill.ConversationID = conv
+		}
+	}
+	result = NormalizeToolResultForStorageWithSpill(result, s.resultMaxBytes, spill)
 	entry.result = result
 	entry.err = err
 	entry.exec.EndTime = &now

internal/mcp/external_manager.go

@@ -77,6 +77,7 @@ type ExternalMCPManager struct {
 	executionService   *ExecutionService
 	toolWaitTimeout    time.Duration
 	toolResultMaxBytes int
+	spillRootDir       string
 	resilience         ExternalMCPResilienceConfig
 	serverRuntimes     map[string]*externalMCPServerRuntime
 	globalSemaphore    chan struct{}
@@ -143,6 +144,20 @@ func (m *ExternalMCPManager) ConfigureToolResultMaxBytes(maxBytes int) {
 	}
 }
 
+// ConfigureToolResultSpillRoot sets the local directory root used when oversized
+// tool results are spilled (aligned with reduction_root_dir; empty → tmp/reduction).
+func (m *ExternalMCPManager) ConfigureToolResultSpillRoot(rootDir string) {
+	if m == nil {
+		return
+	}
+	m.mu.Lock()
+	m.spillRootDir = strings.TrimSpace(rootDir)
+	m.mu.Unlock()
+	if m.executionService != nil {
+		m.executionService.ConfigureToolResultSpillRoot(rootDir)
+	}
+}
+
 // ConfigureToolWaitTimeoutSeconds controls how long an agent-facing tool call
 // waits for an external MCP execution before returning an execution_id that can
 // be polled with wait_tool_execution. seconds<=0 waits until completion.
@@ -704,6 +719,7 @@ func (m *ExternalMCPManager) CallTool(ctx context.Context, toolName string, args
 	if m.executionService == nil {
 		m.executionService = NewExecutionService(m.storage, m.logger)
 		m.executionService.ConfigureToolResultMaxBytes(m.toolResultMaxBytes)
+		m.executionService.ConfigureToolResultSpillRoot(m.spillRootDir)
 	}
 	var ownerUserID string
 	if principal, ok := authctx.PrincipalFromContext(ctx); ok {

internal/mcp/run_context.go

@@ -22,6 +22,8 @@ type EinoExecuteRunRegistry interface {
 type toolRunRegistryCtxKey struct{}
 type einoExecuteRunRegistryCtxKey struct{}
 type mcpConversationIDCtxKey struct{}
+type mcpExecutionIDCtxKey struct{}
+type mcpProjectIDCtxKey struct{}
 
 // WithToolRunRegistry 将登记器注入 ctx（Eino / 原生 Agent 任务 ctx）。
 func WithToolRunRegistry(ctx context.Context, reg ToolRunRegistry) context.Context {
@@ -78,6 +80,48 @@ func MCPConversationIDFromContext(ctx context.Context) string {
 	return v
 }
 
+// WithMCPExecutionID 将当前工具 executionId 注入 ctx，供超长输出落盘文件名对齐。
+func WithMCPExecutionID(ctx context.Context, executionID string) context.Context {
+	if ctx == nil {
+		return nil
+	}
+	id := strings.TrimSpace(executionID)
+	if id == "" {
+		return ctx
+	}
+	return context.WithValue(ctx, mcpExecutionIDCtxKey{}, id)
+}
+
+// MCPExecutionIDFromContext 读取当前工具 executionId。
+func MCPExecutionIDFromContext(ctx context.Context) string {
+	if ctx == nil {
+		return ""
+	}
+	v, _ := ctx.Value(mcpExecutionIDCtxKey{}).(string)
+	return v
+}
+
+// WithMCPProjectID 将项目 ID 注入 ctx，供 reduction/trunc 落盘路径与项目隔离对齐。
+func WithMCPProjectID(ctx context.Context, projectID string) context.Context {
+	if ctx == nil {
+		return nil
+	}
+	id := strings.TrimSpace(projectID)
+	if id == "" {
+		return ctx
+	}
+	return context.WithValue(ctx, mcpProjectIDCtxKey{}, id)
+}
+
+// MCPProjectIDFromContext 读取项目 ID。
+func MCPProjectIDFromContext(ctx context.Context) string {
+	if ctx == nil {
+		return ""
+	}
+	v, _ := ctx.Value(mcpProjectIDCtxKey{}).(string)
+	return v
+}
+
 func notifyToolRunBegin(ctx context.Context, executionID string) {
 	reg := ToolRunRegistryFromContext(ctx)
 	if reg == nil {

internal/mcp/server.go

@@ -56,6 +56,7 @@ type Server struct {
 	executionService       *ExecutionService
 	toolWaitTimeout        time.Duration
 	toolResultMaxBytes     int
+	spillRootDir           string
 }
 
 // SetToolAuthorizer installs the common policy decision point for every
@@ -128,6 +129,20 @@ func (s *Server) ConfigureToolResultMaxBytes(maxBytes int) {
 	}
 }
 
+// ConfigureToolResultSpillRoot sets the local directory root used when oversized
+// tool results are spilled (aligned with reduction_root_dir; empty → tmp/reduction).
+func (s *Server) ConfigureToolResultSpillRoot(rootDir string) {
+	if s == nil {
+		return
+	}
+	s.mu.Lock()
+	s.spillRootDir = strings.TrimSpace(rootDir)
+	s.mu.Unlock()
+	if s.executionService != nil {
+		s.executionService.ConfigureToolResultSpillRoot(rootDir)
+	}
+}
+
 // ConfigureHTTPToolCallTimeoutFromAgentMinutes 将 agent.tool_timeout_minutes 同步到经 HTTP POST /api/mcp 触发的 tools/call。
 // minutes<=0 表示不设置硬性截止时间（与配置「0 不限制」一致）；minutes>0 为该次调用的最长等待时间。
 // 未调用前对 tools/call 使用默认 30 分钟（与历史硬编码一致）。
@@ -900,6 +915,7 @@ func (s *Server) CallTool(ctx context.Context, toolName string, args map[string]
 	if s.executionService == nil {
 		s.executionService = NewExecutionService(s.storage, s.logger)
 		s.executionService.ConfigureToolResultMaxBytes(s.toolResultMaxBytes)
+		s.executionService.ConfigureToolResultSpillRoot(s.spillRootDir)
 	}
 	var ownerUserID string
 	if principal, ok := authctx.PrincipalFromContext(ctx); ok {
@@ -1036,6 +1052,7 @@ func (s *Server) FinishToolExecution(ctx context.Context, executionID, toolName
 
 	s.mu.Lock()
 	maxBytes := s.toolResultMaxBytes
+	spillRoot := s.spillRootDir
 	exec, inMem := s.executions[id]
 	if !inMem || exec == nil {
 		exec = &ToolExecution{
@@ -1063,13 +1080,19 @@ func (s *Server) FinishToolExecution(ctx context.Context, executionID, toolName
 	}
 	exec.Duration = now.Sub(exec.StartTime)
 
+	spill := ToolResultSpillConfig{
+		RootDir:        spillRoot,
+		ProjectID:      MCPProjectIDFromContext(ctx),
+		ConversationID: exec.ConversationID,
+		ExecutionID:    id,
+	}
 	if failed {
 		st, msg := executionStatusAndMessage(invokeErr)
 		exec.Status = st
 		exec.Error = msg
 		if strings.TrimSpace(resultText) != "" {
 			finalResult = &ToolResult{Content: []Content{{Type: "text", Text: resultText}}}
-			finalResult = NormalizeToolResultForStorage(finalResult, maxBytes)
+			finalResult = NormalizeToolResultForStorageWithSpill(finalResult, maxBytes, spill)
 			exec.Result = finalResult
 		}
 	} else {
@@ -1079,7 +1102,7 @@ func (s *Server) FinishToolExecution(ctx context.Context, executionID, toolName
 			text = "（无输出）"
 		}
 		finalResult = &ToolResult{Content: []Content{{Type: "text", Text: text}}}
-		finalResult = NormalizeToolResultForStorage(finalResult, maxBytes)
+		finalResult = NormalizeToolResultForStorageWithSpill(finalResult, maxBytes, spill)
 		exec.Result = finalResult
 	}
 	s.mu.Unlock()
@@ -1116,9 +1139,16 @@ func (s *Server) UpdateToolExecutionResult(executionID string, result *ToolResul
 		return nil
 	}
 	s.mu.Lock()
-	result = NormalizeToolResultForStorage(result, s.toolResultMaxBytes)
+	spill := ToolResultSpillConfig{
+		RootDir:     s.spillRootDir,
+		ExecutionID: executionID,
+	}
 	if exec, ok := s.executions[executionID]; ok && exec != nil {
+		spill.ConversationID = exec.ConversationID
+		result = NormalizeToolResultForStorageWithSpill(result, s.toolResultMaxBytes, spill)
 		exec.Result = result
+	} else {
+		result = NormalizeToolResultForStorageWithSpill(result, s.toolResultMaxBytes, spill)
 	}
 	s.mu.Unlock()
 	if s.storage != nil {

internal/mcp/tool_result_guard.go

@@ -1,12 +1,29 @@
 package mcp
 
-import "fmt"
+import "cyberstrike-ai/internal/tooloutput"
 
 const DefaultToolResultMaxBytes = 12000
 
+// ToolResultSpillConfig controls where oversized tool results are written on disk
+// before the in-memory/DB/agent-facing payload is truncated.
+type ToolResultSpillConfig struct {
+	RootDir        string
+	ProjectID      string
+	ConversationID string
+	ExecutionID    string
+}
+
 // NormalizeToolResultForStorage returns the canonical result used by both the
-// agent-facing response and monitor persistence.
+// agent-facing response and monitor persistence. When maxBytes is exceeded the
+// full text is spilled under the reduction cache tree and replaced with a
+// <persisted-output> notice that includes the file path.
 func NormalizeToolResultForStorage(result *ToolResult, maxBytes int) *ToolResult {
+	return NormalizeToolResultForStorageWithSpill(result, maxBytes, ToolResultSpillConfig{})
+}
+
+// NormalizeToolResultForStorageWithSpill is NormalizeToolResultForStorage with
+// an explicit spill location (conversation/execution scoped).
+func NormalizeToolResultForStorageWithSpill(result *ToolResult, maxBytes int, spill ToolResultSpillConfig) *ToolResult {
 	if result == nil {
 		return nil
 	}
@@ -25,48 +42,14 @@ func NormalizeToolResultForStorage(result *ToolResult, maxBytes int) *ToolResult
 		return out
 	}
 
-	remaining := maxBytes
-	truncated := false
-	for i := range out.Content {
-		if out.Content[i].Type != "text" {
-			continue
-		}
-		if remaining <= 0 {
-			out.Content[i].Text = ""
-			truncated = true
-			continue
-		}
-		text := out.Content[i].Text
-		if len(text) <= remaining {
-			remaining -= len(text)
-			continue
-		}
-		out.Content[i].Text = truncateUTF8Bytes(text, remaining)
-		remaining = 0
-		truncated = true
-	}
-
-	if truncated {
-		marker := fmt.Sprintf("\n\n...[tool output truncated: original %d bytes, kept %d bytes]...", total, maxBytes)
-		textBudget := maxBytes - len(marker)
-		if textBudget < 0 {
-			marker = truncateUTF8Bytes(marker, maxBytes)
-			textBudget = 0
-		}
-		for i := range out.Content {
-			if out.Content[i].Type == "text" {
-				out.Content[i].Text = truncateUTF8Bytes(out.Content[i].Text, textBudget) + marker
-				remaining = 0
-				for j := range out.Content {
-					if j != i && out.Content[j].Type == "text" {
-						out.Content[j].Text = ""
-					}
-				}
-				return out
-			}
-		}
-		out.Content = append(out.Content, Content{Type: "text", Text: marker})
-	}
+	full := ToolResultPlainText(out)
+	bound := tooloutput.BoundWithSpill(full, maxBytes, tooloutput.SpillOpts{
+		RootDir:        spill.RootDir,
+		ProjectID:      spill.ProjectID,
+		ConversationID: spill.ConversationID,
+		ExecutionID:    spill.ExecutionID,
+	})
+	out.Content = []Content{{Type: "text", Text: bound}}
 	return out
 }
 
@@ -80,20 +63,3 @@ func cloneToolResult(in *ToolResult) *ToolResult {
 	}
 	return &out
 }
-
-func truncateUTF8Bytes(s string, maxBytes int) string {
-	if maxBytes <= 0 {
-		return ""
-	}
-	if len(s) <= maxBytes {
-		return s
-	}
-	cut := maxBytes
-	for cut > 0 && (s[cut]&0xC0) == 0x80 {
-		cut--
-	}
-	if cut <= 0 {
-		return ""
-	}
-	return s[:cut]
-}

internal/mcp/tool_result_guard_test.go

@@ -2,6 +2,8 @@ package mcp
 
 import (
 	"context"
+	"os"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -63,12 +65,15 @@ func TestServerCallToolStoresAndReturnsSameGuardedResult(t *testing.T) {
 	storage := newInMemoryMonitorStorage()
 	server := NewServerWithStorage(zap.NewNop(), storage)
 	server.ConfigureToolWaitTimeoutSeconds(0)
-	server.ConfigureToolResultMaxBytes(50)
+	server.ConfigureToolResultMaxBytes(400)
+	spillRoot := t.TempDir()
+	server.ConfigureToolResultSpillRoot(spillRoot)
 	server.RegisterTool(Tool{Name: "big", InputSchema: map[string]interface{}{"type": "object"}}, func(ctx context.Context, args map[string]interface{}) (*ToolResult, error) {
-		return &ToolResult{Content: []Content{{Type: "text", Text: strings.Repeat("x", 100)}}}, nil
+		return &ToolResult{Content: []Content{{Type: "text", Text: strings.Repeat("x", 800)}}}, nil
 	})
 
-	result, executionID, err := server.CallTool(context.Background(), "big", nil)
+	ctx := WithMCPConversationID(context.Background(), "conv-spill")
+	result, executionID, err := server.CallTool(ctx, "big", nil)
 	if err != nil {
 		t.Fatalf("CallTool: %v", err)
 	}
@@ -76,13 +81,29 @@ func TestServerCallToolStoresAndReturnsSameGuardedResult(t *testing.T) {
 		t.Fatal("missing execution id")
 	}
 	returned := ToolResultPlainText(result)
-	if !strings.Contains(returned, "tool output truncated") || strings.Contains(returned, strings.Repeat("x", 100)) {
-		t.Fatalf("returned result was not guarded: %q", returned)
+	if !strings.Contains(returned, "<persisted-output>") || !strings.Contains(returned, "Full output saved to:") {
+		t.Fatalf("returned result was not spilled: %q", returned)
 	}
-	if len(returned) > 50 {
+	if len(returned) > 400 {
 		t.Fatalf("returned result exceeded hard limit: len=%d text=%q", len(returned), returned)
 	}
 
+	spillPath := filepath.Join(spillRoot, "conversations", "conv-spill", "trunc", executionID)
+	abs, err := filepath.Abs(spillPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(returned, abs) {
+		t.Fatalf("missing spill path %q in %q", abs, returned)
+	}
+	body, err := os.ReadFile(abs)
+	if err != nil {
+		t.Fatalf("read spill file: %v", err)
+	}
+	if string(body) != strings.Repeat("x", 800) {
+		t.Fatalf("spill body mismatch: len=%d", len(body))
+	}
+
 	inMem, ok := server.GetExecution(executionID)
 	if !ok || inMem == nil || inMem.Result == nil {
 		t.Fatalf("missing in-memory execution: %#v", inMem)
@@ -101,11 +122,14 @@ func TestServerCallToolStoresAndReturnsSameGuardedResult(t *testing.T) {
 
 func TestExecutionServiceStoresGuardedResult(t *testing.T) {
 	service := NewExecutionService(nil, zap.NewNop())
-	service.ConfigureToolResultMaxBytes(80)
+	service.ConfigureToolResultMaxBytes(400)
+	spillRoot := t.TempDir()
+	service.ConfigureToolResultSpillRoot(spillRoot)
 	handle, err := service.Submit(context.Background(), ExecutionRequest{
-		ToolName: "big",
+		ToolName:       "big",
+		ConversationID: "svc-conv",
 		Run: func(context.Context) (*ToolResult, error) {
-			return &ToolResult{Content: []Content{{Type: "text", Text: strings.Repeat("a", 200)}}}, nil
+			return &ToolResult{Content: []Content{{Type: "text", Text: strings.Repeat("a", 800)}}}, nil
 		},
 	})
 	if err != nil {
@@ -116,10 +140,19 @@ func TestExecutionServiceStoresGuardedResult(t *testing.T) {
 		t.Fatalf("Wait: %v", err)
 	}
 	got := ToolResultPlainText(snap.Execution.Result)
-	if !strings.Contains(got, "tool output truncated") || strings.Contains(got, strings.Repeat("a", 64)) {
-		t.Fatalf("service result was not guarded: %q", got)
+	if !strings.Contains(got, "<persisted-output>") {
+		t.Fatalf("service result was not spilled: %q", got)
 	}
-	if len(got) > 80 {
+	if len(got) > 400 {
 		t.Fatalf("service result exceeded hard limit: len=%d text=%q", len(got), got)
 	}
+	path := filepath.Join(spillRoot, "conversations", "svc-conv", "trunc", handle.ID)
+	abs, _ := filepath.Abs(path)
+	body, err := os.ReadFile(abs)
+	if err != nil {
+		t.Fatalf("read spill: %v", err)
+	}
+	if string(body) != strings.Repeat("a", 800) {
+		t.Fatalf("unexpected spill body len=%d", len(body))
+	}
 }