Coverage for little_loops / subprocess_utils.py: 20%
188 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-04 12:21 -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 dataclasses import dataclass
18from pathlib import Path
20from little_loops.host_runner import resolve_host
22logger = logging.getLogger(__name__)
24# Callback type: (line: str, is_stderr: bool) -> None
25OutputCallback = Callable[[str, bool], None]
27# Process lifecycle callback: (process: Popen) -> None
28ProcessCallback = Callable[[subprocess.Popen[str]], None]
30# Model detection callback: (model: str) -> None
31ModelCallback = Callable[[str], None]
33# Usage callback: (input_tokens: int, output_tokens: int) -> None
34# Kept for back-compat with issue_manager.py and worker_pool.py callers.
35UsageCallback = Callable[[int, int], None]
38@dataclass
39class TokenUsage:
40 """Token usage from a single host-CLI invocation."""
42 input_tokens: int
43 output_tokens: int
44 cache_read_tokens: int
45 cache_creation_tokens: int
46 model: str
49# Detailed usage callback — receives all four token fields plus model ID.
50DetailedUsageCallback = Callable[[TokenUsage], None]
52# Context handoff detection pattern
53CONTEXT_HANDOFF_PATTERN = re.compile(r"CONTEXT_HANDOFF:\s*Ready for fresh session")
54CONTINUATION_PROMPT_PATH = Path(".ll/ll-continue-prompt.md")
56# Sentinel file written when a session ends with high context usage (Option G).
57# Consumed by run_with_continuation; NOT deleted by session-cleanup.sh.
58SENTINEL_PATH = Path(".ll/ll-context-handoff-needed")
60# Chars of captured_stdout to include in Option J guillotine prompt (≈3K tokens).
61_GUILLOTINE_TAIL_CHARS = 12_000
62# Lines of original_command to include for task intent.
63_GUILLOTINE_MAX_TASK_LINES = 20
66def detect_context_handoff(output: str) -> bool:
67 """Check if output contains a context handoff signal.
69 Args:
70 output: Command output to check
72 Returns:
73 True if context handoff was signaled
74 """
75 return bool(CONTEXT_HANDOFF_PATTERN.search(output))
78def read_continuation_prompt(repo_path: Path | None = None) -> str | None:
79 """Read the continuation prompt file if it exists.
81 Args:
82 repo_path: Optional repository root path
84 Returns:
85 Contents of continuation prompt, or None if not found
86 """
87 prompt_path = (repo_path or Path.cwd()) / CONTINUATION_PROMPT_PATH
88 if prompt_path.exists():
89 return prompt_path.read_text()
90 return None
93def read_sentinel(repo_path: Path | None = None) -> dict | None:
94 """Read and consume the context-handoff sentinel file if it exists.
96 The sentinel is written by context-handoff-sentinel.sh (Stop hook) or
97 the Python layer in run_with_continuation when a session ends with high
98 context usage but no CONTEXT_HANDOFF signal.
100 Args:
101 repo_path: Optional repository root path
103 Returns:
104 Parsed sentinel dict, or None if not present
105 """
106 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
107 if not sentinel_path.exists():
108 return None
109 try:
110 data = json.loads(sentinel_path.read_text())
111 sentinel_path.unlink(missing_ok=True)
112 return data
113 except Exception:
114 sentinel_path.unlink(missing_ok=True)
115 return {}
118def write_sentinel(
119 repo_path: Path | None = None,
120 token_count: int = 0,
121 context_limit: int = 200_000,
122) -> None:
123 """Write the context-handoff sentinel file.
125 Args:
126 repo_path: Optional repository root path
127 token_count: Total tokens used in the session
128 context_limit: Context window size
129 """
130 import datetime
132 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
133 usage_percent = int(token_count * 100 / context_limit) if context_limit > 0 else 0
134 try:
135 sentinel_path.parent.mkdir(parents=True, exist_ok=True)
136 sentinel_path.write_text(
137 json.dumps(
138 {
139 "written_at": datetime.datetime.now(datetime.UTC).strftime(
140 "%Y-%m-%dT%H:%M:%SZ"
141 ),
142 "token_count": token_count,
143 "context_limit": context_limit,
144 "usage_percent": usage_percent,
145 }
146 )
147 )
148 except Exception:
149 pass
152def assemble_guillotine_prompt(
153 original_command: str,
154 captured_stdout: str,
155 token_stats: dict,
156) -> str:
157 """Assemble a fresh-session continuation prompt for Option J (parent-side guillotine).
159 Called when context > 90% or "Prompt is too long" is detected with no handoff.
160 The resulting prompt is passed to a BRAND-NEW claude -p session (not --resume),
161 so it starts with 0 tokens.
163 Args:
164 original_command: The original task command / skill invocation
165 captured_stdout: All Claude text output captured so far
166 token_stats: Dict with keys: input_tokens, output_tokens, context_limit,
167 trigger_reason (optional)
169 Returns:
170 Assembled continuation prompt string
171 """
172 task_lines = original_command.strip().splitlines()[:_GUILLOTINE_MAX_TASK_LINES]
173 task_excerpt = "\n".join(task_lines)
174 if len(original_command.strip().splitlines()) > _GUILLOTINE_MAX_TASK_LINES:
175 task_excerpt += f"\n... (truncated to {_GUILLOTINE_MAX_TASK_LINES} lines)"
177 stdout_tail = (captured_stdout or "")[-_GUILLOTINE_TAIL_CHARS:]
178 if not stdout_tail:
179 stdout_tail = "(no output captured before interruption)"
181 input_tokens = token_stats.get("input_tokens", 0)
182 output_tokens = token_stats.get("output_tokens", 0)
183 context_limit = token_stats.get("context_limit", 200_000)
184 trigger_reason = token_stats.get("trigger_reason", "context > 90%")
186 scratch_listing = _list_scratch_files()
188 return f"""\
189⚠ CONTEXT LIMIT REACHED — FRESH SESSION CONTINUATION
191The previous automation session exhausted its context window before completing.
192This fresh session (new context window, starts at 0 tokens) is continuing from
193that interrupted session.
195## Original Task
196{task_excerpt}
198## Session Progress at Interruption
199- Approximate tokens used: {input_tokens + output_tokens:,} / {context_limit:,}
200- Trigger reason: {trigger_reason}
202## Last Session Output (what was happening at interruption)
203{stdout_tail}
205## Scratch Pad Files Available
206{scratch_listing}
208## Instructions for This Session
2091. Do NOT restart from scratch — the previous session made progress (see above)
2102. Read the "Last Session Output" section to understand exactly where we were
2113. Check the scratch pad files before re-running expensive operations
2124. Continue implementation from the interruption point
2135. Complete normally: test, commit, close the issue as usual
214"""
217def _list_scratch_files() -> str:
218 """List files in .loops/tmp/scratch/ with sizes for the guillotine prompt."""
219 scratch_dir = Path(".loops/tmp/scratch")
220 if not scratch_dir.exists():
221 return "None"
222 try:
223 files = sorted(scratch_dir.iterdir())
224 if not files:
225 return "None"
226 lines = []
227 for f in files:
228 try:
229 size_kb = f.stat().st_size // 1024
230 lines.append(f" {f.name} ({size_kb}KB)")
231 except Exception:
232 lines.append(f" {f.name}")
233 return "\n".join(lines)
234 except Exception:
235 return "None"
238def run_claude_command(
239 command: str,
240 timeout: int = 3600,
241 working_dir: Path | None = None,
242 stream_callback: OutputCallback | None = None,
243 on_process_start: ProcessCallback | None = None,
244 on_process_end: ProcessCallback | None = None,
245 idle_timeout: int = 0,
246 on_model_detected: ModelCallback | None = None,
247 on_usage: UsageCallback | None = None,
248 on_usage_detailed: DetailedUsageCallback | None = None,
249 agent: str | None = None,
250 tools: list[str] | None = None,
251 resume_session: bool = False,
252) -> subprocess.CompletedProcess[str]:
253 """Invoke Claude CLI command with real-time output streaming.
255 Args:
256 command: Command to pass to Claude CLI
257 timeout: Timeout in seconds (0 for no timeout)
258 working_dir: Optional working directory for the command
259 stream_callback: Optional callback for streaming output lines.
260 Called with (line, is_stderr) for each line of output.
261 on_process_start: Optional callback invoked after process starts.
262 Receives the Popen object for tracking/management.
263 on_process_end: Optional callback invoked after process completes.
264 Receives the Popen object. Called in finally block.
265 idle_timeout: Kill process if no output for this many seconds (0 to disable).
266 on_model_detected: Optional callback invoked with the model name from the
267 stream-json system/init event. Called at most once per invocation.
268 on_usage: Optional callback invoked with (input_tokens, output_tokens) from
269 the stream-json result event. input_tokens includes cache_read_input_tokens.
270 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass
271 carrying all four token fields (input, output, cache_read, cache_creation)
272 plus the model ID from the stream-json result event.
273 resume_session: If True, passes --continue to the Claude CLI to continue the
274 most recent conversation. Used for the Option E explicit-handoff path.
276 Returns:
277 CompletedProcess with stdout/stderr captured
279 Raises:
280 subprocess.TimeoutExpired: If command exceeds timeout or idle timeout.
281 When triggered by idle timeout, the output field is set to "idle_timeout".
282 """
283 runner = resolve_host()
284 invocation = runner.build_streaming(
285 prompt=command,
286 working_dir=working_dir,
287 resume=resume_session,
288 agent=agent,
289 tools=tools,
290 )
291 cmd_args = [invocation.binary, *invocation.args]
293 env = os.environ.copy()
294 env.update(invocation.env)
295 if "GIT_DIR" in invocation.env:
296 logger.debug("Worktree detected: GIT_DIR=%s", invocation.env["GIT_DIR"])
298 process = subprocess.Popen(
299 cmd_args,
300 stdout=subprocess.PIPE,
301 stderr=subprocess.PIPE,
302 text=True,
303 bufsize=1, # Line buffered
304 cwd=working_dir,
305 env=env,
306 )
308 if on_process_start:
309 on_process_start(process)
311 stdout_lines: list[str] = []
312 stderr_lines: list[str] = []
313 detected_model: str = "unknown"
315 # Use selectors for non-blocking read from both streams
316 with selectors.DefaultSelector() as sel:
317 if process.stdout:
318 sel.register(process.stdout, selectors.EVENT_READ)
319 if process.stderr:
320 sel.register(process.stderr, selectors.EVENT_READ)
322 start_time = time.time()
323 last_output_time = start_time
325 try:
326 while sel.get_map():
327 now = time.time()
328 if timeout and (now - start_time) > timeout:
329 process.kill()
330 try:
331 process.wait(timeout=10)
332 except subprocess.TimeoutExpired:
333 logger.warning(
334 "Process %s did not terminate within 10s after kill",
335 process.pid,
336 )
337 raise subprocess.TimeoutExpired(cmd_args, timeout)
339 if idle_timeout and (now - last_output_time) > idle_timeout:
340 process.kill()
341 try:
342 process.wait(timeout=10)
343 except subprocess.TimeoutExpired:
344 logger.warning(
345 "Process %s did not terminate within 10s after kill",
346 process.pid,
347 )
348 raise subprocess.TimeoutExpired(cmd_args, idle_timeout, output="idle_timeout")
350 ready = sel.select(timeout=1.0)
351 for key, _ in ready:
352 line = key.fileobj.readline() # type: ignore[union-attr]
353 if not line:
354 sel.unregister(key.fileobj)
355 continue
357 last_output_time = time.time()
358 line = line.rstrip("\n")
359 is_stderr = key.fileobj is process.stderr
361 if not is_stderr:
362 try:
363 event = json.loads(line)
364 etype = event.get("type")
365 if etype == "system" and event.get("subtype") == "init":
366 if "model" in event:
367 detected_model = event["model"]
368 if on_model_detected:
369 on_model_detected(event["model"])
370 continue # don't add to stdout_lines
371 elif etype == "assistant":
372 msg = event.get("message", {})
373 text_parts = [
374 block["text"]
375 for block in msg.get("content", [])
376 if block.get("type") == "text"
377 ]
378 text = "\n\n".join(text_parts)
379 if not text:
380 continue
381 for sub_line in text.splitlines() or [""]:
382 stdout_lines.append(sub_line)
383 if stream_callback:
384 stream_callback(sub_line, is_stderr)
385 continue
386 elif etype == "result":
387 usage = event.get("usage", {})
388 if on_usage and usage:
389 on_usage(
390 usage.get("input_tokens", 0)
391 + usage.get("cache_read_input_tokens", 0),
392 usage.get("output_tokens", 0),
393 )
394 if on_usage_detailed and usage:
395 on_usage_detailed(
396 TokenUsage(
397 input_tokens=usage.get("input_tokens", 0),
398 output_tokens=usage.get("output_tokens", 0),
399 cache_read_tokens=usage.get(
400 "cache_read_input_tokens", 0
401 ),
402 cache_creation_tokens=usage.get(
403 "cache_creation_input_tokens", 0
404 ),
405 model=event.get("model", detected_model),
406 )
407 )
408 if event.get("is_error"):
409 error_text = event.get("error") or event.get("result", "")
410 if error_text:
411 stderr_lines.append(f"[result] {error_text}")
412 continue # skip other event types (tool_use, etc.)
413 else:
414 continue # skip other event types (tool_use, etc.)
415 except (json.JSONDecodeError, KeyError, TypeError):
416 pass # non-JSON line: pass through as raw text
418 if is_stderr:
419 stderr_lines.append(line)
420 else:
421 stdout_lines.append(line)
423 if stream_callback:
424 stream_callback(line, is_stderr)
426 try:
427 process.wait(timeout=30)
428 except subprocess.TimeoutExpired:
429 logger.warning(
430 "Process %s did not exit within 30s after streams closed, killing",
431 process.pid,
432 )
433 process.kill()
434 try:
435 process.wait(timeout=10)
436 except subprocess.TimeoutExpired:
437 logger.warning(
438 "Process %s did not terminate within 10s after kill",
439 process.pid,
440 )
441 finally:
442 if on_process_end:
443 on_process_end(process)
445 return subprocess.CompletedProcess(
446 cmd_args,
447 process.returncode if process.returncode is not None else -9,
448 stdout="\n".join(stdout_lines),
449 stderr="\n".join(stderr_lines),
450 )