diff --git a/adk/agent_tool.go b/adk/agent_tool.go index 0d2c40565..2122ea048 100644 --- a/adk/agent_tool.go +++ b/adk/agent_tool.go @@ -26,6 +26,7 @@ import ( "github.com/bytedance/sonic" "github.com/google/uuid" + "github.com/cloudwego/eino/adk/internal/agenttool" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/compose" "github.com/cloudwego/eino/schema" @@ -168,6 +169,12 @@ func (at *typedAgentTool[M]) InvokableRun(ctx context.Context, argumentsInJSON s } gen, enableStreaming := getEmitGeneratorAndEnableStreaming[M](opts) + // A background-capable invocation bounds event forwarding to the caller by a + // "backgrounded" signal and uses a non-panicking send: once the run detaches it + // outlives the parent turn, whose generator is then closed. A plain foreground + // invocation sets neither, so it forwards with a panicking Send — a send on an + // unexpectedly-closed generator is a bug there and must surface. + forwardGate := tool.GetImplSpecificOptions[agenttool.ForwardGate](nil, opts...) var ms *bridgeStore var iter *AsyncIterator[*TypedAgentEvent[M]] var err error @@ -274,7 +281,16 @@ func (at *typedAgentTool[M]) InvokableRun(ctx context.Context, argumentsInJSON s // loop can skip them. stampAgentToolSessionEvent(event, childSessionID) tmp := copyTypedAgentEvent(event) - gen.Send(event) + if forwardGate.Enabled { + // Background-capable: forward only until the run is backgrounded, + // then drop (the run outlives the parent turn). trySend, not Send: + // after detach the parent generator may be closed. + if !backgrounded(forwardGate.Until) { + gen.trySend(event) + } + } else { + gen.Send(event) + } event = tmp } } @@ -331,9 +347,9 @@ type agentToolOptions struct { enableStreaming bool } -// typedAgentToolEventOptions carries the parent runner's event generator for a -// specific message type. This keeps forwarded internal events type-compatible -// with the parent event stream. +// typedAgentToolEventOptions carries the event-forward target (the caller's event +// generator) for a specific message type. This keeps forwarded internal events +// type-compatible with the caller's event stream. type typedAgentToolEventOptions[M MessageType] struct { generator *AsyncGenerator[*TypedAgentEvent[M]] } @@ -345,16 +361,27 @@ func withAgentToolOptions(agentName string, opts []AgentRunOption) tool.Option { }) } -func withAgentToolEventGenerator(gen *AsyncGenerator[*AgentEvent]) tool.Option { - return withTypedAgentToolEventGenerator(gen) -} - -func withTypedAgentToolEventGenerator[M MessageType](gen *AsyncGenerator[*TypedAgentEvent[M]]) tool.Option { +func withTypedAgentToolEventForwardTarget[M MessageType](gen *AsyncGenerator[*TypedAgentEvent[M]]) tool.Option { return tool.WrapImplSpecificOptFn(func(o *typedAgentToolEventOptions[M]) { o.generator = gen }) } +// backgrounded reports whether the done signal has fired (the run has been moved +// to the background). A nil done never fires — the run stays foreground for its +// whole lifetime. +func backgrounded(done <-chan struct{}) bool { + if done == nil { + return false + } + select { + case <-done: + return true + default: + return false + } +} + func getOptionsByAgentName(agentName string, opts []tool.Option) []AgentRunOption { var ret []AgentRunOption for _, opt := range opts { diff --git a/adk/agent_tool_test.go b/adk/agent_tool_test.go index 16e768811..738152cfd 100644 --- a/adk/agent_tool_test.go +++ b/adk/agent_tool_test.go @@ -27,6 +27,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/cloudwego/eino/adk/internal/agenttool" "github.com/cloudwego/eino/components/model" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/compose" @@ -1277,3 +1278,97 @@ func TestCrossTypeAgentToolGracefulError(t *testing.T) { "Error should mention cross-message-type incompatibility") } } + +// --- Event-forward gate --- + +// collectForwarded drains a parent generator's iterator and returns the content of +// every non-streaming message event forwarded to it. Runs in a goroutine; the +// caller closes the generator to end the drain. +func collectForwarded(iter *AsyncIterator[*AgentEvent], out *[]string, done chan<- struct{}) { + for { + e, ok := iter.Next() + if !ok { + close(done) + return + } + if e.Output != nil && e.Output.MessageOutput != nil && !e.Output.MessageOutput.IsStreaming && + e.Output.MessageOutput.Message != nil { + *out = append(*out, e.Output.MessageOutput.Message.Content) + } + } +} + +// With the gate open (done never fires), a background-capable invocation forwards +// the inner agent's events to the parent generator — same as a plain foreground +// agent-as-tool call. +func TestAgentToolEventForwardGate_OpenGateForwards(t *testing.T) { + ctx := context.Background() + sub := &emitEventsAgent{events: []*AgentEvent{ + EventFromMessage(schema.AssistantMessage("child output", nil), nil, schema.Assistant, ""), + }} + at := NewAgentTool(ctx, sub).(tool.InvokableTool) + + iter, gen := NewAsyncIteratorPair[*AgentEvent]() + var got []string + done := make(chan struct{}) + go collectForwarded(iter, &got, done) + + // nil done => never backgrounded => gate stays open the whole run. + _, err := at.InvokableRun(ctx, `{"request":"q"}`, + withTypedAgentToolEventForwardTarget(gen), + agenttool.WithForwardGate(nil)) + require.NoError(t, err) + + gen.Close() + <-done + assert.Contains(t, got, "child output") +} + +// With the gate already closed (the run is backgrounded before it starts), a +// background-capable invocation forwards nothing to the parent generator. +func TestAgentToolEventForwardGate_ClosedGateDropsForwarding(t *testing.T) { + ctx := context.Background() + sub := &emitEventsAgent{events: []*AgentEvent{ + EventFromMessage(schema.AssistantMessage("child output", nil), nil, schema.Assistant, ""), + }} + at := NewAgentTool(ctx, sub).(tool.InvokableTool) + + iter, gen := NewAsyncIteratorPair[*AgentEvent]() + var got []string + done := make(chan struct{}) + go collectForwarded(iter, &got, done) + + backgrounded := make(chan struct{}) + close(backgrounded) // already backgrounded + + _, err := at.InvokableRun(ctx, `{"request":"q"}`, + withTypedAgentToolEventForwardTarget(gen), + agenttool.WithForwardGate(backgrounded)) + require.NoError(t, err) + + gen.Close() + <-done + assert.Empty(t, got) +} + +// On the gated (background-capable) path, forwarding to an already-closed parent +// generator must not panic — a backgrounded run outlives the turn that closes it. +func TestAgentToolEventForwardGate_ClosedParentGeneratorNoPanic(t *testing.T) { + ctx := context.Background() + sub := &emitEventsAgent{events: []*AgentEvent{ + EventFromMessage(schema.AssistantMessage("child output", nil), nil, schema.Assistant, ""), + }} + at := NewAgentTool(ctx, sub).(tool.InvokableTool) + + // A parent generator that is already closed, with the gate still open: the + // gated path uses a non-panicking send, so this must not panic. + _, gen := NewAsyncIteratorPair[*AgentEvent]() + gen.Close() + + require.NotPanics(t, func() { + _, err := at.InvokableRun(ctx, `{"request":"q"}`, + withTypedAgentToolEventForwardTarget(gen), + agenttool.WithForwardGate(nil)) + require.NoError(t, err) + }) +} diff --git a/adk/backgroundtask/manager.go b/adk/backgroundtask/manager.go index edf301ae4..7540d7880 100644 --- a/adk/backgroundtask/manager.go +++ b/adk/backgroundtask/manager.go @@ -288,6 +288,13 @@ type taskRecord struct { // state. Wait selects on it so waiting for one task neither holds m.mu nor is // woken by unrelated tasks finishing. doneCh chan struct{} + // backgrounded is closed exactly once (guarded by bgClosed) when the task moves + // to background execution: before the work starts for an explicit + // RunInBackground launch, or at the auto-background transition (detach). It stays + // open for a run that completes in the foreground. Handed to the work via + // TaskInfo.Backgrounded so foreground-only side effects can stop when it fires. + backgrounded chan struct{} + bgClosed bool } // New creates a new Manager. @@ -545,13 +552,45 @@ func (m *Manager) registerTaskLocked(input *RunInput, id string) (string, error) OutputFile: input.OutputFile, Metadata: cloneMetadata(input.Metadata), }, - doneCh: make(chan struct{}), + doneCh: make(chan struct{}), + backgrounded: make(chan struct{}), } m.sendEventLocked(m.tasks[id], TaskEventCreated) return id, nil } +// markBackgroundedLocked closes a task's backgrounded signal, idempotently. Must be +// called with m.mu held. +func (m *Manager) markBackgroundedLocked(rec *taskRecord) { + if !rec.bgClosed { + rec.bgClosed = true + close(rec.backgrounded) + } +} + +// backgroundedCh returns the task's backgrounded signal channel. Returns a nil +// channel if the task is absent (should not happen for a just-created task); a nil +// channel never fires, which is the correct "stayed foreground" reading. +func (m *Manager) backgroundedCh(id string) <-chan struct{} { + m.mu.Lock() + defer m.mu.Unlock() + if rec, ok := m.tasks[id]; ok { + return rec.backgrounded + } + return nil +} + +// markBackgrounded closes a task's backgrounded signal, idempotently. Used for an +// explicit RunInBackground launch, where the run is background from the start. +func (m *Manager) markBackgrounded(id string) { + m.mu.Lock() + defer m.mu.Unlock() + if rec, ok := m.tasks[id]; ok { + m.markBackgroundedLocked(rec) + } +} + func (m *Manager) closedError() error { return fmt.Errorf("the background task manager has shut down and is no longer accepting new tasks. " + "Do not retry this; finish using any results you already have") @@ -701,6 +740,7 @@ func (m *Manager) detach(id string) bool { return false } rec.task.RunInBackground = true + m.markBackgroundedLocked(rec) m.sendEventLocked(rec, TaskEventBackgrounded) return true } @@ -789,6 +829,14 @@ type TaskInfo struct { // closure is already built. The launcher passes it to MarkOutputFileUnreliable // to report an output-file write failure against this task. ID string + // Backgrounded is closed when this task moves to background execution: before + // the work starts for an explicit RunInBackground launch, or at the + // auto-background transition (foreground deadline reached and permitted). It + // stays open for a run that completes in the foreground. Work uses it to stop + // side effects that are only valid while the run is foreground — most notably + // forwarding events to the launching turn's live stream, which the turn closes + // once it ends. It is never nil. + Backgrounded <-chan struct{} } // WorkFunc performs a single managed execution. It is supplied by the caller @@ -852,6 +900,8 @@ func (m *Manager) Run(ctx context.Context, input *RunInput, work WorkFunc) (*Tas runCtx, cancel := context.WithCancel(detachedCtx{parent: ctx}) m.storeCancelFunc(id, cancel) + bgCh := m.backgroundedCh(id) + // run executes the work and finalizes the task. The terminal outcome lives on // the task record (set by completeTask/failTask), which Run reads back via // taskSnapshot — so run signals completion rather than returning a value. @@ -863,7 +913,7 @@ func (m *Manager) Run(ctx context.Context, input *RunInput, work WorkFunc) (*Tas m.failTask(id, safe.NewPanicErr(p, debug.Stack())) } }() - r, runErr := work(runCtx, TaskInfo{ID: id}) + r, runErr := work(runCtx, TaskInfo{ID: id, Backgrounded: bgCh}) if runErr != nil { m.failTask(id, runErr) } else { @@ -872,8 +922,10 @@ func (m *Manager) Run(ctx context.Context, input *RunInput, work WorkFunc) (*Tas } // Explicit background: run in goroutine, return immediately. createTask already - // marked the task RunInBackground. + // marked the task RunInBackground; signal backgrounded before the work starts so + // the work never treats itself as foreground. if input.RunInBackground { + m.markBackgrounded(id) task := m.taskSnapshot(id) go run() return task, nil @@ -1027,7 +1079,13 @@ func (m *Manager) forwardStream(r *streamRun) { } }() - ws, err := r.work(r.runCtx, TaskInfo{ID: r.id}) + // Explicit background: the run is background from the start, so signal it before + // the work starts (RunStream forces budgetMs=0 for this case). + if r.input.RunInBackground { + m.markBackgrounded(r.id) + } + + ws, err := r.work(r.runCtx, TaskInfo{ID: r.id, Backgrounded: m.backgroundedCh(r.id)}) if err != nil { m.failTask(r.id, err) r.sw.Send("", err) diff --git a/adk/backgroundtask/manager_test.go b/adk/backgroundtask/manager_test.go index c39f7b944..bf60a719f 100644 --- a/adk/backgroundtask/manager_test.go +++ b/adk/backgroundtask/manager_test.go @@ -789,3 +789,80 @@ func TestManager_ContextCancelStopsWork(t *testing.T) { require.True(t, ok) assert.Equal(t, StatusCanceled, task.Status) } + +// --- TaskInfo.Backgrounded signal Tests --- + +// isClosed reports whether a done channel has fired without blocking. +func isClosed(ch <-chan struct{}) bool { + select { + case <-ch: + return true + default: + return false + } +} + +// An explicit RunInBackground launch is background from the start, so the work +// must observe Backgrounded already closed on entry. +func TestManager_Backgrounded_ExplicitClosedBeforeWork(t *testing.T) { + m := New(context.Background(), &Config{}) + defer closeWithTimeout(m) + + seen := make(chan bool, 1) + release := make(chan struct{}) + result, err := m.Run(context.Background(), &RunInput{Description: "bg", RunInBackground: true}, + func(_ context.Context, task TaskInfo) (string, error) { + seen <- isClosed(task.Backgrounded) + <-release + return "ok", nil + }) + require.NoError(t, err) + assert.Equal(t, StatusRunning, result.Status) + + assert.True(t, <-seen, "explicit background work should see Backgrounded already closed") + close(release) + waitTask(t, m, result.ID) +} + +// A run that completes in the foreground is never backgrounded: its Backgrounded +// signal stays open for the whole run. +func TestManager_Backgrounded_ForegroundStaysOpen(t *testing.T) { + m := New(context.Background(), &Config{}) + defer closeWithTimeout(m) + + var duringRun bool + result, err := run(m, "fg", false, func(_ context.Context, task TaskInfo) (string, error) { + duringRun = isClosed(task.Backgrounded) + return "ok", nil + }) + require.NoError(t, err) + assert.Equal(t, StatusCompleted, result.Status) + assert.False(t, duringRun, "foreground work must not see Backgrounded closed") +} + +// A foreground run that is auto-moved to the background at its deadline must have +// its Backgrounded signal close at the transition, so still-running work learns it +// detached. +func TestManager_Backgrounded_AutoBackgroundCloses(t *testing.T) { + m := New(context.Background(), &Config{ForegroundTimeoutMs: intPtr(50), ShouldAutoBackground: allowBackground}) + defer closeWithTimeout(m) + + closedCh := make(chan struct{}) + result, err := m.Run(context.Background(), &RunInput{Description: "slow"}, + func(_ context.Context, task TaskInfo) (string, error) { + // Block until the deadline detaches the run, then confirm the signal fired. + <-task.Backgrounded + close(closedCh) + return "slow result", nil + }) + require.NoError(t, err) + assert.Equal(t, StatusRunning, result.Status) + + select { + case <-closedCh: + case <-time.After(time.Second): + t.Fatal("Backgrounded did not close at the auto-background transition") + } + waitTask(t, m, result.ID) +} + diff --git a/adk/chatmodel.go b/adk/chatmodel.go index f5309eda5..302a6f241 100644 --- a/adk/chatmodel.go +++ b/adk/chatmodel.go @@ -1461,7 +1461,7 @@ func (a *TypedChatModelAgent[M]) buildMessageReActRunFunc(_ context.Context, bc var runOpts []compose.Option runOpts = append(runOpts, mp.composeOpts...) if a.toolsConfig.EmitInternalEvents { - runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withAgentToolEventGenerator(mp.generator)))) + runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withTypedAgentToolEventForwardTarget(mp.generator)))) } if mp.input.EnableStreaming { runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withAgentToolEnableStreaming(true)))) @@ -1602,7 +1602,7 @@ func (a *TypedChatModelAgent[M]) buildAgenticReActRunFunc(_ context.Context, bc var runOpts []compose.Option runOpts = append(runOpts, ap.composeOpts...) if a.toolsConfig.EmitInternalEvents { - runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withTypedAgentToolEventGenerator[*schema.AgenticMessage](ap.generator)))) + runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withTypedAgentToolEventForwardTarget[*schema.AgenticMessage](ap.generator)))) } if ap.input.EnableStreaming { runOpts = append(runOpts, compose.WithToolsNodeOption(compose.WithToolOption(withAgentToolEnableStreaming(true)))) diff --git a/adk/internal/agenttool/options.go b/adk/internal/agenttool/options.go new file mode 100644 index 000000000..440c6c85a --- /dev/null +++ b/adk/internal/agenttool/options.go @@ -0,0 +1,40 @@ +/* + * Copyright 2026 CloudWeGo Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Package agenttool carries ADK-internal AgentTool event-forwarding options +// shared between AgentTool and middleware plumbing. +package agenttool + +import "github.com/cloudwego/eino/components/tool" + +// ForwardGate controls the internal AgentTool event-forwarding gate. +type ForwardGate struct { + // Enabled marks the AgentTool invocation as background-capable: forwarding to + // the caller is bounded by Until and uses a non-panicking send. + Enabled bool + // Until stops forwarding to the caller once closed. Nil means "never + // backgrounded". + Until <-chan struct{} +} + +// WithForwardGate marks an AgentTool invocation as background-capable and +// forwards inner-agent events to the caller until done is closed. +func WithForwardGate(done <-chan struct{}) tool.Option { + return tool.WrapImplSpecificOptFn(func(o *ForwardGate) { + o.Enabled = true + o.Until = done + }) +} diff --git a/adk/middlewares/subagent/agent_tool.go b/adk/middlewares/subagent/agent_tool.go index 6b3a392c5..ac3e2c02d 100644 --- a/adk/middlewares/subagent/agent_tool.go +++ b/adk/middlewares/subagent/agent_tool.go @@ -30,6 +30,7 @@ import ( "github.com/cloudwego/eino/adk/backgroundtask" "github.com/cloudwego/eino/adk/filesystem" "github.com/cloudwego/eino/adk/internal" + "github.com/cloudwego/eino/adk/internal/agenttool" "github.com/cloudwego/eino/components/tool" "github.com/cloudwego/eino/components/tool/utils" "github.com/cloudwego/eino/compose" @@ -118,7 +119,16 @@ func newManagedAgentTool(mgr *backgroundtask.Manager, subAgents map[string]tool. Metadata: map[string]any{MetadataKeySubagentType: in.SubagentType}, OutputFile: outputFile, }, func(workCtx context.Context, task backgroundtask.TaskInfo) (string, error) { - out, runErr := a.InvokableRun(workCtx, params, opts...) + // Bound the inner agent's forwarding to the launching turn's event + // stream by the task's backgrounded signal: a backgrounded run outlives + // the turn (which closes its event generator on turn end), so forwarding + // to it past that point is wrong and unsafe. The AgentTool stops + // forwarding when task.Backgrounded fires — for an explicit + // run_in_background it is already closed here, for an auto-backgrounded + // run it closes at the deadline. Foreground runs never fire it and + // forward for their whole lifetime. + runOpts := append(opts, agenttool.WithForwardGate(task.Backgrounded)) + out, runErr := a.InvokableRun(workCtx, params, runOpts...) if runErr != nil { return "", runErr }