fix(adk): stop forwarding backgrounded sub-agent events to the parent turn

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 {

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)
+	})
+}

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)