Coverage for little_loops / subprocess_utils.py: 15%
186 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-05-11 00:29 -0500
1"""Subprocess utilities for Claude CLI invocation.
3Provides shared functionality for running Claude CLI commands with
4real-time output streaming, timeout handling, and context handoff detection.
5"""
7from __future__ import annotations
9import json
10import logging
11import os
12import re
13import selectors
14import subprocess
15import time
16from collections.abc import Callable
17from pathlib import Path
19logger = logging.getLogger(__name__)
21# Callback type: (line: str, is_stderr: bool) -> None
22OutputCallback = Callable[[str, bool], None]
24# Process lifecycle callback: (process: Popen) -> None
25ProcessCallback = Callable[[subprocess.Popen[str]], None]
27# Model detection callback: (model: str) -> None
28ModelCallback = Callable[[str], None]
30# Usage callback: (input_tokens: int, output_tokens: int) -> None
31UsageCallback = Callable[[int, int], None]
33# Context handoff detection pattern
34CONTEXT_HANDOFF_PATTERN = re.compile(r"CONTEXT_HANDOFF:\s*Ready for fresh session")
35CONTINUATION_PROMPT_PATH = Path(".ll/ll-continue-prompt.md")
37# Sentinel file written when a session ends with high context usage (Option G).
38# Consumed by run_with_continuation; NOT deleted by session-cleanup.sh.
39SENTINEL_PATH = Path(".ll/ll-context-handoff-needed")
41# Chars of captured_stdout to include in Option J guillotine prompt (≈3K tokens).
42_GUILLOTINE_TAIL_CHARS = 12_000
43# Lines of original_command to include for task intent.
44_GUILLOTINE_MAX_TASK_LINES = 20
47def detect_context_handoff(output: str) -> bool:
48 """Check if output contains a context handoff signal.
50 Args:
51 output: Command output to check
53 Returns:
54 True if context handoff was signaled
55 """
56 return bool(CONTEXT_HANDOFF_PATTERN.search(output))
59def read_continuation_prompt(repo_path: Path | None = None) -> str | None:
60 """Read the continuation prompt file if it exists.
62 Args:
63 repo_path: Optional repository root path
65 Returns:
66 Contents of continuation prompt, or None if not found
67 """
68 prompt_path = (repo_path or Path.cwd()) / CONTINUATION_PROMPT_PATH
69 if prompt_path.exists():
70 return prompt_path.read_text()
71 return None
74def read_sentinel(repo_path: Path | None = None) -> dict | None:
75 """Read and consume the context-handoff sentinel file if it exists.
77 The sentinel is written by context-handoff-sentinel.sh (Stop hook) or
78 the Python layer in run_with_continuation when a session ends with high
79 context usage but no CONTEXT_HANDOFF signal.
81 Args:
82 repo_path: Optional repository root path
84 Returns:
85 Parsed sentinel dict, or None if not present
86 """
87 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
88 if not sentinel_path.exists():
89 return None
90 try:
91 data = json.loads(sentinel_path.read_text())
92 sentinel_path.unlink(missing_ok=True)
93 return data
94 except Exception:
95 sentinel_path.unlink(missing_ok=True)
96 return {}
99def write_sentinel(
100 repo_path: Path | None = None,
101 token_count: int = 0,
102 context_limit: int = 200_000,
103) -> None:
104 """Write the context-handoff sentinel file.
106 Args:
107 repo_path: Optional repository root path
108 token_count: Total tokens used in the session
109 context_limit: Context window size
110 """
111 import datetime
113 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
114 usage_percent = int(token_count * 100 / context_limit) if context_limit > 0 else 0
115 try:
116 sentinel_path.parent.mkdir(parents=True, exist_ok=True)
117 sentinel_path.write_text(
118 json.dumps(
119 {
120 "written_at": datetime.datetime.now(datetime.UTC).strftime(
121 "%Y-%m-%dT%H:%M:%SZ"
122 ),
123 "token_count": token_count,
124 "context_limit": context_limit,
125 "usage_percent": usage_percent,
126 }
127 )
128 )
129 except Exception:
130 pass
133def assemble_guillotine_prompt(
134 original_command: str,
135 captured_stdout: str,
136 token_stats: dict,
137) -> str:
138 """Assemble a fresh-session continuation prompt for Option J (parent-side guillotine).
140 Called when context > 90% or "Prompt is too long" is detected with no handoff.
141 The resulting prompt is passed to a BRAND-NEW claude -p session (not --resume),
142 so it starts with 0 tokens.
144 Args:
145 original_command: The original task command / skill invocation
146 captured_stdout: All Claude text output captured so far
147 token_stats: Dict with keys: input_tokens, output_tokens, context_limit,
148 trigger_reason (optional)
150 Returns:
151 Assembled continuation prompt string
152 """
153 task_lines = original_command.strip().splitlines()[:_GUILLOTINE_MAX_TASK_LINES]
154 task_excerpt = "\n".join(task_lines)
155 if len(original_command.strip().splitlines()) > _GUILLOTINE_MAX_TASK_LINES:
156 task_excerpt += f"\n... (truncated to {_GUILLOTINE_MAX_TASK_LINES} lines)"
158 stdout_tail = (captured_stdout or "")[-_GUILLOTINE_TAIL_CHARS:]
159 if not stdout_tail:
160 stdout_tail = "(no output captured before interruption)"
162 input_tokens = token_stats.get("input_tokens", 0)
163 output_tokens = token_stats.get("output_tokens", 0)
164 context_limit = token_stats.get("context_limit", 200_000)
165 trigger_reason = token_stats.get("trigger_reason", "context > 90%")
167 scratch_listing = _list_scratch_files()
169 return f"""\
170⚠ CONTEXT LIMIT REACHED — FRESH SESSION CONTINUATION
172The previous automation session exhausted its context window before completing.
173This fresh session (new context window, starts at 0 tokens) is continuing from
174that interrupted session.
176## Original Task
177{task_excerpt}
179## Session Progress at Interruption
180- Approximate tokens used: {input_tokens + output_tokens:,} / {context_limit:,}
181- Trigger reason: {trigger_reason}
183## Last Session Output (what was happening at interruption)
184{stdout_tail}
186## Scratch Pad Files Available
187{scratch_listing}
189## Instructions for This Session
1901. Do NOT restart from scratch — the previous session made progress (see above)
1912. Read the "Last Session Output" section to understand exactly where we were
1923. Check the scratch pad files before re-running expensive operations
1934. Continue implementation from the interruption point
1945. Complete normally: test, commit, close the issue as usual
195"""
198def _list_scratch_files() -> str:
199 """List files in .loops/tmp/scratch/ with sizes for the guillotine prompt."""
200 scratch_dir = Path(".loops/tmp/scratch")
201 if not scratch_dir.exists():
202 return "None"
203 try:
204 files = sorted(scratch_dir.iterdir())
205 if not files:
206 return "None"
207 lines = []
208 for f in files:
209 try:
210 size_kb = f.stat().st_size // 1024
211 lines.append(f" {f.name} ({size_kb}KB)")
212 except Exception:
213 lines.append(f" {f.name}")
214 return "\n".join(lines)
215 except Exception:
216 return "None"
219def run_claude_command(
220 command: str,
221 timeout: int = 3600,
222 working_dir: Path | None = None,
223 stream_callback: OutputCallback | None = None,
224 on_process_start: ProcessCallback | None = None,
225 on_process_end: ProcessCallback | None = None,
226 idle_timeout: int = 0,
227 on_model_detected: ModelCallback | None = None,
228 on_usage: UsageCallback | None = None,
229 agent: str | None = None,
230 tools: list[str] | None = None,
231 resume_session: bool = False,
232) -> subprocess.CompletedProcess[str]:
233 """Invoke Claude CLI command with real-time output streaming.
235 Args:
236 command: Command to pass to Claude CLI
237 timeout: Timeout in seconds (0 for no timeout)
238 working_dir: Optional working directory for the command
239 stream_callback: Optional callback for streaming output lines.
240 Called with (line, is_stderr) for each line of output.
241 on_process_start: Optional callback invoked after process starts.
242 Receives the Popen object for tracking/management.
243 on_process_end: Optional callback invoked after process completes.
244 Receives the Popen object. Called in finally block.
245 idle_timeout: Kill process if no output for this many seconds (0 to disable).
246 on_model_detected: Optional callback invoked with the model name from the
247 stream-json system/init event. Called at most once per invocation.
248 on_usage: Optional callback invoked with (input_tokens, output_tokens) from
249 the stream-json result event. input_tokens includes cache_read_input_tokens.
250 resume_session: If True, passes --continue to the Claude CLI to continue the
251 most recent conversation. Used for the Option E explicit-handoff path.
253 Returns:
254 CompletedProcess with stdout/stderr captured
256 Raises:
257 subprocess.TimeoutExpired: If command exceeds timeout or idle timeout.
258 When triggered by idle timeout, the output field is set to "idle_timeout".
259 """
260 cmd_args = [
261 "claude",
262 "--dangerously-skip-permissions",
263 "--verbose",
264 "--output-format",
265 "stream-json",
266 ]
267 if resume_session:
268 cmd_args.append("--continue")
269 cmd_args += ["-p", command]
270 if agent:
271 cmd_args += ["--agent", agent]
272 if tools:
273 cmd_args += ["--tools", ",".join(tools)]
275 # Set environment to keep Claude in the project working directory (BUG-007)
276 # This helps prevent file writes from leaking to the main repo in worktrees
277 env = os.environ.copy()
278 env["CLAUDE_BASH_MAINTAIN_PROJECT_WORKING_DIR"] = "1"
280 if working_dir is not None:
281 git_path = Path(working_dir) / ".git"
282 if git_path.is_file(): # worktree: .git is a file, not a directory
283 gitdir_ref = git_path.read_text().strip()
284 if gitdir_ref.startswith("gitdir: "):
285 actual_gitdir = gitdir_ref[8:].strip()
286 # Resolve relative gitdir references to absolute paths
287 resolved = (Path(working_dir) / actual_gitdir).resolve()
288 env["GIT_DIR"] = str(resolved)
289 env["GIT_WORK_TREE"] = str(working_dir)
290 logger.debug("Worktree detected: GIT_DIR=%s", env["GIT_DIR"])
292 process = subprocess.Popen(
293 cmd_args,
294 stdout=subprocess.PIPE,
295 stderr=subprocess.PIPE,
296 text=True,
297 bufsize=1, # Line buffered
298 cwd=working_dir,
299 env=env,
300 )
302 if on_process_start:
303 on_process_start(process)
305 stdout_lines: list[str] = []
306 stderr_lines: list[str] = []
308 # Use selectors for non-blocking read from both streams
309 with selectors.DefaultSelector() as sel:
310 if process.stdout:
311 sel.register(process.stdout, selectors.EVENT_READ)
312 if process.stderr:
313 sel.register(process.stderr, selectors.EVENT_READ)
315 start_time = time.time()
316 last_output_time = start_time
318 try:
319 while sel.get_map():
320 now = time.time()
321 if timeout and (now - start_time) > timeout:
322 process.kill()
323 try:
324 process.wait(timeout=10)
325 except subprocess.TimeoutExpired:
326 logger.warning(
327 "Process %s did not terminate within 10s after kill",
328 process.pid,
329 )
330 raise subprocess.TimeoutExpired(cmd_args, timeout)
332 if idle_timeout and (now - last_output_time) > idle_timeout:
333 process.kill()
334 try:
335 process.wait(timeout=10)
336 except subprocess.TimeoutExpired:
337 logger.warning(
338 "Process %s did not terminate within 10s after kill",
339 process.pid,
340 )
341 raise subprocess.TimeoutExpired(cmd_args, idle_timeout, output="idle_timeout")
343 ready = sel.select(timeout=1.0)
344 for key, _ in ready:
345 line = key.fileobj.readline() # type: ignore[union-attr]
346 if not line:
347 sel.unregister(key.fileobj)
348 continue
350 last_output_time = time.time()
351 line = line.rstrip("\n")
352 is_stderr = key.fileobj is process.stderr
354 if not is_stderr:
355 try:
356 event = json.loads(line)
357 etype = event.get("type")
358 if etype == "system" and event.get("subtype") == "init":
359 if on_model_detected and "model" in event:
360 on_model_detected(event["model"])
361 continue # don't add to stdout_lines
362 elif etype == "assistant":
363 msg = event.get("message", {})
364 text_parts = [
365 block["text"]
366 for block in msg.get("content", [])
367 if block.get("type") == "text"
368 ]
369 text = "\n\n".join(text_parts)
370 if not text:
371 continue
372 for sub_line in text.splitlines() or [""]:
373 stdout_lines.append(sub_line)
374 if stream_callback:
375 stream_callback(sub_line, is_stderr)
376 continue
377 elif etype == "result":
378 usage = event.get("usage", {})
379 if on_usage and usage:
380 on_usage(
381 usage.get("input_tokens", 0)
382 + usage.get("cache_read_input_tokens", 0),
383 usage.get("output_tokens", 0),
384 )
385 if event.get("is_error"):
386 error_text = event.get("error") or event.get("result", "")
387 if error_text:
388 stderr_lines.append(f"[result] {error_text}")
389 continue # skip other event types (tool_use, etc.)
390 else:
391 continue # skip other event types (tool_use, etc.)
392 except (json.JSONDecodeError, KeyError, TypeError):
393 pass # non-JSON line: pass through as raw text
395 if is_stderr:
396 stderr_lines.append(line)
397 else:
398 stdout_lines.append(line)
400 if stream_callback:
401 stream_callback(line, is_stderr)
403 try:
404 process.wait(timeout=30)
405 except subprocess.TimeoutExpired:
406 logger.warning(
407 "Process %s did not exit within 30s after streams closed, killing",
408 process.pid,
409 )
410 process.kill()
411 try:
412 process.wait(timeout=10)
413 except subprocess.TimeoutExpired:
414 logger.warning(
415 "Process %s did not terminate within 10s after kill",
416 process.pid,
417 )
418 finally:
419 if on_process_end:
420 on_process_end(process)
422 return subprocess.CompletedProcess(
423 cmd_args,
424 process.returncode if process.returncode is not None else -9,
425 stdout="\n".join(stdout_lines),
426 stderr="\n".join(stderr_lines),
427 )