mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(learn): claude-cli streams output with idle timeout (#373)
## Description `headroom learn` with the claude-cli backend used `subprocess.run` with a hard 120s wall-clock cap and no liveness signal. A successful long analysis and a hung connection looked identical — exit 0 with "0 recommendations" was the only user-visible signal when the LLM call timed out, which silently hides genuine learnings. This PR makes the CLI backend timeout-aware, with progress detection for claude-cli and configurable wall-clock caps for every backend. Fixes #(issue number) ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - **Streaming claude-cli with idle timeout**: invoke `claude -p --output-format stream-json --verbose` and run a watchdog loop that drains stdout/stderr via reader threads. Each stream-json event resets an idle deadline. Kill the process if no output for `HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS` (default 60s) or if total elapsed exceeds `HEADROOM_LEARN_CLI_TIMEOUT_SECS` (default 300s, was 120s). The final `type:"result"` event carries the assistant response, which is then parsed as JSON. Reader threads (rather than `select`) are used so the watchdog works on Windows where `select` does not support pipe handles. - **Bumped default `_CLI_TIMEOUT` from 120s to 300s** as the hard cap for all CLI backends. The previous 120s was too tight for large digests on slower networks. - **Env-var overrides** via new helper `_resolve_timeout_secs(env_var, default)`: - `HEADROOM_LEARN_CLI_TIMEOUT_SECS` — hard wall-clock cap (all CLI backends) - `HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS` — idle cap (streaming claude-cli only) - Non-positive or non-integer values log a warning and fall back to defaults, so a typo can't disable the timeout. - **gemini-cli and codex-cli** keep `subprocess.run(timeout=hard_cap)` since they do not emit progress events. They benefit from the bumped default and the env-var override. - **CHANGELOG.md** updated under `[Unreleased]` → `### Fixed`. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed (existing repro: 16k-call digest that previously timed out at 120s) New test coverage in `tests/test_learn/test_analyzer.py`: - `test_claude_cli_streams_and_parses_result_event` — happy path, fake Popen yields system/assistant/result events - `test_claude_cli_parses_fenced_result` — markdown fences in the result event still parse - `test_claude_cli_idle_timeout_kills_hang` — `HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS=1` + a hanging stdout iterator triggers the idle watchdog - `test_claude_cli_hard_cap_kills_continuous_chatter` — continuous events with a low hard cap fire the wall-clock kill (proves idle reset alone can't keep a runaway alive) - `test_claude_cli_missing_result_event_raises` — graceful failure when no `result` event is emitted - `test_claude_cli_nonzero_exit_raises` / `test_claude_cli_unparseable_result_raises_with_context` / `test_claude_cli_not_installed_raises` — error paths - Parallel codex-cli error coverage (timeout-honors-env-override included) so the wall-clock path is exercised - `TestResolveTimeoutSecs` — unset / empty / non-integer / non-positive / valid override ## Test Output ``` $ uv run pytest tests/test_learn/test_analyzer.py ============================== 67 passed in 2.14s ============================== $ uv run ruff check headroom/learn/analyzer.py tests/test_learn/test_analyzer.py All checks passed! $ uv run ruff format --check headroom/learn/analyzer.py tests/test_learn/test_analyzer.py 2 files already formatted $ uv run mypy headroom/learn/analyzer.py Success: no issues found in 1 source file ``` ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] I have updated the CHANGELOG.md if applicable ## Additional Notes - The contract assumed for claude-cli stream-json output is: each line is a JSON object with a `type` field; the final event has `type:"result"` with a string `result` field carrying the assistant text. This matches the documented Anthropic CLI behavior. If the contract changes upstream, `_call_claude_cli_streaming` raises a clear "did not emit a final \`result\` event" error rather than silently succeeding. - Backwards-compatible for users without env-var configuration: behavior just becomes "longer hard cap, plus idle watchdog for claude-cli", neither of which can falsely succeed. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gglucass
GitHub
parent
8f374263d3
commit
9bff5752bb
3 changed files with 498 additions and 44 deletions
17
CHANGELOG.md
17
CHANGELOG.md
|
|
@ -175,6 +175,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||
resolution, made PyPI publish failures block GitHub Releases unless
|
||||
`PYPI_SKIP=true`, and added an sdist `LICENSE` invariant.
|
||||
|
||||
- **`headroom learn` with claude-cli no longer fails silently on slow
|
||||
networks or large digests.** The CLI backend timeout was a hard 120s
|
||||
wall-clock cap with no liveness signal: a successful long analysis and
|
||||
a hung connection looked identical, and exit 0 with "no recommendations"
|
||||
was the only user-visible signal. Two changes:
|
||||
(1) **Streaming + idle timeout for claude-cli**: the command now uses
|
||||
`--output-format stream-json --verbose` and a watchdog thread reads
|
||||
events as they arrive. The process is killed only after
|
||||
`HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS` (default 60s) of zero output, or
|
||||
after `HEADROOM_LEARN_CLI_TIMEOUT_SECS` (default 300s, was 120s) total.
|
||||
Long-but-active analyses run to completion; genuine hangs are caught
|
||||
fast. The final `type:"result"` event carries the assistant response.
|
||||
Drains stdout/stderr via reader threads so the watchdog works on
|
||||
Windows too. (2) **Env-var overrides for all CLI backends**:
|
||||
`HEADROOM_LEARN_CLI_TIMEOUT_SECS` is honored by gemini-cli and
|
||||
codex-cli as the wall-clock timeout; idle override applies only to the
|
||||
streaming claude-cli path.
|
||||
- **`Learned: error recovery` section in MEMORY.md no longer bloats with
|
||||
stale, one-shot, or contradictory entries.** The matchers paired up
|
||||
unrelated tool calls (e.g. `state.rs` and `lib.rs` in the same dir
|
||||
|
|
|
|||
|
|
@ -17,8 +17,12 @@ from __future__ import annotations
|
|||
import json
|
||||
import logging
|
||||
import os
|
||||
import queue
|
||||
import shutil
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
import typing
|
||||
|
||||
from .models import (
|
||||
AnalysisResult,
|
||||
|
|
@ -43,9 +47,11 @@ _MODEL_DEFAULTS: list[tuple[str, str]] = [
|
|||
_MAX_DIGEST_TOKENS = 80_000 # Budget for the digest (leave room for prompt + output)
|
||||
|
||||
# CLI tools to try when no API key is set (checked in order).
|
||||
# Each entry: (binary_name, model_identifier, command_prefix)
|
||||
# Each entry: (binary_name, model_identifier, command_prefix). The claude-cli
|
||||
# command uses stream-json output so the analyzer can detect progress and
|
||||
# enforce an idle (rather than wall-clock-only) timeout — see _call_cli_llm.
|
||||
_CLI_BACKENDS: list[tuple[str, str, list[str]]] = [
|
||||
("claude", "claude-cli", ["claude", "-p"]),
|
||||
("claude", "claude-cli", ["claude", "-p", "--output-format", "stream-json", "--verbose"]),
|
||||
("gemini", "gemini-cli", ["gemini", "-p"]),
|
||||
("codex", "codex-cli", ["codex", "exec"]),
|
||||
]
|
||||
|
|
@ -55,7 +61,35 @@ _CLI_MODEL_IDS: set[str] = {model for _, model, _ in _CLI_BACKENDS}
|
|||
|
||||
_USER_PROMPT_PREFIX = "Analyze these coding agent sessions and return JSON recommendations:\n\n" # Shared by _call_cli_llm and _call_llm
|
||||
_MAX_SNIPPET_LEN = 2000 # Max chars of CLI output (stdout/stderr) in error messages
|
||||
_CLI_TIMEOUT = 120 # Subprocess timeout for CLI backends, in seconds
|
||||
# Hard wall-clock cap for CLI backends (seconds). Override with
|
||||
# HEADROOM_LEARN_CLI_TIMEOUT_SECS for slow networks or large digests.
|
||||
_CLI_TIMEOUT = 300
|
||||
# Idle cap (seconds) for streaming claude-cli: kill if no output arrives for
|
||||
# this long. Lets us catch genuine hangs quickly while letting long-but-active
|
||||
# analyses run to completion. Override with HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS.
|
||||
_CLI_IDLE_TIMEOUT = 60
|
||||
|
||||
|
||||
def _resolve_timeout_secs(env_var: str, default: int) -> int:
|
||||
"""Resolve a positive-integer timeout from *env_var* or fall back to *default*.
|
||||
|
||||
Invalid or non-positive values are logged and ignored so a typo in env
|
||||
config can't accidentally disable the timeout.
|
||||
"""
|
||||
raw = os.environ.get(env_var)
|
||||
if raw is None or raw == "":
|
||||
return default
|
||||
try:
|
||||
value = int(raw)
|
||||
except ValueError:
|
||||
logger.warning("Invalid %s=%r — using default %ds", env_var, raw, default)
|
||||
return default
|
||||
if value <= 0:
|
||||
logger.warning(
|
||||
"Invalid %s=%r (must be positive) — using default %ds", env_var, raw, default
|
||||
)
|
||||
return default
|
||||
return value
|
||||
|
||||
|
||||
def _detect_default_model() -> str:
|
||||
|
|
@ -413,9 +447,12 @@ def _call_cli_llm(digest: str, model: str) -> dict:
|
|||
OS ``ARG_MAX`` limits and argument-injection risks.
|
||||
|
||||
CLI invocations:
|
||||
claude-cli → echo <prompt> | claude -p
|
||||
gemini-cli → echo <prompt> | gemini -p
|
||||
codex-cli → echo <prompt> | codex exec
|
||||
claude-cli → claude -p --output-format stream-json --verbose (idle-timeout)
|
||||
gemini-cli → gemini -p (wall-clock timeout)
|
||||
codex-cli → codex exec (wall-clock timeout)
|
||||
|
||||
The claude-cli path streams JSON events, letting the analyzer kill genuine
|
||||
hangs while letting long-but-active analyses run to completion.
|
||||
|
||||
Args:
|
||||
digest: Token-efficient session digest to analyze.
|
||||
|
|
@ -437,6 +474,11 @@ def _call_cli_llm(digest: str, model: str) -> dict:
|
|||
raise ValueError(f"Unknown CLI model: {model}")
|
||||
|
||||
prompt = _SYSTEM_PROMPT + "\n\n" + _USER_PROMPT_PREFIX + digest
|
||||
hard_cap = _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", _CLI_TIMEOUT)
|
||||
|
||||
if model == "claude-cli":
|
||||
idle_cap = _resolve_timeout_secs("HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS", _CLI_IDLE_TIMEOUT)
|
||||
return _call_claude_cli_streaming(cmd, prompt, hard_cap=hard_cap, idle_cap=idle_cap)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
|
|
@ -444,7 +486,7 @@ def _call_cli_llm(digest: str, model: str) -> dict:
|
|||
input=prompt,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=_CLI_TIMEOUT,
|
||||
timeout=hard_cap,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
raise RuntimeError(
|
||||
|
|
@ -453,9 +495,9 @@ def _call_cli_llm(digest: str, model: str) -> dict:
|
|||
) from None
|
||||
except subprocess.TimeoutExpired:
|
||||
raise RuntimeError(
|
||||
f"`{' '.join(cmd)}` did not respond within {_CLI_TIMEOUT}s. "
|
||||
"Check network connectivity or try a different backend with "
|
||||
"--model <litellm-model-name>."
|
||||
f"`{' '.join(cmd)}` did not respond within {hard_cap}s. "
|
||||
"Check network connectivity, raise HEADROOM_LEARN_CLI_TIMEOUT_SECS, "
|
||||
"or try a different backend with --model <litellm-model-name>."
|
||||
) from None
|
||||
|
||||
if result.returncode != 0:
|
||||
|
|
@ -478,6 +520,155 @@ def _call_cli_llm(digest: str, model: str) -> dict:
|
|||
) from exc
|
||||
|
||||
|
||||
def _call_claude_cli_streaming(
|
||||
cmd: list[str], prompt: str, *, hard_cap: int, idle_cap: int
|
||||
) -> dict:
|
||||
"""Run claude-cli with stream-json output and an idle-timeout watchdog.
|
||||
|
||||
Each line of stdout is one JSON event from claude (system/assistant/user/
|
||||
result). Any line resets the idle deadline. The process is killed if no
|
||||
output arrives for *idle_cap* seconds, or if total elapsed exceeds
|
||||
*hard_cap* seconds. The final ``type:"result"`` event carries the assistant
|
||||
response, which is then parsed as JSON.
|
||||
|
||||
Threads (rather than ``select``) drain stdout/stderr so the watchdog works
|
||||
on Windows too, where ``select`` does not support pipe handles.
|
||||
"""
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
cmd,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
bufsize=1, # line-buffered
|
||||
)
|
||||
except FileNotFoundError:
|
||||
raise RuntimeError(
|
||||
f"`{cmd[0]}` not found in PATH. Install it or use a different backend "
|
||||
"with --model <litellm-model-name>."
|
||||
) from None
|
||||
|
||||
assert proc.stdin is not None and proc.stdout is not None and proc.stderr is not None
|
||||
try:
|
||||
proc.stdin.write(prompt)
|
||||
finally:
|
||||
try:
|
||||
proc.stdin.close()
|
||||
except BrokenPipeError: # pragma: no cover — defensive, claude exits before stdin drain
|
||||
pass
|
||||
|
||||
events: queue.Queue[tuple[str, str | None]] = queue.Queue()
|
||||
|
||||
def _pump(stream: typing.IO[str], tag: str) -> None:
|
||||
try:
|
||||
for line in stream:
|
||||
events.put((tag, line))
|
||||
except Exception as exc: # pragma: no cover — defensive
|
||||
logger.debug("stream pump (%s) errored: %s", tag, exc)
|
||||
finally:
|
||||
events.put((tag, None)) # EOF marker
|
||||
|
||||
threading.Thread(target=_pump, args=(proc.stdout, "stdout"), daemon=True).start()
|
||||
threading.Thread(target=_pump, args=(proc.stderr, "stderr"), daemon=True).start()
|
||||
|
||||
start = time.monotonic()
|
||||
last_activity = start
|
||||
stdout_lines: list[str] = []
|
||||
stderr_lines: list[str] = []
|
||||
final_result: str | None = None
|
||||
eofs = 0
|
||||
|
||||
def _kill(reason: str) -> None:
|
||||
proc.kill()
|
||||
try:
|
||||
proc.wait(timeout=5)
|
||||
except (
|
||||
subprocess.TimeoutExpired
|
||||
): # pragma: no cover — defensive, kill normally returns fast
|
||||
pass
|
||||
logger.debug("claude-cli killed: %s", reason)
|
||||
|
||||
while eofs < 2:
|
||||
elapsed = time.monotonic() - start
|
||||
if elapsed > hard_cap:
|
||||
_kill(f"hard cap {hard_cap}s exceeded")
|
||||
raise RuntimeError(
|
||||
f"`{' '.join(cmd)}` exceeded the {hard_cap}s hard cap. "
|
||||
"Raise HEADROOM_LEARN_CLI_TIMEOUT_SECS for slower networks or "
|
||||
"larger digests, or try a different backend with "
|
||||
"--model <litellm-model-name>."
|
||||
)
|
||||
idle_elapsed = time.monotonic() - last_activity
|
||||
if idle_elapsed > idle_cap:
|
||||
_kill(f"idle cap {idle_cap}s exceeded")
|
||||
raise RuntimeError(
|
||||
f"`{' '.join(cmd)}` produced no output for {idle_cap}s. "
|
||||
"Check network connectivity, raise "
|
||||
"HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS, or try a different "
|
||||
"backend with --model <litellm-model-name>."
|
||||
)
|
||||
|
||||
# Block up to 1s waiting for the next event, then re-check deadlines.
|
||||
try:
|
||||
tag, line = events.get(timeout=1.0)
|
||||
except queue.Empty:
|
||||
continue
|
||||
|
||||
if line is None:
|
||||
eofs += 1
|
||||
continue
|
||||
last_activity = time.monotonic()
|
||||
if tag == "stdout":
|
||||
stdout_lines.append(line)
|
||||
event = _parse_stream_event(line)
|
||||
if event is not None and event.get("type") == "result":
|
||||
# Last result event wins if multiple are emitted.
|
||||
result_text = event.get("result")
|
||||
if isinstance(result_text, str):
|
||||
final_result = result_text
|
||||
else:
|
||||
stderr_lines.append(line)
|
||||
|
||||
proc.wait()
|
||||
|
||||
if proc.returncode != 0:
|
||||
stderr_blob = "".join(stderr_lines)[:_MAX_SNIPPET_LEN]
|
||||
raise RuntimeError(f"`{' '.join(cmd)}` failed (exit {proc.returncode}):\n{stderr_blob}")
|
||||
|
||||
stderr_blob = "".join(stderr_lines)
|
||||
if stderr_blob.strip():
|
||||
logger.debug("CLI stderr (exit 0): %s", stderr_blob[:_MAX_SNIPPET_LEN])
|
||||
|
||||
if final_result is None:
|
||||
stdout_snippet = "".join(stdout_lines)[:_MAX_SNIPPET_LEN]
|
||||
raise RuntimeError(
|
||||
f"`{' '.join(cmd)}` did not emit a final `result` event. "
|
||||
f"First {_MAX_SNIPPET_LEN} chars of stdout:\n{stdout_snippet}"
|
||||
)
|
||||
|
||||
try:
|
||||
return _strip_fenced_json(final_result)
|
||||
except json.JSONDecodeError as exc:
|
||||
snippet = final_result[:_MAX_SNIPPET_LEN]
|
||||
raise RuntimeError(
|
||||
f"`{' '.join(cmd)}` returned unparseable output. "
|
||||
f"First {_MAX_SNIPPET_LEN} chars:\n{snippet}"
|
||||
) from exc
|
||||
|
||||
|
||||
def _parse_stream_event(line: str) -> dict | None:
|
||||
"""Parse one line of claude-cli stream-json output, returning None on junk."""
|
||||
line = line.strip()
|
||||
if not line:
|
||||
return None
|
||||
try:
|
||||
parsed = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
return parsed if isinstance(parsed, dict) else None
|
||||
|
||||
|
||||
def _call_llm(digest: str, model: str) -> dict:
|
||||
"""Call LLM with the session digest and return parsed JSON.
|
||||
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
"""Tests for session analyzer — digest builder and LLM-based analysis."""
|
||||
|
||||
import io
|
||||
import json
|
||||
import subprocess
|
||||
import time
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
|
@ -14,6 +16,7 @@ from headroom.learn.analyzer import (
|
|||
_call_llm,
|
||||
_detect_default_model,
|
||||
_parse_llm_response,
|
||||
_resolve_timeout_secs,
|
||||
_strip_fenced_json,
|
||||
)
|
||||
from headroom.learn.models import (
|
||||
|
|
@ -590,21 +593,157 @@ class TestStripFencedJson:
|
|||
_strip_fenced_json("not json at all")
|
||||
|
||||
|
||||
def _fake_claude_popen(
|
||||
*,
|
||||
stdout_lines: list[str],
|
||||
stderr_lines: list[str] | None = None,
|
||||
returncode: int = 0,
|
||||
stdout_delay: float = 0.0,
|
||||
) -> MagicMock:
|
||||
"""Build a Popen mock factory for the streaming claude-cli path.
|
||||
|
||||
Returns a MagicMock that, when called as ``Popen(cmd, ...)``, yields a
|
||||
fake process whose stdout/stderr behave like line-iterable text streams.
|
||||
Each stdout line is sleep(*stdout_delay*)-gated to let tests simulate slow
|
||||
or hung processes.
|
||||
"""
|
||||
if stderr_lines is None:
|
||||
stderr_lines = []
|
||||
|
||||
def _make_iter(lines: list[str], delay: float):
|
||||
def _gen():
|
||||
for line in lines:
|
||||
if delay:
|
||||
time.sleep(delay)
|
||||
yield line
|
||||
|
||||
return _gen()
|
||||
|
||||
factory = MagicMock()
|
||||
|
||||
def _construct(*args, **kwargs):
|
||||
proc = MagicMock()
|
||||
proc.stdin = io.StringIO()
|
||||
proc.stdout = _make_iter(stdout_lines, stdout_delay)
|
||||
proc.stderr = _make_iter(stderr_lines, 0.0)
|
||||
proc.returncode = returncode
|
||||
proc.wait = MagicMock(return_value=returncode)
|
||||
proc.kill = MagicMock()
|
||||
proc.poll = MagicMock(return_value=returncode)
|
||||
return proc
|
||||
|
||||
factory.side_effect = _construct
|
||||
return factory
|
||||
|
||||
|
||||
def _stream_event(event_type: str, **fields) -> str:
|
||||
return json.dumps({"type": event_type, **fields}) + "\n"
|
||||
|
||||
|
||||
def _result_event(text: str) -> str:
|
||||
return _stream_event("result", subtype="success", is_error=False, result=text)
|
||||
|
||||
|
||||
class TestCallCliLlm:
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_claude_cli_success(self, mock_run: MagicMock):
|
||||
mock_run.return_value = MagicMock(
|
||||
returncode=0,
|
||||
stdout='{"context_file_rules": [], "memory_file_rules": []}',
|
||||
stderr="",
|
||||
)
|
||||
result = _call_cli_llm("test digest", "claude-cli")
|
||||
def test_claude_cli_streams_and_parses_result_event(self):
|
||||
stdout = [
|
||||
_stream_event("system", subtype="init"),
|
||||
_stream_event("assistant", message={"content": "thinking..."}),
|
||||
_result_event('{"context_file_rules": [], "memory_file_rules": []}'),
|
||||
]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
) as popen:
|
||||
result = _call_cli_llm("test digest", "claude-cli")
|
||||
assert result == {"context_file_rules": [], "memory_file_rules": []}
|
||||
mock_run.assert_called_once()
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert cmd == ["claude", "-p"]
|
||||
# Prompt passed via stdin, not as an argument
|
||||
assert mock_run.call_args.kwargs.get("input") is not None
|
||||
cmd = popen.call_args[0][0]
|
||||
assert cmd == ["claude", "-p", "--output-format", "stream-json", "--verbose"]
|
||||
|
||||
def test_claude_cli_parses_fenced_result(self):
|
||||
stdout = [
|
||||
_result_event('```json\n{"context_file_rules": [], "memory_file_rules": []}\n```'),
|
||||
]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
):
|
||||
result = _call_cli_llm("test digest", "claude-cli")
|
||||
assert result == {"context_file_rules": [], "memory_file_rules": []}
|
||||
|
||||
def test_claude_cli_idle_timeout_kills_hang(self, monkeypatch):
|
||||
import threading as _threading
|
||||
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS", "1")
|
||||
|
||||
# An iterator that never yields and never EOFs — simulates a hung CLI.
|
||||
# The pump thread blocks in __next__, so no events reach the watchdog.
|
||||
blocked = _threading.Event() # never set
|
||||
|
||||
class _HangingStream:
|
||||
def __iter__(self):
|
||||
return self
|
||||
|
||||
def __next__(self):
|
||||
blocked.wait(timeout=10)
|
||||
raise StopIteration
|
||||
|
||||
def _construct(*args, **kwargs):
|
||||
proc = MagicMock()
|
||||
proc.stdin = io.StringIO()
|
||||
proc.stdout = _HangingStream()
|
||||
proc.stderr = _HangingStream()
|
||||
proc.returncode = 0
|
||||
proc.wait = MagicMock(return_value=0)
|
||||
proc.kill = MagicMock(side_effect=lambda: blocked.set())
|
||||
proc.poll = MagicMock(return_value=None)
|
||||
return proc
|
||||
|
||||
popen = MagicMock(side_effect=_construct)
|
||||
with patch("headroom.learn.analyzer.subprocess.Popen", popen):
|
||||
with pytest.raises(RuntimeError, match="produced no output"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_claude_cli_hard_cap_kills_continuous_chatter(self, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "1")
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_IDLE_TIMEOUT_SECS", "10")
|
||||
|
||||
# Continuous output every 50ms so idle never fires; hard cap should.
|
||||
chatter = [_stream_event("assistant", message={"i": i}) for i in range(1000)]
|
||||
popen = _fake_claude_popen(stdout_lines=chatter, stdout_delay=0.05)
|
||||
with patch("headroom.learn.analyzer.subprocess.Popen", popen):
|
||||
with pytest.raises(RuntimeError, match="exceeded the 1s hard cap"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_claude_cli_missing_result_event_raises(self):
|
||||
stdout = [_stream_event("assistant", message={"content": "no result"})]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
):
|
||||
with pytest.raises(RuntimeError, match="did not emit a final `result` event"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_claude_cli_nonzero_exit_raises(self):
|
||||
popen = _fake_claude_popen(
|
||||
stdout_lines=[],
|
||||
stderr_lines=["Error: auth required\n"],
|
||||
returncode=1,
|
||||
)
|
||||
with patch("headroom.learn.analyzer.subprocess.Popen", popen):
|
||||
with pytest.raises(RuntimeError, match="failed.*exit 1"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_claude_cli_unparseable_result_raises_with_context(self):
|
||||
stdout = [_result_event("This is not JSON at all")]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
):
|
||||
with pytest.raises(RuntimeError, match="unparseable output"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_claude_cli_not_installed_raises(self):
|
||||
popen = MagicMock(side_effect=FileNotFoundError("No such file or directory: 'claude'"))
|
||||
with patch("headroom.learn.analyzer.subprocess.Popen", popen):
|
||||
with pytest.raises(RuntimeError, match="not found in PATH"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_codex_cli_uses_exec(self, mock_run: MagicMock):
|
||||
|
|
@ -630,17 +769,17 @@ class TestCallCliLlm:
|
|||
assert cmd == ["gemini", "-p"]
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_cli_nonzero_exit_raises(self, mock_run: MagicMock):
|
||||
def test_codex_nonzero_exit_raises(self, mock_run: MagicMock):
|
||||
mock_run.return_value = MagicMock(
|
||||
returncode=1,
|
||||
stdout="",
|
||||
stderr="Error: auth required",
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="failed.*exit 1"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_cli_stderr_truncated_in_error(self, mock_run: MagicMock):
|
||||
def test_codex_stderr_truncated_in_error(self, mock_run: MagicMock):
|
||||
long_stderr = "x" * 5000
|
||||
mock_run.return_value = MagicMock(
|
||||
returncode=1,
|
||||
|
|
@ -648,8 +787,7 @@ class TestCallCliLlm:
|
|||
stderr=long_stderr,
|
||||
)
|
||||
with pytest.raises(RuntimeError) as exc_info:
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
# Full 5000-char stderr should not appear in the error message
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
assert long_stderr not in str(exc_info.value)
|
||||
|
||||
def test_unknown_cli_model_raises(self):
|
||||
|
|
@ -657,36 +795,144 @@ class TestCallCliLlm:
|
|||
_call_cli_llm("test digest", "unknown-cli")
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_fenced_output_parsed(self, mock_run: MagicMock):
|
||||
mock_run.return_value = MagicMock(
|
||||
returncode=0,
|
||||
stdout='```json\n{"context_file_rules": [], "memory_file_rules": []}\n```',
|
||||
stderr="",
|
||||
)
|
||||
result = _call_cli_llm("test digest", "claude-cli")
|
||||
assert result == {"context_file_rules": [], "memory_file_rules": []}
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_cli_not_installed_raises(self, mock_run: MagicMock):
|
||||
def test_codex_not_installed_raises(self, mock_run: MagicMock):
|
||||
mock_run.side_effect = FileNotFoundError("No such file or directory: 'codex'")
|
||||
with pytest.raises(RuntimeError, match="not found in PATH"):
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_timeout_raises_runtime_error(self, mock_run: MagicMock):
|
||||
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["claude", "-p"], timeout=120)
|
||||
def test_codex_timeout_raises_runtime_error(self, mock_run: MagicMock):
|
||||
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["codex", "exec"], timeout=300)
|
||||
with pytest.raises(RuntimeError, match="did not respond within"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_unparseable_output_raises_with_context(self, mock_run: MagicMock):
|
||||
def test_codex_timeout_honors_env_override(self, mock_run: MagicMock, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "42")
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout="{}", stderr="")
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
assert mock_run.call_args.kwargs["timeout"] == 42
|
||||
|
||||
@patch("headroom.learn.analyzer.subprocess.run")
|
||||
def test_codex_unparseable_output_raises_with_context(self, mock_run: MagicMock):
|
||||
mock_run.return_value = MagicMock(
|
||||
returncode=0,
|
||||
stdout="This is not JSON at all",
|
||||
stderr="",
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="unparseable output"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
_call_cli_llm("test digest", "codex-cli")
|
||||
|
||||
|
||||
class TestParseStreamEvent:
|
||||
def test_returns_none_for_empty_line(self):
|
||||
from headroom.learn.analyzer import _parse_stream_event
|
||||
|
||||
assert _parse_stream_event("") is None
|
||||
assert _parse_stream_event(" \n") is None
|
||||
|
||||
def test_returns_none_for_invalid_json(self):
|
||||
from headroom.learn.analyzer import _parse_stream_event
|
||||
|
||||
assert _parse_stream_event("not json at all") is None
|
||||
assert _parse_stream_event("{unclosed") is None
|
||||
|
||||
def test_returns_none_for_non_dict_json(self):
|
||||
from headroom.learn.analyzer import _parse_stream_event
|
||||
|
||||
assert _parse_stream_event('"a string"') is None
|
||||
assert _parse_stream_event("[1, 2, 3]") is None
|
||||
|
||||
def test_parses_valid_event(self):
|
||||
from headroom.learn.analyzer import _parse_stream_event
|
||||
|
||||
assert _parse_stream_event('{"type": "result", "result": "x"}') == {
|
||||
"type": "result",
|
||||
"result": "x",
|
||||
}
|
||||
|
||||
|
||||
class TestClaudeCliEdgeCases:
|
||||
"""Coverage for less-traveled branches in the streaming claude-cli path."""
|
||||
|
||||
def test_non_string_result_field_falls_through_to_missing(self):
|
||||
# `result` event present but the `result` field is a dict, not a string.
|
||||
# The watchdog should not store it as final_result, so the path raises
|
||||
# the "did not emit a final result event" error.
|
||||
stdout = [_stream_event("result", subtype="success", result={"unexpected": "shape"})]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
):
|
||||
with pytest.raises(RuntimeError, match="did not emit a final `result` event"):
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
|
||||
def test_stderr_on_success_is_logged_not_raised(self, caplog):
|
||||
import logging
|
||||
|
||||
stdout = [_result_event('{"context_file_rules": [], "memory_file_rules": []}')]
|
||||
stderr_warning = "deprecation: --foo will be removed in v2\n"
|
||||
popen = _fake_claude_popen(stdout_lines=stdout, stderr_lines=[stderr_warning])
|
||||
with caplog.at_level(logging.DEBUG, logger="headroom.learn.analyzer"):
|
||||
with patch("headroom.learn.analyzer.subprocess.Popen", popen):
|
||||
result = _call_cli_llm("test digest", "claude-cli")
|
||||
assert result == {"context_file_rules": [], "memory_file_rules": []}
|
||||
assert any("CLI stderr (exit 0)" in rec.message for rec in caplog.records)
|
||||
|
||||
def test_non_result_stdout_lines_are_buffered_into_snippet_on_failure(self):
|
||||
# If only assistant/system events arrive (no result), the missing-result
|
||||
# error should include a snippet from stdout.
|
||||
stdout = [
|
||||
_stream_event("system", subtype="init"),
|
||||
_stream_event("assistant", message={"content": "thinking..."}),
|
||||
]
|
||||
with patch(
|
||||
"headroom.learn.analyzer.subprocess.Popen", _fake_claude_popen(stdout_lines=stdout)
|
||||
):
|
||||
with pytest.raises(RuntimeError) as exc_info:
|
||||
_call_cli_llm("test digest", "claude-cli")
|
||||
message = str(exc_info.value)
|
||||
assert "did not emit a final `result` event" in message
|
||||
assert "thinking" in message # stdout snippet was included
|
||||
|
||||
def test_resolve_timeout_logs_warning_for_invalid(self, caplog, monkeypatch):
|
||||
import logging
|
||||
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "abc")
|
||||
with caplog.at_level(logging.WARNING, logger="headroom.learn.analyzer"):
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
assert any(
|
||||
"Invalid HEADROOM_LEARN_CLI_TIMEOUT_SECS" in rec.message for rec in caplog.records
|
||||
)
|
||||
|
||||
def test_resolve_timeout_logs_warning_for_non_positive(self, caplog, monkeypatch):
|
||||
import logging
|
||||
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "-5")
|
||||
with caplog.at_level(logging.WARNING, logger="headroom.learn.analyzer"):
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
assert any("must be positive" in rec.message for rec in caplog.records)
|
||||
|
||||
|
||||
class TestResolveTimeoutSecs:
|
||||
def test_uses_default_when_unset(self, monkeypatch):
|
||||
monkeypatch.delenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", raising=False)
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
|
||||
def test_uses_default_when_empty(self, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "")
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
|
||||
def test_uses_default_for_non_integer(self, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "not-a-number")
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
|
||||
def test_uses_default_for_non_positive(self, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "0")
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 300
|
||||
|
||||
def test_returns_overridden_value(self, monkeypatch):
|
||||
monkeypatch.setenv("HEADROOM_LEARN_CLI_TIMEOUT_SECS", "777")
|
||||
assert _resolve_timeout_secs("HEADROOM_LEARN_CLI_TIMEOUT_SECS", 300) == 777
|
||||
|
||||
|
||||
class TestCallLlmRouting:
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue