Replace ListLike Protocol with list[T] type annotations (#562)

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

Author: simonrosenberg

openhands/sdk/context/view.py

@@ -1,3 +1,4 @@
+from collections.abc import Sequence
 from logging import getLogger
 from typing import overload
 
@@ -12,7 +13,6 @@
 from openhands.sdk.event.base import EventBase, EventID
 from openhands.sdk.event.llm_convertible import ActionEvent, ObservationBaseEvent
 from openhands.sdk.event.types import ToolCallID
-from openhands.sdk.utils.protocol import ListLike
 
 
 logger = getLogger(__name__)
@@ -143,7 +143,7 @@ def _should_keep_event(
             return True
 
     @staticmethod
-    def from_events(events: ListLike[EventBase]) -> "View":
+    def from_events(events: Sequence[EventBase]) -> "View":
         """Create a view from a list of events, respecting the semantics of any
         condensation events.
         """

openhands/sdk/conversation/__init__.py

@@ -1,6 +1,7 @@
 from openhands.sdk.conversation.base import BaseConversation
 from openhands.sdk.conversation.conversation import Conversation
-from openhands.sdk.conversation.event_store import EventLog, ListLike
+from openhands.sdk.conversation.event_store import EventLog
+from openhands.sdk.conversation.events_list_base import EventsListBase
 from openhands.sdk.conversation.secrets_manager import SecretsManager
 from openhands.sdk.conversation.state import ConversationState
 from openhands.sdk.conversation.stuck_detector import StuckDetector
@@ -17,5 +18,5 @@
     "SecretsManager",
     "StuckDetector",
     "EventLog",
-    "ListLike",
+    "EventsListBase",
 ]

openhands/sdk/conversation/base.py

@@ -4,19 +4,18 @@
 from typing import TYPE_CHECKING, Protocol
 
 from openhands.sdk.conversation.conversation_stats import ConversationStats
+from openhands.sdk.conversation.events_list_base import EventsListBase
 from openhands.sdk.conversation.secrets_manager import SecretValue
 from openhands.sdk.conversation.types import ConversationCallbackType, ConversationID
 from openhands.sdk.llm.message import Message
 from openhands.sdk.security.confirmation_policy import (
     ConfirmationPolicyBase,
     NeverConfirm,
 )
-from openhands.sdk.utils.protocol import ListLike
 
 
 if TYPE_CHECKING:
     from openhands.sdk.conversation.state import AgentExecutionStatus
-    from openhands.sdk.event.base import EventBase
 
 
 class ConversationStateProtocol(Protocol):
@@ -28,7 +27,7 @@ def id(self) -> ConversationID:
         ...
 
     @property
-    def events(self) -> ListLike["EventBase"]:
+    def events(self) -> EventsListBase:
         """Access to the events list."""
         ...
 

openhands/sdk/conversation/event_store.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterator
 from typing import SupportsIndex, overload
 
+from openhands.sdk.conversation.events_list_base import EventsListBase
 from openhands.sdk.conversation.persistence_const import (
     EVENT_FILE_PATTERN,
     EVENT_NAME_RE,
@@ -11,13 +12,12 @@
 from openhands.sdk.event import EventBase, EventID
 from openhands.sdk.io import FileStore
 from openhands.sdk.logger import get_logger
-from openhands.sdk.utils.protocol import ListLike
 
 
 logger = get_logger(__name__)
 
 
-class EventLog(ListLike[EventBase]):
+class EventLog(EventsListBase):
     def __init__(self, fs: FileStore, dir_path: str = EVENTS_DIR) -> None:
         self._fs = fs
         self._dir = dir_path
@@ -41,15 +41,19 @@ def get_id(self, idx: int) -> EventID:
         return self._idx_to_id[idx]
 
     @overload
-    def __getitem__(self, idx: SupportsIndex, /) -> EventBase: ...
+    def __getitem__(self, idx: int) -> EventBase: ...
+
     @overload
-    def __getitem__(self, idx: slice, /) -> list[EventBase]: ...
+    def __getitem__(self, idx: slice) -> list[EventBase]: ...
 
-    def __getitem__(self, idx: SupportsIndex | slice, /) -> EventBase | list[EventBase]:
+    def __getitem__(self, idx: SupportsIndex | slice) -> EventBase | list[EventBase]:
         if isinstance(idx, slice):
             start, stop, step = idx.indices(self._length)
-            return [self[i] for i in range(start, stop, step)]
+            return [self._get_single_item(i) for i in range(start, stop, step)]
         # idx is int-like (SupportsIndex)
+        return self._get_single_item(idx)
+
+    def _get_single_item(self, idx: SupportsIndex) -> EventBase:
         i = operator.index(idx)
         if i < 0:
             i += self._length
@@ -73,8 +77,8 @@ def __iter__(self) -> Iterator[EventBase]:
                 self._id_to_idx.setdefault(evt_id, i)
             yield evt
 
-    def append(self, item: EventBase) -> None:
-        evt_id = item.id
+    def append(self, event: EventBase) -> None:
+        evt_id = event.id
         # Check for duplicate ID
         if evt_id in self._id_to_idx:
             existing_idx = self._id_to_idx[evt_id]
@@ -83,7 +87,7 @@ def append(self, item: EventBase) -> None:
             )
 
         path = self._path(self._length, event_id=evt_id)
-        self._fs.write(path, item.model_dump_json(exclude_none=True))
+        self._fs.write(path, event.model_dump_json(exclude_none=True))
         self._idx_to_id[self._length] = evt_id
         self._id_to_idx[evt_id] = self._length
         self._length += 1

openhands/sdk/conversation/events_list_base.py

@@ -0,0 +1,17 @@
+from abc import ABC, abstractmethod
+from collections.abc import Sequence
+
+from openhands.sdk.event import EventBase
+
+
+class EventsListBase(Sequence[EventBase], ABC):
+    """Abstract base class for event lists that can be appended to.
+
+    This provides a common interface for both local EventLog and remote
+    RemoteEventsList implementations, avoiding circular imports in protocols.
+    """
+
+    @abstractmethod
+    def append(self, event: EventBase) -> None:
+        """Add a new event to the list."""
+        ...

openhands/sdk/conversation/impl/remote_conversation.py

@@ -11,6 +11,7 @@
 from openhands.sdk.agent.base import AgentBase
 from openhands.sdk.conversation.base import BaseConversation, ConversationStateProtocol
 from openhands.sdk.conversation.conversation_stats import ConversationStats
+from openhands.sdk.conversation.events_list_base import EventsListBase
 from openhands.sdk.conversation.secrets_manager import SecretValue
 from openhands.sdk.conversation.state import AgentExecutionStatus
 from openhands.sdk.conversation.types import ConversationCallbackType, ConversationID
@@ -21,7 +22,6 @@
 from openhands.sdk.security.confirmation_policy import (
     ConfirmationPolicyBase,
 )
-from openhands.sdk.utils.protocol import ListLike
 
 
 logger = get_logger(__name__)
@@ -101,7 +101,7 @@ async def _client_loop(self) -> None:
                 delay = min(delay * 2, 30.0)
 
 
-class RemoteEventsList(ListLike[EventBase]):
+class RemoteEventsList(EventsListBase):
     """A list-like, read-only view of remote conversation events.
 
     On first access it fetches existing events from the server. Afterwards,
@@ -155,6 +155,10 @@ def add_event(self, event: EventBase) -> None:
                 self._cached_event_ids.add(event.id)
                 logger.debug(f"Added event {event.id} to local cache")
 
+    def append(self, event: EventBase) -> None:
+        """Add a new event to the list (for compatibility with EventLog interface)."""
+        self.add_event(event)
+
     def create_default_callback(self) -> ConversationCallbackType:
         """Create a default callback that adds events to this list."""
 
@@ -167,30 +171,19 @@ def __len__(self) -> int:
         return len(self._cached_events)
 
     @overload
-    def __getitem__(self, index: SupportsIndex, /) -> EventBase: ...
+    def __getitem__(self, index: int) -> EventBase: ...
 
     @overload
-    def __getitem__(self, index: slice, /) -> list[EventBase]: ...
+    def __getitem__(self, index: slice) -> list[EventBase]: ...
 
-    def __getitem__(
-        self,
-        index: SupportsIndex | slice,
-        /,
-    ) -> EventBase | list[EventBase]:
+    def __getitem__(self, index: SupportsIndex | slice) -> EventBase | list[EventBase]:
         with self._lock:
             return self._cached_events[index]
 
     def __iter__(self):
         with self._lock:
             return iter(self._cached_events)
 
-    def append(self, event: EventBase) -> None:
-        # For remote conversations, events are added via API calls
-        # This method is here for interface compatibility but shouldn't be used directly
-        raise NotImplementedError(
-            "Cannot directly append events to remote conversation"
-        )
-
 
 class RemoteState(ConversationStateProtocol):
     """A state-like interface for accessing remote conversation state."""

openhands/sdk/conversation/state.py

@@ -1,5 +1,6 @@
 # state.py
 import json
+from collections.abc import Sequence
 from enum import Enum
 from typing import TYPE_CHECKING
 
@@ -21,7 +22,6 @@
     NeverConfirm,
 )
 from openhands.sdk.utils.models import OpenHandsModel
-from openhands.sdk.utils.protocol import ListLike
 
 
 logger = get_logger(__name__)
@@ -107,9 +107,9 @@ def model_post_init(self, __context) -> None:
         # Initialize FIFOLock
         FIFOLock.__init__(self)
 
-    # ===== Public "events" facade (ListLike[Event]) =====
+    # ===== Public "events" facade (Sequence[Event]) =====
     @property
-    def events(self) -> ListLike[EventBase]:
+    def events(self) -> EventLog:
         return self._events
 
     @property
@@ -231,7 +231,7 @@ def __setattr__(self, name, value):
                 raise e
 
     @staticmethod
-    def get_unmatched_actions(events: ListLike[EventBase]) -> list[ActionEvent]:
+    def get_unmatched_actions(events: Sequence[EventBase]) -> list[ActionEvent]:
         """Find actions in the event history that don't have matching observations.
 
         This method identifies ActionEvents that don't have corresponding

openhands/sdk/utils/__init__.py

@@ -1,6 +1,5 @@
 """Utility functions for the OpenHands SDK."""
 
-from .protocol import ListLike
 from .truncate import (
     DEFAULT_TEXT_CONTENT_LIMIT,
     DEFAULT_TRUNCATE_NOTICE,
@@ -12,5 +11,4 @@
     "DEFAULT_TEXT_CONTENT_LIMIT",
     "DEFAULT_TRUNCATE_NOTICE",
     "maybe_truncate",
-    "ListLike",
 ]

openhands/sdk/utils/protocol.py

@@ -18,7 +18,6 @@
     MetricsSnapshot,
     TextContent,
 )
-from openhands.sdk.utils.protocol import ListLike
 
 
 def message_event(content: str) -> MessageEvent:
@@ -84,13 +83,13 @@ def test_should_condense(mock_llm: LLM) -> None:
 
     # Create events below the threshold
     small_events = [message_event(f"Event {i}") for i in range(max_size)]
-    small_view = View.from_events(cast(ListLike[EventBase], small_events))
+    small_view = View.from_events(small_events)
 
     assert not condenser.should_condense(small_view)
 
     # Create events above the threshold
     large_events = [message_event(f"Event {i}") for i in range(max_size + 1)]
-    large_view = View.from_events(cast(ListLike[EventBase], large_events))
+    large_view = View.from_events(large_events)
 
     assert condenser.should_condense(large_view)
 
@@ -101,7 +100,7 @@ def test_condense_returns_view_when_no_condensation_needed(mock_llm: LLM) -> Non
     condenser = LLMSummarizingCondenser(llm=mock_llm, max_size=max_size)
 
     events: list[EventBase] = [message_event(f"Event {i}") for i in range(max_size)]
-    view = View.from_events(cast(ListLike[EventBase], events))
+    view = View.from_events(events)
 
     result = condenser.condense(view)
 
@@ -123,7 +122,7 @@ def test_condense_returns_condensation_when_needed(mock_llm: LLM) -> None:
     cast(Any, mock_llm).set_mock_response_content("Summary of forgotten events")
 
     events: list[EventBase] = [message_event(f"Event {i}") for i in range(max_size + 1)]
-    view = View.from_events(cast(ListLike[EventBase], events))
+    view = View.from_events(events)
 
     result = condenser.condense(view)
 
@@ -160,7 +159,7 @@ def test_get_condensation_with_previous_summary(mock_llm: LLM) -> None:
         events[:keep_first] + [condensation] + events[keep_first:]
     )
 
-    view = View.from_events(cast(ListLike[EventBase], events_with_condensation))
+    view = View.from_events(events_with_condensation)
 
     result = condenser.get_condensation(view)