• Closes: Gemini Enterprise integration fails with "unrecognized class method: streaming_agent_run_with_events" #770
• Related: Agent Engine support #749

Problem:

Gemini Enterprise / AgentSpace invokes ADK agents on Agent Engine using the streaming_agent_run_with_events method. ADK Go did not fully match the request and response contract used by the Python ADK Agent Engine implementation.

Observed issues:

• Gemini Enterprise sends the actual request payload inside input.request_json.
• The embedded session_id is a Gemini Enterprise / Discovery Engine session resource, not a backend ADK session ID.
• Passing that external session resource directly to VertexAISessionService fails because user-provided session IDs are not supported.
• After backend session creation was handled, Gemini Enterprise could invoke the agent, but responses were not rendered correctly because the stream response shape differed from Python ADK.
• When response rendering began working, Gemini Enterprise displayed duplicate answer text because both partial and final SSE events were emitted.

Solution:

This PR adds Gemini Enterprise / AgentSpace compatibility for streaming_agent_run_with_events in ADK Go:

• Registers streaming_agent_run_with_events as an Agent Engine stream handler.

• Decodes the wrapped input.request_json payload.

• Treats the first Gemini Enterprise session resource as a bootstrap identifier.

• Checks whether the incoming session_id is already a backend ADK session ID.

• Creates a generated backend ADK session when the incoming session_id is not already a backend ADK session.

• Returns the generated backend ADK session ID in the response envelope so Gemini Enterprise can use it on later turns.

• Reuses the returned backend ADK session ID on subsequent Gemini Enterprise requests.

• Emits the Python ADK-style response envelope for streaming_agent_run_with_events:

  {
    "events": [],
    "artifacts": [],
    "session_id": "..."
  }

• Uses non-SSE run mode for streaming_agent_run_with_events to avoid Gemini Enterprise rendering both partial and final text as duplicate answer content.

• Leaves existing async_stream_query behavior unchanged.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

Passed locally:

go test ./server/agentengine/...
go test ./session/vertexai

Coverage added for:

• decoding input.request_json
• creating a backend ADK session from the initial Gemini Enterprise session resource
• reusing a returned backend ADK session ID
• AgentSpace metadata for streaming_agent_run_with_events
• Python-style response envelope for AgentSpace streams
• ensuring streaming_agent_run_with_events uses non-SSE mode to avoid duplicate Gemini Enterprise response text
• preserving VertexAISessionService behavior that rejects caller-provided session IDs on create

Manual End-to-End (E2E) Tests:

Manually tested with a deployed ADK Go agent invoked from Gemini Enterprise / AgentSpace.

Verified:

• Gemini Enterprise calls streaming_agent_run_with_events.
• The request arrives with input.request_json.
• ADK creates a backend ADK session ID from the first Gemini Enterprise session resource.
• Gemini Enterprise can continue the conversation using the returned backend ADK session ID.
• The backend ADK session ID is returned in the AgentSpace response envelope as session_id.
• The ADK UI shows the expected session, user message, and agent response.
• Gemini Enterprise renders the agent response after switching to the Python-style stream envelope.
• Duplicate response text is resolved by using non-SSE mode for streaming_agent_run_with_events.

Checklist

• I have read the CONTRIBUTING.md document.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules.

Additional context

This behavior was compared against the Vertex AI Python SDK AdkApp.streaming_agent_run_with_events implementation. The Go implementation now follows the same high-level contract for Gemini Enterprise / AgentSpace invocation: wrapped request input, generated backend ADK session IDs returned in the response envelope, backend session IDs reused on later turns, and response events wrapped in an AgentSpace-compatible envelope.