feat: provide access to session events from executor callbacks (#521)

* provide access to session events from executor callbacks

* gemini nitpicks

* test cleanup

Author: yarolegovich

server/adka2a/executor.go

@@ -110,7 +110,13 @@ func (e *Executor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q
 	if err != nil {
 		return fmt.Errorf("a2a message conversion failed: %w", err)
 	}
-	r, err := runner.New(e.config.RunnerConfig)
+
+	runnerCfg, executorPlugin, err := withExecutorPlugin(e.config.RunnerConfig)
+	if err != nil {
+		return fmt.Errorf("failed to install a2a-executor plugin: %w", err)
+	}
+
+	r, err := runner.New(runnerCfg)
 	if err != nil {
 		return fmt.Errorf("failed to create a runner: %w", err)
 	}
@@ -137,10 +143,10 @@ func (e *Executor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q
 
 	invocationMeta := toInvocationMeta(ctx, e.config, reqCtx)
 
-	session, err := e.prepareSession(ctx, invocationMeta)
+	err = e.prepareSession(ctx, invocationMeta)
 	if err != nil {
 		event := toTaskFailedUpdateEvent(reqCtx, err, invocationMeta.eventMeta)
-		execCtx := newExecutorContext(ctx, invocationMeta, emptySessionState{}, content)
+		execCtx := newExecutorContext(ctx, invocationMeta, executorPlugin, content)
 		return e.writeFinalTaskStatus(execCtx, queue, event, err)
 	}
 
@@ -151,7 +157,7 @@ func (e *Executor) Execute(ctx context.Context, reqCtx *a2asrv.RequestContext, q
 	}
 
 	processor := newEventProcessor(reqCtx, invocationMeta, e.config.GenAIPartConverter)
-	executorContext := newExecutorContext(ctx, invocationMeta, session.State(), content)
+	executorContext := newExecutorContext(ctx, invocationMeta, executorPlugin, content)
 	return e.process(executorContext, r, processor, queue)
 }
 
@@ -209,26 +215,26 @@ func (e *Executor) writeFinalTaskStatus(ctx ExecutorContext, queue eventqueue.Qu
 	return nil
 }
 
-func (e *Executor) prepareSession(ctx context.Context, meta invocationMeta) (session.Session, error) {
+func (e *Executor) prepareSession(ctx context.Context, meta invocationMeta) error {
 	service := e.config.RunnerConfig.SessionService
 
-	getResp, err := service.Get(ctx, &session.GetRequest{
+	_, err := service.Get(ctx, &session.GetRequest{
 		AppName:   e.config.RunnerConfig.AppName,
 		UserID:    meta.userID,
 		SessionID: meta.sessionID,
 	})
 	if err == nil {
-		return getResp.Session, nil
+		return nil
 	}
 
-	createResp, err := service.Create(ctx, &session.CreateRequest{
+	_, err = service.Create(ctx, &session.CreateRequest{
 		AppName:   e.config.RunnerConfig.AppName,
 		UserID:    meta.userID,
 		SessionID: meta.sessionID,
 		State:     make(map[string]any),
 	})
 	if err != nil {
-		return nil, fmt.Errorf("failed to create a session: %w", err)
+		return fmt.Errorf("failed to create a session: %w", err)
 	}
-	return createResp.Session, nil
+	return nil
 }

server/adka2a/executor_context.go

@@ -38,6 +38,8 @@ type ExecutorContext interface {
 	AgentName() string
 	// ReadonlyState provides a view of the current session state.
 	ReadonlyState() session.ReadonlyState
+	// Events provides a readonly view of the current session events.
+	Events() session.Events
 	// UserContent is a converted A2A message which is passed to runner.Run.
 	UserContent() *genai.Content
 	// RequestContext containts information about the original A2A Request, the current task and related tasks.
@@ -47,15 +49,15 @@ type ExecutorContext interface {
 type executorContext struct {
 	context.Context
 	meta        invocationMeta
-	session     session.ReadonlyState
+	plugin      *executorPlugin
 	userContent *genai.Content
 }
 
-func newExecutorContext(ctx context.Context, meta invocationMeta, session session.ReadonlyState, userContent *genai.Content) ExecutorContext {
+func newExecutorContext(ctx context.Context, meta invocationMeta, plugin *executorPlugin, userContent *genai.Content) ExecutorContext {
 	return &executorContext{
 		Context:     ctx,
 		meta:        meta,
-		session:     session,
+		plugin:      plugin,
 		userContent: userContent,
 	}
 }
@@ -73,7 +75,19 @@ func (ec *executorContext) AgentName() string {
 }
 
 func (ec *executorContext) ReadonlyState() session.ReadonlyState {
-	return ec.session
+	session := ec.plugin.invocationSession
+	if session == nil {
+		return emptySessionState{}
+	}
+	return session.State()
+}
+
+func (ec *executorContext) Events() session.Events {
+	session := ec.plugin.invocationSession
+	if session == nil {
+		return emptySessionEvents{}
+	}
+	return session.Events()
 }
 
 func (ec *executorContext) RequestContext() *a2asrv.RequestContext {
@@ -93,3 +107,17 @@ func (emptySessionState) Get(string) (any, error) {
 func (emptySessionState) All() iter.Seq2[string, any] {
 	return func(yield func(string, any) bool) {}
 }
+
+type emptySessionEvents struct{}
+
+func (emptySessionEvents) At(i int) *session.Event {
+	return nil
+}
+
+func (emptySessionEvents) All() iter.Seq[*session.Event] {
+	return func(yield func(*session.Event) bool) {}
+}
+
+func (emptySessionEvents) Len() int {
+	return 0
+}

server/adka2a/executor_plugin.go

@@ -0,0 +1,57 @@
+// Copyright 2026 Google LLC
+//
+// 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 adka2a
+
+import (
+	"slices"
+
+	"google.golang.org/genai"
+
+	"google.golang.org/adk/agent"
+	"google.golang.org/adk/plugin"
+	"google.golang.org/adk/runner"
+	"google.golang.org/adk/session"
+)
+
+type executorPlugin struct {
+	plugin *plugin.Plugin
+
+	invocationSession session.Session
+}
+
+func withExecutorPlugin(cfg runner.Config) (runner.Config, *executorPlugin, error) {
+	executorPlugin, err := newExecutorPlugin()
+	if err != nil {
+		return cfg, nil, err
+	}
+	cfg.PluginConfig.Plugins = append(slices.Clone(cfg.PluginConfig.Plugins), executorPlugin.plugin)
+	return cfg, executorPlugin, nil
+}
+
+func newExecutorPlugin() (*executorPlugin, error) {
+	execPlugin := &executorPlugin{}
+	plugin, err := plugin.New(plugin.Config{
+		Name: "a2a-executor",
+		BeforeRunCallback: func(ic agent.InvocationContext) (*genai.Content, error) {
+			execPlugin.invocationSession = ic.Session()
+			return nil, nil
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	execPlugin.plugin = plugin
+	return execPlugin, nil
+}

server/adka2a/executor_test.go

@@ -404,6 +404,27 @@ func TestExecutor_Callbacks(t *testing.T) {
 				newFinalStatusUpdate(task, a2a.TaskStateCompleted, nil),
 			},
 		},
+		{
+			name: "can access session events",
+			events: []*session.Event{
+				{LLMResponse: modelResponseFromParts(genai.NewPartFromText("Hello"))},
+				{LLMResponse: modelResponseFromParts(genai.NewPartFromText(", world!"))},
+			},
+			afterExecution: func(ctx ExecutorContext, finalUpdate *a2a.TaskStatusUpdateEvent, err error) error {
+				eventCount := ctx.Events().Len()
+				finalUpdate.Status.Message = a2a.NewMessage(a2a.MessageRoleAgent, a2a.TextPart{Text: fmt.Sprintf("event count = %d", eventCount)})
+				return nil
+			},
+			wantEvents: []a2a.Event{
+				a2a.NewStatusUpdateEvent(task, a2a.TaskStateWorking, nil),
+				a2a.NewArtifactEvent(task, a2a.TextPart{Text: "Hello"}),
+				a2a.NewArtifactUpdateEvent(task, a2a.NewArtifactID(), a2a.TextPart{Text: ", world!"}),
+				newArtifactLastChunkEvent(task),
+				newFinalStatusUpdate(task, a2a.TaskStateCompleted,
+					a2a.NewMessage(a2a.MessageRoleAgent, a2a.TextPart{Text: "event count = 3"}),
+				),
+			},
+		},
 		{
 			name: "abort execution",
 			events: []*session.Event{