#!/usr/bin/env python3 """Tests for the watchdog: python3 test-resume-stalled-sessions.py""" import json import os import tempfile import unittest import uuid from datetime import datetime, timedelta from importlib import util from pathlib import Path spec = util.spec_from_file_location("watchdog", Path(__file__).with_name("resume-stalled-sessions.py")) watchdog = util.module_from_spec(spec) spec.loader.exec_module(watchdog) # Verbatim from Claude Code's own limit banners, so a wording it really emits # cannot silently stop matching. CLEARS_ON_ITS_OWN = [ "You've hit your session limit · resets 7pm (Europe/Stockholm)", "You've hit your weekly limit · resets Mon 12:00am", "You've hit your Opus limit · resets 7pm", "You've hit your Sonnet limit · resets 7pm", "You've hit your Fable 5 limit · resets 7pm", "You've hit your fast limit · resets in 12m", "You've hit your monthly spend limit · raise it at claude.ai/settings/usage?from=cc_cli_limit_message", "You've hit your individual spend limit · run /usage-credits to raise it, or visit claude.ai/admin-settings/usage", "You've hit your org's monthly spend limit · ask your admin to raise it at claude.ai/settings/usage", "You've hit your usage credit limit · raise it at claude.ai/settings/usage", "You're out of usage credits", "You're out of extra usage", "Your org is out of usage · add funds to continue", "Fable 5 requires usage credits", "usage limit reached — check plan", "rate limited — wait and retry", "API overloaded — wait and retry", "API unavailable — retry", "API error — see detail", "fetch failed", "socket hang up", "ECONNRESET", ] # An entitlement someone else controls, or a decision left to the user: no wait # clears either, so resuming only burns the daily budget. STAYS_PARKED = [ "Your seat type doesn't include usage credits", "Your seat type doesn't include extra usage", "Your usage allocation has been disabled by your admin", "Your group's usage limit is set to $0", "This service is disabled for your org", "login required — run /login", "request too large — /compact or trim", "invalid API request — see detail", "input needed: which branch should I target?", "waiting for the migration to be approved", "needs your decision on the schema", "You've hit your session limit · waiting for your go-ahead", ] class Resumable(unittest.TestCase): def test_recoverable_causes(self): for detail in CLEARS_ON_ITS_OWN: with self.subTest(detail=detail): self.assertTrue(watchdog.resumable(detail)) def test_human_or_permanent_causes(self): for detail in STAYS_PARKED: with self.subTest(detail=detail): self.assertFalse(watchdog.resumable(detail)) class ResetMoment(unittest.TestCase): def setUp(self): self.blocked_at = datetime(2026, 8, 4, 9, 0).astimezone() def test_spend_and_credit_blocks_retry_hourly(self): for detail in [ "You've hit your monthly spend limit · raise it at claude.ai/settings/usage?from=cc_cli_limit_message", "You've hit your usage credit limit · raise it at claude.ai/settings/usage", "You're out of usage credits", "Your org is out of usage · add funds to continue", ]: with self.subTest(detail=detail): self.assertEqual( watchdog.reset_moment(detail, self.blocked_at), self.blocked_at + timedelta(hours=1), ) def test_stated_reset_wins_when_it_falls_inside_the_window(self): detail = "You've hit your session limit · resets 11am" self.assertEqual( watchdog.reset_moment(detail, self.blocked_at), self.blocked_at.replace(hour=11), ) def test_unstated_reset_falls_back_to_the_limit_window(self): self.assertEqual( watchdog.reset_moment("You've hit your session limit", self.blocked_at), self.blocked_at + watchdog.SESSION_WINDOW, ) self.assertEqual( watchdog.reset_moment("You've hit your weekly limit", self.blocked_at), self.blocked_at + watchdog.WEEKLY_WINDOW, ) def test_transient_failures_retry_at_once(self): self.assertIsNone(watchdog.reset_moment("API error — see detail", self.blocked_at)) class MemoryStrain(unittest.TestCase): """A host with no room left must not be given another session to run.""" def meminfo(self, available_kb, swap_total_kb=4194300, swap_free_kb=4194300): temp = tempfile.NamedTemporaryFile("w", suffix=".meminfo", delete=False) self.addCleanup(os.unlink, temp.name) temp.write( "MemTotal: 32877496 kB\n" "MemFree: 1048576 kB\n" f"MemAvailable: {available_kb} kB\n" f"SwapTotal: {swap_total_kb} kB\n" f"SwapFree: {swap_free_kb} kB\n" ) temp.close() return Path(temp.name) def test_healthy_host_takes_another_session(self): self.assertIsNone(watchdog.memory_strain(2048, 50, self.meminfo(30083284))) def test_low_memory_holds_off(self): strain = watchdog.memory_strain(2048, 50, self.meminfo(1048576)) self.assertIn("1024 MB available", strain) def test_swap_in_heavy_use_holds_off(self): strain = watchdog.memory_strain(2048, 50, self.meminfo(30083284, swap_free_kb=1048575)) self.assertIn("swap 75% used", strain) def test_swapless_host_is_judged_on_memory_alone(self): self.assertIsNone( watchdog.memory_strain(2048, 50, self.meminfo(30083284, swap_total_kb=0, swap_free_kb=0)) ) def test_host_that_does_not_report_memory_is_not_blocked(self): self.assertIsNone(watchdog.memory_strain(2048, 50, Path("/nonexistent/meminfo"))) kernel_without_memavailable = self.meminfo(0) kernel_without_memavailable.write_text("MemTotal: 32877496 kB\n") self.assertIsNone(watchdog.memory_strain(2048, 50, kernel_without_memavailable)) class Candidates(unittest.TestCase): """The whole path: a spend-limited background job with no process left.""" SPEND = "You've hit your monthly spend limit · raise it at claude.ai/settings/usage?from=cc_cli_limit_message" def setUp(self): temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) self.jobs = Path(temp.name) self.now = datetime.now().astimezone() def blocked_job(self, detail, blocked_ago): job_id, session_id = uuid.uuid7().hex[:8], str(uuid.uuid7()) state = self.jobs / job_id / "state.json" state.parent.mkdir() state.write_text(json.dumps({"detail": detail, "state": "blocked"})) stamp = (self.now - blocked_ago).timestamp() os.utime(state, (stamp, stamp)) return {"cwd": str(self.jobs), "id": job_id, "kind": "background", "sessionId": session_id} def test_spend_block_is_picked_up_once_its_hour_is_up(self): session = self.blocked_job(self.SPEND, timedelta(hours=2)) found = watchdog.candidates([session], self.jobs, self.now) self.assertEqual([job["id"] for job in found], [session["id"]]) self.assertEqual(found[0]["state_file"], self.jobs / session["id"] / "state.json") def test_fresh_spend_block_waits(self): session = self.blocked_job(self.SPEND, timedelta(minutes=10)) self.assertEqual(watchdog.candidates([session], self.jobs, self.now), []) class MarkSuperseded(unittest.TestCase): """The husk keeps its conversation, so its name must not read as the live one.""" def setUp(self): temp = tempfile.TemporaryDirectory() self.addCleanup(temp.cleanup) self.state = Path(temp.name) / "state.json" def write(self, **fields): self.state.write_text(json.dumps(fields)) def read(self): return json.loads(self.state.read_text()) def test_name_gains_the_prefix(self): self.write(name="rewrite larvitsmpp typescript esm", state="blocked") self.assertIsNone(watchdog.mark_superseded(self.state, "X ")) self.assertEqual(self.read()["name"], "X rewrite larvitsmpp typescript esm") def test_the_respawn_name_stays_in_step(self): self.write(name="fejkdata", respawnFlags=["--agent", "claude", "--name", "fejkdata"]) watchdog.mark_superseded(self.state, "X ") self.assertEqual(self.read()["respawnFlags"], ["--agent", "claude", "--name", "X fejkdata"]) def test_a_husk_is_prefixed_only_once(self): self.write(name="X fejkdata") self.assertIsNone(watchdog.mark_superseded(self.state, "X ")) self.assertEqual(self.read()["name"], "X fejkdata") def test_an_unnamed_job_is_left_alone(self): self.write(state="blocked") self.assertIsNone(watchdog.mark_superseded(self.state, "X ")) self.assertNotIn("name", self.read()) def test_unwritable_state_is_reported_rather_than_raised(self): self.assertIsNotNone(watchdog.mark_superseded(self.state.with_name("gone.json"), "X ")) if __name__ == "__main__": unittest.main()