Address PR review comments for tool result uniqueness

- Fix type safety: Use list[TextContent | ImageContent] instead of
  list[TextContent] with type ignore
- Use model_copy() instead of model_dump() for observation merging
- Add UUID suffix to merged event IDs to prevent potential collisions
- Extract _group_observations_by_tool_call() helper to eliminate
  duplicate iteration between transform() and enforce()
- Replace unittest mocks with real ObservationEvent instances in tests

Co-authored-by: openhands <openhands@all-hands.dev>

Author: openhands-agent

openhands-sdk/openhands/sdk/context/view/properties/tool_result_uniqueness.py

@@ -10,6 +10,7 @@
 information for the LLM.
 """
 
+import uuid
 from collections import defaultdict
 from collections.abc import Sequence
 
@@ -24,7 +25,7 @@
     ObservationEvent,
     ToolCallID,
 )
-from openhands.sdk.llm import TextContent
+from openhands.sdk.llm import ImageContent, TextContent
 
 
 def _create_merged_observation(
@@ -49,21 +50,18 @@ def _create_merged_observation(
 
     # Create new content list with error context prepended
     original_content = list(obs_event.observation.content)
-    merged_content: list[TextContent] = [TextContent(text=error_prefix)]
-    merged_content.extend(original_content)  # type: ignore[arg-type]
+    merged_content: list[TextContent | ImageContent] = [TextContent(text=error_prefix)]
+    merged_content.extend(original_content)
 
-    # Create a new observation with merged content
-    # We need to preserve all fields from the original observation
-    obs_data = obs_event.observation.model_dump()
-    obs_data["content"] = merged_content
-
-    # Create the new observation using the same class as the original
-    merged_observation = obs_event.observation.__class__(**obs_data)
+    # Create a new observation with merged content using model_copy
+    merged_observation = obs_event.observation.model_copy(
+        update={"content": merged_content}
+    )
 
     # Create new ObservationEvent with a unique ID
-    # ID format: "{original_id}-merged" to ensure uniqueness
+    # ID format: "{original_id}-merged-{uuid}" to ensure uniqueness
     return ObservationEvent(
-        id=f"{obs_event.id}-merged",
+        id=f"{obs_event.id}-merged-{uuid.uuid4().hex[:8]}",
         tool_name=obs_event.tool_name,
         tool_call_id=obs_event.tool_call_id,
         observation=merged_observation,
@@ -72,6 +70,26 @@ def _create_merged_observation(
     )
 
 
+def _group_observations_by_tool_call(
+    events: list[LLMConvertibleEvent],
+) -> dict[ToolCallID, list[ObservationBaseEvent]]:
+    """Group observations by their tool_call_id.
+
+    Args:
+        events: The list of events to process.
+
+    Returns:
+        A mapping from tool_call_id to list of observations with that ID.
+    """
+    observations_by_tool_call: dict[ToolCallID, list[ObservationBaseEvent]] = (
+        defaultdict(list)
+    )
+    for event in events:
+        if isinstance(event, ObservationBaseEvent):
+            observations_by_tool_call[event.tool_call_id].append(event)
+    return observations_by_tool_call
+
+
 class ToolResultUniquenessProperty(ViewPropertyBase):
     """Each tool_call_id must have exactly one tool result.
 
@@ -92,15 +110,9 @@ def transform(
         (typically from a restart scenario), merge the error context into the
         observation so the LLM has full context about what happened.
         """
-        # Group observations by tool_call_id
-        observations_by_tool_call: dict[ToolCallID, list[ObservationBaseEvent]] = (
-            defaultdict(list)
+        observations_by_tool_call = _group_observations_by_tool_call(
+            current_view_events
         )
-
-        for event in current_view_events:
-            if isinstance(event, ObservationBaseEvent):
-                observations_by_tool_call[event.tool_call_id].append(event)
-
         transforms: dict[EventID, LLMConvertibleEvent] = {}
 
         for observations in observations_by_tool_call.values():
@@ -132,15 +144,9 @@ def enforce(
         removes the remaining duplicate events (the original AgentErrorEvents and
         any other duplicates).
         """
-        # Group observations by tool_call_id
-        observations_by_tool_call: dict[ToolCallID, list[ObservationBaseEvent]] = (
-            defaultdict(list)
+        observations_by_tool_call = _group_observations_by_tool_call(
+            current_view_events
         )
-
-        for event in current_view_events:
-            if isinstance(event, ObservationBaseEvent):
-                observations_by_tool_call[event.tool_call_id].append(event)
-
         events_to_remove: set[EventID] = set()
 
         for observations in observations_by_tool_call.values():

tests/sdk/context/view/properties/test_tool_result_uniqueness.py

@@ -6,15 +6,12 @@
 - Duplicates are removed after merging
 """
 
-from unittest.mock import create_autospec
-
 from openhands.sdk.context.view.properties.tool_result_uniqueness import (
     ToolResultUniquenessProperty,
     _create_merged_observation,
 )
 from openhands.sdk.event.base import LLMConvertibleEvent
 from openhands.sdk.event.llm_convertible import (
-    ActionEvent,
     AgentErrorEvent,
     ObservationEvent,
     UserRejectObservation,
@@ -63,7 +60,8 @@ def test_merges_single_error_into_observation(self) -> None:
 
         merged = _create_merged_observation(obs, [error])
 
-        assert merged.id == "obs_1-merged"
+        # ID format: {original_id}-merged-{uuid}
+        assert merged.id.startswith("obs_1-merged-")
         assert merged.tool_call_id == "call_1"
         assert merged.tool_name == "terminal"
         # Check that error context is prepended
@@ -129,7 +127,8 @@ def test_transforms_when_error_and_observation_exist(self) -> None:
         assert "obs_1" in transforms
         merged = transforms["obs_1"]
         assert isinstance(merged, ObservationEvent)
-        assert merged.id == "obs_1-merged"
+        # ID format: {original_id}-merged-{uuid}
+        assert merged.id.startswith("obs_1-merged-")
         # Error content should be in merged observation
         content_text = "".join(
             c.text
@@ -173,13 +172,8 @@ def test_empty_list(self) -> None:
 
     def test_no_duplicates(self) -> None:
         """Test enforce when there are no duplicate tool_call_ids."""
-        obs1 = create_autospec(ObservationEvent, instance=True)
-        obs1.tool_call_id = "call_1"
-        obs1.id = "obs_1"
-
-        obs2 = create_autospec(ObservationEvent, instance=True)
-        obs2.tool_call_id = "call_2"
-        obs2.id = "obs_2"
+        obs1 = create_observation_event("obs_1", "call_1", "Result 1")
+        obs2 = create_observation_event("obs_2", "call_2", "Result 2")
 
         events: list[LLMConvertibleEvent] = [
             message_event("Start"),
@@ -193,13 +187,8 @@ def test_no_duplicates(self) -> None:
 
     def test_duplicate_observation_events(self) -> None:
         """Test that duplicate ObservationEvents keep the later one."""
-        obs1 = create_autospec(ObservationEvent, instance=True)
-        obs1.tool_call_id = "call_1"
-        obs1.id = "obs_1"
-
-        obs2 = create_autospec(ObservationEvent, instance=True)
-        obs2.tool_call_id = "call_1"  # Same tool_call_id!
-        obs2.id = "obs_2"
+        obs1 = create_observation_event("obs_1", "call_1", "First result")
+        obs2 = create_observation_event("obs_2", "call_1", "Second result")
 
         events: list[LLMConvertibleEvent] = [obs1, obs2]
 
@@ -216,10 +205,7 @@ def test_observation_event_preferred_over_agent_error(self) -> None:
             tool_call_id="call_1",
             error="Restart occurred while tool was running",
         )
-
-        obs = create_autospec(ObservationEvent, instance=True)
-        obs.tool_call_id = "call_1"  # Same tool_call_id!
-        obs.id = "obs_1"
+        obs = create_observation_event("obs_1", "call_1", "Actual result")
 
         events: list[LLMConvertibleEvent] = [agent_error, obs]
 
@@ -229,24 +215,15 @@ def test_observation_event_preferred_over_agent_error(self) -> None:
 
     def test_agent_error_before_observation_event(self) -> None:
         """Test AgentErrorEvent followed by ObservationEvent (restart scenario)."""
-        action = create_autospec(ActionEvent, instance=True)
-        action.tool_call_id = "call_1"
-        action.id = "action_1"
-        action.llm_response_id = "response_1"
-
         agent_error = AgentErrorEvent(
             tool_name="terminal",
             tool_call_id="call_1",
             error="A restart occurred while this tool was in progress.",
         )
-
-        obs = create_autospec(ObservationEvent, instance=True)
-        obs.tool_call_id = "call_1"
-        obs.id = "obs_1"
+        obs = create_observation_event("obs_1", "call_1", "Actual result")
 
         events: list[LLMConvertibleEvent] = [
             message_event("User message"),
-            action,
             agent_error,
             obs,  # Actual result arrives later
         ]
@@ -283,10 +260,7 @@ def test_user_reject_observation_handling(self) -> None:
             action_id="action_1",
             rejection_reason="User rejected",
         )
-
-        obs = create_autospec(ObservationEvent, instance=True)
-        obs.tool_call_id = "call_1"
-        obs.id = "obs_1"
+        obs = create_observation_event("obs_1", "call_1", "Actual result")
 
         events: list[LLMConvertibleEvent] = [reject, obs]
 
@@ -302,14 +276,10 @@ def test_mixed_scenario_multiple_tool_calls(self) -> None:
             tool_call_id="call_1",
             error="Restart error",
         )
-        obs1 = create_autospec(ObservationEvent, instance=True)
-        obs1.tool_call_id = "call_1"
-        obs1.id = "obs_1"
+        obs1 = create_observation_event("obs_1", "call_1", "Result 1")
 
         # Tool call 2: single observation (no duplicate)
-        obs2 = create_autospec(ObservationEvent, instance=True)
-        obs2.tool_call_id = "call_2"
-        obs2.id = "obs_2"
+        obs2 = create_observation_event("obs_2", "call_2", "Result 2")
 
         # Tool call 3: single error (no duplicate)
         error3 = AgentErrorEvent(
@@ -340,9 +310,7 @@ def setup_method(self) -> None:
 
     def test_complete_indices_returned(self) -> None:
         """Test that manipulation indices are complete (no restrictions)."""
-        obs = create_autospec(ObservationEvent, instance=True)
-        obs.tool_call_id = "call_1"
-        obs.id = "obs_1"
+        obs = create_observation_event("obs_1", "call_1", "Result")
 
         events: list[LLMConvertibleEvent] = [
             message_event("Start"),