diff --git a/internal/multiagent/eino_single_runner.go b/internal/multiagent/eino_single_runner.go index 6a1b494b..cb5f0743 100644 --- a/internal/multiagent/eino_single_runner.go +++ b/internal/multiagent/eino_single_runner.go @@ -120,7 +120,7 @@ func RunEinoSingleChatModelAgent( } if einoSkillMW != nil { if einoFSTools && einoLoc != nil { - fsMw, fsErr := subAgentAgenticFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, einoSingleAgentName, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + fsMw, fsErr := subAgentAgenticFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, einoSingleAgentName, conversationID, projectID, ma.EinoMiddleware.ReductionRootDir, toolMaxBytesFromMW(&ma.EinoMiddleware), mcpExecBinder, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) if fsErr != nil { return nil, fmt.Errorf("eino single filesystem 中间件: %w", fsErr) } diff --git a/internal/multiagent/eino_skills.go b/internal/multiagent/eino_skills.go index 9cbaec95..2340c7b6 100644 --- a/internal/multiagent/eino_skills.go +++ b/internal/multiagent/eino_skills.go @@ -2,6 +2,7 @@ package multiagent import ( "context" + "encoding/json" "fmt" "os" "path/filepath" @@ -10,11 +11,13 @@ import ( "cyberstrike-ai/internal/config" "cyberstrike-ai/internal/einomcp" "cyberstrike-ai/internal/security" + "cyberstrike-ai/internal/tooloutput" localbk "github.com/cloudwego/eino-ext/adk/backend/local" "github.com/cloudwego/eino/adk" "github.com/cloudwego/eino/adk/middlewares/filesystem" "github.com/cloudwego/eino/adk/middlewares/skill" + "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/schema" "go.uber.org/zap" ) @@ -100,6 +103,11 @@ func subAgentAgenticFilesystemMiddleware( loc *localbk.Local, invokeNotify *einomcp.ToolInvokeNotifyHolder, einoAgentName string, + conversationID string, + projectID string, + reductionRootDir string, + toolMaxBytes int, + binder *MCPExecutionBinder, beginMonitor func(toolCallID, command string) string, appendPartialMonitor func(executionID, toolCallID, chunk string), registerCancelMonitor func(executionID string, cancel context.CancelFunc), @@ -113,7 +121,7 @@ func subAgentAgenticFilesystemMiddleware( if loc == nil { return nil, nil } - return filesystem.NewTyped[*schema.AgenticMessage](ctx, &filesystem.MiddlewareConfig{ + mw, err := filesystem.NewTyped[*schema.AgenticMessage](ctx, &filesystem.MiddlewareConfig{ Backend: loc, StreamingShell: &einoStreamingShellWrap{ inner: security.NewEinoStreamingShell(), @@ -130,6 +138,71 @@ func subAgentAgenticFilesystemMiddleware( shellNoOutputTimeoutSec: shellNoOutputTimeoutSec, }, }) + if err != nil { + return nil, err + } + return &einoAgenticFilesystemToolMiddleware{ + TypedChatModelAgentMiddleware: mw, + conversationID: conversationID, + projectID: projectID, + reductionRootDir: reductionRootDir, + toolMaxBytes: toolMaxBytes, + binder: binder, + }, nil +} + +type einoAgenticFilesystemToolMiddleware struct { + adk.TypedChatModelAgentMiddleware[*schema.AgenticMessage] + conversationID string + projectID string + reductionRootDir string + toolMaxBytes int + binder *MCPExecutionBinder +} + +func (m *einoAgenticFilesystemToolMiddleware) WrapInvokableToolCall(ctx context.Context, endpoint adk.InvokableToolCallEndpoint, tCtx *adk.ToolContext) (adk.InvokableToolCallEndpoint, error) { + wrapped, err := m.TypedChatModelAgentMiddleware.WrapInvokableToolCall(ctx, endpoint, tCtx) + if err != nil { + return nil, err + } + if tCtx == nil || !isBuiltinEinoADKFilesystemToolName(tCtx.Name) { + return wrapped, nil + } + return func(ctx context.Context, argumentsInJSON string, opts ...tool.Option) (string, error) { + args := parseToolArgumentsObject(argumentsInJSON) + if len(args) > 0 && m.binder != nil { + m.binder.BindArguments(tCtx.CallID, args) + } + result, runErr := wrapped(ctx, argumentsInJSON, opts...) + if runErr != nil { + return result, runErr + } + return m.boundToolResult(tCtx.CallID, result), nil + }, nil +} + +func (m *einoAgenticFilesystemToolMiddleware) boundToolResult(toolCallID, result string) string { + if m == nil || m.toolMaxBytes <= 0 || len(result) <= m.toolMaxBytes { + return result + } + return tooloutput.BoundWithSpill(result, m.toolMaxBytes, tooloutput.SpillOpts{ + RootDir: m.reductionRootDir, + ProjectID: m.projectID, + ConversationID: m.conversationID, + ExecutionID: toolCallID, + }) +} + +func parseToolArgumentsObject(raw string) map[string]interface{} { + raw = strings.TrimSpace(raw) + if raw == "" || raw == "{}" || raw == "null" { + return nil + } + var args map[string]interface{} + if err := json.Unmarshal([]byte(raw), &args); err != nil || len(args) == 0 { + return nil + } + return args } // agentToolTimeoutMinutes 返回 agent.tool_timeout_minutes（与 executeToolViaMCP 一致）；cfg 为 nil 时 0。 diff --git a/internal/multiagent/eino_tool_result_progress_emitter.go b/internal/multiagent/eino_tool_result_progress_emitter.go index b425d471..f4be40ec 100644 --- a/internal/multiagent/eino_tool_result_progress_emitter.go +++ b/internal/multiagent/eino_tool_result_progress_emitter.go @@ -115,7 +115,22 @@ func (e *einoToolResultProgressEmitter) Emit(ctx context.Context, toolName, cont if e.executeStdoutDup != nil { e.executeStdoutDup.Record(toolName, content, displayIsErr) } - recordEinoADKFilesystemToolMonitor(ctx, e.filesystemMonitorAgent, e.filesystemMonitorRecord, e.mcpExecutionBinder, toolName, toolCallID, e.messages(), content, displayIsErr) + if args := e.toolCallArguments(toolCallID, toolName); len(args) > 0 { + data["argumentsObj"] = args + data["arguments"] = mustMarshalToolArguments(args) + } + if execID := recordEinoADKFilesystemToolMonitor(ctx, e.filesystemMonitorAgent, e.filesystemMonitorRecord, e.mcpExecutionBinder, toolName, toolCallID, e.messages(), content, displayIsErr); execID != "" { + if stored := e.filesystemMonitorAgent.MCPExecutionResultText(execID); strings.TrimSpace(stored) != "" { + content = stored + if len(content) > 200 { + preview = content[:200] + "..." + } else { + preview = content + } + data["result"] = content + data["resultPreview"] = preview + } + } if e.filesystemMonitorAgent != nil && e.mcpExecutionBinder != nil { if execID := e.mcpExecutionBinder.ExecutionID(toolCallID); execID != "" { e.filesystemMonitorAgent.UpdateMCPExecutionDisplayResult(execID, content) @@ -152,3 +167,15 @@ func (e *einoToolResultProgressEmitter) messages() []adk.Message { } return e.runMessages.Messages() } + +func (e *einoToolResultProgressEmitter) toolCallArguments(toolCallID, toolName string) map[string]interface{} { + if e == nil { + return nil + } + if e.mcpExecutionBinder != nil { + if args := e.mcpExecutionBinder.Arguments(toolCallID); len(args) > 0 { + return args + } + } + return toolCallArgsFromAccumulated(e.messages(), toolCallID, toolName) +} diff --git a/internal/multiagent/eino_tool_result_progress_emitter_test.go b/internal/multiagent/eino_tool_result_progress_emitter_test.go index 71470d1b..ada417f5 100644 --- a/internal/multiagent/eino_tool_result_progress_emitter_test.go +++ b/internal/multiagent/eino_tool_result_progress_emitter_test.go @@ -1,6 +1,11 @@ package multiagent -import "testing" +import ( + "testing" + + "github.com/cloudwego/eino/adk" + "github.com/cloudwego/eino/schema" +) func TestEinoToolResultProgressEmitterInfersPendingAndDedupes(t *testing.T) { var events []map[string]interface{} @@ -135,3 +140,44 @@ func TestEinoToolResultProgressEmitterTruncatesPreview(t *testing.T) { t.Fatalf("preview = %q len=%d", got, len(got)) } } + +func TestEinoToolResultProgressEmitterBackfillsArgumentsFromRunMessages(t *testing.T) { + var data map[string]interface{} + progress := func(eventType, _ string, raw interface{}) { + if eventType == "tool_result" { + data, _ = raw.(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/requirements.txt"}`, + }, + }}, + }, + }) + emitter := newEinoToolResultProgressEmitter(einoToolResultProgressEmitterConfig{ + ConversationID: "conv-1", + Progress: progress, + RunMessages: runMessages, + }) + + if !emitter.Emit(nil, "read_file", "ok", "call-read", false, "lead") { + t.Fatal("expected tool result emit") + } + args, ok := data["argumentsObj"].(map[string]interface{}) + if !ok { + t.Fatalf("argumentsObj = %#v", data["argumentsObj"]) + } + if args["path"] != "/tmp/requirements.txt" { + t.Fatalf("path = %#v, want /tmp/requirements.txt", args["path"]) + } + if data["arguments"] != `{"path":"/tmp/requirements.txt"}` { + t.Fatalf("arguments = %#v", data["arguments"]) + } +} diff --git a/internal/multiagent/mcp_execution_binder.go b/internal/multiagent/mcp_execution_binder.go index 79fec45b..e9f55ff7 100644 --- a/internal/multiagent/mcp_execution_binder.go +++ b/internal/multiagent/mcp_execution_binder.go @@ -7,12 +7,16 @@ import ( // MCPExecutionBinder maps ADK toolCallID → MCP monitor execution ID for a single agent run. type MCPExecutionBinder struct { - mu sync.RWMutex - byToolCall map[string]string + mu sync.RWMutex + byToolCall map[string]string + argsByToolCall map[string]map[string]interface{} } func NewMCPExecutionBinder() *MCPExecutionBinder { - return &MCPExecutionBinder{byToolCall: make(map[string]string)} + return &MCPExecutionBinder{ + byToolCall: make(map[string]string), + argsByToolCall: make(map[string]map[string]interface{}), + } } func (b *MCPExecutionBinder) Bind(toolCallID, executionID string) { @@ -29,6 +33,19 @@ func (b *MCPExecutionBinder) Bind(toolCallID, executionID string) { b.mu.Unlock() } +func (b *MCPExecutionBinder) BindArguments(toolCallID string, args map[string]interface{}) { + if b == nil || len(args) == 0 { + return + } + tid := strings.TrimSpace(toolCallID) + if tid == "" { + return + } + b.mu.Lock() + b.argsByToolCall[tid] = cloneToolArgs(args) + b.mu.Unlock() +} + func (b *MCPExecutionBinder) ExecutionID(toolCallID string) string { if b == nil { return "" @@ -38,3 +55,24 @@ func (b *MCPExecutionBinder) ExecutionID(toolCallID string) string { defer b.mu.RUnlock() return b.byToolCall[tid] } + +func (b *MCPExecutionBinder) Arguments(toolCallID string) map[string]interface{} { + if b == nil { + return nil + } + tid := strings.TrimSpace(toolCallID) + b.mu.RLock() + defer b.mu.RUnlock() + return cloneToolArgs(b.argsByToolCall[tid]) +} + +func cloneToolArgs(args map[string]interface{}) map[string]interface{} { + if len(args) == 0 { + return nil + } + out := make(map[string]interface{}, len(args)) + for k, v := range args { + out[k] = v + } + return out +} diff --git a/internal/multiagent/mcp_execution_binder_test.go b/internal/multiagent/mcp_execution_binder_test.go index e2f142fd..97d31d53 100644 --- a/internal/multiagent/mcp_execution_binder_test.go +++ b/internal/multiagent/mcp_execution_binder_test.go @@ -9,9 +9,13 @@ import ( func TestMCPExecutionBinder(t *testing.T) { b := NewMCPExecutionBinder() b.Bind("call-1", "exec-1") + b.BindArguments("call-1", map[string]interface{}{"file_path": "/tmp/a.txt"}) if got := b.ExecutionID("call-1"); got != "exec-1" { t.Fatalf("expected exec-1, got %q", got) } + if got := b.Arguments("call-1"); got["file_path"] != "/tmp/a.txt" { + t.Fatalf("arguments = %#v", got) + } if got := b.ExecutionID("missing"); got != "" { t.Fatalf("expected empty, got %q", got) } diff --git a/internal/multiagent/runner.go b/internal/multiagent/runner.go index 1052e3a8..10cb1da8 100644 --- a/internal/multiagent/runner.go +++ b/internal/multiagent/runner.go @@ -52,6 +52,7 @@ type RunResult struct { type toolCallPendingInfo struct { ToolCallID string ToolName string + Arguments map[string]interface{} EinoAgent string EinoRole string } @@ -232,7 +233,7 @@ func RunDeepAgent( } if agenticSkillMW != nil { if agenticFSTools && agenticLoc != nil { - subFs, fsErr := subAgentAgenticFilesystemMiddleware(ctx, agenticLoc, toolInvokeNotify, id, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + subFs, fsErr := subAgentAgenticFilesystemMiddleware(ctx, agenticLoc, toolInvokeNotify, id, conversationID, projectID, ma.EinoMiddleware.ReductionRootDir, toolMaxBytesFromMW(&ma.EinoMiddleware), mcpExecBinder, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) if fsErr != nil { return nil, fmt.Errorf("子代理 %q filesystem 中间件: %w", id, fsErr) } @@ -492,7 +493,7 @@ func RunDeepAgent( } var peFsMw adk.TypedChatModelAgentMiddleware[*schema.AgenticMessage] if agenticSkillMW != nil && agenticFSTools && agenticLoc != nil { - peFsMw, err = subAgentAgenticFilesystemMiddleware(ctx, agenticLoc, toolInvokeNotify, "executor", einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + peFsMw, err = subAgentAgenticFilesystemMiddleware(ctx, agenticLoc, toolInvokeNotify, "executor", conversationID, projectID, ma.EinoMiddleware.ReductionRootDir, toolMaxBytesFromMW(&ma.EinoMiddleware), mcpExecBinder, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) if err != nil { return nil, fmt.Errorf("plan_execute agentic filesystem 中间件: %w", err) } @@ -994,6 +995,7 @@ func emitToolCallsFromMessage( markPending(toolCallPendingInfo{ ToolCallID: toolCallID, ToolName: display, + Arguments: argsObj, EinoAgent: agentName, EinoRole: role, })