(fix): Condensation events to reconstruct contexts added to event stream

openhands/agenthub/codeact_agent/codeact_agent.py

@@ -11,8 +11,10 @@
     Action,
     AgentFinishAction,
 )
+from openhands.events.event import Event
 from openhands.llm.llm import LLM
 from openhands.memory.condenser import Condenser
+from openhands.memory.condenser.condenser import Condensation, View
 from openhands.memory.conversation_memory import ConversationMemory
 from openhands.runtime.plugins import (
     AgentSkillsRequirement,
@@ -92,6 +94,7 @@ def reset(self) -> None:
 
     def step(self, state: State) -> Action:
         """Performs one step using the CodeAct Agent.
+
         This includes gathering info on previous steps and prompting the model to make a command to execute.
 
         Parameters:
@@ -113,8 +116,23 @@ def step(self, state: State) -> Action:
         if latest_user_message and latest_user_message.content.strip() == '/exit':
             return AgentFinishAction()
 
-        # prepare what we want to send to the LLM
-        messages = self._get_messages(state)
+        # Condense the events from the state. If we get a view we'll pass those
+        # to the conversation manager for processing, but if we get a condensation
+        # event we'll just return that instead of an action. The controller will
+        # immediately ask the agent to step again with the new view.
+        condensed_history: list[Event] = []
+        match self.condenser.condensed_history(state):
+            case View(events=events):
+                condensed_history = events
+
+            case Condensation(action=condensation_action):
+                return condensation_action
+
+        logger.debug(
+            f'Processing {len(condensed_history)} events from a total of {len(state.history)} events'
+        )
+
+        messages = self._get_messages(condensed_history)
         params: dict = {
             'messages': self.llm.format_messages_for_llm(messages),
         }
@@ -127,7 +145,7 @@ def step(self, state: State) -> Action:
             self.pending_actions.append(action)
         return self.pending_actions.popleft()
 
-    def _get_messages(self, state: State) -> list[Message]:
+    def _get_messages(self, events: list[Event]) -> list[Message]:
         """Constructs the message history for the LLM conversation.
 
         This method builds a structured conversation history by processing events from the state
@@ -143,7 +161,7 @@ def _get_messages(self, state: State) -> list[Message]:
         6. Adds environment reminders for non-function-calling mode
 
         Args:
-            state (State): The current state object containing conversation history and other metadata
+            events: The list of events to convert to messages
 
         Returns:
             list[Message]: A list of formatted messages ready for LLM consumption, including:
@@ -167,13 +185,6 @@ def _get_messages(self, state: State) -> list[Message]:
             with_caching=self.llm.is_caching_prompt_active()
         )
 
-        # Condense the events from the state.
-        events = self.condenser.condensed_history(state)
-
-        logger.debug(
-            f'Processing {len(events)} events from a total of {len(state.history)} events'
-        )
-
         # Use ConversationMemory to process events
         messages = self.conversation_memory.process_events(
             condensed_history=events,

openhands/controller/agent_controller.py

@@ -54,7 +54,7 @@
     MessageAction,
     NullAction,
 )
-from openhands.events.action.agent import RecallAction
+from openhands.events.action.agent import CondensationAction, RecallAction
 from openhands.events.event import Event
 from openhands.events.observation import (
     AgentCondensationObservation,
@@ -305,6 +305,8 @@ def should_step(self, event: Event) -> bool:
                 return True
             if isinstance(event, AgentDelegateAction):
                 return True
+            if isinstance(event, CondensationAction):
+                return True
             return False
         if isinstance(event, Observation):
             if (

openhands/core/schema/action.py

@@ -81,3 +81,6 @@ class ActionType(str, Enum):
 
     RECALL = 'recall'
     """Retrieves content from a user workspace, microagent, or other source."""
+
+    CONDENSATION = 'condensation'
+    """Condenses a list of events into a summary."""

openhands/events/action/agent.py

@@ -111,3 +111,26 @@ def __str__(self) -> str:
         ret = '**RecallAction**\n'
         ret += f'QUERY: {self.query[:50]}'
         return ret
+
+
+@dataclass
+class CondensationAction(Action):
+    """This action indicates a condensation of the conversation history is happening."""
+
+    action: str = ActionType.CONDENSATION
+
+    condenser_cls: str = ''
+    """The class of the condenser producing the condensation."""
+
+    forgotten_event_ids: list[int] = field(default_factory=list)
+    """The IDs of the events that are being forgotten (removed from the `View` given to the LLM)."""
+
+    considered_event_ids: list[int] = field(default_factory=list)
+    """The IDs of the events that are being considered for condensation."""
+
+    summary: str | None = None
+    """An optional summary of the events being forgotten."""
+
+    @property
+    def message(self) -> str:
+        return f'Summary: {self.summary}'

openhands/events/serialization/action.py

@@ -8,6 +8,7 @@
     AgentRejectAction,
     AgentThinkAction,
     ChangeAgentStateAction,
+    CondensationAction,
     RecallAction,
 )
 from openhands.events.action.browse import BrowseInteractiveAction, BrowseURLAction
@@ -39,6 +40,7 @@
     RecallAction,
     ChangeAgentStateAction,
     MessageAction,
+    CondensationAction,
 )
 
 ACTION_TYPE_TO_CLASS = {action_class.action: action_class for action_class in actions}  # type: ignore[attr-defined]

openhands/memory/condenser/__init__.py

@@ -1,4 +1,15 @@
 import openhands.memory.condenser.impl  # noqa F401 (we import this to get the condensers registered)
-from openhands.memory.condenser.condenser import Condenser, get_condensation_metadata
+from openhands.memory.condenser.condenser import (
+    Condenser,
+    get_condensation_metadata,
+    View,
+    Condensation,
+)
 
-__all__ = ['Condenser', 'get_condensation_metadata', 'CONDENSER_REGISTRY']
+__all__ = [
+    'Condenser',
+    'get_condensation_metadata',
+    'CONDENSER_REGISTRY',
+    'View',
+    'Condensation',
+]

openhands/memory/condenser/condenser.py

@@ -2,12 +2,13 @@
 
 from abc import ABC, abstractmethod
 from contextlib import contextmanager
-from typing import Any
+from typing import Any, overload
 
-from typing_extensions import override
+from pydantic import BaseModel
 
 from openhands.controller.state.state import State
 from openhands.core.config.condenser_config import CondenserConfig
+from openhands.events.action.agent import CondensationAction
 from openhands.events.event import Event
 
 CONDENSER_METADATA_KEY = 'condenser_meta'
@@ -32,17 +33,54 @@ def get_condensation_metadata(state: State) -> list[dict[str, Any]]:
 """Registry of condenser configurations to their corresponding condenser classes."""
 
 
+class View(BaseModel):
+    """Linearly ordered view of events.
+
+    Produced by a condenser to indicate the included events are ready to process as LLM input.
+    """
+
+    events: list[Event]
+
+    def __len__(self) -> int:
+        return len(self.events)
+
+    def __iter__(self):
+        return iter(self.events)
+
+    # To preserve list-like indexing, we ideally support slicing and position-based indexing.
+    # The only challenge with that is switching the return type based on the input type -- we
+    # can mark the different signatures for MyPy with `@overload` decorators.
+
+    @overload
+    def __getitem__(self, key: slice) -> list[Event]: ...
+
+    @overload
+    def __getitem__(self, key: int) -> Event: ...
+
+    def __getitem__(self, key: int | slice) -> Event | list[Event]:
+        if isinstance(key, slice):
+            start, stop, step = key.indices(len(self))
+            return [self[i] for i in range(start, stop, step)]
+        elif isinstance(key, int):
+            return self.events[key]
+        else:
+            raise ValueError(f'Invalid key type: {type(key)}')
+
+
+class Condensation(BaseModel):
+    """Produced by a condenser to indicate the history has been condensed."""
+
+    action: CondensationAction
+
+
 class Condenser(ABC):
     """Abstract condenser interface.
 
     Condensers take a list of `Event` objects and reduce them into a potentially smaller list.
 
     Agents can use condensers to reduce the amount of events they need to consider when deciding which action to take. To use a condenser, agents can call the `condensed_history` method on the current `State` being considered and use the results instead of the full history.
 
-    Example usage::
-
-        condenser = Condenser.from_config(condenser_config)
-        events = condenser.condensed_history(state)
+    If the condenser returns a `Condensation` instead of a `View`, the agent should return `Condensation.action` instead of producing its own action. On the next agent step the condenser will use that condensation event to produce a new `View`.
     """
 
     def __init__(self):
@@ -82,7 +120,7 @@ def metadata_batch(self, state: State):
             self.write_metadata(state)
 
     @abstractmethod
-    def condense(self, events: list[Event]) -> list[Event]:
+    def condense(self, events: list[Event]) -> View | Condensation:
         """Condense a sequence of events into a potentially smaller list.
 
         New condenser strategies should override this method to implement their own condensation logic. Call `self.add_metadata` in the implementation to record any relevant per-condensation diagnostic information.
@@ -91,10 +129,10 @@ def condense(self, events: list[Event]) -> list[Event]:
             events: A list of events representing the entire history of the agent.
 
         Returns:
-            list[Event]: An event sequence representing a condensed history of the agent.
+            View | Condensation: A condensed view of the events or an event indicating the history has been condensed.
         """
 
-    def condensed_history(self, state: State) -> list[Event]:
+    def condensed_history(self, state: State) -> View | Condensation:
         """Condense the state's history."""
         with self.metadata_batch(state):
             return self.condense(state.history)
@@ -140,39 +178,32 @@ def from_config(cls, config: CondenserConfig) -> Condenser:
 class RollingCondenser(Condenser, ABC):
     """Base class for a specialized condenser strategy that applies condensation to a rolling history.
 
-    The rolling history is computed by appending new events to the most recent condensation. For example, the sequence of calls::
-
-        assert state.history == [event1, event2, event3]
-        condensation = condenser.condensed_history(state)
-
-        # ...new events are added to the state...
-
-        assert state.history == [event1, event2, event3, event4, event5]
-        condenser.condensed_history(state)
+    The rolling history is generated by `get_view`, which analyzes all events in the history and produces a `View` object representing what will be sent to the LLM.
 
-    will result in second call to `condensed_history` passing `condensation + [event4, event5]` to the `condense` method.
+    If `should_condense` says so, the condenser is then responsible for generating a `Condensation` object from the `View` object. This will be added to the event history which should -- when given to `get_view` -- produce the condensed `View` to be passed to the LLM.
     """
 
-    def __init__(self) -> None:
-        self._condensation: list[Event] = []
-        self._last_history_length: int = 0
-
-        super().__init__()
+    @abstractmethod
+    def should_condense(self, view: View) -> bool:
+        """Determine if a view should be condensed."""
 
-    @override
-    def condensed_history(self, state: State) -> list[Event]:
-        # The history should grow monotonically -- if it doesn't, something has
-        # truncated the history and we need to reset our tracking.
-        if len(state.history) < self._last_history_length:
-            self._condensation = []
-            self._last_history_length = 0
+    @abstractmethod
+    def get_view(self, events: list[Event]) -> View:
+        """Get the view from a list of events."""
 
-        new_events = state.history[self._last_history_length :]
+    @abstractmethod
+    def get_condensation(self, view: View) -> Condensation:
+        """Get the condensation from a view."""
 
-        with self.metadata_batch(state):
-            results = self.condense(self._condensation + new_events)
+    def condense(self, events: list[Event]) -> View | Condensation:
+        # Convert the state to a view. This might require some condenser-specific logic.
+        view = self.get_view(events)
 
-        self._condensation = results
-        self._last_history_length = len(state.history)
+        # If we trigger the condenser-specific condensation threshold, compute and return
+        # the condensation.
+        if self.should_condense(view):
+            return self.get_condensation(view)
 
-        return results
+        # Otherwise we're safe to just return the view.
+        else:
+            return view

openhands/memory/condenser/impl/amortized_forgetting_condenser.py

@@ -1,8 +1,13 @@
 from __future__ import annotations
 
 from openhands.core.config.condenser_config import AmortizedForgettingCondenserConfig
+from openhands.events.action.agent import CondensationAction
 from openhands.events.event import Event
-from openhands.memory.condenser.condenser import RollingCondenser
+from openhands.memory.condenser.condenser import (
+    Condensation,
+    RollingCondenser,
+    View,
+)
 
 
 class AmortizedForgettingCondenser(RollingCondenser):
@@ -32,18 +37,44 @@ def __init__(self, max_size: int = 100, keep_first: int = 0):
 
         super().__init__()
 
-    def condense(self, events: list[Event]) -> list[Event]:
-        """Apply the amortized forgetting strategy to the given list of events."""
-        if len(events) <= self.max_size:
-            return events
+    def get_view(self, events: list[Event]) -> View:
+        # Get all non-condensation events
+        result_events = []
+        forgotten_event_ids = []
 
+        for event in events:
+            if isinstance(event, CondensationAction):
+                forgotten_event_ids.extend(event.forgotten_event_ids)
+            else:
+                result_events.append(event)
+
+        return View(
+            events=[
+                event for event in result_events if event.id not in forgotten_event_ids
+            ]
+        )
+
+    def get_condensation(self, view: View) -> Condensation:
         target_size = self.max_size // 2
-        head = events[: self.keep_first]
+        head = view[: self.keep_first]
 
         events_from_tail = target_size - len(head)
-        tail = events[-events_from_tail:]
+        tail = view[-events_from_tail:]
+
+        events_to_keep = head + tail
+
+        event = CondensationAction(
+            forgotten_event_ids=[
+                event.id for event in view if event not in events_to_keep
+            ],
+            considered_event_ids=[event.id for event in view],
+            condenser_cls=self.__class__.__name__,
+        )
+
+        return Condensation(action=event)
 
-        return head + tail
+    def should_condense(self, view: View) -> bool:
+        return len(view) > self.max_size
 
     @classmethod
     def from_config(