feat(agents): forward ManagedAgent instruction as system_instruction

Resolve ManagedAgent.instruction (with {placeholder} state injection for
strings; InstructionProvider callables bypass injection) and forward it to
the Managed Agents API as the interaction's system_instruction on every
turn, including chained turns. An empty instruction sends nothing.

Co-authored-by: Haran Rajkumar <haranrk@google.com>
PiperOrigin-RevId: 951220238

Author: haranrk

src/google/adk/agents/_managed_agent.py

@@ -51,6 +51,7 @@
 from ..utils.content_utils import to_user_content
 from ..utils.context_utils import Aclosing
 from ..utils.env_utils import is_enterprise_mode_enabled
+from ..utils.instructions_utils import inject_session_state
 from ..utils.instructions_utils import InstructionProvider
 from .base_agent import BaseAgent
 from .context import Context
@@ -394,6 +395,15 @@ async def _run_async_impl(
     )
     interaction_tools = await self._resolve_backend_tools(ctx)
 
+    raw_si, bypass_state_injection = await self.canonical_instruction(
+        ReadonlyContext(ctx)
+    )
+    system_instruction = raw_si
+    if not bypass_state_injection:
+      system_instruction = await inject_session_state(
+          raw_si, ReadonlyContext(ctx)
+      )
+
     create_kwargs: dict[str, Any] = {
         'agent': self.agent_id,
         'input': input_steps,
@@ -411,6 +421,8 @@ async def _run_async_impl(
       create_kwargs['agent_config'] = self.agent_config
     if prev_interaction_id:
       create_kwargs['previous_interaction_id'] = prev_interaction_id
+    if system_instruction:
+      create_kwargs['system_instruction'] = system_instruction
 
     # Request-time header merge, parity with google_llm.generate_content_async:
     # combine any RunConfig headers with ADK tracking headers, non-destructively.
@@ -436,7 +448,7 @@ async def _run_async_impl(
         build_interactions_request_log(
             model=self.agent_id,
             input_steps=input_steps,
-            system_instruction=None,
+            system_instruction=system_instruction or None,
             tools=interaction_tools if interaction_tools else None,
             generation_config=None,
             previous_interaction_id=prev_interaction_id,

tests/unittests/agents/test_managed_agent.py

@@ -1089,3 +1089,121 @@ async def provider(ctx):
 
   assert text == 'async provider'
   assert bypass is True
+
+
+def test_instruction_defaults_to_empty_and_is_omitted():
+  client = _RecordingClient([[]])
+  agent = ManagedAgent(name='mgr', agent_id='agents/a', api_client=client)
+
+  asyncio.run(_drain(agent._run_async_impl(_user_ctx('hi'))))
+
+  assert agent.instruction == ''
+  assert 'system_instruction' not in client.aio.interactions.calls[0]
+
+
+def test_string_instruction_forwarded_as_system_instruction():
+  client = _RecordingClient([[]])
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction='You are a terse assistant.',
+      api_client=client,
+  )
+
+  asyncio.run(_drain(agent._run_async_impl(_user_ctx('hi'))))
+
+  assert (
+      client.aio.interactions.calls[0]['system_instruction']
+      == 'You are a terse assistant.'
+  )
+
+
+def test_sync_instruction_provider_forwarded_and_bypasses_injection():
+  client = _RecordingClient([[]])
+
+  # The '{name}' must be left literal: providers bypass state injection.
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction=lambda ctx: 'Persona for {name}',
+      api_client=client,
+  )
+
+  asyncio.run(_drain(agent._run_async_impl(_user_ctx('hi'))))
+
+  assert (
+      client.aio.interactions.calls[0]['system_instruction']
+      == 'Persona for {name}'
+  )
+
+
+def test_async_instruction_provider_forwarded():
+  client = _RecordingClient([[]])
+
+  async def provider(ctx):
+    return 'Async persona.'
+
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction=provider,
+      api_client=client,
+  )
+
+  asyncio.run(_drain(agent._run_async_impl(_user_ctx('hi'))))
+
+  assert (
+      client.aio.interactions.calls[0]['system_instruction'] == 'Async persona.'
+  )
+
+
+def test_instruction_sent_on_chained_turn():
+  prior = Event(
+      author='mgr', interaction_id='int_prev', environment_id='env_prev'
+  )
+  client = _RecordingClient([[]])
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction='Stay in character.',
+      api_client=client,
+  )
+  ctx = _user_ctx('again', session_events=[prior])
+
+  asyncio.run(_drain(agent._run_async_impl(ctx)))
+
+  create_kwargs = client.aio.interactions.calls[0]
+  assert create_kwargs['previous_interaction_id'] == 'int_prev'
+  assert create_kwargs['system_instruction'] == 'Stay in character.'
+
+
+def test_string_instruction_injects_session_state():
+  from google.adk.agents.invocation_context import InvocationContext
+  from google.adk.sessions.in_memory_session_service import InMemorySessionService
+  from google.adk.sessions.session import Session
+
+  client = _RecordingClient([[]])
+  agent = ManagedAgent(
+      name='mgr',
+      agent_id='agents/a',
+      instruction='Discuss {topic}.',
+      api_client=client,
+  )
+  session = Session(
+      app_name='test', user_id='user', id='s1', state={'topic': 'volcanoes'}
+  )
+  ctx = InvocationContext(
+      invocation_id='inv1',
+      session=session,
+      session_service=InMemorySessionService(),
+      user_content=genai_types.Content(
+          role='user', parts=[genai_types.Part(text='hi')]
+      ),
+  )
+
+  asyncio.run(_drain(agent._run_async_impl(ctx)))
+
+  assert (
+      client.aio.interactions.calls[0]['system_instruction']
+      == 'Discuss volcanoes.'
+  )