improve ps1 for env commands

openhands/events/observation/commands.py

@@ -12,7 +12,7 @@
 CMD_OUTPUT_PS1_BEGIN = '###PS1JSON###\n'
 CMD_OUTPUT_PS1_END = '\n###PS1END###'
 CMD_OUTPUT_METADATA_PS1_REGEX = re.compile(
-    f'{CMD_OUTPUT_PS1_BEGIN}(.*?){CMD_OUTPUT_PS1_END}', re.DOTALL
+    f'^{CMD_OUTPUT_PS1_BEGIN}(.*?){CMD_OUTPUT_PS1_END}', re.DOTALL | re.MULTILINE
 )
 
 
@@ -49,7 +49,14 @@ def to_ps1_prompt(cls) -> str:
 
     @classmethod
     def matches_ps1_metadata(cls, string: str) -> list[re.Match[str]]:
-        return list(CMD_OUTPUT_METADATA_PS1_REGEX.finditer(string))
+        matches = []
+        for match in CMD_OUTPUT_METADATA_PS1_REGEX.finditer(string):
+            try:
+                json.loads(match.group(1))  # Try to parse as JSON
+                matches.append(match)
+            except json.JSONDecodeError:
+                continue  # Skip if not valid JSON
+        return matches
 
     @classmethod
     def from_ps1_match(cls, match: re.Match[str]) -> Self:

openhands/runtime/utils/bash.py

@@ -3,6 +3,7 @@
 
 import libtmux
 
+from openhands.core.logger import openhands_logger as logger
 from openhands.events.action import CmdRunAction
 from openhands.events.observation.commands import (
     CMD_OUTPUT_PS1_END,
@@ -58,6 +59,7 @@ def __init__(
         # Store the last command for interactive input handling
         self.prev_status: BashCommandStatus | None = None
         self.prev_output: str = ''
+        logger.debug(f'Bash session initialized with work dir: {work_dir}')
 
     def close(self):
         self.session.kill_session()

tests/unit/test_bash_session.py

@@ -186,6 +186,19 @@ def test_empty_command_errors():
     session.close()
 
 
+def test_env_command():
+    session = BashSession(work_dir=os.getcwd())
+
+    # Test empty command without previous command
+    obs = session.execute(CmdRunAction('env'))
+    logger.info(obs, extra={'msg_type': 'OBSERVATION'})
+    assert 'PS1=###PS1JSON###' in obs.content
+    assert 'PS2=' in obs.content
+    assert obs.metadata.exit_code == 0
+    assert session.prev_status == BashCommandStatus.COMPLETED
+    session.close()
+
+
 def test_command_output_continuation():
     session = BashSession(work_dir=os.getcwd(), no_change_timeout_seconds=2)