fix: escape curly braces in agent instructions to prevent context variable errors (#1394)

## Summary

Fixes #1382

When an agent's instructions contain curly braces (e.g., `{repo}`,
`{branch}`), ADK's `inject_session_state` tries to resolve them as
context variable references. Since kagent doesn't use ADK context
variables, this causes a `KeyError`.

**Root cause:** ADK's `LlmAgent.canonical_instruction()` returns
`bypass_state_injection=False` when `instruction` is a string,
triggering `inject_session_state` which uses regex `{+[^{}]*}+` to find
and resolve `{text}` patterns as state variables.

**Fix:** Wrap the instruction string in a callable
(`InstructionProvider`) before passing to the ADK `Agent` constructor.
When instruction is callable, ADK sets `bypass_state_injection=True`,
completely skipping the template processing. The `_configure_memory`
method is also updated to properly handle callable instructions when
appending memory usage instructions.

## Changes

- `python/packages/kagent-adk/src/kagent/adk/types.py`: Wrap
`self.instruction` in an `InstructionProvider` callable in `to_agent()`.
Update `_configure_memory()` to compose callables when appending memory
instructions.
-
`python/packages/kagent-adk/tests/unittests/test_instruction_escaping.py`:
7 unit tests covering curly braces, JSON-like content, nested braces,
empty instructions, and verifying the callable bypass mechanism.

## Test plan

- [x] All 7 new unit tests pass
- [x] Existing test suite unaffected (pre-existing env import issues for
other tests are unrelated)
- [ ] CI pipeline validation

Signed-off-by: opspawn <agent@opspawn.com>

---------

Signed-off-by: opspawn <agent@opspawn.com>
Signed-off-by: fl-sean03 <jmhbh@users.noreply.github.com>
Signed-off-by: Sean Florez <sean@opspawn.com>
Signed-off-by: fl-sean03 <fl-sean03@users.noreply.github.com>
Co-authored-by: fl-sean03 <jmhbh@users.noreply.github.com>
Co-authored-by: Sean Florez <sean@opspawn.com>
Co-authored-by: fl-sean03 <fl-sean03@users.noreply.github.com>

Author: 3 people

python/packages/kagent-adk/src/kagent/adk/types.py

@@ -369,11 +369,17 @@ async def rewrite_url_to_proxy(request: httpx.Request) -> None:
         code_executor = SandboxedLocalCodeExecutor() if self.execute_code else None
         model = _create_llm_from_model_config(self.model)
 
+        # Use static_instruction to bypass ADK's session state injection
+        # (inject_session_state). ADK treats curly braces like {repo} as
+        # context variable references and raises KeyError when they aren't
+        # found in session state. static_instruction is sent directly to
+        # the model without any placeholder processing.
+        # See: https://github.com/kagent-dev/kagent/issues/1382
         agent = Agent(
             name=name,
             model=model,
             description=self.description,
-            instruction=self.instruction,
+            static_instruction=self.instruction,
             tools=tools,
             code_executor=code_executor,
         )
@@ -399,11 +405,18 @@ def _configure_memory(self, agent: Agent) -> None:
             agent.tools.append(LoadMemoryTool())
             agent.tools.append(SaveMemoryTool())
 
-            agent.instruction = (
-                f"{agent.instruction}\n\n"
-                "You have long-term memory: use save_memory to store important findings, learnings, and user preferences. "
+            memory_suffix = (
+                "\n\nYou have long-term memory: use save_memory to store important findings, learnings, and user preferences. "
                 "When you need more context or are unsure, use load_memory to search past conversations for relevant information."
             )
+            # Append memory instructions to static_instruction so they stay
+            # in the system prompt (same as the main instruction) instead of
+            # being injected as a separate user message, which would confuse
+            # the LLM conversation flow and break mock-based e2e tests.
+            if agent.static_instruction:
+                agent.static_instruction += memory_suffix
+            else:
+                agent.instruction += memory_suffix
 
             # Define auto-save callback
             async def auto_save_session_to_memory_callback(callback_context: CallbackContext):

python/packages/kagent-adk/tests/unittests/test_instruction_escaping.py

@@ -0,0 +1,136 @@
+"""Tests for curly brace handling in agent instructions.
+
+Verifies that agent prompts containing curly braces (like {repo}) don't cause
+KeyError from ADK's session state injection, by using static_instruction which
+bypasses placeholder processing entirely. See:
+https://github.com/kagent-dev/kagent/issues/1382
+"""
+
+import asyncio
+import sys
+from unittest.mock import MagicMock
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def _isolate_types_import(monkeypatch):
+    """Ensure kagent.adk.types can be imported without the heavy dependency chain.
+
+    kagent.adk.__init__ pulls in kagent.core (tracing, opentelemetry, etc.)
+    which may not be installed in the test environment. We mock the missing
+    modules so the types module itself can be imported directly.
+    """
+    stubs = [
+        "kagent.core",
+        "kagent.core.a2a",
+        "kagent.core.tracing",
+        "kagent.core.tracing._span_processor",
+        "agentsts",
+        "agentsts.adk",
+    ]
+    for mod_name in stubs:
+        if mod_name not in sys.modules:
+            monkeypatch.setitem(sys.modules, mod_name, MagicMock())
+
+
+def _make_agent_config(instruction: str):
+    """Create a minimal AgentConfig with the given instruction."""
+    import kagent.adk.types as types_mod
+
+    return types_mod.AgentConfig(
+        model={"type": "gemini", "model": "gemini-2.0-flash"},
+        description="test agent",
+        instruction=instruction,
+    )
+
+
+class TestStaticInstruction:
+    """Tests that curly braces in agent instructions are handled via static_instruction."""
+
+    def test_instruction_with_curly_braces_creates_agent(self):
+        """Agent with {repo} in instruction should not raise KeyError."""
+        config = _make_agent_config("Clone the repo {repo} and run tests on branch {branch}.")
+        agent = config.to_agent("test_agent")
+        assert agent is not None
+        assert agent.name == "test_agent"
+
+    def test_instruction_goes_to_static_instruction(self):
+        """Instruction should be set as static_instruction to bypass state injection."""
+        original = "Use {variable} in prompt"
+        config = _make_agent_config(original)
+        agent = config.to_agent("test_agent")
+        assert agent.static_instruction == original
+
+    def test_dynamic_instruction_is_empty_without_memory(self):
+        """Without memory, the dynamic instruction field should not be set."""
+        config = _make_agent_config("Deploy to {environment}")
+        agent = config.to_agent("test_agent")
+        assert not agent.instruction
+
+    def test_instruction_without_braces_works(self):
+        """Instructions without curly braces should still work normally."""
+        config = _make_agent_config("Just a normal instruction without braces.")
+        agent = config.to_agent("test_agent")
+        assert agent.static_instruction == "Just a normal instruction without braces."
+
+    def test_instruction_with_nested_braces(self):
+        """Instructions with nested or multiple braces should be preserved."""
+        original = "Format: {{key}}, single: {value}, mixed: {a} and {{b}}"
+        config = _make_agent_config(original)
+        agent = config.to_agent("test_agent")
+        assert agent.static_instruction == original
+
+    def test_instruction_with_json_like_content(self):
+        """Instructions containing JSON-like content should be preserved."""
+        original = 'Return output as JSON: {"status": "ok", "data": {items}}'
+        config = _make_agent_config(original)
+        agent = config.to_agent("test_agent")
+        assert agent.static_instruction == original
+
+    def test_empty_instruction(self):
+        """Empty instruction should still work."""
+        config = _make_agent_config("")
+        agent = config.to_agent("test_agent")
+        assert agent.static_instruction == ""
+
+
+class TestStaticInstructionBypass:
+    """Tests that static_instruction bypasses inject_session_state in ADK.
+
+    ADK's _build_instructions() passes static_instruction directly to
+    _transformers.t_content() without any variable substitution, while
+    the instruction field goes through _process_agent_instruction() which
+    calls inject_session_state(). This is the mechanism that prevents
+    KeyError on {repo}-style placeholders.
+    """
+
+    def test_inject_session_state_raises_on_unresolved_variable(self):
+        """inject_session_state raises KeyError for {repo} when state is empty.
+
+        This is the original bug: ADK's state injection treats {repo} as a
+        context variable reference and raises KeyError when it's not in session
+        state. Using static_instruction prevents this path from executing.
+        """
+        from google.adk.utils.instructions_utils import inject_session_state
+
+        mock_ctx = MagicMock()
+        mock_ctx._invocation_context.session.state = {}
+        with pytest.raises(KeyError, match="repo"):
+            asyncio.get_event_loop().run_until_complete(inject_session_state("Clone {repo}", mock_ctx))
+
+    def test_raw_string_instruction_would_not_bypass(self):
+        """A raw string in the instruction field would set bypass_state_injection=False.
+
+        This proves the fix is necessary: without static_instruction, ADK
+        would attempt state injection on the instruction field and raise
+        KeyError on unresolved {variables}.
+        """
+        config = _make_agent_config("Safe instruction")
+        agent = config.to_agent("test_agent")
+        # Temporarily set instruction to a raw string to verify ADK behavior
+        agent.instruction = "Raw string with {repo}"
+        mock_ctx = MagicMock()
+        text, bypass = asyncio.get_event_loop().run_until_complete(agent.canonical_instruction(mock_ctx))
+        assert bypass is False, "raw string in instruction must set bypass_state_injection=False"
+        assert text == "Raw string with {repo}"