Coverage for little_loops / subprocess_utils.py: 17%

1"""Subprocess utilities for Claude CLI invocation. 

2 

3Provides shared functionality for running Claude CLI commands with 

4real-time output streaming, timeout handling, and context handoff detection. 

5""" 

6 

7from __future__ import annotations 

8 

9import json 

10import logging 

11import os 

12import re 

13import selectors 

14import subprocess 

15import time 

16from collections.abc import Callable 

17from pathlib import Path 

18 

19logger = logging.getLogger(__name__) 

20 

21# Callback type: (line: str, is_stderr: bool) -> None 

22OutputCallback = Callable[[str, bool], None] 

23 

24# Process lifecycle callback: (process: Popen) -> None 

25ProcessCallback = Callable[[subprocess.Popen[str]], None] 

26 

27# Model detection callback: (model: str) -> None 

28ModelCallback = Callable[[str], None] 

29 

30# Context handoff detection pattern 

31CONTEXT_HANDOFF_PATTERN = re.compile(r"CONTEXT_HANDOFF:\s*Ready for fresh session") 

32CONTINUATION_PROMPT_PATH = Path(".ll/ll-continue-prompt.md") 

33 

34 

35def detect_context_handoff(output: str) -> bool: 

36 """Check if output contains a context handoff signal. 

37 

38 Args: 

39 output: Command output to check 

40 

41 Returns: 

42 True if context handoff was signaled 

43 """ 

44 return bool(CONTEXT_HANDOFF_PATTERN.search(output)) 

45 

46 

47def read_continuation_prompt(repo_path: Path | None = None) -> str | None: 

48 """Read the continuation prompt file if it exists. 

49 

50 Args: 

51 repo_path: Optional repository root path 

52 

53 Returns: 

54 Contents of continuation prompt, or None if not found 

55 """ 

56 prompt_path = (repo_path or Path.cwd()) / CONTINUATION_PROMPT_PATH 

57 if prompt_path.exists(): 

58 return prompt_path.read_text() 

59 return None 

60 

61 

62def run_claude_command( 

63 command: str, 

64 timeout: int = 3600, 

65 working_dir: Path | None = None, 

66 stream_callback: OutputCallback | None = None, 

67 on_process_start: ProcessCallback | None = None, 

68 on_process_end: ProcessCallback | None = None, 

69 idle_timeout: int = 0, 

70 on_model_detected: ModelCallback | None = None, 

71 agent: str | None = None, 

72 tools: list[str] | None = None, 

73) -> subprocess.CompletedProcess[str]: 

74 """Invoke Claude CLI command with real-time output streaming. 

75 

76 Args: 

77 command: Command to pass to Claude CLI 

78 timeout: Timeout in seconds (0 for no timeout) 

79 working_dir: Optional working directory for the command 

80 stream_callback: Optional callback for streaming output lines. 

81 Called with (line, is_stderr) for each line of output. 

82 on_process_start: Optional callback invoked after process starts. 

83 Receives the Popen object for tracking/management. 

84 on_process_end: Optional callback invoked after process completes. 

85 Receives the Popen object. Called in finally block. 

86 idle_timeout: Kill process if no output for this many seconds (0 to disable). 

87 on_model_detected: Optional callback invoked with the model name from the 

88 stream-json system/init event. Called at most once per invocation. 

89 

90 Returns: 

91 CompletedProcess with stdout/stderr captured 

92 

93 Raises: 

94 subprocess.TimeoutExpired: If command exceeds timeout or idle timeout. 

95 When triggered by idle timeout, the output field is set to "idle_timeout". 

96 """ 

97 cmd_args = [ 

98 "claude", 

99 "--dangerously-skip-permissions", 

100 "--verbose", 

101 "--output-format", 

102 "stream-json", 

103 "-p", 

104 command, 

105 ] 

106 if agent: 

107 cmd_args += ["--agent", agent] 

108 if tools: 

109 cmd_args += ["--tools", ",".join(tools)] 

110 

111 # Set environment to keep Claude in the project working directory (BUG-007) 

112 # This helps prevent file writes from leaking to the main repo in worktrees 

113 env = os.environ.copy() 

114 env["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1" 

115 

116 if working_dir is not None: 

117 git_path = Path(working_dir) / ".git" 

118 if git_path.is_file(): # worktree: .git is a file, not a directory 

119 gitdir_ref = git_path.read_text().strip() 

120 if gitdir_ref.startswith("gitdir: "): 

121 actual_gitdir = gitdir_ref[8:].strip() 

122 # Resolve relative gitdir references to absolute paths 

123 resolved = (Path(working_dir) / actual_gitdir).resolve() 

124 env["GIT_DIR"] = str(resolved) 

125 env["GIT_WORK_TREE"] = str(working_dir) 

126 logger.debug("Worktree detected: GIT_DIR=%s", env["GIT_DIR"]) 

127 

128 process = subprocess.Popen( 

129 cmd_args, 

130 stdout=subprocess.PIPE, 

131 stderr=subprocess.PIPE, 

132 text=True, 

133 bufsize=1, # Line buffered 

134 cwd=working_dir, 

135 env=env, 

136 ) 

137 

138 if on_process_start: 

139 on_process_start(process) 

140 

141 stdout_lines: list[str] = [] 

142 stderr_lines: list[str] = [] 

143 

144 # Use selectors for non-blocking read from both streams 

145 with selectors.DefaultSelector() as sel: 

146 if process.stdout: 

147 sel.register(process.stdout, selectors.EVENT_READ) 

148 if process.stderr: 

149 sel.register(process.stderr, selectors.EVENT_READ) 

150 

151 start_time = time.time() 

152 last_output_time = start_time 

153 

154 try: 

155 while sel.get_map(): 

156 now = time.time() 

157 if timeout and (now - start_time) > timeout: 

158 process.kill() 

159 try: 

160 process.wait(timeout=10) 

161 except subprocess.TimeoutExpired: 

162 logger.warning( 

163 "Process %s did not terminate within 10s after kill", 

164 process.pid, 

165 ) 

166 raise subprocess.TimeoutExpired(cmd_args, timeout) 

167 

168 if idle_timeout and (now - last_output_time) > idle_timeout: 

169 process.kill() 

170 try: 

171 process.wait(timeout=10) 

172 except subprocess.TimeoutExpired: 

173 logger.warning( 

174 "Process %s did not terminate within 10s after kill", 

175 process.pid, 

176 ) 

177 raise subprocess.TimeoutExpired(cmd_args, idle_timeout, output="idle_timeout") 

178 

179 ready = sel.select(timeout=1.0) 

180 for key, _ in ready: 

181 line = key.fileobj.readline() # type: ignore[union-attr] 

182 if not line: 

183 sel.unregister(key.fileobj) 

184 continue 

185 

186 last_output_time = time.time() 

187 line = line.rstrip("\n") 

188 is_stderr = key.fileobj is process.stderr 

189 

190 if not is_stderr: 

191 try: 

192 event = json.loads(line) 

193 etype = event.get("type") 

194 if etype == "system" and event.get("subtype") == "init": 

195 if on_model_detected and "model" in event: 

196 on_model_detected(event["model"]) 

197 continue # don't add to stdout_lines 

198 elif etype == "assistant": 

199 msg = event.get("message", {}) 

200 text_parts = [ 

201 block["text"] 

202 for block in msg.get("content", []) 

203 if block.get("type") == "text" 

204 ] 

205 line = "".join(text_parts) 

206 if not line: 

207 continue 

208 else: 

209 continue # skip other event types (result, tool_use, etc.) 

210 except (json.JSONDecodeError, KeyError, TypeError): 

211 pass # non-JSON line: pass through as raw text 

212 

213 if is_stderr: 

214 stderr_lines.append(line) 

215 else: 

216 stdout_lines.append(line) 

217 

218 if stream_callback: 

219 stream_callback(line, is_stderr) 

220 

221 try: 

222 process.wait(timeout=30) 

223 except subprocess.TimeoutExpired: 

224 logger.warning( 

225 "Process %s did not exit within 30s after streams closed, killing", 

226 process.pid, 

227 ) 

228 process.kill() 

229 try: 

230 process.wait(timeout=10) 

231 except subprocess.TimeoutExpired: 

232 logger.warning( 

233 "Process %s did not terminate within 10s after kill", 

234 process.pid, 

235 ) 

236 finally: 

237 if on_process_end: 

238 on_process_end(process) 

239 

240 return subprocess.CompletedProcess( 

241 cmd_args, 

242 process.returncode if process.returncode is not None else -9, 

243 stdout="\n".join(stdout_lines), 

244 stderr="\n".join(stderr_lines), 

245 )