"""Tests for src/claude_runner.py (steering — persistent Claude processes). Fully offline: `tests/fake_claude.py` is a real subprocess (no mocked `subprocess.Popen`) speaking a minimal stream-json dialect, so steering timing over real OS pipes is genuinely exercised rather than assumed. """ import threading import time from pathlib import Path import pytest from src import claude_runner, claude_session from src.claude_runner import SteerStatus FAKE_CLAUDE = Path(__file__).parent / "fake_claude.py" def _wait_until(predicate, timeout=5.0, interval=0.02): deadline = time.monotonic() + timeout while time.monotonic() < deadline: if predicate(): return time.sleep(interval) raise AssertionError("condition not met within timeout") @pytest.fixture(autouse=True) def _reset_registry(): """H2: the registry is module-level global state (like claude_session._session_locks) — clear it around every test so fake processes from one test never leak into the next.""" claude_runner.reset_registry_for_tests() yield claude_runner.reset_registry_for_tests() @pytest.fixture def make_proc(monkeypatch): """Factory for a ClaudeProcess wired to tests/fake_claude.py, with automatic cleanup of every process it creates.""" created: list[claude_runner.ClaudeProcess] = [] monkeypatch.setattr(claude_runner, "CLAUDE_BIN", str(FAKE_CLAUDE)) def _factory(scenario: str, channel_id: str = "test-channel", session_id_env: str = "fake-session-1", model: str = "sonnet"): monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", scenario) monkeypatch.setenv("FAKE_CLAUDE_SESSION_ID", session_id_env) proc = claude_runner.ClaudeProcess(channel_id, model=model) created.append(proc) return proc yield _factory for p in created: try: p.stop() except Exception: pass # --------------------------------------------------------------------------- # T1 — normal turn shape + rate limit + timeout (Etapa 1 contract) # --------------------------------------------------------------------------- def test_run_turn_matches_run_claude_dict_shape(make_proc): proc = make_proc("normal") result = proc.run_turn("hello") expected_keys = { "result", "session_id", "usage", "total_cost_usd", "cost_usd", "duration_ms", "num_turns", "intermediate_count", "subtype", "is_error", } assert set(result.keys()) == expected_keys assert result["is_error"] is False assert result["session_id"] == "fake-session-1" def test_run_turn_wraps_external_content(make_proc): """T1 (-k external_content selects this test and the steer one below).""" proc = make_proc("normal") result = proc.run_turn("plain text from Marius") assert "[EXTERNAL CONTENT]" in result["result"] assert "[END EXTERNAL CONTENT]" in result["result"] assert "plain text from Marius" in result["result"] def test_timeout_raises_exact_message(make_proc): proc = make_proc("timeout") with pytest.raises(TimeoutError, match=r"Claude CLI timed out after 1s"): proc.run_turn("hi", timeout=1) def test_rate_limit_raises_runtime_error_router_can_detect(make_proc): """T2/C4/T6: the exact failure mode that must keep `_local_fallback_reply` alive — router.py/scheduler.py both gate on `is_rate_limit_error`.""" proc = make_proc("rate_limit") with pytest.raises(RuntimeError) as excinfo: proc.run_turn("hi") assert claude_session.is_rate_limit_error(str(excinfo.value)) def test_non_rate_limit_error_is_returned_not_raised(make_proc): """C4: the parser/wrapper must stay non-throwing for any `is_error` that isn't a rate limit (PlanningSession-style callers retry on subtype).""" proc = make_proc("generic_error") result = proc.run_turn("hi") # must NOT raise assert result["is_error"] is True assert result["subtype"] == "error_max_turns" # --------------------------------------------------------------------------- # Steering — mid-turn message lands on the same stdin # --------------------------------------------------------------------------- def test_steering_mid_turn_reflects_second_message(make_proc): proc = make_proc("steer") texts: list[str] = [] holder: dict = {} def _go(): holder["result"] = proc.run_turn("first message", on_text=texts.append) t = threading.Thread(target=_go) t.start() # Wait for the fake's first assistant block — proves it's now blocked # on its own stdin.readline(), genuinely waiting for the steer. _wait_until(lambda: len(texts) >= 1) assert proc.inflight is True outcome = proc.steer("second message") t.join(timeout=5) assert outcome.status == SteerStatus.STEERED assert outcome.turn is None result = holder["result"] assert "steered:" in result["result"] assert result["num_turns"] == 2 assert proc.inflight is False def test_steer_wraps_external_content(make_proc): """T1 (-k external_content).""" proc = make_proc("steer") texts: list[str] = [] holder: dict = {} def _go(): holder["result"] = proc.run_turn("first", on_text=texts.append) t = threading.Thread(target=_go) t.start() _wait_until(lambda: len(texts) >= 1) proc.steer("stai, schimbare de plan") t.join(timeout=5) assert "[EXTERNAL CONTENT]" in holder["result"]["result"] assert "stai, schimbare de plan" in holder["result"]["result"] def test_steer_pending_texts_available_on_turn_failure(make_proc): """C3: if the turn a steer landed in fails, the steered text must be exposed for re-dispatch, never silently dropped.""" proc = make_proc("timeout") # sleeps forever after init — never sends a result def _go(): try: proc.run_turn("first", timeout=1) except TimeoutError: pass t = threading.Thread(target=_go) t.start() # session_id arrives (via the init event) right after spawn, well # before the fake's 600s sleep would ever resolve — a reliable # "the turn is genuinely in flight" signal for this scenario. _wait_until(lambda: proc.session_id == "fake-session-1") assert proc.inflight is True outcome = proc.steer("please don't get lost") assert outcome.status == SteerStatus.STEERED t.join(timeout=5) # watchdog kills the process after 1s -> TimeoutError pending = proc.pop_pending_steers() assert pending == ["please don't get lost"] # popped once -> empty on a second call assert proc.pop_pending_steers() == [] # --------------------------------------------------------------------------- # C1 — RAN_AS_TURN: steer() finds no in-flight turn # --------------------------------------------------------------------------- def test_steer_without_inflight_turn_becomes_ran_as_turn(make_proc): proc = make_proc("normal") first = proc.run_turn("warm up") assert first["is_error"] is False assert proc.inflight is False outcome = proc.steer("out of band message") assert outcome.status == SteerStatus.RAN_AS_TURN assert outcome.turn is not None assert "out of band message" in outcome.turn["result"] # D4: the caller must treat `turn` as the response and never re-dispatch # by calling run_turn() again for the same text — nothing to assert # mechanically here beyond the API shape making that the only sane path. # --------------------------------------------------------------------------- # T7 — BrokenPipeError on steer falls back to a normal turn # --------------------------------------------------------------------------- def test_steer_broken_pipe_falls_back_to_turn(make_proc, monkeypatch): proc = make_proc("normal") proc.run_turn("warm up") # spawns the process proc.proc.kill() proc.proc.wait() monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "normal") outcome = proc.steer("please deliver me") assert outcome.status == SteerStatus.PROCESS_DEAD assert outcome.turn is not None assert "please deliver me" in outcome.turn["result"] # --------------------------------------------------------------------------- # M2 — timeout kills the process; next turn respawns with --resume # --------------------------------------------------------------------------- def test_respawn_with_resume_after_timeout_kill(make_proc, monkeypatch): proc = make_proc("timeout") with pytest.raises(TimeoutError): proc.run_turn("hi", timeout=1) # session_id came from the `system`/`init` event, captured before the # turn ever got its (never-arriving) result line. assert proc.session_id == "fake-session-1" assert not proc.alive() dead_pid = proc.proc.pid cmd = proc._build_cmd() assert "--resume" in cmd assert cmd[cmd.index("--resume") + 1] == "fake-session-1" monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "normal") result = proc.run_turn("again", timeout=5) assert result["is_error"] is False assert proc.proc.pid != dead_pid # actually respawned, not reused # --------------------------------------------------------------------------- # T3 — stderr pipe must never deadlock a turn # --------------------------------------------------------------------------- def test_big_stderr_does_not_deadlock(make_proc): proc = make_proc("big_stderr") result = proc.run_turn("hello", timeout=15) assert result["is_error"] is False assert len(proc._stderr_buf) > 0 assert len(proc._stderr_buf) <= 50 # deque(maxlen=50) # --------------------------------------------------------------------------- # RunnerRegistry — max_live degradation, reaper resilience # --------------------------------------------------------------------------- def test_max_live_degrades_second_channel_to_one_shot(): registry = claude_runner.RunnerRegistry(max_live=1, idle_minutes=20) proc_a = registry.get("channel-a") assert proc_a is not None proc_b = registry.get("channel-b") assert proc_b is None # T10: degrade, don't evict (X6) and don't block # the same channel is never punished by its own occupied slot: assert registry.get("channel-a") is proc_a registry.stop_all() def test_reaper_never_stops_inflight_process(): registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1) proc = registry.get("channel-a") proc.inflight = True proc.last_active = time.monotonic() - 3600 registry._reap_once() assert registry.live_count() == 1 registry.stop_all() def test_reaper_stops_idle_process(): registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1) proc = registry.get("channel-a") proc.inflight = False proc.last_active = time.monotonic() - 3600 registry._reap_once() assert registry.live_count() == 0 def test_reaper_survives_stop_exception(monkeypatch): """T11: a dead reaper is a silent RAM leak — one process's stop() raising must not kill the reap loop or the process.""" registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=1) proc = registry.get("channel-a") proc.inflight = False proc.last_active = time.monotonic() - 3600 def _boom(): raise RuntimeError("stop() exploded") monkeypatch.setattr(proc, "stop", _boom) registry._reap_once() # must not raise assert registry.live_count() == 0 # bookkeeping removed regardless def test_stop_all_clears_registry(make_proc): proc = make_proc("normal") registry = claude_runner.RunnerRegistry(max_live=2, idle_minutes=20) registry._procs["test-channel"] = proc proc.run_turn("hello") # actually spawns assert proc.alive() registry.stop_all() assert registry.live_count() == 0 assert not proc.alive() def test_get_registry_singleton_and_reset(): r1 = claude_runner.get_registry(max_live=3, idle_minutes=5) r2 = claude_runner.get_registry(max_live=99) # ignored once created assert r1 is r2 assert r1.max_live == 3 claude_runner.reset_registry_for_tests() r3 = claude_runner.get_registry(max_live=7) assert r3 is not r1 assert r3.max_live == 7 def test_max_live_non_positive_logs_and_always_degrades(caplog): with caplog.at_level("WARNING"): registry = claude_runner.RunnerRegistry(max_live=0, idle_minutes=20) assert any("max_live" in rec.message for rec in caplog.records) assert registry.get("any-channel") is None