"""Tests for View.from_events summary offset handling. This module tests that the summary insertion during condensation does not break action-observation pairs or action batches. """ import pytest from openhands.sdk.context.view import View from openhands.sdk.event import ( ActionEvent, Condensation, CondensationSummaryEvent, MessageEvent, ObservationEvent, ) from openhands.sdk.event.llm_convertible import SystemPromptEvent from openhands.sdk.llm import Message, MessageToolCall, TextContent from openhands.sdk.tool.schema import Action, Observation # --------------------------------------------------------------------------- # Test fixtures and helpers # --------------------------------------------------------------------------- class MockAction(Action): """Mock action for testing.""" param: str = "test" class MockObservation(Observation): """Mock observation for testing.""" result: str = "test result" @property def to_llm_content(self): return [TextContent(text=self.result)] def create_system_prompt_event() -> SystemPromptEvent: """Create a system prompt event.""" return SystemPromptEvent( system_prompt=TextContent(text="You are a helpful assistant."), tools=[], ) def create_message_event(role: str = "user", text: str = "Hello") -> MessageEvent: """Create a message event.""" return MessageEvent( source="user" if role == "user" else "agent", llm_message=Message( role=role, content=[TextContent(text=text)], ), ) def create_action_event( tool_call_id: str, llm_response_id: str, tool_name: str = "test_tool", ) -> ActionEvent: """Create an action event.""" return ActionEvent( thought=[TextContent(text="I will do this action")], action=MockAction(), tool_name=tool_name, tool_call_id=tool_call_id, tool_call=MessageToolCall( id=tool_call_id, name=tool_name, arguments='{"param": "test"}', origin="completion", ), llm_response_id=llm_response_id, ) def create_observation_event( tool_call_id: str, action_id: str, tool_name: str = "test_tool", ) -> ObservationEvent: """Create an observation event.""" return ObservationEvent( observation=MockObservation(), action_id=action_id, tool_name=tool_name, tool_call_id=tool_call_id, ) def is_summary_between_action_and_observation(events: list) -> bool: """Check if a CondensationSummaryEvent is between an action and its observation.""" for i, event in enumerate(events): if isinstance(event, CondensationSummaryEvent): # Check if previous event is an action if i > 0 and isinstance(events[i - 1], ActionEvent): action = events[i - 1] action_tool_call_id = action.tool_call_id # Check if any later event is the matching observation for j in range(i + 1, len(events)): if isinstance(events[j], ObservationEvent): if events[j].tool_call_id == action_tool_call_id: return True return False def is_summary_between_batch_actions(events: list) -> bool: """Check if a CondensationSummaryEvent is between actions of the same batch.""" for i, event in enumerate(events): if isinstance(event, CondensationSummaryEvent): if i > 0 and i < len(events) - 1: prev_event = events[i - 1] next_event = events[i + 1] if isinstance(prev_event, ActionEvent) and isinstance( next_event, ActionEvent ): if prev_event.llm_response_id == next_event.llm_response_id: return True return False def print_view_debug_info(events: list, test_name: str = "") -> None: """Print debug information about the view's event structure. This helps diagnose issues with summary insertion position. """ from openhands.sdk.event.llm_convertible import ObservationBaseEvent print(f"\n{'='*60}") if test_name: print(f"[DEBUG] Test: {test_name}") print(f"[DEBUG] Event list ({len(events)} events):") for i, evt in enumerate(events): evt_type = type(evt).__name__ extra_info = "" if isinstance(evt, ActionEvent): extra_info = f" [tool_call_id={evt.tool_call_id}, llm_response_id={evt.llm_response_id[:8]}...]" elif isinstance(evt, ObservationBaseEvent): extra_info = f" [tool_call_id={evt.tool_call_id}]" elif isinstance(evt, CondensationSummaryEvent): extra_info = " <-- SUMMARY INSERTED HERE" print(f" [{i}] {evt_type}{extra_info}") # Check for issues is_bad_action_obs = is_summary_between_action_and_observation(events) is_bad_batch = is_summary_between_batch_actions(events) if is_bad_action_obs: print(f"[DEBUG] ⚠️ WARNING: Summary is between action and observation!") if is_bad_batch: print(f"[DEBUG] ⚠️ WARNING: Summary is between actions of same batch!") if not is_bad_action_obs and not is_bad_batch: print(f"[DEBUG] ✓ Summary position is safe") print(f"{'='*60}\n") # --------------------------------------------------------------------------- # Tests for summary offset bug # --------------------------------------------------------------------------- class TestSummaryOffsetBug: """Tests demonstrating and fixing the summary_offset bug. The bug occurs when: 1. Condensation uses a fixed summary_offset (e.g., keep_first=4) 2. After filtering forgotten events, the kept_events list structure differs 3. The summary ends up being inserted between an action and its observation """ def test_summary_inserted_between_action_and_observation_bug(self): """Test that reproduces the bug where summary is inserted between action and observation. This simulates the exact scenario from the production debug output: - After filtering orphaned observations, the kept_events structure changes - summary_offset lands between an action and its observation """ # Simulating the exact production scenario: # Original events before any condensation had action-obs pairs # After condensation filtering, an orphaned obs appears at start # and summary_offset=4 ends up pointing between action and obs sys_prompt = create_system_prompt_event() user_msg = create_message_event("user", "Help me with a task") # Create action-observation pairs action1 = create_action_event( tool_call_id="tool_call_1", llm_response_id="response_1", ) obs1 = create_observation_event( tool_call_id="tool_call_1", action_id=action1.id, ) action2 = create_action_event( tool_call_id="tool_call_2", llm_response_id="response_2", ) obs2 = create_observation_event( tool_call_id="tool_call_2", action_id=action2.id, ) action3 = create_action_event( tool_call_id="tool_call_3", llm_response_id="response_3", ) obs3 = create_observation_event( tool_call_id="tool_call_3", action_id=action3.id, ) # Create event list: [sys, user, a1, o1, a2, o2, a3, o3] events = [sys_prompt, user_msg, action1, obs1, action2, obs2, action3, obs3] # Scenario: Condensation with summary_offset=3 will land between action1 and obs1 # Events: [0:sys, 1:user, 2:a1, 3:o1, ...] # Offset 3 lands at o1, pushing it to position 4 # Result: [sys, user, a1, SUMMARY, o1, a2, o2, a3, o3] # This breaks the action1-obs1 pairing! condensation = Condensation( forgotten_event_ids=[], summary="This is a summary of previous events.", summary_offset=3, # Lands between action1 and obs1 llm_response_id="condenser_response", ) all_events = events + [condensation] view = View.from_events(all_events) # Print debug info print_view_debug_info( view.events, "test_summary_inserted_between_action_and_observation_bug" ) # Check if summary is between action and observation is_bad = is_summary_between_action_and_observation(view.events) # The bug is confirmed if summary is between action and observation if is_bad: pytest.fail( "BUG CONFIRMED: Summary was inserted between action and observation. " "This breaks LLM API message ordering requirements." ) def test_summary_offset_with_batch_atomicity(self): """Test that summary is not inserted between actions of the same batch. When LLM returns parallel tool calls, all actions have the same llm_response_id. The summary should not be inserted between them. """ sys_prompt = create_system_prompt_event() user_msg = create_message_event("user", "Do multiple things") # Create a batch of parallel tool calls (same llm_response_id) batch_response_id = "parallel_response" action1 = create_action_event( tool_call_id="batch_call_1", llm_response_id=batch_response_id, ) action2 = create_action_event( tool_call_id="batch_call_2", llm_response_id=batch_response_id, ) # Observations come after all actions obs1 = create_observation_event( tool_call_id="batch_call_1", action_id=action1.id, ) obs2 = create_observation_event( tool_call_id="batch_call_2", action_id=action2.id, ) events = [sys_prompt, user_msg, action1, action2, obs1, obs2] # Create condensation with offset that lands between the two actions condensation = Condensation( forgotten_event_ids=[], summary="Summary text", summary_offset=3, # Between action1 and action2 llm_response_id="condenser_response", ) all_events = events + [condensation] view = View.from_events(all_events) # Print debug info print_view_debug_info(view.events, "test_summary_offset_with_batch_atomicity") is_bad = is_summary_between_batch_actions(view.events) if is_bad: pytest.fail( "BUG: Summary was inserted between actions of the same batch. " "This breaks the message structure for parallel tool calls." ) def test_summary_offset_clamping(self): """Test that summary_offset is clamped to valid range.""" sys_prompt = create_system_prompt_event() user_msg = create_message_event("user", "Hello") events = [sys_prompt, user_msg] # Create condensation with offset larger than event count condensation = Condensation( forgotten_event_ids=[], summary="Summary text", summary_offset=100, # Way larger than event count llm_response_id="condenser_response", ) all_events = events + [condensation] # This should not crash view = View.from_events(all_events) # Summary should be at the end assert isinstance(view.events[-1], CondensationSummaryEvent) def test_normal_condensation_preserves_order(self): """Test that normal condensation with safe offset preserves event order.""" sys_prompt = create_system_prompt_event() user_msg = create_message_event("user", "Help me") action1 = create_action_event( tool_call_id="call_1", llm_response_id="resp_1", ) obs1 = create_observation_event( tool_call_id="call_1", action_id=action1.id, ) action2 = create_action_event( tool_call_id="call_2", llm_response_id="resp_2", ) obs2 = create_observation_event( tool_call_id="call_2", action_id=action2.id, ) events = [sys_prompt, user_msg, action1, obs1, action2, obs2] # Forget action1 and obs1, keep rest condensation = Condensation( forgotten_event_ids=[action1.id, obs1.id], summary="Summary of forgotten events", summary_offset=2, # After sys_prompt and user_msg llm_response_id="condenser_response", ) all_events = events + [condensation] view = View.from_events(all_events) # Expected: [sys_prompt, user_msg, SUMMARY, action2, obs2] assert len(view.events) == 5 # Verify structure assert isinstance(view.events[0], SystemPromptEvent) assert isinstance(view.events[1], MessageEvent) assert isinstance(view.events[2], CondensationSummaryEvent) assert isinstance(view.events[3], ActionEvent) assert isinstance(view.events[4], ObservationEvent) # Verify action-observation pairing assert view.events[3].tool_call_id == view.events[4].tool_call_id # Verify no unsafe insertion assert not is_summary_between_action_and_observation(view.events) assert not is_summary_between_batch_actions(view.events) class TestFilterUnmatchedToolCalls: """Tests for the filter_unmatched_tool_calls functionality.""" def test_orphaned_observation_is_filtered(self): """Test that observations without matching actions are filtered.""" sys_prompt = create_system_prompt_event() # Create orphaned observation (no matching action) orphaned_obs = create_observation_event( tool_call_id="orphaned", action_id="nonexistent", ) # Create valid action-observation pair action = create_action_event( tool_call_id="valid", llm_response_id="resp", ) obs = create_observation_event( tool_call_id="valid", action_id=action.id, ) events = [sys_prompt, orphaned_obs, action, obs] view = View.from_events(events) # Orphaned observation should be filtered out assert len(view.events) == 3 assert not any( isinstance(e, ObservationEvent) and e.tool_call_id == "orphaned" for e in view.events ) def test_orphaned_action_is_filtered(self): """Test that actions without matching observations are filtered.""" sys_prompt = create_system_prompt_event() # Create orphaned action (no matching observation) orphaned_action = create_action_event( tool_call_id="orphaned", llm_response_id="resp", ) # Create valid action-observation pair action = create_action_event( tool_call_id="valid", llm_response_id="resp2", ) obs = create_observation_event( tool_call_id="valid", action_id=action.id, ) events = [sys_prompt, orphaned_action, action, obs] view = View.from_events(events) # Orphaned action should be filtered out assert len(view.events) == 3 assert not any( isinstance(e, ActionEvent) and e.tool_call_id == "orphaned" for e in view.events )