hook

"""Exercise keeper state/locking with fake time and server, never real Claude."""
import os
from pathlib import Path
import subprocess
import signal
import sys
import tempfile
import time
import unittest


SOURCE = Path(__file__).resolve().parents[1] / "claude-rc"
HARNESS = r'''
source "$RC_TEST_SOURCE"
STATE_DIR="$RC_TEST_DIR"
STATE="$STATE_DIR/state"
LOG="$STATE_DIR/log"
PROJECT_DIR="$STATE_DIR"
now() { cat "$STATE_DIR/time"; }
running_pid() { [ -f "$STATE_DIR/running" ] && echo 12345; }
prereqs_ok() { [ -f "$STATE_DIR/prereqs" ]; }
has_cron() { return 1; }
cron_register() { :; }
spawn_server() {
  echo "$(now)" >> "$STATE_DIR/starts"
  if [ -f "$STATE_DIR/connect" ]; then
    sleep 0.1
    touch "$STATE_DIR/running"
  fi
}
main "$@"
'''


class RecoveryTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="claude-rc-test-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.env = dict(os.environ, RC_TEST_SOURCE=str(SOURCE), RC_TEST_DIR=str(self.root))
        self.env.pop("CLAUDE_RC_DISABLE", None)
        self.write_state(state="enabled")
        self.at(10000)

    def write_state(self, **state):
        (self.root / "state").write_text("".join(f"{k}={v}\n" for k, v in state.items()))

    def state(self):
        return dict(line.split("=", 1) for line in (self.root / "state").read_text().splitlines())

    def at(self, stamp):
        (self.root / "time").write_text(str(stamp))

    def run_keeper(self, verb="check", expected=0):
        p = subprocess.run(["bash", "-c", HARNESS, "test-keeper", verb],
                           env=self.env, capture_output=True, text=True, timeout=15)
        self.assertEqual(p.returncode, expected, p.stdout + p.stderr)
        return p.stdout

    def starts(self):
        p = self.root / "starts"
        return p.read_text().splitlines() if p.exists() else []

    def exhaust(self):
        for t in range(10000, 10500, 100):
            self.at(t)
            self.run_keeper()
        self.assertEqual(len(self.starts()), 5)
        self.assertEqual(self.state()["state"], "enabled")
        self.at(10500)
        self.run_keeper(expected=2)
        self.assertEqual(self.state()["state"], "failed")
        self.assertEqual(self.state()["retry_after"], "11300")

    def test_cooldown_retries_without_enable_and_limits_repeated_failure(self):
        self.exhaust()
        self.at(11299)
        self.run_keeper(expected=2)
        self.assertEqual(len(self.starts()), 5)
        self.at(11300)
        self.run_keeper()
        self.assertEqual(len(self.starts()), 6)
        self.at(11301)
        self.run_keeper(expected=2)
        self.assertEqual(self.state()["retry_after"], "12200")
        self.at(12200)
        self.run_keeper()
        self.assertEqual(len(self.starts()), 7)

    def test_reconnect_stays_up_without_duplicate(self):
        self.exhaust()
        self.at(11300)
        (self.root / "connect").touch()
        self.run_keeper()
        self.assertEqual(self.state()["state"], "enabled")
        for t in [11301, 11400, 14000]:
            self.at(t)
            self.run_keeper()
        self.assertEqual(len(self.starts()), 6)
        self.assertNotIn("retry_after", self.state())

    def test_old_failed_state_recovers(self):
        self.write_state(state="failed", failed_at=500,
                         launches="100,200,300,400,500", last_launch=500)
        self.run_keeper()
        self.assertEqual(self.state()["state"], "enabled")
        self.assertEqual(len(self.starts()), 1)

    def test_live_fifth_attempt_repairs_old_false_failure(self):
        self.write_state(state="failed", retry_after=20000)
        (self.root / "running").touch()
        self.run_keeper()
        self.assertEqual(self.state()["state"], "enabled")
        self.assertEqual(self.starts(), [])

    def test_disabled_declined_and_unasked_do_not_launch(self):
        for state in ["disabled", "declined", "unasked"]:
            with self.subTest(state=state):
                self.write_state(state=state)
                self.run_keeper()
                self.assertEqual(self.state()["state"], state)
                self.assertEqual(self.starts(), [])

    def test_prompt_hook_obeys_same_cooldown(self):
        self.exhaust()
        self.assertIn("cooling down", self.run_keeper("hook"))
        self.assertEqual(len(self.starts()), 5)
        self.assertEqual(self.run_keeper("hook"), "")
        self.at(11300)
        self.assertIn("Starting Claude Remote Control", self.run_keeper("hook"))
        self.assertEqual(len(self.starts()), 6)

    def test_hook_turns_it_on_silently_once_signed_in(self):
        # 1.3.8 (John 2026-09-28): never ask; the first prompt after sign-in turns it on.
        self.write_state(state="unasked")
        (self.root / "prereqs").touch()
        self.assertEqual(self.run_keeper("hook"), "")
        self.assertEqual(self.state()["state"], "enabled")
        self.assertEqual(self.state()["enabled_by"], "auto")
        self.assertEqual(len(self.starts()), 1)

    def test_hook_stays_silent_and_off_before_sign_in(self):
        self.write_state(state="unasked")
        self.assertEqual(self.run_keeper("hook"), "")
        self.assertEqual(self.state()["state"], "unasked")
        self.assertEqual(self.starts(), [])

    def test_hook_never_overrides_a_human_off(self):
        (self.root / "prereqs").touch()
        for state in ["disabled", "declined"]:
            with self.subTest(state=state):
                self.write_state(state=state)
                self.assertEqual(self.run_keeper("hook"), "")
                self.assertEqual(self.state()["state"], state)
                self.assertEqual(self.starts(), [])

    def test_concurrent_watchdogs_launch_once(self):
        (self.root / "connect").touch()
        processes = [subprocess.Popen(["bash", "-c", HARNESS, "test-keeper", "check"],
                                      env=self.env, stdout=subprocess.PIPE,
                                      stderr=subprocess.PIPE, text=True) for _ in range(8)]
        for p in processes:
            out, err = p.communicate(timeout=15)
            self.assertEqual(p.returncode, 0, out + err)
        self.assertEqual(len(self.starts()), 1)

    def test_clock_rollback_does_not_latch_cooldown(self):
        self.exhaust()
        self.at(9000)
        self.run_keeper()
        self.assertEqual(self.state()["state"], "enabled")

    def test_status_explains_automatic_retry(self):
        self.exhaust()
        self.assertIn("next watchdog check after this time", self.run_keeper("status"))

    def test_real_launch_pipeline_releases_watchdog_lock(self):
        # Use the real setsid/nohup/pipe code, substituting only the Claude binary.
        binary = self.root / "bin" / "claude"
        binary.parent.mkdir()
        binary.write_text(f"#!{sys.executable}\n" + r'''
import os, pathlib, time
p = pathlib.Path(os.environ["RC_TEST_DIR"])
(p / "child-lock").write_text(str(pathlib.Path("/proc/self/fd/9").exists()))
(p / "child-group").write_text(str(os.getpgrp()))
time.sleep(30)
''')
        binary.chmod(0o755)
        harness = HARNESS.split("spawn_server()", 1)[0] + r'''
export PATH="$STATE_DIR/bin:$PATH"
main check
'''
        group_file = self.root / "child-group"
        try:
            p = subprocess.run(["bash", "-c", harness], env=self.env,
                               capture_output=True, text=True, timeout=10)
            self.assertEqual(p.returncode, 0, p.stdout + p.stderr)
            deadline = time.monotonic() + 5
            while not group_file.exists() and time.monotonic() < deadline:
                time.sleep(0.02)
            self.assertTrue(group_file.exists(), "fake daemon did not start")
            self.assertEqual((self.root / "child-lock").read_text(), "False")
            p = subprocess.run(["flock", "-n", str(self.root / "claude-rc.lock"), "true"])
            self.assertEqual(p.returncode, 0, "daemon retained the watchdog lock")
        finally:
            if group_file.exists():
                group = int(group_file.read_text())
                if group != os.getpgrp():
                    try:
                        os.killpg(group, signal.SIGTERM)
                    except ProcessLookupError:
                        pass


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