No first-class way to serve the ADK data plane (`streamQuery`/sessions) **and** A2A 1.0 on a single Agent Engine

## 🔴 Required Information

**Describe the Bug:**

`AdkApp` and `A2aAgent` (the `vertexai.agent_engines` / `google-cloud-aiplatform`
templates) are **mutually exclusive** on a single Agent Engine, so there is no
supported way to deploy one reasoning engine that serves both the ADK data plane
and A2A:

- `AdkApp.register_operations()` returns only the ADK data-plane modes
(`""`, `async`, `stream`, `async_stream` → `stream_query`, sessions, artifacts, …).
- `A2aAgent.register_operations()` returns only `a2a_extension`
(`on_message_send`, `on_get_task`, …).

An agent that needs its ADK data plane (e.g. it relies on `:streamQuery`, or on
the durable self-`:asyncQuery` query-job pattern) **and** wants to be reachable
by external A2A clients must hand-build a custom hybrid template. We built and
verified one, but this should be first-class.

This is **distinct from** two existing issues: #6274 (upgrade `a2a-sdk` to 1.x)
fixes the version conflict but does not provide a hybrid deployment; #6220
(`adk deploy` doesn't provision `:asyncQuery`) is orthogonal. Neither removes the
need for the workaround here. (Treating this as a bug rather than a feature: the
platform already accepts a hybrid spec — see Observed Behavior — there is just no
template/option that produces one.)

**Steps to Reproduce:**

Mutual exclusivity is visible with no deploy and no credentials — just inspect
what each template registers:

1. Install `google-adk` and `google-cloud-aiplatform` (+ `a2a-sdk` 1.x).
2. Run the snippet under "Minimal Reproduction Code" below.
3. Observe that `AdkApp` registers only the ADK modes and `A2aAgent` registers
only `a2a_extension` — with no option/flag/helper to get both on one engine.

**Expected Behavior:**

A first-class way to expose A2A **alongside** the ADK data plane on one Agent
Engine — e.g. an `AdkApp(..., a2a=True, agent_card=...)` option, or a supported
composition helper / template — so one engine exposes both `:streamQuery`
(+ sessions, + the durable self-`:asyncQuery`) and `/a2a`.

**Observed Behavior:**

Each template exposes only its own half:

```
AdkApp.register_operations() -> ['', 'async', 'stream', 'async_stream'] # no A2A
A2aAgent.register_operations()-> ['a2a_extension'] # no data plane
```

The platform itself *does* accept a hybrid: our hand-built subclass (an `AdkApp`
that composes an `A2aAgent` around the same ADK `Runner` and merges
`register_operations()`), deployed via the GenAI client, produced a single engine
whose spec has both surfaces, and both `:streamQuery` and `/a2a` (`message:send`)
respond on it:

```
class methods: {'': 4, 'async': 13, 'stream': 1, 'async_stream': 2, 'a2a_extension': 9}
top-level spec.agent_card: present
```

So the gap is purely the missing template/option, not a backend limitation.

**Environment Details:**

- ADK Library Version (pip show google-adk): 2.2.0 (2.3.0 still pins `a2a-sdk>=0.3.4,<0.4`)
- Desktop OS: Linux
- Python Version (python -V): 3.13

**Model Information:**

- Are you using LiteLLM: No
- Which model is being used: gemini-2.5-flash (irrelevant — this is a deploy/templating issue)

---

## 🟡 Optional Information

**Additional Context:**

*Use case.* An ADK orchestrator on Agent Engine that (1) runs long-running work
as durable query jobs it submits against itself (`run_query_job` / `:asyncQuery`,
which needs the ADK `stream_query` op registered on the engine) and (2) must be
reachable from a UI over the A2A protocol. A pure `A2aAgent` satisfies (2) but
drops the ADK ops, breaking (1); a pure `AdkApp` satisfies (1) but has no A2A.

*What we had to build.* Subclass `AdkApp`, compose an `A2aAgent` around the same
`Runner`, and merge the op maps:

```python
class A2aAdkApp(AdkApp):
def register_operations(self):
ops = dict(super().register_operations()) # ADK data plane
ops["a2a_extension"] = [ # + A2A surface
"on_message_send", "on_get_task", "on_list_tasks", "on_cancel_task",
"on_create_task_push_notification_config",
"on_get_task_push_notification_config",
"on_list_task_push_notification_configs",
"on_delete_task_push_notification_config",
"on_get_extended_agent_card",
]
return ops

def set_up(self):
super().set_up() # builds the ADK Runner
a2a = A2aAgent(agent_card=self.agent_card,
agent_executor_builder=lambda: )
a2a.set_up()
# delegate the on_* methods + rest_routes + agent_card to `a2a`
```

Deployed via the **GenAI client** (`client.agent_engines.update/create`), because
only that path writes the top-level `spec.agent_card`; the
`vertexai.agent_engines.create/update` proto (`ReasoningEngineSpec`) in this SDK
version has no `agent_card` field.

*Related version conflict (mostly covered by #6274).* We could not use
`google.adk.a2a` (`A2aAgentExecutor`): `google-adk` pins `a2a-sdk>=0.3.4,<0.4`,
but the Agent Engine A2A runtime and `A2aAgent` template are a2a-sdk 1.x /
protocol 1.0 (`create_agent_card` emits `supported_interfaces` at `HTTP+JSON`/`1.0`;
the handler is `validate_version(PROTOCOL_VERSION_1_0)`). So we installed
`a2a-sdk[http-server]>=1.0,<2`, avoided `google.adk.a2a`, and wrapped an ADK
**core** `Runner` in a hand-written a2a-sdk 1.x `AgentExecutor`. If #6274 lands,
that executor could be ADK's own — but the hybrid registration/deploy is still
needed.

*Papercuts encountered (may be separate/docs):*

1. **`A2A-Version: 1.0` header is mandatory.** Missing → read as `0.3` and
rejected (`400 VERSION_NOT_SUPPORTED`, "A2A version '0.3' is not supported by
this handler. Expected version '1.0'."). The a2a-sdk `ClientFactory` sets it
via `setdefault`, so SDK clients are fine; hand-rolled HTTP must add it.
2. **Executor event ordering.** Emitting a `TaskStatusUpdateEvent`
(`TaskUpdater.start_work()`) before enqueuing the initial `Task` →
`500 INVALID_AGENT_RESPONSE` ("Agent should enqueue Task before
TaskStatusUpdateEvent event"). A clearer error / guardrail would help.
3. **Default in-memory task store.** `A2aAgent` defaults to `InMemoryTaskStore`,
so a `tasks/{id}` GET after the request returns `404 TASK_NOT_FOUND`
(surprising on a multi-instance serverless runtime; `message:send` returns the
completed task synchronously, so it's survivable).
4. **`adk api_server --a2a` silently mounts no A2A routes** (`UnboundLocalError`
on `json` in `get_fast_api_app`). Filed separately — see #6332.

*Request.* Provide one of: an `AdkApp(..., a2a=True, agent_card=...)` option that
merges `a2a_extension` into `register_operations()` and wires an
`A2aAgentExecutor` around the app's `Runner`; or a supported composition helper
(e.g. `to_a2a(app, keep_adk_ops=True)`) / Agent-Engine template yielding a
deployable object with both surfaces; plus clearer runtime errors for papercuts
(1)–(2). The templates live in `google-cloud-aiplatform` while `to_a2a` /
`A2aAgentExecutor` live in `adk-python`, so the fix may span both repos; filing
here to start the conversation.

**Minimal Reproduction Code:**

```python
from google.adk.agents import LlmAgent
from vertexai.agent_engines import AdkApp
from vertexai.agent_engines.templates.a2a import A2aAgent, create_agent_card
from a2a.types import AgentSkill

agent = LlmAgent(name="hello_agent", model="gemini-2.0-flash", instruction="hi")

adk = AdkApp(agent=agent)
print("AdkApp :", list(adk.register_operations().keys()))
# -> ['', 'async', 'stream', 'async_stream']

card = create_agent_card(
agent_name="hello_agent", description="hi",
skills=[AgentSkill(id="s", name="s", description="s", tags=[])],
)
a2a = A2aAgent(agent_card=card, agent_executor_builder=lambda: None)
print("A2aAgent:", list(a2a.register_operations().keys()))
# -> ['a2a_extension']
```

Also relevant: `google-cloud-aiplatform` 1.159.0, `a2a-sdk` 1.1.0 (protocol 1.0),
target Vertex AI Agent Engine (`us-central1`).

**How often has this issue occurred?:**

- Always (100%) — the templates always register only their own surface.