claude-session-watchdog/test-resume-stalled-sessions.py

#!/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 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"]])

    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), [])


if __name__ == "__main__":
    unittest.main()