Coverage for little_loops / subprocess_utils.py: 20%
198 statements
« prev ^ index » next coverage.py v7.12.0, created at 2026-06-08 15:34 -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 signal
15import subprocess
16import time
17from collections.abc import Callable
18from dataclasses import dataclass
19from pathlib import Path
21from little_loops.host_runner import resolve_host
23logger = logging.getLogger(__name__)
25# Callback type: (line: str, is_stderr: bool) -> None
26OutputCallback = Callable[[str, bool], None]
28# Process lifecycle callback: (process: Popen) -> None
29ProcessCallback = Callable[[subprocess.Popen[str]], None]
31# Model detection callback: (model: str) -> None
32ModelCallback = Callable[[str], None]
34# Usage callback: (input_tokens: int, output_tokens: int) -> None
35# Kept for back-compat with issue_manager.py and worker_pool.py callers.
36UsageCallback = Callable[[int, int], None]
39@dataclass
40class TokenUsage:
41 """Token usage from a single host-CLI invocation."""
43 input_tokens: int
44 output_tokens: int
45 cache_read_tokens: int
46 cache_creation_tokens: int
47 model: str
50# Detailed usage callback — receives all four token fields plus model ID.
51DetailedUsageCallback = Callable[[TokenUsage], None]
53# Context handoff detection pattern
54CONTEXT_HANDOFF_PATTERN = re.compile(r"CONTEXT_HANDOFF:\s*Ready for fresh session")
55CONTINUATION_PROMPT_PATH = Path(".ll/ll-continue-prompt.md")
57# Sentinel file written when a session ends with high context usage (Option G).
58# Consumed by run_with_continuation; NOT deleted by session-cleanup.sh.
59SENTINEL_PATH = Path(".ll/ll-context-handoff-needed")
61# Chars of captured_stdout to include in Option J guillotine prompt (≈3K tokens).
62_GUILLOTINE_TAIL_CHARS = 12_000
63# Lines of original_command to include for task intent.
64_GUILLOTINE_MAX_TASK_LINES = 20
67def detect_context_handoff(output: str) -> bool:
68 """Check if output contains a context handoff signal.
70 Args:
71 output: Command output to check
73 Returns:
74 True if context handoff was signaled
75 """
76 return bool(CONTEXT_HANDOFF_PATTERN.search(output))
79def read_continuation_prompt(repo_path: Path | None = None) -> str | None:
80 """Read the continuation prompt file if it exists.
82 Args:
83 repo_path: Optional repository root path
85 Returns:
86 Contents of continuation prompt, or None if not found
87 """
88 prompt_path = (repo_path or Path.cwd()) / CONTINUATION_PROMPT_PATH
89 if prompt_path.exists():
90 return prompt_path.read_text()
91 return None
94def read_sentinel(repo_path: Path | None = None) -> dict | None:
95 """Read and consume the context-handoff sentinel file if it exists.
97 The sentinel is written by context-handoff-sentinel.sh (Stop hook) or
98 the Python layer in run_with_continuation when a session ends with high
99 context usage but no CONTEXT_HANDOFF signal.
101 Args:
102 repo_path: Optional repository root path
104 Returns:
105 Parsed sentinel dict, or None if not present
106 """
107 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
108 if not sentinel_path.exists():
109 return None
110 try:
111 data = json.loads(sentinel_path.read_text())
112 sentinel_path.unlink(missing_ok=True)
113 return data
114 except Exception:
115 sentinel_path.unlink(missing_ok=True)
116 return {}
119def write_sentinel(
120 repo_path: Path | None = None,
121 token_count: int = 0,
122 context_limit: int = 200_000,
123) -> None:
124 """Write the context-handoff sentinel file.
126 Args:
127 repo_path: Optional repository root path
128 token_count: Total tokens used in the session
129 context_limit: Context window size
130 """
131 import datetime
133 sentinel_path = (repo_path or Path.cwd()) / SENTINEL_PATH
134 usage_percent = int(token_count * 100 / context_limit) if context_limit > 0 else 0
135 try:
136 sentinel_path.parent.mkdir(parents=True, exist_ok=True)
137 sentinel_path.write_text(
138 json.dumps(
139 {
140 "written_at": datetime.datetime.now(datetime.UTC).strftime(
141 "%Y-%m-%dT%H:%M:%SZ"
142 ),
143 "token_count": token_count,
144 "context_limit": context_limit,
145 "usage_percent": usage_percent,
146 }
147 )
148 )
149 except Exception:
150 pass
153def assemble_guillotine_prompt(
154 original_command: str,
155 captured_stdout: str,
156 token_stats: dict,
157) -> str:
158 """Assemble a fresh-session continuation prompt for Option J (parent-side guillotine).
160 Called when context > 90% or "Prompt is too long" is detected with no handoff.
161 The resulting prompt is passed to a BRAND-NEW claude -p session (not --resume),
162 so it starts with 0 tokens.
164 Args:
165 original_command: The original task command / skill invocation
166 captured_stdout: All Claude text output captured so far
167 token_stats: Dict with keys: input_tokens, output_tokens, context_limit,
168 trigger_reason (optional)
170 Returns:
171 Assembled continuation prompt string
172 """
173 task_lines = original_command.strip().splitlines()[:_GUILLOTINE_MAX_TASK_LINES]
174 task_excerpt = "\n".join(task_lines)
175 if len(original_command.strip().splitlines()) > _GUILLOTINE_MAX_TASK_LINES:
176 task_excerpt += f"\n... (truncated to {_GUILLOTINE_MAX_TASK_LINES} lines)"
178 stdout_tail = (captured_stdout or "")[-_GUILLOTINE_TAIL_CHARS:]
179 if not stdout_tail:
180 stdout_tail = "(no output captured before interruption)"
182 input_tokens = token_stats.get("input_tokens", 0)
183 output_tokens = token_stats.get("output_tokens", 0)
184 context_limit = token_stats.get("context_limit", 200_000)
185 trigger_reason = token_stats.get("trigger_reason", "context > 90%")
187 scratch_listing = _list_scratch_files()
189 return f"""\
190⚠ CONTEXT LIMIT REACHED — FRESH SESSION CONTINUATION
192The previous automation session exhausted its context window before completing.
193This fresh session (new context window, starts at 0 tokens) is continuing from
194that interrupted session.
196## Original Task
197{task_excerpt}
199## Session Progress at Interruption
200- Approximate tokens used: {input_tokens + output_tokens:,} / {context_limit:,}
201- Trigger reason: {trigger_reason}
203## Last Session Output (what was happening at interruption)
204{stdout_tail}
206## Scratch Pad Files Available
207{scratch_listing}
209## Instructions for This Session
2101. Do NOT restart from scratch — the previous session made progress (see above)
2112. Read the "Last Session Output" section to understand exactly where we were
2123. Check the scratch pad files before re-running expensive operations
2134. Continue implementation from the interruption point
2145. Complete normally: test, commit, close the issue as usual
215"""
218def _list_scratch_files() -> str:
219 """List files in .loops/tmp/scratch/ with sizes for the guillotine prompt."""
220 scratch_dir = Path(".loops/tmp/scratch")
221 if not scratch_dir.exists():
222 return "None"
223 try:
224 files = sorted(scratch_dir.iterdir())
225 if not files:
226 return "None"
227 lines = []
228 for f in files:
229 try:
230 size_kb = f.stat().st_size // 1024
231 lines.append(f" {f.name} ({size_kb}KB)")
232 except Exception:
233 lines.append(f" {f.name}")
234 return "\n".join(lines)
235 except Exception:
236 return "None"
239def _kill_process_group(process: subprocess.Popen) -> None:
240 """Send SIGKILL to the process group; fall back to single-PID kill on error.
242 Uses os.getpgid / os.killpg (POSIX) so background Workflow/Task children
243 launched by the session are reaped together with the main process.
244 AttributeError catches platforms where os.killpg is absent (Windows).
245 """
246 try:
247 os.killpg(os.getpgid(process.pid), signal.SIGKILL)
248 except (ProcessLookupError, PermissionError, AttributeError):
249 process.kill()
252def run_claude_command(
253 command: str,
254 timeout: int = 3600,
255 working_dir: Path | None = None,
256 stream_callback: OutputCallback | None = None,
257 on_process_start: ProcessCallback | None = None,
258 on_process_end: ProcessCallback | None = None,
259 idle_timeout: int = 0,
260 on_model_detected: ModelCallback | None = None,
261 on_usage: UsageCallback | None = None,
262 on_usage_detailed: DetailedUsageCallback | None = None,
263 agent: str | None = None,
264 tools: list[str] | None = None,
265 resume_session: bool = False,
266) -> subprocess.CompletedProcess[str]:
267 """Invoke Claude CLI command with real-time output streaming.
269 Args:
270 command: Command to pass to Claude CLI
271 timeout: Timeout in seconds (0 for no timeout)
272 working_dir: Optional working directory for the command
273 stream_callback: Optional callback for streaming output lines.
274 Called with (line, is_stderr) for each line of output.
275 on_process_start: Optional callback invoked after process starts.
276 Receives the Popen object for tracking/management.
277 on_process_end: Optional callback invoked after process completes.
278 Receives the Popen object. Called in finally block.
279 idle_timeout: Kill process if no output for this many seconds (0 to disable).
280 on_model_detected: Optional callback invoked with the model name from the
281 stream-json system/init event. Called at most once per invocation.
282 on_usage: Optional callback invoked with (input_tokens, output_tokens) from
283 the stream-json result event. input_tokens includes cache_read_input_tokens.
284 on_usage_detailed: Optional callback invoked with a TokenUsage dataclass
285 carrying all four token fields (input, output, cache_read, cache_creation)
286 plus the model ID from the stream-json result event.
287 resume_session: If True, passes --continue to the Claude CLI to continue the
288 most recent conversation. Used for the Option E explicit-handoff path.
290 Returns:
291 CompletedProcess with stdout/stderr captured
293 Raises:
294 subprocess.TimeoutExpired: If command exceeds timeout or idle timeout.
295 When triggered by idle timeout, the output field is set to "idle_timeout".
296 """
297 runner = resolve_host()
298 invocation = runner.build_streaming(
299 prompt=command,
300 working_dir=working_dir,
301 resume=resume_session,
302 agent=agent,
303 tools=tools,
304 )
305 cmd_args = [invocation.binary, *invocation.args]
307 env = os.environ.copy()
308 env.update(invocation.env)
309 if "GIT_DIR" in invocation.env:
310 logger.debug("Worktree detected: GIT_DIR=%s", invocation.env["GIT_DIR"])
312 process = subprocess.Popen(
313 cmd_args,
314 stdout=subprocess.PIPE,
315 stderr=subprocess.PIPE,
316 text=True,
317 bufsize=1, # Line buffered
318 cwd=working_dir,
319 env=env,
320 start_new_session=True,
321 )
323 if on_process_start:
324 on_process_start(process)
326 stdout_lines: list[str] = []
327 stderr_lines: list[str] = []
328 detected_model: str = "unknown"
330 # Use selectors for non-blocking read from both streams
331 with selectors.DefaultSelector() as sel:
332 if process.stdout:
333 sel.register(process.stdout, selectors.EVENT_READ)
334 if process.stderr:
335 sel.register(process.stderr, selectors.EVENT_READ)
337 start_time = time.time()
338 last_output_time = start_time
339 # End-of-turn detection: the stream-json "result" event is the canonical
340 # signal that the headless `claude -p` session is done. We break on it
341 # instead of waiting for pipe EOF, because background Workflow/Task child
342 # processes inherit the stdout/stderr write-ends and a pipe only reports
343 # EOF when the *last* writer closes it — so EOF may never arrive even
344 # though the turn finished, hanging the reader until the wall-clock
345 # timeout fires on a successful run.
346 result_seen = False
348 try:
349 while sel.get_map():
350 now = time.time()
351 if timeout and (now - start_time) > timeout:
352 _kill_process_group(process)
353 try:
354 process.wait(timeout=10)
355 except subprocess.TimeoutExpired:
356 logger.warning(
357 "Process %s did not terminate within 10s after kill",
358 process.pid,
359 )
360 raise subprocess.TimeoutExpired(cmd_args, timeout)
362 if idle_timeout and (now - last_output_time) > idle_timeout:
363 _kill_process_group(process)
364 try:
365 process.wait(timeout=10)
366 except subprocess.TimeoutExpired:
367 logger.warning(
368 "Process %s did not terminate within 10s after kill",
369 process.pid,
370 )
371 raise subprocess.TimeoutExpired(cmd_args, idle_timeout, output="idle_timeout")
373 ready = sel.select(timeout=1.0)
374 for key, _ in ready:
375 line = key.fileobj.readline() # type: ignore[union-attr]
376 if not line:
377 sel.unregister(key.fileobj)
378 continue
380 last_output_time = time.time()
381 line = line.rstrip("\n")
382 is_stderr = key.fileobj is process.stderr
384 if not is_stderr:
385 try:
386 event = json.loads(line)
387 etype = event.get("type")
388 if etype == "system" and event.get("subtype") == "init":
389 if "model" in event:
390 detected_model = event["model"]
391 if on_model_detected:
392 on_model_detected(event["model"])
393 continue # don't add to stdout_lines
394 elif etype == "assistant":
395 msg = event.get("message", {})
396 text_parts = [
397 block["text"]
398 for block in msg.get("content", [])
399 if block.get("type") == "text"
400 ]
401 text = "\n\n".join(text_parts)
402 if not text:
403 continue
404 for sub_line in text.splitlines() or [""]:
405 stdout_lines.append(sub_line)
406 if stream_callback:
407 stream_callback(sub_line, is_stderr)
408 continue
409 elif etype == "result":
410 usage = event.get("usage", {})
411 if on_usage and usage:
412 on_usage(
413 usage.get("input_tokens", 0)
414 + usage.get("cache_read_input_tokens", 0),
415 usage.get("output_tokens", 0),
416 )
417 if on_usage_detailed and usage:
418 on_usage_detailed(
419 TokenUsage(
420 input_tokens=usage.get("input_tokens", 0),
421 output_tokens=usage.get("output_tokens", 0),
422 cache_read_tokens=usage.get(
423 "cache_read_input_tokens", 0
424 ),
425 cache_creation_tokens=usage.get(
426 "cache_creation_input_tokens", 0
427 ),
428 model=event.get("model", detected_model),
429 )
430 )
431 if event.get("is_error"):
432 error_text = event.get("error") or event.get("result", "")
433 if error_text:
434 stderr_lines.append(f"[result] {error_text}")
435 # Turn is done. Finish draining the current ready
436 # batch (so trailing buffered lines aren't lost),
437 # then break the loop below instead of blocking on
438 # a pipe EOF that inherited background-task FDs may
439 # never deliver.
440 result_seen = True
441 continue # skip other event types (tool_use, etc.)
442 else:
443 continue # skip other event types (tool_use, etc.)
444 except (json.JSONDecodeError, KeyError, TypeError):
445 pass # non-JSON line: pass through as raw text
447 if is_stderr:
448 stderr_lines.append(line)
449 else:
450 stdout_lines.append(line)
452 if stream_callback:
453 stream_callback(line, is_stderr)
455 # The "result" event ended the turn and the current ready batch
456 # has now been fully drained; stop reading rather than blocking
457 # for a pipe EOF that may never arrive.
458 if result_seen:
459 break
461 try:
462 process.wait(timeout=30)
463 except subprocess.TimeoutExpired:
464 logger.warning(
465 "Process %s did not exit within 30s after streams closed, killing",
466 process.pid,
467 )
468 _kill_process_group(process)
469 try:
470 process.wait(timeout=10)
471 except subprocess.TimeoutExpired:
472 logger.warning(
473 "Process %s did not terminate within 10s after kill",
474 process.pid,
475 )
476 finally:
477 if on_process_end:
478 on_process_end(process)
480 return subprocess.CompletedProcess(
481 cmd_args,
482 process.returncode if process.returncode is not None else -9,
483 stdout="\n".join(stdout_lines),
484 stderr="\n".join(stderr_lines),
485 )