feat(adk): identify individual agent invocations

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()
 

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...)

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

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.
 	//

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)

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

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 {

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,

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 {