Repository: google/adk-go
File: agent/remoteagent/a2a_agent.go (Go)

Code before:
			if cbResp, cbErr := processor.runAfterA2ARequestCallbacks(ctx, event, err); cbResp != nil || cbErr != nil {
				if cbErr != nil {
					yield(nil, cbErr)
					return
				}
				event = cbResp
				err = nil
			}

			if err != nil {
				yield(nil, err)
				return
			}

			if event != nil { // an event might be skipped
				if intermediate := processor.aggregatePartial(ctx, event); intermediate != nil {
					if !yield(intermediate, nil) {
						return
					}
				}
				if !yield(event, nil) {
					return
				}
			}
		}

		if ctx.RunConfig().StreamingMode == agent.StreamingModeNone {
			a2aEvent, a2aErr := client.SendMessage(ctx, req)
			processEvent(a2aEvent, a2aErr)
			return
		}

		for a2aEvent, a2aErr := range client.SendStreamingMessage(ctx, req) {
			processEvent(a2aEvent, a2aErr)
		}
	}
}

func newMessage(ctx agent.InvocationContext) (*a2a.Message, error) {
	events := ctx.Session().Events()
	if userFnCall := getUserFunctionCallAt(events, events.Len()-1); userFnCall != nil {
		event := userFnCall.event
		parts, err := adka2a.ToA2AParts(event.Content.Parts, event.LongRunningToolIDs)
		if err != nil {
			return nil, fmt.Errorf("event part conversion failed: %w", err)
		}
		msg := a2a.NewMessage(a2a.MessageRoleUser, parts...)
		msg.TaskID = userFnCall.taskID
		msg.ContextID = userFnCall.contextID
		return msg, nil
	}

Review comment (suggestion):
the fix looks good, thanks!

can you please extend test case struct to have `noStreaming: bool` in either TestRemoteAgent_ADK2ADK or TestRemoteAgent_ADK2A2A (a2a_agent_test.go).

then copy one of the existing testcases and modify wantResponses to see that it only emits a single event which is an aggregation of all the events emitted on remote if invoked in non-streaming mode.

Code after:
			if cbResp, cbErr := processor.runAfterA2ARequestCallbacks(ctx, event, err); cbResp != nil || cbErr != nil {
				if cbErr != nil {
					return yield(nil, cbErr)
				}
				event = cbResp
				err = nil
			}

			if err != nil {
				return yield(nil, err)
			}

			if event != nil { // an event might be skipped
				if intermediate := processor.aggregatePartial(ctx, event); intermediate != nil {
					if !yield(intermediate, nil) {
						return false
					}
				}
				if !yield(event, nil) {
					return false
				}
			}
			return true
		}

		if ctx.RunConfig().StreamingMode == agent.StreamingModeNone {
			a2aEvent, a2aErr := client.SendMessage(ctx, req)
			processEvent(a2aEvent, a2aErr)
			return
		}

		for a2aEvent, a2aErr := range client.SendStreamingMessage(ctx, req) {
			if !processEvent(a2aEvent, a2aErr) {
				return
			}
		}
	}
}

func newMessage(ctx agent.InvocationContext) (*a2a.Message, error) {
	events := ctx.Session().Events()
	if userFnCall := getUserFunctionCallAt(events, events.Len()-1); userFnCall != nil {
		event := userFnCall.event
		parts, err := adka2a.ToA2AParts(event.Content.Parts, event.LongRunningToolIDs)
		if err != nil {
			return nil, fmt.Errorf("event part conversion failed: %w", err)
		}
		msg := a2a.NewMessage(a2a.MessageRoleUser, parts...)
		msg.TaskID = userFnCall.taskID
		msg.ContextID = userFnCall.contextID
		return msg, nil