Coverage for src / openenv / docker / runtime.py: 96.39%

1"""Runtime inspection helpers for running bot containers.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import subprocess 

7from dataclasses import dataclass, field 

8 

9from openenv.core.errors import CommandError 

10 

11 

12SNAPSHOT_SCRIPT_TEMPLATE = """\ 

13import json 

14from pathlib import Path 

15 

16skills_root = Path({workspace!r}) / "skills" 

17payload = [] 

18if skills_root.exists(): 

19 for skill_dir in sorted(path for path in skills_root.iterdir() if path.is_dir()): 

20 files = {{}} 

21 for file_path in sorted(path for path in skill_dir.rglob("*") if path.is_file()): 

22 relative_path = file_path.relative_to(skill_dir).as_posix() 

23 files[relative_path] = file_path.read_text(encoding="utf-8", errors="replace") 

24 payload.append({{"name": skill_dir.name, "files": files}}) 

25print(json.dumps(payload, sort_keys=True)) 

26""" 

27 

28 

29@dataclass(slots=True) 

30class CapturedSkill: 

31 """A skill snapshot collected from a running container.""" 

32 

33 name: str 

34 description: str 

35 content: str 

36 source: str | None = None 

37 assets: dict[str, str] = field(default_factory=dict) 

38 

39 

40def list_running_container_names() -> set[str]: 

41 """Return the names of running Docker containers.""" 

42 stdout = _run_command( 

43 ["docker", "ps", "--format", "{{.Names}}"], 

44 unavailable_message=( 

45 "Docker is not available on PATH. Install Docker or Docker Desktop " 

46 "before listing running bots." 

47 ), 

48 failure_message="Failed to list running Docker containers.", 

49 ) 

50 return {line.strip() for line in stdout.splitlines() if line.strip()} 

51 

52 

53def fetch_container_logs(container_name: str, *, tail: int = 120) -> str: 

54 """Return recent logs for a running container.""" 

55 return _run_command( 

56 ["docker", "logs", "--tail", str(tail), container_name], 

57 unavailable_message=( 

58 "Docker is not available on PATH. Install Docker or Docker Desktop " 

59 "before reading bot logs." 

60 ), 

61 failure_message=f"Failed to read logs for container `{container_name}`.", 

62 ) 

63 

64 

65def snapshot_installed_skills( 

66 container_name: str, 

67 *, 

68 workspace: str, 

69) -> list[CapturedSkill]: 

70 """Snapshot installed skills from a running bot container.""" 

71 stdout = _run_command( 

72 [ 

73 "docker", 

74 "exec", 

75 container_name, 

76 "python", 

77 "-c", 

78 SNAPSHOT_SCRIPT_TEMPLATE.format(workspace=workspace), 

79 ], 

80 unavailable_message=( 

81 "Docker is not available on PATH. Install Docker or Docker Desktop " 

82 "before creating a skill snapshot." 

83 ), 

84 failure_message=f"Failed to snapshot installed skills for `{container_name}`.", 

85 ) 

86 try: 

87 payload = json.loads(stdout.strip() or "[]") 

88 except json.JSONDecodeError as exc: 

89 raise CommandError( 

90 f"Container `{container_name}` returned an unreadable skill snapshot payload." 

91 ) from exc 

92 if not isinstance(payload, list): 

93 raise CommandError( 

94 f"Container `{container_name}` returned an invalid skill snapshot payload." 

95 ) 

96 

97 snapshots: list[CapturedSkill] = [] 

98 for item in payload: 

99 if not isinstance(item, dict): 

100 continue 

101 name = str(item.get("name", "")).strip() 

102 files = item.get("files", {}) 

103 if not name or not isinstance(files, dict): 

104 continue 

105 skill_md = files.get("SKILL.md") 

106 if not isinstance(skill_md, str) or not skill_md.strip(): 

107 continue 

108 assets = { 

109 path: content 

110 for path, content in sorted(files.items()) 

111 if path != "SKILL.md" and isinstance(path, str) and isinstance(content, str) 

112 } 

113 frontmatter = _parse_frontmatter(skill_md) 

114 snapshots.append( 

115 CapturedSkill( 

116 name=name, 

117 description=frontmatter.get( 

118 "description", 

119 f"Snapshotted skill from running container {container_name}", 

120 ), 

121 content=skill_md, 

122 source=frontmatter.get("source"), 

123 assets=assets, 

124 ) 

125 ) 

126 return snapshots 

127 

128 

129def _run_command( 

130 command: list[str], 

131 *, 

132 unavailable_message: str, 

133 failure_message: str, 

134) -> str: 

135 """Execute a Docker command and normalize transport and process failures.""" 

136 try: 

137 completed = subprocess.run( 

138 command, 

139 check=True, 

140 capture_output=True, 

141 text=True, 

142 ) 

143 except OSError as exc: 

144 raise CommandError(unavailable_message) from exc 

145 except subprocess.CalledProcessError as exc: 

146 stderr = (exc.stderr or "").strip() 

147 details = f" Docker said: {stderr}" if stderr else "" 

148 raise CommandError(f"{failure_message}{details}", exit_code=exc.returncode) from exc 

149 return completed.stdout 

150 

151 

152def _parse_frontmatter(content: str) -> dict[str, str]: 

153 """Extract a simple YAML-frontmatter key/value mapping from `SKILL.md` content.""" 

154 lines = content.splitlines() 

155 if not lines or lines[0].strip() != "---": 

156 return {} 

157 fields: dict[str, str] = {} 

158 for line in lines[1:]: 158 ↛ 166line 158 didn't jump to line 166 because the loop on line 158 didn't complete

159 stripped = line.strip() 

160 if stripped == "---": 

161 break 

162 if ":" not in line: 162 ↛ 163line 162 didn't jump to line 163 because the condition on line 162 was never true

163 continue 

164 key, value = line.split(":", 1) 

165 fields[key.strip().lower()] = value.strip() 

166 return fields