diff --git a/internal/multiagent/eino_filesystem_tool_monitor.go b/internal/multiagent/eino_filesystem_tool_monitor.go index d69fcc5c..8e3a8cb5 100644 --- a/internal/multiagent/eino_filesystem_tool_monitor.go +++ b/internal/multiagent/eino_filesystem_tool_monitor.go @@ -64,6 +64,17 @@ func toolCallArgsFromAccumulated(msgs []adk.Message, toolCallID, expectToolName return map[string]interface{}{} } +func mustMarshalToolArguments(args map[string]interface{}) string { + if len(args) == 0 { + return "{}" + } + raw, err := json.Marshal(args) + if err != nil { + return "{}" + } + return string(raw) +} + // beginEinoADKFilesystemToolMonitor 在 Eino ADK filesystem 工具开始调用时写入 running 状态。 func beginEinoADKFilesystemToolMonitor( ctx context.Context, @@ -71,6 +82,7 @@ func beginEinoADKFilesystemToolMonitor( rec einomcp.ExecutionRecorder, binder *MCPExecutionBinder, toolCallID, toolName string, + args map[string]interface{}, ) { if ag == nil || rec == nil { return @@ -87,7 +99,7 @@ func beginEinoADKFilesystemToolMonitor( return } storedName := "eino_fs::" + strings.ToLower(name) - id := ag.BeginLocalToolExecution(ctx, storedName, map[string]interface{}{}) + id := ag.BeginLocalToolExecution(ctx, storedName, args) if id == "" { return } @@ -108,18 +120,21 @@ func recordEinoADKFilesystemToolMonitor( msgs []adk.Message, resultText string, isErr bool, -) { +) string { if ag == nil || rec == nil { - return + return "" } name := strings.TrimSpace(toolName) if name == "" || strings.EqualFold(name, "execute") { - return + return "" } if !isBuiltinEinoADKFilesystemToolName(name) { - return + return "" } args := toolCallArgsFromAccumulated(msgs, toolCallID, name) + if len(args) == 0 && binder != nil { + args = binder.Arguments(toolCallID) + } storedName := "eino_fs::" + strings.ToLower(name) var invErr error if isErr { @@ -138,4 +153,5 @@ func recordEinoADKFilesystemToolMonitor( if id != "" && execID == "" { rec(id, toolCallID) } + return id } diff --git a/internal/multiagent/eino_filesystem_tool_monitor_test.go b/internal/multiagent/eino_filesystem_tool_monitor_test.go index 332e81f4..1beaa6ae 100644 --- a/internal/multiagent/eino_filesystem_tool_monitor_test.go +++ b/internal/multiagent/eino_filesystem_tool_monitor_test.go @@ -2,6 +2,7 @@ package multiagent import ( "context" + "strings" "testing" "cyberstrike-ai/internal/agent" @@ -10,6 +11,7 @@ import ( "cyberstrike-ai/internal/mcp" "github.com/cloudwego/eino/adk" + "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" "go.uber.org/zap" ) @@ -26,7 +28,7 @@ func TestEinoADKFilesystemToolMonitorBindsFinishesAndUpdatesDisplayResult(t *tes recorded = append(recorded, executionID+"|"+toolCallID) }) - beginEinoADKFilesystemToolMonitor(ctx, ag, rec, binder, "call-read", "read_file") + beginEinoADKFilesystemToolMonitor(ctx, ag, rec, binder, "call-read", "read_file", map[string]interface{}{"path": "/tmp/secret.txt"}) execID := binder.ExecutionID("call-read") if execID == "" { t.Fatal("expected begin to bind execution id") @@ -38,6 +40,9 @@ func TestEinoADKFilesystemToolMonitorBindsFinishesAndUpdatesDisplayResult(t *tes if len(recorded) != 1 || recorded[0] != execID+"|call-read" { t.Fatalf("recorded begin ids = %#v", recorded) } + if got, _ := exec.Arguments["path"].(string); got != "/tmp/secret.txt" { + t.Fatalf("begin execution args = %#v", exec.Arguments) + } runMessages := newEinoRunMessageAccumulator([]adk.Message{ &schema.Message{ @@ -80,3 +85,86 @@ func TestEinoADKFilesystemToolMonitorBindsFinishesAndUpdatesDisplayResult(t *tes t.Fatalf("finish should reuse existing execution without recording a second id, got %#v", recorded) } } + +func TestEinoADKFilesystemToolMonitorSpillsLargeReadFileResultForProgress(t *testing.T) { + t.Parallel() + ctx := context.Background() + logger := zap.NewNop() + server := mcp.NewServer(logger) + server.ConfigureToolResultMaxBytes(400) + server.ConfigureToolResultSpillRoot(t.TempDir()) + ag := agent.NewAgent(&config.OpenAIConfig{}, &config.AgentConfig{}, server, nil, logger, 1) + binder := NewMCPExecutionBinder() + rec := einomcp.ExecutionRecorder(func(executionID, toolCallID string) {}) + var event map[string]interface{} + + runMessages := newEinoRunMessageAccumulator([]adk.Message{ + &schema.Message{ + Role: schema.Assistant, + ToolCalls: []schema.ToolCall{{ + ID: "call-read", + Type: "function", + Function: schema.FunctionCall{ + Name: "read_file", + Arguments: `{"path":"/tmp/large.txt"}`, + }, + }}, + }, + }) + beginEinoADKFilesystemToolMonitor(ctx, ag, rec, binder, "call-read", "read_file", map[string]interface{}{"path": "/tmp/large.txt"}) + emitter := newEinoToolResultProgressEmitter(einoToolResultProgressEmitterConfig{ + ConversationID: "conv-1", + RunMessages: runMessages, + FilesystemMonitorAgent: ag, + FilesystemMonitorRecord: rec, + MCPExecutionBinder: binder, + Progress: func(eventType, _ string, data interface{}) { + if eventType == "tool_result" { + event, _ = data.(map[string]interface{}) + } + }, + }) + + if !emitter.Emit(ctx, "read_file", strings.Repeat("0123456789", 100), "call-read", false, "lead") { + t.Fatal("expected tool result emit") + } + result, _ := event["result"].(string) + if !strings.Contains(result, "") || !strings.Contains(result, "Full output saved to:") { + t.Fatalf("large read_file result was not spilled in progress event: %q", result) + } + if len(result) > 400 { + t.Fatalf("progress result exceeded configured max: len=%d text=%q", len(result), result) + } +} + +func TestEinoAgenticFilesystemWrapperCapturesArgumentsAndSpillsResult(t *testing.T) { + t.Parallel() + binder := NewMCPExecutionBinder() + mw := &einoAgenticFilesystemToolMiddleware{ + TypedChatModelAgentMiddleware: &adk.TypedBaseChatModelAgentMiddleware[*schema.AgenticMessage]{}, + conversationID: "conv-1", + toolMaxBytes: 400, + reductionRootDir: t.TempDir(), + binder: binder, + } + endpoint, err := mw.WrapInvokableToolCall(context.Background(), func(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) { + return strings.Repeat("0123456789", 100), nil + }, &adk.ToolContext{Name: "read_file", CallID: "call-read"}) + if err != nil { + t.Fatalf("WrapInvokableToolCall: %v", err) + } + result, err := endpoint(context.Background(), `{"file_path":"/tmp/requirements.txt","limit":2000}`) + if err != nil { + t.Fatalf("endpoint: %v", err) + } + args := binder.Arguments("call-read") + if args["file_path"] != "/tmp/requirements.txt" { + t.Fatalf("captured args = %#v", args) + } + if !strings.Contains(result, "") || !strings.Contains(result, "Full output saved to:") { + t.Fatalf("expected persisted-output summary, got %q", result) + } + if len(result) > 400 { + t.Fatalf("summary exceeded max bytes: len=%d", len(result)) + } +} diff --git a/internal/multiagent/eino_run_event_drain.go b/internal/multiagent/eino_run_event_drain.go index 5333caf7..4aacc70b 100644 --- a/internal/multiagent/eino_run_event_drain.go +++ b/internal/multiagent/eino_run_event_drain.go @@ -228,6 +228,7 @@ func (d *einoRunEventDrain) markPendingWithMonitor(tc toolCallPendingInfo) { d.cfg.MCPExecutionBinder, tc.ToolCallID, tc.ToolName, + tc.Arguments, ) }