import json
import os
import threading
import time
import unittest
import urllib.error
import urllib.request
from http.server import HTTPServer, ThreadingHTTPServer
from unittest.mock import patch

import companion.server as server


class AttentionStateTests(unittest.TestCase):
    def setUp(self):
        server.NTFY_TOPIC = ""
        with server._attention_lock:
            server._attention_records.clear()
            server._attention_seq = 0
        server._invalidate_usage_cache()

    def post_event(self, **overrides):
        payload = {
            "provider": "Codex",
            "event": "PermissionRequest",
            "session_id": "test-session",
            "cwd": "/Users/jameslopez/projects/CodexBar-ESP32",
        }
        payload.update(overrides)
        return server.ingest_attention_event(payload)

    def codex_attention(self):
        return server.get_status_payload()["providers"][1]["attention"]

    def test_permission_request_sets_needs_user_with_project_name(self):
        ok, response = self.post_event(provider="codex")

        self.assertTrue(ok)
        self.assertEqual(response, {"ok": True})
        attention = self.codex_attention()
        self.assertEqual(attention["state"], "needs_user")
        self.assertEqual(attention["count"], 1)
        self.assertEqual(attention["primary_project"], "CodexBar-ESP32")
        self.assertEqual(attention["projects"], ["CodexBar-ESP32"])
        self.assertEqual(attention["reason"], "permission_request")

    def test_session_end_clears_attention_record(self):
        self.post_event()
        ok, response = self.post_event(event="SessionEnd")

        self.assertTrue(ok)
        self.assertEqual(response, {"ok": True})
        self.assertEqual(self.codex_attention()["state"], "idle")

    def test_stop_defaults_to_done(self):
        ok, _ = self.post_event(event="Stop")

        self.assertTrue(ok)
        attention = self.codex_attention()
        self.assertEqual(attention["state"], "done")
        self.assertEqual(attention["reason"], "stop")

    def test_stop_with_question_is_classified_as_needs_user(self):
        ok, _ = self.post_event(
            event="Stop",
            last_assistant_message="I can implement either approach. Which option do you prefer?",
        )

        self.assertTrue(ok)
        attention = self.codex_attention()
        self.assertEqual(attention["state"], "needs_user")
        self.assertEqual(attention["reason"], "possible_question")

    def test_stop_with_choice_list_is_classified_as_needs_user(self):
        ok, _ = self.post_event(
            event="Stop",
            last_assistant_message="Possible next steps:\n1. Wire hooks globally\n2. Test on the device",
        )

        self.assertTrue(ok)
        attention = self.codex_attention()
        self.assertEqual(attention["state"], "needs_user")
        self.assertEqual(attention["reason"], "possible_question")

    def test_done_record_expires_after_timeout(self):
        self.post_event(event="Stop")

        with server._attention_lock:
            for rec in server._attention_records.values():
                rec["updated_at"] = 0

        self.assertEqual(self.codex_attention()["state"], "idle")

    def test_notification_attention_expires_quickly(self):
        self.post_event(event="Notification")

        with server._attention_lock:
            for rec in server._attention_records.values():
                rec["updated_at"] = int(time.time()) - server.ATTENTION_REASON_TIMEOUTS_SEC["notification"] - 1

        self.assertEqual(self.codex_attention()["state"], "idle")

    def test_auto_approval_notification_stays_running(self):
        self.post_event(event="SessionStart")
        ok, _ = self.post_event(
            event="Notification",
            message="Automatic approval review approved: Bash(ls -la)",
        )

        self.assertTrue(ok)
        self.assertEqual(self.codex_attention()["state"], "running")

    def test_auto_approval_notification_does_not_send_ntfy(self):
        self.post_event(event="SessionStart")
        fired = []
        original = server._send_ntfy_async

        def capture(*args, **kwargs):
            fired.append((args, kwargs))
            return original(*args, **kwargs)

        with patch.object(server, "_send_ntfy_async", side_effect=capture):
            self.post_event(
                event="Notification",
                message="Automatic approval review approved: Bash(echo hi)",
            )

        self.assertEqual(fired, [])

    def test_stop_after_running_sends_done_notification(self):
        self.post_event(event="SessionStart")
        fired = []
        with patch.object(server, "_send_ntfy_async", side_effect=lambda *a, **kw: fired.append((a, kw))):
            self.post_event(event="Stop")

        self.assertEqual(len(fired), 1)
        _, kwargs = fired[0]
        self.assertEqual(kwargs.get("kind"), "done")

    def test_stop_does_not_send_done_notification_when_disabled(self):
        original = server.NTFY_NOTIFY_DONE
        server.NTFY_NOTIFY_DONE = False
        try:
            self.post_event(event="SessionStart")
            fired = []
            with patch.object(server, "_send_ntfy_async", side_effect=lambda *a, **kw: fired.append((a, kw))):
                self.post_event(event="Stop")
            self.assertEqual(fired, [])
        finally:
            server.NTFY_NOTIFY_DONE = original

    def test_needs_user_notification_has_high_priority(self):
        captured_headers = {}

        class FakeResp:
            def __enter__(self): return self
            def __exit__(self, *a): return False

        class ImmediateThread:
            def __init__(self, target, daemon=False): self.target = target
            def start(self): self.target()

        def capture(req, timeout=0):
            captured_headers.update(req.headers)
            return FakeResp()

        with patch.object(server, "NTFY_TOPIC", "test-topic"), \
             patch.object(server, "NTFY_SSH_HOST", ""), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_ntfy_async("Claude", "MyProject", "notification")

        self.assertEqual(captured_headers.get("Priority"), "high")

    def test_done_notification_has_low_priority(self):
        captured_headers = {}

        class FakeResp:
            def __enter__(self): return self
            def __exit__(self, *a): return False

        class ImmediateThread:
            def __init__(self, target, daemon=False): self.target = target
            def start(self): self.target()

        def capture(req, timeout=0):
            captured_headers.update(req.headers)
            return FakeResp()

        with patch.object(server, "NTFY_TOPIC", "test-topic"), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_ntfy_async("Claude", "MyProject", "stop", kind="done")

        self.assertEqual(captured_headers.get("Priority"), "low")

    def test_genuine_notification_still_sets_needs_user(self):
        ok, _ = self.post_event(event="Notification")

        self.assertTrue(ok)
        self.assertEqual(self.codex_attention()["state"], "needs_user")
        self.assertEqual(self.codex_attention()["reason"], "notification")

    def test_permission_request_uses_longer_needs_user_timeout(self):
        self.post_event(event="PermissionRequest")

        with server._attention_lock:
            for rec in server._attention_records.values():
                rec["updated_at"] = int(time.time()) - server.ATTENTION_REASON_TIMEOUTS_SEC["notification"] - 1

        self.assertEqual(self.codex_attention()["state"], "needs_user")

    def test_new_session_start_replaces_old_same_project_record(self):
        self.post_event(event="Stop", session_id="old-session")
        self.post_event(event="SessionStart", session_id="new-session")

        attention = self.codex_attention()
        self.assertEqual(attention["state"], "running")
        self.assertEqual(attention["count"], 1)
        with server._attention_lock:
            self.assertEqual(len(server._attention_records), 1)

    def test_multiple_waiting_projects_are_counted_and_project_list_is_capped(self):
        for idx in range(4):
            self.post_event(
                session_id=f"session-{idx}",
                cwd=f"/Users/jameslopez/projects/project-{idx}",
            )

        attention = self.codex_attention()
        self.assertEqual(attention["state"], "needs_user")
        self.assertEqual(attention["count"], 4)
        self.assertEqual(len(attention["projects"]), 3)
        self.assertEqual(attention["primary_project"], "project-3")

    def test_unknown_provider_is_rejected(self):
        ok, response = self.post_event(provider="ChatGPT")

        self.assertFalse(ok)
        self.assertEqual(response["ok"], False)
        self.assertIn("provider", response["error"])

    def test_lm_studio_completion_is_accepted(self):
        ok, response = self.post_event(provider="lmstudio", event="TaskComplete")

        self.assertTrue(ok)
        self.assertEqual(response, {"ok": True})

    def test_lm_studio_completion_sends_done_notification_without_prior_start(self):
        fired = []
        with patch.object(server, "_send_ntfy_async", side_effect=lambda *a, **kw: fired.append((a, kw))):
            self.post_event(provider="LM Studio", event="TaskComplete")

        self.assertEqual(len(fired), 1)
        args, kwargs = fired[0]
        self.assertEqual(args[:3], ("LM Studio", "CodexBar-ESP32", "task_complete"))
        self.assertEqual(kwargs.get("kind"), "done")

    def test_lm_studio_is_not_returned_in_status_payload(self):
        self.post_event(provider="LM Studio", event="TaskComplete")

        names = [provider["name"] for provider in server.get_status_payload()["providers"]]
        self.assertEqual(names, ["Claude", "Codex"])

    def test_ntfy_includes_ssh_action_when_configured(self):
        captured_headers = {}

        class FakeResp:
            def __enter__(self): return self
            def __exit__(self, *a): return False

        class ImmediateThread:
            def __init__(self, target, daemon=False): self.target = target
            def start(self): self.target()

        def capture(req, timeout=0):
            captured_headers.update(req.headers)
            return FakeResp()

        with patch.object(server, "NTFY_TOPIC", "test-topic"), \
             patch.object(server, "NTFY_SSH_HOST", "jamess-mac-mini.tail02a1a0.ts.net"), \
             patch.object(server, "NTFY_SSH_USER", "james"), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_ntfy_async("Claude", "MyProject", "notification")

        self.assertIn("Actions", captured_headers)
        self.assertIn("ssh://james@jamess-mac-mini.tail02a1a0.ts.net", captured_headers["Actions"])

    def test_ntfy_omits_ssh_action_when_not_configured(self):
        captured_headers = {}

        class FakeResp:
            def __enter__(self): return self
            def __exit__(self, *a): return False

        class ImmediateThread:
            def __init__(self, target, daemon=False): self.target = target
            def start(self): self.target()

        def capture(req, timeout=0):
            captured_headers.update(req.headers)
            return FakeResp()

        with patch.object(server, "NTFY_TOPIC", "test-topic"), \
             patch.object(server, "NTFY_SSH_HOST", ""), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_ntfy_async("Claude", "MyProject", "notification")

        self.assertNotIn("Actions", captured_headers)

    def test_startup_does_not_kill_existing_process_by_default(self):
        with patch.dict(os.environ, {}, clear=False):
            os.environ.pop("CODEXBAR_REPLACE_EXISTING", None)
            with patch.object(server, "_pids_on_port", return_value=[12345]), \
                 patch.object(server.os, "getpid", return_value=99999), \
                 patch.object(server.os, "kill") as mock_kill:
                self.assertFalse(server._replace_existing_if_requested())
                mock_kill.assert_not_called()

    def test_startup_can_replace_existing_process_when_requested(self):
        with patch.dict(os.environ, {"CODEXBAR_REPLACE_EXISTING": "1"}), \
             patch.object(server, "_pids_on_port", return_value=[12345]), \
             patch.object(server.os, "getpid", return_value=99999), \
             patch.object(server.os, "kill") as mock_kill, \
             patch.object(server.time, "time", side_effect=[0, 4]):
            self.assertFalse(server._replace_existing_if_requested())
            mock_kill.assert_called_once()

    def test_startup_can_replace_existing_process_when_port_clears(self):
        with patch.dict(os.environ, {"CODEXBAR_REPLACE_EXISTING": "1"}), \
             patch.object(server, "_pids_on_port", side_effect=[[12345], []]), \
             patch.object(server.os, "getpid", return_value=99999), \
             patch.object(server.os, "kill") as mock_kill:
            self.assertTrue(server._replace_existing_if_requested())
            mock_kill.assert_called_once()


class AttentionHttpTests(unittest.TestCase):
    def setUp(self):
        server.NTFY_TOPIC = ""
        with server._attention_lock:
            server._attention_records.clear()
            server._attention_seq = 0
        server._invalidate_usage_cache()
        self.httpd = HTTPServer(("127.0.0.1", 0), server.Handler)
        self.port = self.httpd.server_address[1]
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()

    def tearDown(self):
        self.httpd.shutdown()
        self.httpd.server_close()
        self.thread.join(timeout=2)

    def request_json(self, path, payload=None, method=None):
        url = f"http://127.0.0.1:{self.port}{path}"
        data = None if payload is None else json.dumps(payload).encode()
        headers = {"Content-Type": "application/json"} if payload is not None else {}
        req = urllib.request.Request(url, data=data, headers=headers, method=method)
        with urllib.request.urlopen(req, timeout=5) as resp:
            return resp.status, json.loads(resp.read())

    def test_post_attention_then_get_status(self):
        status, body = self.request_json(
            "/attention",
            {
                "provider": "Claude",
                "event": "PermissionRequest",
                "session_id": "http-session",
                "cwd": "/Users/jameslopez/projects/CodexBar-ESP32",
            },
            method="POST",
        )
        self.assertEqual(status, 200)
        self.assertEqual(body, {"ok": True})

        status, body = self.request_json("/status")
        self.assertEqual(status, 200)
        claude = body["providers"][0]["attention"]
        self.assertEqual(claude["state"], "needs_user")
        self.assertEqual(claude["primary_project"], "CodexBar-ESP32")

    def test_post_attention_rejects_bad_json(self):
        req = urllib.request.Request(
            f"http://127.0.0.1:{self.port}/attention",
            data=b"{",
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with self.assertRaises(urllib.error.HTTPError) as ctx:
            urllib.request.urlopen(req, timeout=5)

        self.assertEqual(ctx.exception.code, 400)
        body = json.loads(ctx.exception.read())
        ctx.exception.close()
        self.assertFalse(body["ok"])
        self.assertIn("invalid JSON", body["error"])

    def test_status_does_not_fetch_usage_credentials_or_provider_apis(self):
        with patch.object(server, "get_access_token", side_effect=AssertionError("should not fetch usage")):
            status, body = self.request_json("/status")

        self.assertEqual(status, 200)
        self.assertTrue(body["ok"])
        self.assertEqual(body["providers"][0]["attention"]["state"], "idle")

    def test_health_responds_while_usage_fetch_is_slow(self):
        self.httpd.shutdown()
        self.httpd.server_close()
        self.thread.join(timeout=2)

        server._invalidate_usage_cache()
        self.httpd = ThreadingHTTPServer(("127.0.0.1", 0), server.Handler)
        self.port = self.httpd.server_address[1]
        self.thread = threading.Thread(target=self.httpd.serve_forever, daemon=True)
        self.thread.start()

        def slow_keychain():
            time.sleep(1)
            raise RuntimeError("simulated slow wake")

        usage_error = []

        def request_usage():
            try:
                self.request_json("/usage")
            except Exception as exc:
                usage_error.append(exc)

        with patch.object(server, "get_access_token", side_effect=slow_keychain):
            usage_thread = threading.Thread(target=request_usage)
            usage_thread.start()
            time.sleep(0.1)

            with urllib.request.urlopen(f"http://127.0.0.1:{self.port}/health", timeout=1) as resp:
                self.assertEqual(resp.status, 200)
                self.assertEqual(resp.read(), b"ok")


TWO_DAYS = 2 * 86400


class PaceDetectionTests(unittest.TestCase):
    def setUp(self):
        server.NTFY_TOPIC = ""
        server.NTFY_USAGE_RESET_TOPIC = ""
        server.UNDERPACE_PCT_THRESHOLD = 25
        server.UNDERPACE_MIN_REMAINING_SEC = TWO_DAYS
        server.UNDERPACE_COOLDOWN_SEC = 1800
        server.UNDERPACE_WINDOW = "7d"
        with server._underpace_lock:
            server._underpace_last_fired = 0
        with server._usage_reset_lock:
            server._usage_reset_windows.clear()
        with server._attention_lock:
            server._attention_records.clear()
            server._attention_seq = 0

    def _providers(self, pct, resets_in_sec=3 * 86400, duration_sec=7 * 86400):
        return [{"name": "Claude", "windows": [{"label": "7d", "pct": pct, "resets_in_sec": resets_in_sec, "duration_sec": duration_sec}]}]

    def _reset_provider(self, pct, resets_in_sec, duration_sec, label=None):
        label = label or ("5h" if duration_sec == 18000 else "5d")
        return [{"name": "Claude", "windows": [{"label": label, "pct": pct, "resets_in_sec": resets_in_sec, "duration_sec": duration_sec}]}]

    def _reset_provider_named(self, provider, pct, resets_in_sec, duration_sec, label=None):
        label = label or ("5h" if duration_sec == 18000 else "7d")
        return [{"name": provider, "windows": [{"label": label, "pct": pct, "resets_in_sec": resets_in_sec, "duration_sec": duration_sec}]}]

    # ── _compute_pace (still used for /pace display) ──────────────────────

    def test_compute_pace_on_pace(self):
        # 40% used with 3h remaining of 5h → elapsed 2h/5h = 40% expected → pace 1.0
        pace = server._compute_pace(40, 10800, 18000)
        self.assertAlmostEqual(pace, 1.0, places=2)

    def test_compute_pace_underpace(self):
        # 5% used with 3h remaining of 5h → elapsed 2h/5h = 40% → expected 40% → pace 0.125
        pace = server._compute_pace(5, 10800, 18000)
        self.assertAlmostEqual(pace, 0.125, places=3)

    def test_compute_pace_overpace(self):
        # 80% used with 4h remaining → expected 20% → pace 4.0
        pace = server._compute_pace(80, 14400, 18000)
        self.assertAlmostEqual(pace, 4.0, places=2)

    def test_compute_pace_returns_none_at_start_of_window(self):
        self.assertIsNone(server._compute_pace(0, 18000, 18000))

    def test_compute_pace_returns_none_bad_duration(self):
        self.assertIsNone(server._compute_pace(50, 3600, 0))

    # ── _check_underpace trigger condition ────────────────────────────────

    def _fire(self, **kwargs):
        notifications = []
        with patch.object(server, "_send_underpace_ntfy_async", side_effect=lambda r, n, p=0: notifications.append((r, p))):
            server._check_underpace(self._providers(**kwargs))
        return notifications

    def test_fires_when_under_25pct_and_2d_remaining(self):
        hits = self._fire(pct=12, resets_in_sec=3 * 86400)
        self.assertEqual(len(hits), 1)
        self.assertEqual(hits[0], (3 * 86400, 12))

    def test_fires_at_zero_pct(self):
        self.assertEqual(len(self._fire(pct=0, resets_in_sec=3 * 86400)), 1)

    def test_does_not_fire_at_25pct(self):
        self.assertEqual(self._fire(pct=25, resets_in_sec=3 * 86400), [])

    def test_does_not_fire_above_threshold(self):
        self.assertEqual(self._fire(pct=50, resets_in_sec=3 * 86400), [])

    def test_does_not_fire_when_less_than_2d_remaining(self):
        self.assertEqual(self._fire(pct=10, resets_in_sec=86400 - 1), [])

    def test_fires_at_exactly_2d_remaining(self):
        self.assertEqual(len(self._fire(pct=10, resets_in_sec=TWO_DAYS)), 1)

    def test_respects_cooldown(self):
        with server._underpace_lock:
            server._underpace_last_fired = time.time() - 10
        self.assertEqual(self._fire(pct=10, resets_in_sec=3 * 86400), [])

    def test_fires_again_after_cooldown_expires(self):
        with server._underpace_lock:
            server._underpace_last_fired = time.time() - server.UNDERPACE_COOLDOWN_SEC - 1
        self.assertEqual(len(self._fire(pct=10, resets_in_sec=3 * 86400)), 1)

    # ── get_pace_payload ──────────────────────────────────────────────────

    def test_get_pace_payload_idle_trigger_true(self):
        providers = self._providers(pct=10, resets_in_sec=3 * 86400)
        payload = server.get_pace_payload(providers)
        self.assertTrue(payload["ok"])
        self.assertEqual(payload["pct_threshold"], 25)
        w = payload["windows"][0]
        self.assertEqual(w["label"], "7d")
        self.assertTrue(w["idle_trigger"])
        self.assertIn("pace", w)

    def test_get_pace_payload_idle_trigger_false_at_threshold(self):
        providers = self._providers(pct=25, resets_in_sec=3 * 86400)
        self.assertFalse(server.get_pace_payload(providers)["windows"][0]["idle_trigger"])

    def test_get_pace_payload_idle_trigger_false_low_remaining(self):
        providers = self._providers(pct=10, resets_in_sec=86400)
        self.assertFalse(server.get_pace_payload(providers)["windows"][0]["idle_trigger"])

    # ── usage reset ntfy notifications ───────────────────────────────────

    def _reset_notifications_after(self, before, after, sleeping=False):
        notifications = []
        with patch.object(server, "_is_sleeping_hour", return_value=sleeping), \
             patch.object(server, "_send_usage_reset_ntfy_async", side_effect=lambda *args: notifications.append(args)):
            server._check_usage_resets(before)
            server._check_usage_resets(after)
        return notifications

    def test_reset_detector_does_not_notify_on_first_observation(self):
        notifications = []
        with patch.object(server, "_send_usage_reset_ntfy_async", side_effect=lambda *args: notifications.append(args)):
            server._check_usage_resets(self._reset_provider(pct=0, resets_in_sec=5 * 86400, duration_sec=5 * 86400))

        self.assertEqual(notifications, [])

    def test_long_window_reset_notifies_on_provider_early_reset(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=73, resets_in_sec=2 * 86400, duration_sec=5 * 86400, label="5d"),
            self._reset_provider(pct=0, resets_in_sec=5 * 86400, duration_sec=5 * 86400, label="5d"),
        )

        self.assertEqual(hits, [("Claude", "5d", 5 * 86400, 27, True)])

    def test_long_window_expected_reset_uses_quieter_notification(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=73, resets_in_sec=60, duration_sec=5 * 86400, label="5d"),
            self._reset_provider(pct=0, resets_in_sec=5 * 86400, duration_sec=5 * 86400, label="5d"),
        )

        self.assertEqual(hits, [("Claude", "5d", 5 * 86400, 27, False)])

    def test_seven_day_long_window_reset_is_covered(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=85, resets_in_sec=60, duration_sec=7 * 86400, label="7d"),
            self._reset_provider(pct=0, resets_in_sec=7 * 86400, duration_sec=7 * 86400, label="7d"),
        )

        self.assertEqual(hits, [("Claude", "7d", 7 * 86400, 15, False)])

    def test_simultaneous_five_hour_and_weekly_reset_marks_weekly_early(self):
        before = [{
            "name": "Claude",
            "windows": [
                {"label": "5h", "pct": 70, "resets_in_sec": 60, "duration_sec": 18000},
                {"label": "7d", "pct": 17, "resets_in_sec": -1, "duration_sec": 7 * 86400},
            ],
        }]
        after = [{
            "name": "Claude",
            "windows": [
                {"label": "5h", "pct": 0, "resets_in_sec": 18000, "duration_sec": 18000},
                {"label": "7d", "pct": 0, "resets_in_sec": 7 * 86400, "duration_sec": 7 * 86400},
            ],
        }]

        hits = self._reset_notifications_after(before, after)

        self.assertEqual(hits, [
            ("Claude", "5h", 18000, 30, False),
            ("Claude", "7d", 7 * 86400, 83, True),
        ])

    def test_codex_reset_uses_same_provider_window_detector(self):
        hits = self._reset_notifications_after(
            self._reset_provider_named("Codex", pct=85, resets_in_sec=60, duration_sec=7 * 86400, label="7d"),
            self._reset_provider_named("Codex", pct=0, resets_in_sec=7 * 86400, duration_sec=7 * 86400, label="7d"),
        )

        self.assertEqual(hits, [("Codex", "7d", 7 * 86400, 15, False)])

    def test_five_hour_reset_notifies_when_awake_and_previous_remaining_under_40(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=70, resets_in_sec=60, duration_sec=18000, label="5h"),
            self._reset_provider(pct=0, resets_in_sec=18000, duration_sec=18000, label="5h"),
        )

        self.assertEqual(hits, [("Claude", "5h", 18000, 30, False)])

    def test_five_hour_reset_does_not_notify_during_sleeping_hours(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=70, resets_in_sec=60, duration_sec=18000, label="5h"),
            self._reset_provider(pct=0, resets_in_sec=18000, duration_sec=18000, label="5h"),
            sleeping=True,
        )

        self.assertEqual(hits, [])

    def test_five_hour_reset_requires_previous_remaining_under_40(self):
        hits = self._reset_notifications_after(
            self._reset_provider(pct=60, resets_in_sec=60, duration_sec=18000, label="5h"),
            self._reset_provider(pct=0, resets_in_sec=18000, duration_sec=18000, label="5h"),
        )

        self.assertEqual(hits, [])

    def test_reset_ntfy_message_distinguishes_early_and_expected(self):
        messages = []

        class FakeResponse:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

        class ImmediateThread:
            def __init__(self, target, daemon=False):
                self.target = target

            def start(self):
                self.target()

        def capture(req, timeout=0):
            messages.append(req.data.decode())
            return FakeResponse()

        with patch.object(server, "NTFY_USAGE_RESET_TOPIC", "test-topic"), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target, daemon)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_usage_reset_ntfy_async("Claude", "5d", 5 * 86400, 27, True)
            server._send_usage_reset_ntfy_async("Claude", "5d", 5 * 86400, 27, False)

        self.assertEqual(messages[0], "Claude 5d usage reset early surprise 🎁")
        self.assertEqual(messages[1], "Claude 5d usage refreshed 🔄 You had 27% remaining.")

    def test_reset_ntfy_uses_dedicated_topic(self):
        urls = []

        class FakeResponse:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

        class ImmediateThread:
            def __init__(self, target, daemon=False):
                self.target = target

            def start(self):
                self.target()

        def capture(req, timeout=0):
            urls.append(req.full_url)
            return FakeResponse()

        with patch.object(server, "NTFY_TOPIC", "attention-topic"), \
             patch.object(server, "NTFY_USAGE_RESET_TOPIC", "reset topic"), \
             patch.object(server.threading, "Thread", side_effect=lambda target, daemon=False: ImmediateThread(target, daemon)), \
             patch.object(server.urllib.request, "urlopen", side_effect=capture):
            server._send_usage_reset_ntfy_async("Codex", "7d", 7 * 86400, 15, False)

        self.assertEqual(urls, ["https://ntfy.sh/reset%20topic"])


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