diff --git a/docs/guides/README.md b/docs/guides/README.md
index e12fe22f070..ccefce218c0 100644
--- a/docs/guides/README.md
+++ b/docs/guides/README.md
@@ -6,6 +6,7 @@ This directory contains specific developer guides for the ADK Python implementat
 
 ### Agents
 * [BaseAgent](agents/base_agent/index.md) - The foundational base class for custom agents, container orchestrators, and lifecycle callbacks.
+* [CallerPrincipal](agents/caller_principal/index.md) - The caller identity a serving layer establishes for an invocation, and how tool confirmation uses it.
 * [Context](agents/context/index.md) - The runtime interface for state, artifacts, memory, credentials, and dynamic execution.
 * [Creating Agents with Configurations](agents/config/index.md) - Building and wiring multi-agent graphs from external YAML configuration files.
 * [LlmAgent](agents/llm_agent/index.md) - The primary conversational reasoning agent orchestrating models, tools, and workflows.
diff --git a/docs/guides/agents/caller_principal/index.md b/docs/guides/agents/caller_principal/index.md
new file mode 100644
index 00000000000..eb815f65063
--- /dev/null
+++ b/docs/guides/agents/caller_principal/index.md
@@ -0,0 +1,188 @@
+# CallerPrincipal
+
+`CallerPrincipal` records whether a serving layer authenticated the caller of
+an invocation, and as whom. The human-in-the-loop tool confirmation flow reads
+it to decide whether an approval that arrived over the wire is one the agent
+may act on.
+
+## Introduction
+
+A tool that requires confirmation pauses the agent until a user approves the
+call. When the agent is served remotely, the approval arrives as a message on
+the same channel as everything else the remote caller sends, so without a
+principal the framework cannot tell an operator's approval from a remote peer
+approving the dangerous tool call it just caused. `CallerPrincipal` carries the
+one fact that decides this: whether the layer that received the request
+verified who sent it.
+
+The principal lives on `InvocationContext.caller_principal`. `Runner.run_async`
+accepts it as a keyword argument, and the A2A executor sets it for every
+request it serves, so an application that serves agents over A2A gets it
+without writing any code. An application with its own serving layer sets it
+from its own authenticator.
+
+The design keeps the principal separate from the transport on purpose. A
+transport is a proxy for identity, and a proxy for identity fails in both
+directions: it refuses authenticated callers because of how they connected,
+and it admits anyone who reaches a path the proxy does not cover. Asking the
+authentication question directly avoids both failures.
+
+## Get started
+
+This example serves an agent behind an application's own request handler. The
+handler runs its authenticator first, then tells the runner what it found. A
+confirmation-gated tool is included so that the principal has something to
+decide.
+
+```python
+from google.adk.agents import CallerPrincipal
+from google.adk.agents import LlmAgent
+from google.adk.runners import InMemoryRunner
+from google.adk.tools.bash_tool import ExecuteBashTool
+
+agent = LlmAgent(name="ops_agent", tools=[ExecuteBashTool()])
+runner = InMemoryRunner(agent=agent, app_name="ops")
+
+
+async def handle_request(request, new_message):
+  # Your authenticator runs before this point. Report what it established;
+  # do not infer anything from the message the caller sent.
+  user = request.authenticated_user
+  principal = CallerPrincipal(
+      authenticated=user is not None,
+      user_name=user.name if user else None,
+      source="gateway",
+  )
+  async for event in runner.run_async(
+      user_id=request.user_id,
+      session_id=request.session_id,
+      new_message=new_message,
+      caller_principal=principal,
+  ):
+    yield event
+```
+
+An in-process caller, such as a test or a command-line tool that calls
+`runner.run_async` directly, does not pass a principal. Leaving it unset says
+that no remote trust boundary was crossed, which is true for that caller.
+
+## How it works
+
+`InvocationContext.caller_principal` has three meaningful states, and the tool
+confirmation flow treats each differently.
+
+| State | Meaning | Tool confirmation |
+| :--- | :--- | :--- |
+| `None` | The invocation started in process. Nothing had to vouch for the caller, because the caller is the operator. | Honored. |
+| `authenticated=False` | A serving layer handled the request and could not say who sent it. | Warned about by default; refused when strict mode is on. |
+| `authenticated=True` | The serving layer verified the caller. | Honored. |
+
+A refusal is explicit rather than silent. The confirmation is delivered to the
+tool with `confirmed=False`, which is the same state a user's decline produces,
+so a tool that follows the confirmation contract returns its rejection response
+and the turn ends with a reason the caller can see. The pending call never
+hangs waiting for an approval that will not come.
+
+The principal is derived only from what the serving layer established. It is
+never read from message content, from event authorship, or from transport
+metadata, because a remote caller controls all three and could set any of them
+to whatever the framework wanted to see.
+
+### A2A
+
+`A2aAgentExecutor` builds the principal from the A2A server's call context on
+every request. The caller is authenticated exactly when the A2A server
+authenticated it, which is also the case in which the invocation's user id is
+the authenticated user name rather than a generated `A2A_USER_` value. The two
+cannot disagree about the same request.
+
+An A2A server that runs without an authenticator produces an unauthenticated
+principal for every request. With the default settings that yields a warning
+per approval and the approval is honored, which is the behavior those
+deployments had before the principal existed.
+
+### Strict mode
+
+Strict mode refuses confirmations from unauthenticated callers instead of
+warning about them. It is the `STRICT_CALLER_PRINCIPAL` feature and is off by
+default, because turning it on changes what an existing deployment does with an
+approval, and an upgrade should not do that on its own. The intent is to make
+strict mode the default at the next major version.
+
+Turn it on with the environment variable ADK uses for every feature:
+
+```bash
+export ADK_ENABLE_STRICT_CALLER_PRINCIPAL=1
+```
+
+Or programmatically, when environment variables are not practical in your
+deployment:
+
+```python
+from google.adk.features import FeatureName
+from google.adk.features import override_feature_enabled
+
+override_feature_enabled(FeatureName.STRICT_CALLER_PRINCIPAL, True)
+```
+
+Strict mode never affects an invocation with no principal. An in-process
+caller crossed no boundary, so there is nothing to refuse.
+
+## Configuration options
+
+`CallerPrincipal` is a frozen model with three fields. It rejects unknown
+fields, so a typo cannot silently produce a principal that says nothing.
+
+| Option | Type | Default | Description |
+| :--- | :--- | :--- | :--- |
+| `authenticated` | `bool` | required | Whether a serving-layer authenticator verified the caller. |
+| `user_name` | `str \| None` | `None` | The verified identity when `authenticated` is `True`. |
+| `source` | `str \| None` | `None` | A short label for the serving edge that set the principal. |
+
+`authenticated` is the only field a trust decision reads. Set it to `True` only
+when your authenticator actually verified the caller. A request that carries a
+name it asserted about itself, without verification, is not authenticated, and
+saying otherwise reopens the problem the principal exists to close.
+
+`user_name` is informational. It is the identity the authenticator established,
+kept alongside the decision so that a log line or an audit record can say who
+approved a tool call. Leave it `None` when `authenticated` is `False`; a name
+without verification behind it is a claim, not an identity.
+
+`source` is a label such as `"a2a"` or `"gateway"` that names the edge that
+set the principal. It appears in the confirmation flow's log messages so that
+an operator reading a warning can tell which entry point produced it. Nothing
+makes a trust decision on it, so a transport cannot become a proxy for identity
+through this field either.
+
+## Advanced applications
+
+### Custom A2A request converters
+
+`A2aAgentExecutorConfig.request_converter` lets an application replace the
+function that turns an A2A request into a run request. The executor sets the
+principal from the server's call context after the converter returns,
+regardless of what the converter produced. A converter can therefore neither
+drop the principal nor assert one the server never established, which keeps
+the trust decision structural rather than dependent on every converter
+remembering to make it.
+
+### Serving layers with more than one entry point
+
+An application that exposes an agent through several edges can give each a
+distinct `source` and let them share the same rule for `authenticated`. The
+confirmation flow's log messages then identify the edge, and the trust
+decision stays uniform across all of them.
+
+## Limitations
+
+- The principal describes the caller of the invocation as a whole. ADK does
+  not verify the identity behind each individual event within a session.
+- Only the tool confirmation flow reads the principal today. Other consumers
+  can read `InvocationContext.caller_principal`, but the framework does not
+  yet gate anything else on it.
+- The synchronous `Runner.run` wrapper and live sessions do not accept a
+  principal. Both behave as in-process callers.
+- Setting a principal does not authenticate anything by itself. The ADK API
+  server's `/run` endpoint still requires its own authentication; the principal
+  records the outcome of authentication, it does not perform it.
diff --git a/src/google/adk/a2a/converters/request_converter.py b/src/google/adk/a2a/converters/request_converter.py
index 363b8f10a62..87318f0c846 100644
--- a/src/google/adk/a2a/converters/request_converter.py
+++ b/src/google/adk/a2a/converters/request_converter.py
@@ -23,6 +23,7 @@
 from pydantic import BaseModel
 
 from .. import _compat
+from ...agents._caller_principal import CallerPrincipal
 from ...agents.run_config import RunConfig
 from ..experimental import a2a_experimental
 from .part_converter import A2APartToGenAIPartConverter
@@ -41,6 +42,7 @@ class AgentRunRequest(BaseModel):
   new_message: Optional[genai_types.Content] = None
   state_delta: Optional[dict[str, Any]] = None
   run_config: Optional[RunConfig] = None
+  caller_principal: Optional[CallerPrincipal] = None
 
 
 A2ARequestToAgentRunRequestConverter = Callable[
@@ -77,6 +79,37 @@ def _get_user_id(request: RequestContext) -> str:
   return f'A2A_USER_{request.context_id}'
 
 
+def build_caller_principal(request: RequestContext) -> CallerPrincipal:
+  """Records what the A2A serving layer established about the caller.
+
+  ``_get_user_id`` above already reads ``call_context.user``, but it collapses
+  the result to a bare string and drops the one bit that matters downstream:
+  whether an authenticator produced that name or whether it was synthesized
+  from the caller-supplied context id. This keeps that bit.
+
+  Args:
+    request: The incoming request context from the A2A server.
+
+  Returns:
+    An authenticated principal when the A2A server authenticated the caller,
+    and an unauthenticated one otherwise. Never ``None``: an A2A request has
+    crossed a remote trust boundary either way, and ``None`` has to keep
+    meaning that no boundary was crossed at all.
+  """
+  user = request.call_context.user if request.call_context else None
+  user_name = getattr(user, 'user_name', None) if user is not None else None
+  # Authenticated exactly when _get_user_id above takes its first branch, so
+  # the principal and the user id can never disagree about the same request,
+  # plus one extra check: an a2a User that reports is_authenticated False is
+  # not vouched for even if it carries a name. Stricter by one condition,
+  # never looser.
+  if not isinstance(user_name, str) or not user_name:
+    return CallerPrincipal(authenticated=False, source='a2a')
+  if getattr(user, 'is_authenticated', True) is False:
+    return CallerPrincipal(authenticated=False, source='a2a')
+  return CallerPrincipal(authenticated=True, user_name=user_name, source='a2a')
+
+
 @a2a_experimental
 def convert_a2a_request_to_agent_run_request(
     request: RequestContext,
@@ -118,4 +151,5 @@ def convert_a2a_request_to_agent_run_request(
           parts=output_parts,
       ),
       run_config=RunConfig(custom_metadata=custom_metadata),
+      caller_principal=build_caller_principal(request),
   )
diff --git a/src/google/adk/a2a/executor/a2a_agent_executor.py b/src/google/adk/a2a/executor/a2a_agent_executor.py
index 44375e1cb86..b13e458b443 100644
--- a/src/google/adk/a2a/executor/a2a_agent_executor.py
+++ b/src/google/adk/a2a/executor/a2a_agent_executor.py
@@ -34,6 +34,7 @@
 from ...utils.context_utils import Aclosing
 from ..agent.interceptors.new_integration_extension import _NEW_A2A_ADK_INTEGRATION_EXTENSION
 from ..converters.request_converter import AgentRunRequest
+from ..converters.request_converter import build_caller_principal
 from ..converters.utils import _get_adk_metadata_key
 from ..experimental import a2a_experimental
 from .a2a_agent_executor_impl import _A2aAgentExecutor as ExecutorImpl
@@ -189,6 +190,11 @@ async def _handle_request(
         context,
         self._config.a2a_part_converter,
     )
+    # The request converter is pluggable, so the principal is re-derived here
+    # from the RequestContext the A2A server handed us. A custom converter must
+    # not be able to drop the principal, and must not be able to assert one
+    # the server never established.
+    run_request.caller_principal = build_caller_principal(context)
 
     # ensure the session exists
     session = await self._prepare_session(context, run_request, runner)
@@ -198,6 +204,7 @@ async def _handle_request(
         session=session,
         new_message=run_request.new_message,
         run_config=run_request.run_config,
+        caller_principal=run_request.caller_principal,
     )
 
     executor_context = ExecutorContext(
diff --git a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py b/src/google/adk/a2a/executor/a2a_agent_executor_impl.py
index c34ecf8d03d..fcd5d05c4ce 100644
--- a/src/google/adk/a2a/executor/a2a_agent_executor_impl.py
+++ b/src/google/adk/a2a/executor/a2a_agent_executor_impl.py
@@ -37,6 +37,7 @@
 from ..converters.long_running_functions import handle_user_input
 from ..converters.long_running_functions import LongRunningFunctions
 from ..converters.request_converter import AgentRunRequest
+from ..converters.request_converter import build_caller_principal
 from ..converters.utils import _get_adk_metadata_key
 from ..experimental import a2a_experimental
 from .config import A2aAgentExecutorConfig
@@ -102,6 +103,11 @@ async def execute(
           context,
           self._config.a2a_part_converter,
       )
+      # The request converter is pluggable, so the principal is re-derived
+      # here from the RequestContext the A2A server handed us. A custom
+      # converter must not be able to drop the principal, and must not be
+      # able to assert one the server never established.
+      run_request.caller_principal = build_caller_principal(context)
       session_id = await self._resolve_session(run_request, runner)
 
       executor_context = ExecutorContext(
diff --git a/src/google/adk/agents/__init__.py b/src/google/adk/agents/__init__.py
index 6a2f464913e..ecbc54abf6c 100644
--- a/src/google/adk/agents/__init__.py
+++ b/src/google/adk/agents/__init__.py
@@ -42,6 +42,7 @@
     'Agent': '.llm_agent',
     'BaseAgent': '.base_agent',
     'BaseAgentConfig': '.base_agent_config',
+    'CallerPrincipal': '._caller_principal',
     'Context': '.context',
     'InvocationContext': '.invocation_context',
     'LiveRequest': '.live_request_queue',
@@ -68,6 +69,7 @@
     'McpInstructionProvider',
     'ParallelAgent',
     'SequentialAgent',
+    'CallerPrincipal',
     'InvocationContext',
     'LiveRequest',
     'LiveRequestQueue',
diff --git a/src/google/adk/agents/_caller_principal.py b/src/google/adk/agents/_caller_principal.py
new file mode 100644
index 00000000000..b9ab4b5d5ab
--- /dev/null
+++ b/src/google/adk/agents/_caller_principal.py
@@ -0,0 +1,58 @@
+# 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.
+
+"""The caller identity that the serving layer established for an invocation."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel
+from pydantic import ConfigDict
+
+
+class CallerPrincipal(BaseModel):
+  """Who sent this invocation, and whether a serving layer vouched for them.
+
+  A serving edge (for example the A2A executor) sets this from the
+  authentication it actually performed on the inbound request. It is never
+  derived from message content, event authorship, or transport metadata,
+  because the caller controls all of those.
+
+  The confirmation flow uses it to decide whether a human-in-the-loop approval
+  can be honored. Three states are meaningful:
+
+  - ``None`` on the invocation context: no remote trust boundary was crossed
+    (for example an in-process ``Runner.run_async`` call). The caller is the
+    operator by construction.
+  - ``authenticated=True``: the edge verified the caller's identity.
+  - ``authenticated=False``: the edge saw the request but could not vouch for
+    who sent it.
+  """
+
+  model_config = ConfigDict(extra="forbid", frozen=True)
+  """The pydantic model config."""
+
+  authenticated: bool
+  """True only if a serving-layer authenticator verified the caller."""
+
+  user_name: Optional[str] = None
+  """The verified identity when ``authenticated`` is True, else None."""
+
+  source: Optional[str] = None
+  """Short label for the edge that set this, for example ``"a2a"``.
+
+  Informational only. Trust decisions must key off ``authenticated``, never
+  off the source, so a transport can not be used as a proxy for identity.
+  """
diff --git a/src/google/adk/agents/invocation_context.py b/src/google/adk/agents/invocation_context.py
index 55b6e7bbb37..ae80378cfca 100644
--- a/src/google/adk/agents/invocation_context.py
+++ b/src/google/adk/agents/invocation_context.py
@@ -42,6 +42,7 @@
 from ..sessions.session import Session
 from ..tools.base_tool import BaseTool
 from ..workflow._base_node import BaseNode
+from ._caller_principal import CallerPrincipal
 from .base_agent import BaseAgent
 from .base_agent import BaseAgentState
 from .context_cache_config import ContextCacheConfig
@@ -212,6 +213,18 @@ class InvocationContext(BaseModel):
   run_config: RunConfig | None = None
   """Configurations for live agents under this invocation."""
 
+  caller_principal: CallerPrincipal | None = None
+  """Who the serving layer says submitted this invocation.
+
+  ``None`` means no remote trust boundary was crossed for this invocation: it
+  was started in process (for example ``Runner.run_async`` called directly), so
+  the caller is the operator by construction.
+
+  A serving edge sets this from the authentication it actually performed on the
+  inbound request. It is never derived from message content, event authorship,
+  or transport metadata, because a remote caller controls all of those.
+  """
+
   resumability_config: ResumabilityConfig | None = None
   """The resumability config that applies to all agents under this invocation."""
 
diff --git a/src/google/adk/features/_feature_registry.py b/src/google/adk/features/_feature_registry.py
index e4259206624..0be5bc00d6f 100644
--- a/src/google/adk/features/_feature_registry.py
+++ b/src/google/adk/features/_feature_registry.py
@@ -69,6 +69,7 @@ class FeatureName(str, Enum):
   SPANNER_ADMIN_TOOLSET = "SPANNER_ADMIN_TOOLSET"
   SPANNER_TOOL_SETTINGS = "SPANNER_TOOL_SETTINGS"
   SPANNER_VECTOR_STORE = "SPANNER_VECTOR_STORE"
+  STRICT_CALLER_PRINCIPAL = "STRICT_CALLER_PRINCIPAL"
   TOOL_CONFIG = "TOOL_CONFIG"
   TOOL_CONFIRMATION = "TOOL_CONFIRMATION"
   PLUGGABLE_AUTH = "PLUGGABLE_AUTH"
@@ -220,6 +221,14 @@ class FeatureConfig:
     FeatureName.SPANNER_VECTOR_STORE: FeatureConfig(
         FeatureStage.EXPERIMENTAL, default_on=True
     ),
+    # Refuse, rather than warn about, a human-in-the-loop tool confirmation
+    # from a caller the serving layer did not authenticate. Off by default so
+    # that upgrading cannot break a deployment that runs without an
+    # authenticator; flipping default_on is the whole of the planned change
+    # at the next major version.
+    FeatureName.STRICT_CALLER_PRINCIPAL: FeatureConfig(
+        FeatureStage.EXPERIMENTAL, default_on=False
+    ),
     FeatureName.TOOL_CONFIG: FeatureConfig(
         FeatureStage.EXPERIMENTAL, default_on=True
     ),
diff --git a/src/google/adk/flows/llm_flows/tools/_confirmation.py b/src/google/adk/flows/llm_flows/tools/_confirmation.py
index 1a5455ac5ae..e1cc5a255f3 100644
--- a/src/google/adk/flows/llm_flows/tools/_confirmation.py
+++ b/src/google/adk/flows/llm_flows/tools/_confirmation.py
@@ -26,6 +26,8 @@
 from ....agents.invocation_context import InvocationContext
 from ....agents.readonly_context import ReadonlyContext
 from ....events.event import Event
+from ....features import FeatureName
+from ....features import is_feature_enabled
 from ....models.llm_request import LlmRequest
 from ....tools.base_tool import BaseTool
 from ....tools.tool_confirmation import ToolConfirmation
@@ -254,6 +256,74 @@ def _map_confirmation_to_original_fc_ids(
   return mapping
 
 
+def _apply_caller_principal_gate(
+    invocation_context: InvocationContext,
+    confirmations_by_fc_id: dict[str, ToolConfirmation],
+) -> dict[str, ToolConfirmation]:
+  """Decides whether the caller of this invocation may approve a tool call.
+
+  Three states, and only the middle one is a refusal:
+
+  - No principal: nothing vouched for the caller because nothing had to. The
+    invocation was started in process, so the caller is the operator.
+  - Principal present, not authenticated: a serving layer handled this request
+    and could not say who sent it, so the approval is not known to be the
+    operator's.
+  - Principal present and authenticated: the serving layer verified the caller.
+
+  The question asked here is deliberately about authentication and not about
+  transport. A transport is a proxy for identity, and a proxy for identity
+  fails in both directions: it refuses authenticated peers that happen to
+  arrive over the wire, and it admits anyone who reaches an ungated path.
+
+  A refusal rewrites the confirmation to ``confirmed=False`` instead of
+  dropping it. Dropping it leaves the ``adk_request_confirmation`` call pending
+  with nothing left to resolve it, which is what made an earlier attempt at
+  this guard stall every human-in-the-loop tool. ``confirmed=False`` is a state
+  the framework already has a contract for -- it is what a human decline
+  produces -- so the tool returns its rejection response and the turn ends with
+  a reason the caller can see.
+
+  Strict mode is the STRICT_CALLER_PRINCIPAL feature, off by default for now
+  so that upgrading cannot break a working deployment that runs its A2A server
+  without an authenticator; those get a warning instead. The intent is to flip
+  the registry default at the next major version.
+
+  Args:
+    invocation_context: Current invocation context.
+    confirmations_by_fc_id: Confirmations parsed from the last user event.
+
+  Returns:
+    The confirmations to act on, with refused ones forced to
+    ``confirmed=False``.
+  """
+  principal = invocation_context.caller_principal
+  if principal is None or principal.authenticated:
+    return confirmations_by_fc_id
+
+  if not is_feature_enabled(FeatureName.STRICT_CALLER_PRINCIPAL):
+    logger.warning(
+        "Honoring a tool confirmation from an unauthenticated caller"
+        " (principal source %r). The serving layer could not say who sent this"
+        " approval, so it is not known to be the operator's. Enable the"
+        " STRICT_CALLER_PRINCIPAL feature to refuse these instead; that is"
+        " intended to become the default in a future major version.",
+        principal.source,
+    )
+    return confirmations_by_fc_id
+
+  logger.error(
+      "Refusing a tool confirmation from an unauthenticated caller (principal"
+      " source %r). Enable authentication on the serving layer, or disable the"
+      " STRICT_CALLER_PRINCIPAL feature to downgrade this to a warning.",
+      principal.source,
+  )
+  return {
+      confirmation_fc_id: confirmation.model_copy(update={"confirmed": False})
+      for confirmation_fc_id, confirmation in confirmations_by_fc_id.items()
+  }
+
+
 class _RequestConfirmationLlmRequestProcessor(BaseLlmRequestProcessor):
   """Handles tool confirmation information to build the LLM request."""
 
@@ -323,6 +393,13 @@ async def run_async(
     if not confirmations_by_fc_id:
       return
 
+    # An approval is only worth acting on if it came from the operator this
+    # agent answers to. Everything above this point establishes that a
+    # confirmation was sent; this establishes who sent it.
+    confirmations_by_fc_id = _apply_caller_principal_gate(
+        invocation_context, confirmations_by_fc_id
+    )
+
     # Resolve all canonical tools and build tools_dict. Deliberately after the
     # dedup above so a consumed confirmation does not force a toolset
     # resolution, which can be a remote call for e.g. MCP toolsets.
diff --git a/src/google/adk/runners.py b/src/google/adk/runners.py
index 55f26166550..0b26424b731 100644
--- a/src/google/adk/runners.py
+++ b/src/google/adk/runners.py
@@ -36,6 +36,7 @@
 from opentelemetry import context
 from typing_extensions import Self
 
+from .agents._caller_principal import CallerPrincipal
 from .agents.base_agent import BaseAgent
 from .agents.context_cache_config import ContextCacheConfig
 from .agents.invocation_context import InvocationContext
@@ -1044,6 +1045,7 @@ async def run_async(
       new_message: Optional[types.Content] = None,
       state_delta: Optional[dict[str, Any]] = None,
       run_config: Optional[RunConfig] = None,
+      caller_principal: Optional[CallerPrincipal] = None,
       yield_user_message: bool = False,
   ) -> AsyncGenerator[Event, None]:
     """Main entry method to run the agent in this runner.
@@ -1062,6 +1064,10 @@ async def run_async(
       new_message: A new message to append to the session.
       state_delta: Optional state changes to apply to the session.
       run_config: The run config for the agent.
+      caller_principal: Set by a serving layer to record whether it
+        authenticated the caller of this invocation, and as whom. Leave it
+        unset for in-process callers: no remote trust boundary is crossed, so
+        there is nothing to vouch for.
       yield_user_message: If True, yield the user message event before
         agent/node events.
 
@@ -1221,6 +1227,7 @@ async def _run_with_trace(
               run_config=run_config,
               state_delta=state_delta,
               invocation_id=invocation_id,
+              caller_principal=caller_principal,
           )
         else:
           invocation_id = self._resolve_invocation_id(
@@ -1236,6 +1243,7 @@ async def _run_with_trace(
                 new_message=new_message,
                 run_config=run_config,
                 state_delta=state_delta,
+                caller_principal=caller_principal,
             )
           else:
             invocation_context = (
@@ -1245,6 +1253,7 @@ async def _run_with_trace(
                     invocation_id=invocation_id,
                     run_config=run_config,
                     state_delta=state_delta,
+                    caller_principal=caller_principal,
                 )
             )
             active_agent = invocation_context.agent
@@ -1900,6 +1909,7 @@ async def _setup_context_for_new_invocation(
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
       invocation_id: Optional[str] = None,
+      caller_principal: Optional[CallerPrincipal] = None,
   ) -> InvocationContext:
     """Sets up the context for a new invocation.
 
@@ -1909,6 +1919,8 @@ async def _setup_context_for_new_invocation(
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
       invocation_id: Optional invocation identifier.
+      caller_principal: Optional caller identity established by a serving
+        layer.
 
     Returns:
       The invocation context for the new invocation.
@@ -1919,6 +1931,7 @@ async def _setup_context_for_new_invocation(
         new_message=new_message,
         run_config=run_config,
         invocation_id=invocation_id,
+        caller_principal=caller_principal,
     )
     # Step 2: Handle new message, by running callbacks and appending to
     # session.
@@ -1947,6 +1960,7 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: str,
       run_config: RunConfig,
       state_delta: Optional[dict[str, Any]],
+      caller_principal: Optional[CallerPrincipal] = None,
   ) -> InvocationContext:
     """Sets up the context for a resumed invocation.
 
@@ -1956,6 +1970,8 @@ async def _setup_context_for_resumed_invocation(
       invocation_id: The invocation id to resume.
       run_config: The run config of the agent.
       state_delta: Optional state changes to apply to the session.
+      caller_principal: Optional caller identity established by a serving
+        layer.
 
     Returns:
       The invocation context for the resumed invocation.
@@ -1981,6 +1997,7 @@ async def _setup_context_for_resumed_invocation(
         new_message=user_message,
         run_config=run_config,
         invocation_id=invocation_id,
+        caller_principal=caller_principal,
     )
     # Step 3: Maybe handle new message.
     if new_message:
@@ -2031,6 +2048,7 @@ def _new_invocation_context(
       new_message: Optional[types.Content] = None,
       live_request_queue: Optional[LiveRequestQueue] = None,
       run_config: Optional[RunConfig] = None,
+      caller_principal: Optional[CallerPrincipal] = None,
   ) -> InvocationContext:
     """Creates a new invocation context.
 
@@ -2040,6 +2058,8 @@ def _new_invocation_context(
         new_message: The new message for the context.
         live_request_queue: The live request queue for the context.
         run_config: The run config for the context.
+        caller_principal: The caller identity a serving layer established for
+            this invocation, or None when no serving layer was involved.
 
     Returns:
         The new invocation context.
@@ -2076,6 +2096,7 @@ def _new_invocation_context(
         user_content=new_message,
         live_request_queue=live_request_queue,
         run_config=run_config,
+        caller_principal=caller_principal,
         resumability_config=self.resumability_config,
     )
 
diff --git a/tests/unittests/a2a/converters/test_request_converter.py b/tests/unittests/a2a/converters/test_request_converter.py
index e3481b31ca7..8a4a0790095 100644
--- a/tests/unittests/a2a/converters/test_request_converter.py
+++ b/tests/unittests/a2a/converters/test_request_converter.py
@@ -17,6 +17,7 @@
 from a2a.server.agent_execution import RequestContext
 from google.adk.a2a import _compat
 from google.adk.a2a.converters.request_converter import _get_user_id
+from google.adk.a2a.converters.request_converter import build_caller_principal
 from google.adk.a2a.converters.request_converter import convert_a2a_request_to_agent_run_request
 from google.adk.runners import RunConfig
 from google.genai import types as genai_types
@@ -457,3 +458,112 @@ def test_end_to_end_conversion_with_fallback_user(self):
     assert result.new_message.role == "user"
     assert result.new_message.parts == [mock_genai_part]
     assert isinstance(result.run_config, RunConfig)
+
+
+class TestBuildCallerPrincipal:
+  """Test cases for build_caller_principal."""
+
+  def _request_with_user(self, user):
+    """Builds a RequestContext whose call context carries the given user."""
+    mock_call_context = Mock()
+    mock_call_context.user = user
+    request = Mock(spec=RequestContext)
+    request.call_context = mock_call_context
+    request.context_id = "test_context"
+    return request
+
+  def test_authenticated_when_call_context_names_the_user(self):
+    """The serving layer verified the caller, so the principal says so."""
+    mock_user = Mock()
+    mock_user.user_name = "authenticated_user"
+
+    principal = build_caller_principal(self._request_with_user(mock_user))
+
+    assert principal.authenticated
+    assert principal.user_name == "authenticated_user"
+    assert principal.source == "a2a"
+
+  def test_unauthenticated_without_call_context(self):
+    """No call context means no authenticator ran."""
+    request = Mock(spec=RequestContext)
+    request.call_context = None
+    request.context_id = "test_context"
+
+    principal = build_caller_principal(request)
+
+    assert not principal.authenticated
+    assert principal.user_name is None
+    assert principal.source == "a2a"
+
+  def test_unauthenticated_when_call_context_has_no_user(self):
+    """A call context without a user vouches for nobody."""
+    principal = build_caller_principal(self._request_with_user(None))
+
+    assert not principal.authenticated
+    assert principal.user_name is None
+
+  def test_unauthenticated_with_empty_user_name(self):
+    """An a2a UnauthenticatedUser reports an empty name."""
+    mock_user = Mock()
+    mock_user.user_name = ""
+
+    principal = build_caller_principal(self._request_with_user(mock_user))
+
+    assert not principal.authenticated
+    assert principal.user_name is None
+
+  def test_unauthenticated_when_the_user_says_so_despite_a_name(self):
+    """is_authenticated False wins over a populated name."""
+    mock_user = Mock()
+    mock_user.user_name = "looks_real"
+    mock_user.is_authenticated = False
+
+    principal = build_caller_principal(self._request_with_user(mock_user))
+
+    assert not principal.authenticated
+    assert principal.user_name is None
+
+  @pytest.mark.parametrize(
+      "user_name, expect_authenticated",
+      [("real_user", True), ("", False), (None, False)],
+  )
+  def test_principal_never_disagrees_with_get_user_id(
+      self, user_name, expect_authenticated
+  ):
+    """The principal is authenticated exactly when _get_user_id trusts the name.
+
+    Both functions read the same call context. Deriving them from the same
+    condition is what keeps them from drifting apart, which is the failure
+    mode behind the original bug: an authentication fact computed and then
+    thrown away.
+    """
+    mock_user = Mock()
+    mock_user.user_name = user_name
+    request = self._request_with_user(mock_user)
+
+    principal = build_caller_principal(request)
+    used_call_context_name = _get_user_id(request) == user_name
+
+    assert principal.authenticated is expect_authenticated
+    assert principal.authenticated is used_call_context_name
+
+  def test_conversion_attaches_the_principal(self):
+    """The converter must put the principal on the AgentRunRequest."""
+    mock_message = Mock()
+    mock_message.parts = []
+    mock_user = Mock()
+    mock_user.user_name = "authenticated_user"
+    mock_call_context = Mock()
+    mock_call_context.user = mock_user
+
+    request = Mock(spec=RequestContext)
+    request.message = mock_message
+    request.context_id = "test_context"
+    request.call_context = mock_call_context
+    request.metadata = None
+
+    result = convert_a2a_request_to_agent_run_request(request, Mock())
+
+    assert result.caller_principal is not None
+    assert result.caller_principal.authenticated
+    assert result.caller_principal.user_name == "authenticated_user"
diff --git a/tests/unittests/flows/llm_flows/tools/test_confirmation.py b/tests/unittests/flows/llm_flows/tools/test_confirmation.py
index 0acd73915bf..b36447e4964 100644
--- a/tests/unittests/flows/llm_flows/tools/test_confirmation.py
+++ b/tests/unittests/flows/llm_flows/tools/test_confirmation.py
@@ -14,6 +14,7 @@
 
 from unittest.mock import patch
 
+from google.adk.agents._caller_principal import CallerPrincipal
 from google.adk.agents.llm_agent import LlmAgent
 from google.adk.events.event import Event
 from google.adk.events.event_actions import EventActions
@@ -1319,3 +1320,221 @@ async def test_resolve_confirmation_targets_requires_adk_name():
 
   assert set(tool_confirmation_dict) == {"requested_fc_id"}
   assert set(original_fcs_dict) == {"requested_fc_id"}
+
+
+def _build_pending_confirmation_events(agent_name: str) -> list[Event]:
+  """Builds a gated tool call followed by an approval on the user turn.
+
+  Args:
+    agent_name: Author to use for the agent-authored events.
+
+  Returns:
+    The session events, in order.
+  """
+  original_function_call = types.FunctionCall(
+      name=MOCK_TOOL_NAME, args={"param1": "test"}, id=MOCK_FUNCTION_CALL_ID
+  )
+  tool_confirmation = ToolConfirmation(confirmed=False, hint="test hint")
+  confirmation_request = types.FunctionCall(
+      name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+      args={
+          "originalFunctionCall": original_function_call.model_dump(
+              exclude_none=True, by_alias=True
+          ),
+          "toolConfirmation": tool_confirmation.model_dump(
+              by_alias=True, exclude_none=True
+          ),
+      },
+      id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
+  )
+  approval = types.FunctionResponse(
+      name=functions.REQUEST_CONFIRMATION_FUNCTION_CALL_NAME,
+      id=MOCK_CONFIRMATION_FUNCTION_CALL_ID,
+      response={"response": ToolConfirmation(confirmed=True).model_dump_json()},
+  )
+  return [
+      Event(
+          author=agent_name,
+          content=types.Content(
+              parts=[types.Part(function_call=original_function_call)]
+          ),
+      ),
+      Event(
+          author=agent_name,
+          content=types.Content(
+              parts=[types.Part(function_call=confirmation_request)]
+          ),
+      ),
+      Event(
+          author="user",
+          content=types.Content(parts=[types.Part(function_response=approval)]),
+      ),
+  ]
+
+
+async def _run_with_caller_principal(caller_principal):
+  """Runs the processor over a pending approval under one caller principal.
+
+  Args:
+    caller_principal: The principal to put on the invocation context.
+
+  Returns:
+    A tuple of the yielded events and the ToolConfirmation the processor
+    handed to tool execution, or None for the confirmation if execution was
+    never reached.
+  """
+  agent = LlmAgent(
+      name="test_agent",
+      tools=[FunctionTool(mock_tool, require_confirmation=True)],
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent
+  )
+  invocation_context.caller_principal = caller_principal
+  invocation_context.session.events.extend(
+      _build_pending_confirmation_events(agent.name)
+  )
+
+  resolved_event = Event(
+      author="agent",
+      content=types.Content(
+          parts=[
+              types.Part(
+                  function_response=types.FunctionResponse(
+                      name=MOCK_TOOL_NAME,
+                      id=MOCK_FUNCTION_CALL_ID,
+                      response={"result": "Mock tool result with test"},
+                  )
+              )
+          ]
+      ),
+  )
+
+  with patch(
+      "google.adk.flows.llm_flows.functions.handle_function_call_list_async"
+  ) as mock_handle_function_call_list_async:
+    mock_handle_function_call_list_async.return_value = resolved_event
+
+    events = []
+    async for event in request_processor.run_async(
+        invocation_context, LlmRequest()
+    ):
+      events.append(event)
+
+    if not mock_handle_function_call_list_async.call_args:
+      return events, None
+    args, _ = mock_handle_function_call_list_async.call_args
+    return events, args[4][MOCK_FUNCTION_CALL_ID]
+
+
+@pytest.mark.asyncio
+async def test_confirmation_honored_without_caller_principal():
+  """No principal means no remote boundary was crossed, so nothing changes."""
+  events, confirmation = await _run_with_caller_principal(None)
+
+  assert len(events) == 1
+  assert confirmation is not None
+  assert confirmation.confirmed
+
+
+@pytest.mark.asyncio
+async def test_confirmation_honored_for_authenticated_caller():
+  """An authenticated remote caller keeps its approval.
+
+  This is the half a transport-keyed guard gets wrong in the permissive
+  direction being wrong the other way: arriving over A2A is not by itself a
+  reason to refuse.
+  """
+  events, confirmation = await _run_with_caller_principal(
+      CallerPrincipal(authenticated=True, user_name="alice", source="a2a")
+  )
+
+  assert len(events) == 1
+  assert confirmation is not None
+  assert confirmation.confirmed
+
+
+@pytest.mark.asyncio
+async def test_confirmation_honored_for_unauthenticated_caller_by_default():
+  """Default is warn, not refuse.
+
+  Upgrading must not break a deployment that runs its A2A server without an
+  authenticator. Those operators get a log line; the behavior is unchanged
+  until they opt in.
+  """
+  events, confirmation = await _run_with_caller_principal(
+      CallerPrincipal(authenticated=False, source="a2a")
+  )
+
+  assert len(events) == 1
+  assert confirmation is not None
+  assert confirmation.confirmed
+
+
+@pytest.mark.asyncio
+async def test_confirmation_refused_for_unauthenticated_caller_when_strict(
+    monkeypatch,
+):
+  """Strict mode refuses, and the refusal is explicit rather than a stall.
+
+  The call still reaches tool execution, carrying confirmed=False. That is the
+  state a human decline produces, so the pending adk_request_confirmation call
+  resolves with a rejection the caller can see. Dropping the confirmation
+  instead is what made the earlier transport-keyed guard hang every
+  human-in-the-loop tool.
+  """
+  monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1")
+
+  events, confirmation = await _run_with_caller_principal(
+      CallerPrincipal(authenticated=False, source="a2a")
+  )
+
+  assert len(events) == 1
+  assert confirmation is not None
+  assert not confirmation.confirmed
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_does_not_touch_in_process_callers(monkeypatch):
+  """Strict mode must not refuse an invocation that crossed no boundary."""
+  monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1")
+
+  events, confirmation = await _run_with_caller_principal(None)
+
+  assert len(events) == 1
+  assert confirmation is not None
+  assert confirmation.confirmed
+
+
+@pytest.mark.asyncio
+async def test_strict_mode_keeps_multi_turn_reentry_a_noop(monkeypatch):
+  """Re-entry within a turn stays a no-op, principal or not.
+
+  The processor re-runs on every LLM step and the approval stays the last user
+  event for the rest of the turn. The gate deliberately runs after the
+  consumed-confirmation dedup, so a second pass over an approval that was
+  already acted on yields nothing rather than refusing or re-executing it.
+  This is the regression that got the previous guard reverted.
+  """
+  monkeypatch.setenv("ADK_ENABLE_STRICT_CALLER_PRINCIPAL", "1")
+  agent = LlmAgent(
+      name="test_agent",
+      tools=[FunctionTool(mock_tool, require_confirmation=False)],
+  )
+  invocation_context = await testing_utils.create_invocation_context(
+      agent=agent
+  )
+  invocation_context.caller_principal = CallerPrincipal(
+      authenticated=False, source="a2a"
+  )
+  invocation_context.session.events.extend(
+      _build_consumed_dynamic_confirmation_events(agent.name)
+  )
+
+  events = []
+  async for event in request_processor.run_async(
+      invocation_context, LlmRequest()
+  ):
+    events.append(event)
+
+  assert not events