feat: add rerun_actions method to replay conversation actions

openhands-sdk/openhands/sdk/conversation/impl/local_conversation.py

@@ -2,6 +2,7 @@
 import uuid
 from collections.abc import Mapping
 from pathlib import Path
+from typing import Any
 
 from openhands.sdk.agent.base import AgentBase
 from openhands.sdk.context.prompts.prompt import render_template
@@ -25,6 +26,7 @@
     DefaultConversationVisualizer,
 )
 from openhands.sdk.event import (
+    ActionEvent,
     CondensationRequest,
     MessageEvent,
     PauseEvent,
@@ -928,6 +930,121 @@ def condense(self) -> None:
 
         logger.info("Condensation request processed")
 
+    def rerun_actions(
+        self,
+        rerun_log_path: str | Path | None = None,
+    ) -> bool:
+        """Re-execute all actions from the conversation's event history.
+
+        This method iterates through all ActionEvents in the conversation and
+        re-executes them using their original action parameters. Execution
+        stops immediately if any tool call fails.
+
+        WARNING: This is an advanced feature intended for specific use cases
+        such as reproducing environment state from a saved conversation. Many
+        tool operations are NOT idempotent:
+
+        - File operations may fail if files already exist or were deleted
+        - Terminal commands may have different effects on changed state
+        - API calls may have side effects or return different results
+        - Browser state may differ from the original session
+
+        Use this method only when you understand that:
+        1. Results may differ from the original conversation
+        2. Some actions may fail due to changed environment state
+        3. The workspace should typically be reset before rerunning
+
+        Args:
+            rerun_log_path: Optional path to save a rerun event log. If provided,
+                the log will contain the actions and their new observations in
+                chronological order (as JSONL format).
+
+        Returns:
+            True if all actions executed successfully, False if any action failed.
+
+        Raises:
+            KeyError: If a tool from the original conversation is not available.
+                This is a configuration error (different from execution failure).
+        """
+        # Ensure agent is initialized (loads plugins and initializes tools)
+        self._ensure_agent_ready()
+
+        rerun_log: list[dict[str, Any]] = []
+        action_count = 0
+
+        for event in self._state.events:
+            if not isinstance(event, ActionEvent):
+                continue
+            if event.action is None:
+                # Skip actions that failed validation during original run
+                continue
+
+            action_count += 1
+            tool_name = event.tool_name
+
+            # Get the tool from the agent's tools_map
+            tool = self.agent.tools_map.get(tool_name)
+            if tool is None:
+                available_tools = list(self.agent.tools_map.keys())
+                raise KeyError(
+                    f"Tool '{tool_name}' not found during rerun. "
+                    f"Available tools: {available_tools}. "
+                    f"Ensure the agent is configured with the same tools as the "
+                    f"original conversation."
+                )
+
+            if not tool.executor:
+                logger.warning(
+                    f"Skipping action {action_count}: "
+                    f"tool '{tool_name}' has no executor"
+                )
+                continue
+
+            # Execute the tool with the original action
+            try:
+                logger.info(f"Rerunning action {action_count}: {tool_name}")
+                observation = tool(event.action, self)
+
+                # Log the action and observation
+                rerun_log.append(
+                    {
+                        "action_index": action_count,
+                        "tool_name": tool_name,
+                        "action": event.action.model_dump(),
+                        "observation": observation.model_dump(),
+                    }
+                )
+            except Exception as e:
+                logger.error(
+                    f"Action {action_count} ({tool_name}) failed during rerun: {e}"
+                )
+                # Save partial log before returning failure
+                if rerun_log_path is not None:
+                    self._save_rerun_log(rerun_log_path, rerun_log)
+                return False
+
+        logger.info(f"Rerun complete: {action_count} actions processed successfully")
+
+        # Save the complete rerun log
+        if rerun_log_path is not None:
+            self._save_rerun_log(rerun_log_path, rerun_log)
+
+        return True
+
+    def _save_rerun_log(
+        self,
+        path: str | Path,
+        rerun_log: list[dict[str, Any]],
+    ) -> None:
+        """Save the rerun log to a file in JSONL format."""
+        import json
+
+        log_path = Path(path)
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+        with log_path.open("w") as f:
+            for entry in rerun_log:
+                f.write(json.dumps(entry) + "\n")
+
     def execute_tool(self, tool_name: str, action: Action) -> Observation:
         """Execute a tool directly without going through the agent loop.