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