From bccb324d2b93c4b767e50efcb139c3a81f976392 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=85=AC=E6=98=8E?= <83812544+Ed1s0nZ@users.noreply.github.com> Date: Wed, 22 Jul 2026 14:17:23 +0800 Subject: [PATCH] Add files via upload --- internal/multiagent/eino_execute_monitor.go | 23 ++++- .../multiagent/eino_execute_streaming_wrap.go | 94 ++++++++++++++++--- .../eino_execute_streaming_wrap_test.go | 75 +++++++++++++++ internal/multiagent/eino_single_runner.go | 4 +- internal/multiagent/eino_skills.go | 15 +++ internal/multiagent/runner.go | 10 +- 6 files changed, 202 insertions(+), 19 deletions(-) diff --git a/internal/multiagent/eino_execute_monitor.go b/internal/multiagent/eino_execute_monitor.go index 7450a344..6630b94b 100644 --- a/internal/multiagent/eino_execute_monitor.go +++ b/internal/multiagent/eino_execute_monitor.go @@ -12,6 +12,9 @@ import ( // 与 CallTool 路径一致，使监控页能展示「执行中」状态。 func newEinoExecuteMonitorCallbacks(ctx context.Context, ag *agent.Agent, recorder einomcp.ExecutionRecorder) ( begin func(toolCallID, command string) string, + appendPartial func(executionID, toolCallID, chunk string), + registerCancel func(executionID string, cancel context.CancelFunc), + unregisterCancel func(executionID string), finish func(executionID, toolCallID, command, stdout string, success bool, invokeErr error), ) { begin = func(toolCallID, command string) string { @@ -25,6 +28,24 @@ func newEinoExecuteMonitorCallbacks(ctx context.Context, ag *agent.Agent, record } return id } + appendPartial = func(executionID, toolCallID, chunk string) { + if ag == nil || executionID == "" || chunk == "" { + return + } + ag.AppendLocalToolExecutionPartialOutput(executionID, chunk) + } + registerCancel = func(executionID string, cancel context.CancelFunc) { + if ag == nil || executionID == "" || cancel == nil { + return + } + ag.RegisterLocalToolExecutionCancel(executionID, cancel) + } + unregisterCancel = func(executionID string) { + if ag == nil || executionID == "" { + return + } + ag.UnregisterLocalToolExecutionCancel(executionID) + } finish = func(executionID, toolCallID, command, stdout string, success bool, invokeErr error) { if ag == nil { return @@ -43,5 +64,5 @@ func newEinoExecuteMonitorCallbacks(ctx context.Context, ag *agent.Agent, record recorder(id, toolCallID) } } - return begin, finish + return begin, appendPartial, registerCancel, unregisterCancel, finish } diff --git a/internal/multiagent/eino_execute_streaming_wrap.go b/internal/multiagent/eino_execute_streaming_wrap.go index 96afbf3a..0e7af38b 100644 --- a/internal/multiagent/eino_execute_streaming_wrap.go +++ b/internal/multiagent/eino_execute_streaming_wrap.go @@ -63,11 +63,16 @@ type einoStreamingShellWrap struct { outputChunk func(toolName, toolCallID, chunk string) // toolTimeoutMinutes 与 agent.tool_timeout_minutes 对齐；>0 时对单次 execute 套用 context 超时（与 MCP 工具经 executeToolViaMCP 行为一致）。0 表示仅依赖上层 ctx（如整任务 10h 上限）。 toolTimeoutMinutes int + // toolWaitTimeoutSeconds 与 agent.tool_wait_timeout_seconds 对齐；>0 时本轮等待到期后返回 execution_id，shell 继续后台运行。 + toolWaitTimeoutSeconds int // shellNoOutputTimeoutSec：无任何输出时的空闲秒数；0=关闭。 shellNoOutputTimeoutSec int // beginMonitor 在 execute 开始时写入 running 状态；finishMonitor 在流结束后更新为 completed/failed。 - beginMonitor func(toolCallID, command string) string - finishMonitor func(executionID, toolCallID, command, stdout string, success bool, invokeErr error) + beginMonitor func(toolCallID, command string) string + appendPartialMonitor func(executionID, toolCallID, chunk string) + registerCancelMonitor func(executionID string, cancel context.CancelFunc) + unregisterCancelMonitor func(executionID string) + finishMonitor func(executionID, toolCallID, command, stdout string, success bool, invokeErr error) } func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *filesystem.ExecuteRequest) (*schema.StreamReader[*filesystem.ExecuteResponse], error) { @@ -104,6 +109,9 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi if w.toolTimeoutMinutes > 0 { execCtx, timeoutCancel = context.WithTimeout(execCtx, time.Duration(w.toolTimeoutMinutes)*time.Minute) } + if monitorExecID != "" && w.registerCancelMonitor != nil { + w.registerCancelMonitor(monitorExecID, execCancel) + } if execReg != nil && convID != "" { execReg.RegisterActiveEinoExecute(convID, execCancel) } @@ -116,6 +124,9 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi if execCancel != nil { execCancel() } + if monitorExecID != "" && w.unregisterCancelMonitor != nil { + w.unregisterCancelMonitor(monitorExecID) + } if einoExecuteRecvErrIsToolTimeout(err, execCtx) { hint := "\n\n" + einoExecuteTimeoutUserHint() + "\n" if w.finishMonitor != nil { @@ -146,7 +157,7 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi outR, outW := schema.Pipe[*filesystem.ExecuteResponse](32) - go func(inner *schema.StreamReader[*filesystem.ExecuteResponse], command string, cancel context.CancelFunc, timeoutCleanup context.CancelFunc, tctx context.Context, conversationID string, reg mcp.EinoExecuteRunRegistry, toolReg mcp.ToolRunRegistry, execID string, toolCallID string, noOutputSec int) { + go func(inner *schema.StreamReader[*filesystem.ExecuteResponse], command string, cancel context.CancelFunc, timeoutCleanup context.CancelFunc, tctx context.Context, conversationID string, reg mcp.EinoExecuteRunRegistry, toolReg mcp.ToolRunRegistry, execID string, toolCallID string, noOutputSec int, waitTimeoutSec int) { var innerCloseOnce sync.Once closeInner := func() { innerCloseOnce.Do(func() { inner.Close() }) @@ -164,6 +175,9 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi if toolReg != nil && conversationID != "" && execID != "" { defer toolReg.UnregisterRunningTool(conversationID, execID) } + if w.unregisterCancelMonitor != nil && execID != "" { + defer w.unregisterCancelMonitor(execID) + } // ctx 取消时关闭内层流，避免 amass 等长时间无换行输出时 Recv 永久阻塞。 stopWatch := make(chan struct{}) @@ -181,11 +195,30 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi var invokeErr error exitCode := 0 hasExitCode := false + softReturned := false + var outCloseOnce sync.Once + closeOut := func() { + outCloseOnce.Do(func() { outW.Close() }) + } + defer closeOut() + sendOut := func(resp *filesystem.ExecuteResponse, err error) bool { + if softReturned { + return false + } + return outW.Send(resp, err) + } idleWatch := security.NewShellInactivityWatch(noOutputSec) if idleWatch != nil { defer idleWatch.Stop() } + var waitTimeoutCh <-chan time.Time + var waitTimer *time.Timer + if waitTimeoutSec > 0 { + waitTimer = time.NewTimer(time.Duration(waitTimeoutSec) * time.Second) + waitTimeoutCh = waitTimer.C + defer waitTimer.Stop() + } type execRecvMsg struct { resp *filesystem.ExecuteResponse @@ -206,8 +239,11 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi success = false invokeErr = fmt.Errorf("shell inactivity timeout (%ds)", idleWatch.Sec) msg := security.ShellNoOutputTimeoutMessage(idleWatch.Sec) - _ = outW.Send(&filesystem.ExecuteResponse{Output: msg}, nil) + _ = sendOut(&filesystem.ExecuteResponse{Output: msg}, nil) sb.WriteString(msg) + if w.appendPartialMonitor != nil && execID != "" { + w.appendPartialMonitor(execID, toolCallID, msg) + } if w.outputChunk != nil && toolCallID != "" { w.outputChunk("execute", toolCallID, msg) } @@ -227,6 +263,14 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi case <-idleCh: fireInactivityTimeout() break recvLoop + case <-waitTimeoutCh: + if execID != "" && !softReturned { + msg := einoExecuteSoftWaitTimeoutResult(execID, waitTimeoutSec) + _ = outW.Send(&filesystem.ExecuteResponse{Output: msg}, nil) + softReturned = true + closeOut() + } + waitTimeoutCh = nil case msg := <-recvCh: rerr := msg.err resp := msg.resp @@ -244,7 +288,7 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi invokeErr = context.Canceled break recvLoop } - _ = outW.Send(nil, rerr) + _ = sendOut(nil, rerr) break recvLoop } if resp != nil { @@ -263,11 +307,14 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi } sb.WriteString(resp.Output) appended = resp.Output + if w.appendPartialMonitor != nil && execID != "" { + w.appendPartialMonitor(execID, toolCallID, appended) + } } if w.outputChunk != nil && strings.TrimSpace(appended) != "" { w.outputChunk("execute", toolCallID, appended) } - if outW.Send(resp, nil) { + if sendOut(resp, nil) { success = false invokeErr = fmt.Errorf("execute stream closed by consumer") break recvLoop @@ -304,7 +351,10 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi // ADK 从本 Pipe 拼出 tool 消息正文；仅 Notify 尾标不会进入模型上下文。超时句写入流，与 UI 一致。 if invokeErr != nil && errors.Is(invokeErr, context.DeadlineExceeded) { hint := "\n\n" + einoExecuteTimeoutUserHint() + "\n" - _ = outW.Send(&filesystem.ExecuteResponse{Output: hint}, nil) + _ = sendOut(&filesystem.ExecuteResponse{Output: hint}, nil) + if w.appendPartialMonitor != nil && execID != "" { + w.appendPartialMonitor(execID, toolCallID, hint) + } if w.outputChunk != nil && tid != "" { w.outputChunk("execute", tid, hint) } @@ -313,9 +363,9 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi // 中断时循环内已逐行写入 stdout；此处只追加 USER INTERRUPT NOTE，避免整段输出重复。 if invokeErr != nil && errors.Is(invokeErr, context.Canceled) && abortNote != "" { if partialStreamed != "" { - _ = outW.Send(&filesystem.ExecuteResponse{Output: "\n\n" + mcp.AbortNoteBannerForModel + "\n" + abortNote}, nil) + _ = sendOut(&filesystem.ExecuteResponse{Output: "\n\n" + mcp.AbortNoteBannerForModel + "\n" + abortNote}, nil) } else if text := strings.TrimSpace(sb.String()); text != "" { - _ = outW.Send(&filesystem.ExecuteResponse{Output: text + "\n"}, nil) + _ = sendOut(&filesystem.ExecuteResponse{Output: text + "\n"}, nil) } } rawOutput := sb.String() @@ -323,7 +373,10 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi if !success && hasExitCode && exitCode != 0 { statusLine := security.ExecuteFailureStatusLine(exitCode) if !strings.Contains(rawOutput, "命令执行失败:") { - _ = outW.Send(&filesystem.ExecuteResponse{Output: statusLine}, nil) + _ = sendOut(&filesystem.ExecuteResponse{Output: statusLine}, nil) + if w.appendPartialMonitor != nil && execID != "" { + w.appendPartialMonitor(execID, toolCallID, statusLine) + } sb.WriteString(statusLine) } fireBody = einomcp.ToolErrorPrefix + security.FormatCommandFailureResult(exitCode, rawOutput) @@ -332,10 +385,25 @@ func (w *einoStreamingShellWrap) ExecuteStreaming(ctx context.Context, input *fi w.finishMonitor(execID, toolCallID, command, sb.String(), success, invokeErr) } if w.invokeNotify != nil { - w.invokeNotify.Fire(toolCallID, "execute", agentTag, success, fireBody, invokeErr) + if !softReturned { + w.invokeNotify.Fire(toolCallID, "execute", agentTag, success, fireBody, invokeErr) + } } - outW.Close() - }(sr, userCmd, execCancel, timeoutCancel, execCtx, convID, execReg, toolRunReg, monitorExecID, tid, w.shellNoOutputTimeoutSec) + }(sr, userCmd, execCancel, timeoutCancel, execCtx, convID, execReg, toolRunReg, monitorExecID, tid, w.shellNoOutputTimeoutSec, w.toolWaitTimeoutSeconds) return outR, nil } + +func einoExecuteSoftWaitTimeoutResult(executionID string, waitTimeoutSec int) string { + waitText := "configured wait timeout" + if waitTimeoutSec > 0 { + waitText = fmt.Sprintf("%ds", waitTimeoutSec) + } + return fmt.Sprintf(`工具已提交到后台执行，当前仍在运行。 + +execution_id: %s +status: running +wait_timeout: %s + +你可以继续推理、改用其他工具，或调用 get_tool_execution / wait_tool_execution 读取 partial_output 并继续等待；也可以调用 cancel_tool_execution 取消。`, executionID, waitText) +} diff --git a/internal/multiagent/eino_execute_streaming_wrap_test.go b/internal/multiagent/eino_execute_streaming_wrap_test.go index 43d0e819..54549b0c 100644 --- a/internal/multiagent/eino_execute_streaming_wrap_test.go +++ b/internal/multiagent/eino_execute_streaming_wrap_test.go @@ -153,6 +153,81 @@ func TestEinoStreamingShellWrap_InactivityAfterPartialOutput(t *testing.T) { } } +func TestEinoStreamingShellWrap_SoftWaitTimeoutReturnsExecutionIDAndKeepsRunning(t *testing.T) { + inner := &mockStreamingShellPartialThenHang{} + partialCh := make(chan string, 4) + cancelCh := make(chan context.CancelFunc, 1) + unregistered := make(chan string, 1) + wrap := &einoStreamingShellWrap{ + inner: inner, + toolWaitTimeoutSeconds: 1, + beginMonitor: func(toolCallID, command string) string { + return "exec-soft-wait" + }, + appendPartialMonitor: func(executionID, toolCallID, chunk string) { + partialCh <- chunk + }, + registerCancelMonitor: func(executionID string, cancel context.CancelFunc) { + if executionID == "exec-soft-wait" { + cancelCh <- cancel + } + }, + unregisterCancelMonitor: func(executionID string) { + unregistered <- executionID + }, + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + sr, err := wrap.ExecuteStreaming(ctx, &filesystem.ExecuteRequest{Command: "sudo whoami"}) + if err != nil { + t.Fatalf("ExecuteStreaming: %v", err) + } + defer sr.Close() + + var got strings.Builder + for { + resp, rerr := sr.Recv() + if errors.Is(rerr, io.EOF) { + break + } + if rerr != nil { + t.Fatalf("recv: %v", rerr) + } + if resp != nil { + got.WriteString(resp.Output) + } + } + body := got.String() + if !strings.Contains(body, "execution_id: exec-soft-wait") || !strings.Contains(body, "status: running") { + t.Fatalf("expected background execution marker, got: %q", body) + } + select { + case chunk := <-partialCh: + if !strings.Contains(chunk, "[sudo] password") { + t.Fatalf("unexpected partial chunk: %q", chunk) + } + default: + t.Fatal("expected streamed partial output before soft wait return") + } + if !inner.called { + t.Fatal("inner shell did not run") + } + select { + case registeredCancel := <-cancelCh: + registeredCancel() + case <-time.After(time.Second): + t.Fatal("expected execution cancel registration") + } + select { + case id := <-unregistered: + if id != "exec-soft-wait" { + t.Fatalf("unexpected unregistered id: %q", id) + } + case <-time.After(time.Second): + t.Fatal("expected execution cancel unregister") + } +} + type mockStreamingShellHanging struct { called bool } diff --git a/internal/multiagent/eino_single_runner.go b/internal/multiagent/eino_single_runner.go index 756d9c45..ea7d255b 100644 --- a/internal/multiagent/eino_single_runner.go +++ b/internal/multiagent/eino_single_runner.go @@ -82,7 +82,7 @@ func RunEinoSingleChatModelAgent( } toolInvokeNotify := einomcp.NewToolInvokeNotifyHolder() - einoExecBegin, einoExecFinish := newEinoExecuteMonitorCallbacks(ctx, ag, recorder) + einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish := newEinoExecuteMonitorCallbacks(ctx, ag, recorder) mainDefs := ag.ToolsForRole(roleTools) mainTools, err := einomcp.ToolsFromDefinitions(ag, holder, mainDefs, recorder, nil, toolInvokeNotify, einoSingleAgentName) if err != nil { @@ -139,7 +139,7 @@ func RunEinoSingleChatModelAgent( } if einoSkillMW != nil { if einoFSTools && einoLoc != nil { - fsMw, fsErr := subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, einoSingleAgentName, einoExecBegin, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + fsMw, fsErr := subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, einoSingleAgentName, 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 e5fc5c75..724562ba 100644 --- a/internal/multiagent/eino_skills.go +++ b/internal/multiagent/eino_skills.go @@ -107,8 +107,12 @@ func subAgentFilesystemMiddleware( invokeNotify *einomcp.ToolInvokeNotifyHolder, einoAgentName string, beginMonitor func(toolCallID, command string) string, + appendPartialMonitor func(executionID, toolCallID, chunk string), + registerCancelMonitor func(executionID string, cancel context.CancelFunc), + unregisterCancelMonitor func(executionID string), finishMonitor func(executionID, toolCallID, command, stdout string, success bool, invokeErr error), toolTimeoutMinutes int, + toolWaitTimeoutSeconds int, shellNoOutputTimeoutSec int, outputChunk func(toolName, toolCallID, chunk string), ) (adk.ChatModelAgentMiddleware, error) { @@ -123,8 +127,12 @@ func subAgentFilesystemMiddleware( einoAgentName: strings.TrimSpace(einoAgentName), outputChunk: outputChunk, beginMonitor: beginMonitor, + appendPartialMonitor: appendPartialMonitor, + registerCancelMonitor: registerCancelMonitor, + unregisterCancelMonitor: unregisterCancelMonitor, finishMonitor: finishMonitor, toolTimeoutMinutes: toolTimeoutMinutes, + toolWaitTimeoutSeconds: toolWaitTimeoutSeconds, shellNoOutputTimeoutSec: shellNoOutputTimeoutSec, }, }) @@ -138,6 +146,13 @@ func agentToolTimeoutMinutes(cfg *config.Config) int { return cfg.Agent.ToolTimeoutMinutes } +func agentToolWaitTimeoutSeconds(cfg *config.Config) int { + if cfg == nil { + return 0 + } + return cfg.Agent.ToolWaitTimeoutSeconds +} + // agentShellNoOutputTimeoutSeconds：0=默认 300s（5 分钟）；-1=关闭；>0=自定义秒数。 func agentShellNoOutputTimeoutSeconds(cfg *config.Config) int { if cfg == nil { diff --git a/internal/multiagent/runner.go b/internal/multiagent/runner.go index 5e328d08..b41b02ea 100644 --- a/internal/multiagent/runner.go +++ b/internal/multiagent/runner.go @@ -135,7 +135,7 @@ func RunDeepAgent( mcpIDs = append(mcpIDs, id) mcpIDsMu.Unlock() } - einoExecBegin, einoExecFinish := newEinoExecuteMonitorCallbacks(ctx, ag, recorder) + einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish := newEinoExecuteMonitorCallbacks(ctx, ag, recorder) // 与单代理流式一致：在 response_start / response_delta 的 data 中带当前 mcpExecutionIds，供主聊天绑定复制与展示。 snapshotMCPIDs := func() []string { @@ -240,7 +240,7 @@ func RunDeepAgent( } if einoSkillMW != nil { if einoFSTools && einoLoc != nil { - subFs, fsErr := subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, id, einoExecBegin, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + subFs, fsErr := subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, id, einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) if fsErr != nil { return nil, fmt.Errorf("子代理 %q filesystem 中间件: %w", id, fsErr) } @@ -387,8 +387,12 @@ func RunDeepAgent( einoAgentName: orchestratorName, outputChunk: nil, beginMonitor: einoExecBegin, + appendPartialMonitor: einoExecAppendPartial, + registerCancelMonitor: einoExecRegisterCancel, + unregisterCancelMonitor: einoExecUnregisterCancel, finishMonitor: einoExecFinish, toolTimeoutMinutes: agentToolTimeoutMinutes(appCfg), + toolWaitTimeoutSeconds: agentToolWaitTimeoutSeconds(appCfg), shellNoOutputTimeoutSec: agentShellNoOutputTimeoutSeconds(appCfg), } } @@ -486,7 +490,7 @@ func RunDeepAgent( // 构建 filesystem 中间件（与 Deep sub-agent 一致） var peFsMw adk.ChatModelAgentMiddleware if einoSkillMW != nil && einoFSTools && einoLoc != nil { - peFsMw, err = subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, "executor", einoExecBegin, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) + peFsMw, err = subAgentFilesystemMiddleware(ctx, einoLoc, toolInvokeNotify, "executor", einoExecBegin, einoExecAppendPartial, einoExecRegisterCancel, einoExecUnregisterCancel, einoExecFinish, agentToolTimeoutMinutes(appCfg), agentToolWaitTimeoutSeconds(appCfg), agentShellNoOutputTimeoutSeconds(appCfg), nil) if err != nil { return nil, fmt.Errorf("plan_execute filesystem 中间件: %w", err) }