Add helper function

Author: mwildehahn

dbostest.sqlite

@@ -359,6 +359,14 @@ def __init__(
         self._entered_count = 0
         self._exit_stack = None
 
+    def _get_instructions_literal_and_functions(
+        self,
+    ) -> tuple[str | None, list[_system_prompt.SystemPromptRunner[AgentDepsT]]]:
+        instructions, instructions_functions = self._instructions, self._instructions_functions
+        if override_instructions := self._override_instructions.get():
+            instructions, instructions_functions = self._instructions_literal_and_functions(override_instructions.value)
+        return instructions, instructions_functions
+
     def _instructions_literal_and_functions(
         self,
         instructions: InstructionsInput,
@@ -597,10 +605,7 @@ async def main():
         usage_limits = usage_limits or _usage.UsageLimits()
 
         async def get_instructions(run_context: RunContext[AgentDepsT]) -> str | None:
-            literal, functions = self._instructions, self._instructions_functions
-            if override := self._override_instructions.get():
-                literal, functions = self._instructions_literal_and_functions(override.value)
-
+            literal, functions = self._get_instructions_literal_and_functions()
             parts = [
                 literal,
                 *[await func.run(run_context) for func in functions],
@@ -642,12 +647,7 @@ async def get_instructions(run_context: RunContext[AgentDepsT]) -> str | None:
             instrumentation_settings=instrumentation_settings,
         )
 
-        instructions_for_node, instructions_functions_for_node = self._instructions, self._instructions_functions
-        if override_instructions := self._override_instructions.get():
-            instructions_for_node, instructions_functions_for_node = self._instructions_literal_and_functions(
-                override_instructions.value
-            )
-
+        instructions_for_node, instructions_functions_for_node = self._get_instructions_literal_and_functions()
         start_node = _agent_graph.UserPromptNode[AgentDepsT](
             user_prompt=user_prompt,
             deferred_tool_results=deferred_tool_results,
@@ -771,9 +771,9 @@ def override(
             tools_token = None
 
         if _utils.is_set(instructions):
-            ins_token = self._override_instructions.set(_utils.Some(instructions))
+            instructions_token = self._override_instructions.set(_utils.Some(instructions))
         else:
-            ins_token = None
+            instructions_token = None
 
         try:
             yield
@@ -786,8 +786,8 @@ def override(
                 self._override_toolsets.reset(toolsets_token)
             if tools_token is not None:
                 self._override_tools.reset(tools_token)
-            if ins_token is not None:
-                self._override_instructions.reset(ins_token)
+            if instructions_token is not None:
+                self._override_instructions.reset(instructions_token)
 
     @overload
     def instructions(

pydantic_ai_slim/pydantic_ai/agent/__init__.py

@@ -10,6 +10,7 @@
     ModelMessage,
     ModelRequest,
     ModelRequestPart,
+    ModelResponse,
     SystemPromptPart,
 )
 from pydantic_ai.models.test import TestModel
@@ -29,6 +30,20 @@ def _system_prompt_texts(parts: Sequence[ModelRequestPart]) -> list[str]:
     return [p.content for p in parts if isinstance(p, SystemPromptPart)]
 
 
+def test_first_request_skips_non_requests():
+    """Helper ignores non-request messages until it finds a request."""
+    response = ModelResponse(parts=())
+    request = ModelRequest(parts=())
+    assert _first_request([response, request]) is request
+
+
+def test_first_request_raises_without_model_request():
+    """Helper raises when no model request is present."""
+    response = ModelResponse(parts=())
+    with pytest.raises(AssertionError, match='no ModelRequest found'):
+        _first_request([response])
+
+
 def test_override_instructions_basic():
     """Test that override can override instructions."""
     agent = Agent('test')
@@ -144,6 +159,18 @@ def override_fn() -> str:
     assert 'BASE' not in req.instructions
 
 
+def test_override_instructions_ignores_unknown_types():
+    """Override ignores instruction entries it does not understand."""
+    agent = Agent('test')
+
+    with agent.override(instructions=['ONLY_LITERAL', object()]):
+        with capture_run_messages() as messages:
+            agent.run_sync('Hi', model=TestModel(custom_output_text='ok'))
+
+    req = _first_request(messages)
+    assert req.instructions == 'ONLY_LITERAL'
+
+
 @pytest.mark.anyio
 async def test_override_concurrent_isolation():
     """Test that concurrent overrides are isolated from each other."""