diff --git a/adk/agent_tool_test.go b/adk/agent_tool_test.go index 785ad995a..dfd3bfc58 100644 --- a/adk/agent_tool_test.go +++ b/adk/agent_tool_test.go @@ -763,6 +763,54 @@ func TestNestedAgentTool_RunPath(t *testing.T) { } } +func TestAgentToolInvocationIDDistinguishesRepeatedRuns(t *testing.T) { + ctx := context.Background() + agentTool := NewAgentTool(ctx, &invocationIDAgent{}).(tool.InvokableTool) + + first := invokeAgentToolAndCollectEvents(t, ctx, agentTool) + second := invokeAgentToolAndCollectEvents(t, ctx, agentTool) + + require.Len(t, first, 2) + require.Len(t, second, 2) + require.NotEmpty(t, first[0].InvocationID) + require.Equal(t, first[0].InvocationID, first[1].InvocationID) + require.Equal(t, second[0].InvocationID, second[1].InvocationID) + require.Equal(t, first[0].RunPath, second[0].RunPath) + require.NotEqual(t, first[0].InvocationID, second[0].InvocationID) +} + +type invocationIDAgent struct{} + +func (*invocationIDAgent) Name(context.Context) string { return "repeated_agent" } +func (*invocationIDAgent) Description(context.Context) string { return "emits two events" } +func (*invocationIDAgent) Run(context.Context, *AgentInput, ...AgentRunOption) *AsyncIterator[*AgentEvent] { + iter, gen := NewAsyncIteratorPair[*AgentEvent]() + go func() { + defer gen.Close() + gen.Send(EventFromMessage(schema.AssistantMessage("first", nil), nil, schema.Assistant, "")) + gen.Send(EventFromMessage(schema.AssistantMessage("second", nil), nil, schema.Assistant, "")) + }() + return iter +} + +func invokeAgentToolAndCollectEvents(t *testing.T, ctx context.Context, agentTool tool.InvokableTool) []*AgentEvent { + t.Helper() + + iter, gen := NewAsyncIteratorPair[*AgentEvent]() + _, err := agentTool.InvokableRun(ctx, `{"request":"test"}`, withAgentToolEventGenerator(gen)) + require.NoError(t, err) + gen.Close() + + var events []*AgentEvent + for { + event, ok := iter.Next() + if !ok { + return events + } + events = append(events, event) + } +} + func TestNestedAgentTool_NoInternalEventsWhenDisabled(t *testing.T) { ctx := context.Background() diff --git a/adk/deterministic_transfer.go b/adk/deterministic_transfer.go index ce5b20093..226fe3b7e 100644 --- a/adk/deterministic_transfer.go +++ b/adk/deterministic_transfer.go @@ -181,9 +181,10 @@ func runFlowAgentWithIsolatedSession(ctx context.Context, fa *flowAgent, input * } ctx = setRunCtx(ctx, &runContext{ - Session: isolatedSession, - RootInput: parentRunCtx.RootInput, - RunPath: parentRunCtx.RunPath, + Session: isolatedSession, + RootInput: parentRunCtx.RootInput, + RunPath: parentRunCtx.RunPath, + InvocationID: parentRunCtx.InvocationID, }) iter := fa.Run(ctx, input, options...) @@ -218,9 +219,10 @@ func resumeFlowAgentWithIsolatedSession(ctx context.Context, fa *flowAgent, info } ctx = setRunCtx(ctx, &runContext{ - Session: isolatedSession, - RootInput: parentRunCtx.RootInput, - RunPath: parentRunCtx.RunPath, + Session: isolatedSession, + RootInput: parentRunCtx.RootInput, + RunPath: parentRunCtx.RunPath, + InvocationID: parentRunCtx.InvocationID, }) iter := fa.Resume(ctx, info, opts...) diff --git a/adk/flow.go b/adk/flow.go index 8b06992c3..f8fcc6687 100644 --- a/adk/flow.go +++ b/adk/flow.go @@ -495,7 +495,7 @@ func (a *flowAgent) run( panicErr := recover() if panicErr != nil { e := safe.NewPanicErr(panicErr, debug.Stack()) - generator.Send(&AgentEvent{Err: e}) + generator.Send(&AgentEvent{InvocationID: runCtx.InvocationID, Err: e}) } cbGen.Close() @@ -509,6 +509,9 @@ func (a *flowAgent) run( break } + if event.InvocationID == "" { + event.InvocationID = runCtx.InvocationID + } // RunPath ownership: the eino framework sets RunPath exactly once. // If event.RunPath is already set (e.g., by agentTool), we don't modify it. // If event.RunPath is nil/empty, we set it to the current runCtx.RunPath. @@ -562,7 +565,7 @@ func (a *flowAgent) run( if agentToRun == nil { e := fmt.Errorf("transfer failed: agent '%s' not found when transferring from '%s'", destName, a.Name(ctxForSubAgents)) - generator.Send(&AgentEvent{Err: e}) + generator.Send(&AgentEvent{InvocationID: runCtx.InvocationID, Err: e}) return } @@ -741,7 +744,7 @@ func (a *typedFlowAgent[M]) run( panicErr := recover() if panicErr != nil { e := safe.NewPanicErr(panicErr, debug.Stack()) - generator.Send(&TypedAgentEvent[M]{Err: e}) + generator.Send(&TypedAgentEvent[M]{InvocationID: runCtx.InvocationID, Err: e}) } agenticCbGen.Close() @@ -754,6 +757,9 @@ func (a *typedFlowAgent[M]) run( break } + if event.InvocationID == "" { + event.InvocationID = runCtx.InvocationID + } if len(event.RunPath) == 0 { event.AgentName = a.Name(ctx) event.RunPath = runCtx.RunPath diff --git a/adk/interface.go b/adk/interface.go index 8905950d9..38e635147 100644 --- a/adk/interface.go +++ b/adk/interface.go @@ -419,6 +419,15 @@ type runStepSerialization struct { type TypedAgentEvent[M MessageType] struct { AgentName string + // InvocationID uniquely identifies one physical agent execution attempt. + // All events emitted by that invocation share the same ID. Repeated, + // parallel, and nested invocations receive different IDs even when their + // RunPath values are identical. + // + // InvocationID is framework-generated. Use SessionEvent.TurnID instead + // when events need to be grouped across resume attempts in one logical turn. + InvocationID string + // RunPath represents the execution path from root agent to the current event source. // This field is managed entirely by the framework and cannot be set by end-users. // diff --git a/adk/interrupt_test.go b/adk/interrupt_test.go index 480c0f8f7..ad09f42fd 100644 --- a/adk/interrupt_test.go +++ b/adk/interrupt_test.go @@ -586,6 +586,11 @@ func TestWorkflowInterrupt(t *testing.T) { } assert.Equal(t, 2, len(events)) + assert.NotEmpty(t, events[0].InvocationID) + assert.NotEmpty(t, events[1].InvocationID) + assert.NotEqual(t, events[0].InvocationID, events[1].InvocationID) + messageEvents[0].InvocationID = events[0].InvocationID + messageEvents[1].InvocationID = events[1].InvocationID assert.Equal(t, messageEvents, events) }) @@ -931,6 +936,11 @@ func TestWorkflowInterrupt(t *testing.T) { }, } assert.Equal(t, 2, len(events)) + assert.NotEmpty(t, events[0].InvocationID) + assert.NotEmpty(t, events[1].InvocationID) + assert.NotEqual(t, events[0].InvocationID, events[1].InvocationID) + loopFinalMessageEvents[0].InvocationID = events[0].InvocationID + loopFinalMessageEvents[1].InvocationID = events[1].InvocationID assert.Equal(t, loopFinalMessageEvents, events) }) @@ -991,8 +1001,23 @@ func TestWorkflowInterrupt(t *testing.T) { }, } - assert.Contains(t, events, parallelMessageEvents[0]) - assert.Contains(t, events, parallelMessageEvents[1]) + eventByAgentName := make(map[string]*AgentEvent, len(events)) + for _, event := range events { + assert.NotEmpty(t, event.InvocationID) + eventByAgentName[event.AgentName] = event + } + sa3Event, sa3OK := eventByAgentName["sa3"] + sa4Event, sa4OK := eventByAgentName["sa4"] + if assert.True(t, sa3OK) && assert.True(t, sa4OK) { + assert.NotEqual(t, sa3Event.InvocationID, sa4Event.InvocationID) + } + for _, expected := range parallelMessageEvents { + actual := eventByAgentName[expected.AgentName] + if assert.NotNil(t, actual) { + expected.InvocationID = actual.InvocationID + assert.Equal(t, expected, actual) + } + } assert.NotNil(t, interruptEvent) assert.Equal(t, "parallel agent", interruptEvent.AgentName) diff --git a/adk/runctx.go b/adk/runctx.go index dd42226af..6ce0decc7 100644 --- a/adk/runctx.go +++ b/adk/runctx.go @@ -27,6 +27,8 @@ import ( "sync" "time" + "github.com/google/uuid" + "github.com/cloudwego/eino/schema" ) @@ -344,8 +346,9 @@ func (rs *runSession) getValue(key string) (any, bool) { } type runContext struct { - RootInput *AgentInput - RunPath []RunStep + RootInput *AgentInput + RunPath []RunStep + InvocationID string AgenticRootInput any @@ -361,6 +364,7 @@ func (rc *runContext) deepCopy() *runContext { RootInput: rc.RootInput, AgenticRootInput: rc.AgenticRootInput, RunPath: make([]RunStep, len(rc.RunPath)), + InvocationID: rc.InvocationID, Session: rc.Session, } @@ -392,6 +396,7 @@ func initRunCtx(ctx context.Context, agentName string, input *AgentInput) (conte } runCtx.RunPath = append(runCtx.RunPath, RunStep{agentName: agentName}) + runCtx.InvocationID = uuid.NewString() if runCtx.isRoot() && input != nil { runCtx.RootInput = input } @@ -408,6 +413,7 @@ func initTypedRunCtx[M MessageType](ctx context.Context, agentName string, input } runCtx.RunPath = append(runCtx.RunPath, RunStep{agentName: agentName}) + runCtx.InvocationID = uuid.NewString() if runCtx.isRoot() && input != nil { var zero M if _, ok := any(zero).(*schema.Message); ok { @@ -498,9 +504,10 @@ func forkRunCtx(ctx context.Context) context.Context { // Create a new runContext for the child lane, pointing to the new session. childRunCtx := &runContext{ - RootInput: parentRunCtx.RootInput, - RunPath: make([]RunStep, len(parentRunCtx.RunPath)), - Session: childSession, + RootInput: parentRunCtx.RootInput, + RunPath: make([]RunStep, len(parentRunCtx.RunPath)), + InvocationID: parentRunCtx.InvocationID, + Session: childSession, } copy(childRunCtx.RunPath, parentRunCtx.RunPath) @@ -522,6 +529,9 @@ func updateRunPathOnly(ctx context.Context, agentNames ...string) context.Contex for _, agentName := range agentNames { runCtx.RunPath = append(runCtx.RunPath, RunStep{agentName: agentName}) } + if len(agentNames) > 0 { + runCtx.InvocationID = uuid.NewString() + } return setRunCtx(ctx, runCtx) } diff --git a/adk/runctx_test.go b/adk/runctx_test.go index bef1f44eb..a274a41d6 100644 --- a/adk/runctx_test.go +++ b/adk/runctx_test.go @@ -427,6 +427,23 @@ func TestForkJoinRunCtx(t *testing.T) { assert.Equal(t, []string{"A", "B", "C1", "D", "E", "F"}, getEventNames(mainRunCtx.Session.getEvents()), "After F") } +func TestRunContextInvocationID(t *testing.T) { + firstCtx, firstRunCtx := initRunCtx(context.Background(), "agent", nil) + _, secondRunCtx := initRunCtx(context.Background(), "agent", nil) + + assert.NotEmpty(t, firstRunCtx.InvocationID) + assert.NotEmpty(t, secondRunCtx.InvocationID) + assert.NotEqual(t, firstRunCtx.InvocationID, secondRunCtx.InvocationID) + + firstChild := getRunCtx(updateRunPathOnly(firstCtx, "child")) + secondChild := getRunCtx(updateRunPathOnly(firstCtx, "child")) + + assert.Equal(t, firstChild.RunPath, secondChild.RunPath) + assert.NotEmpty(t, firstChild.InvocationID) + assert.NotEmpty(t, secondChild.InvocationID) + assert.NotEqual(t, firstChild.InvocationID, secondChild.InvocationID) +} + // makeStreamingEventWrapper creates an agentEventWrapper with a streaming MessageOutput // whose stream yields the given message then terminates with streamErr (or io.EOF if nil). func makeStreamingEventWrapper(msg Message, streamErr error) *agentEventWrapper { diff --git a/adk/runner.go b/adk/runner.go index a7d722e6f..28824b451 100644 --- a/adk/runner.go +++ b/adk/runner.go @@ -314,9 +314,10 @@ func typedRunnerHandleIterImpl[M MessageType](enableStreaming bool, store CheckP interruptSignal = event.Action.internalInterrupted interruptContexts := core.ToInterruptContexts(interruptSignal, allowedAddressSegmentTypes) event = &TypedAgentEvent[M]{ - AgentName: event.AgentName, - RunPath: event.RunPath, - Output: event.Output, + AgentName: event.AgentName, + InvocationID: event.InvocationID, + RunPath: event.RunPath, + Output: event.Output, Action: &AgentAction{ Interrupted: &InterruptInfo{ Data: event.Action.Interrupted.Data, diff --git a/adk/utils.go b/adk/utils.go index 991abd083..f62f72017 100644 --- a/adk/utils.go +++ b/adk/utils.go @@ -239,10 +239,11 @@ func copyTypedAgentEvent[M MessageType](ae *TypedAgentEvent[M]) *TypedAgentEvent copy(rp, ae.RunPath) copied := &TypedAgentEvent[M]{ - AgentName: ae.AgentName, - RunPath: rp, - Action: ae.Action, - Err: ae.Err, + AgentName: ae.AgentName, + InvocationID: ae.InvocationID, + RunPath: rp, + Action: ae.Action, + Err: ae.Err, } if ae.Output == nil {