Coverage for src / osiris_cli / shell_session.py: 0%

1#!/usr/bin/env python3 

2""" 

3Persistent shell session used by the run_shell_command tool. 

4 

5Letta Code keeps a running shell so commands can build on previous state. 

6This module provides a lightweight approximation by tracking the working 

7directory and executing commands within it. 

8""" 

9 

10from __future__ import annotations 

11 

12import os 

13import subprocess 

14from dataclasses import dataclass, field 

15from pathlib import Path 

16from typing import Any, Dict, Optional 

17 

18 

19@dataclass 

20class PersistentShell: 

21 cwd: Path = field(default_factory=Path.cwd) 

22 env: Dict[str, str] = field(default_factory=lambda: os.environ.copy()) 

23 default_timeout: int = 120 

24 

25 def run(self, command: str, timeout: Optional[int] = None) -> Dict[str, Any]: 

26 command = command.strip() 

27 if not command: 

28 return { 

29 "command": command, 

30 "stdout": "", 

31 "stderr": "No command provided.", 

32 "exit_code": 1, 

33 "status": "error", 

34 } 

35 timeout = timeout or self.default_timeout 

36 

37 if command.startswith("cd "): 

38 return self._handle_cd(command) 

39 

40 proc = subprocess.run( 

41 command, 

42 shell=True, 

43 cwd=str(self.cwd), 

44 env=self.env, 

45 capture_output=True, 

46 text=True, 

47 timeout=timeout, 

48 ) 

49 stdout = proc.stdout.strip() 

50 stderr = proc.stderr.strip() 

51 status = "success" if proc.returncode == 0 else "error" 

52 return { 

53 "command": command, 

54 "stdout": stdout, 

55 "stderr": stderr, 

56 "exit_code": proc.returncode, 

57 "status": status, 

58 } 

59 

60 def _handle_cd(self, command: str) -> Dict[str, Any]: 

61 _, _, target = command.partition(" ") 

62 target = target.strip() or "~" 

63 expanded = os.path.expanduser(target) 

64 new_path = (self.cwd / expanded).resolve() if not os.path.isabs(expanded) else Path(expanded).resolve() 

65 

66 if not new_path.exists() or not new_path.is_dir(): 

67 return { 

68 "command": command, 

69 "stdout": "", 

70 "stderr": f"Directory not found: {target}", 

71 "exit_code": 1, 

72 "status": "error", 

73 } 

74 

75 self.cwd = new_path 

76 return { 

77 "command": command, 

78 "stdout": f"Changed directory to {self.cwd}", 

79 "stderr": "", 

80 "exit_code": 0, 

81 "status": "success", 

82 } 

83 

84 

85persistent_shell = PersistentShell()