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