From 605aa1341c06859f186d198dd89c061de04f26d5 Mon Sep 17 00:00:00 2001 From: Wei Alexander Xin Date: Wed, 15 Jul 2026 02:21:23 -0400 Subject: [PATCH 1/2] test(guard): stabilize daemon spawn race fixtures Model the mocked daemon as live and bypass process-table probing in claim-focused contention tests. Keep freshness handling covered by a deterministic dead-PID test, so scheduler delays cannot turn fake child PIDs into apparent second spawns. --- tests/test_guard_race_2026_05_18.py | 8 ++++++++ tests/test_spawn_lock.py | 32 +++++++++++++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/tests/test_guard_race_2026_05_18.py b/tests/test_guard_race_2026_05_18.py index 61d2063a..3bd2138f 100644 --- a/tests/test_guard_race_2026_05_18.py +++ b/tests/test_guard_race_2026_05_18.py @@ -88,12 +88,20 @@ def _fake_popen(cmd_parts, **kwargs): fake_pid = 900_000 + worker_index return _DummyProc(fake_pid) + def _fake_is_process_alive(pid: int) -> bool: + """Model the spawned fake guard as alive for the contention check.""" + return pid > 0 + # Mock subprocess.Popen so no real daemon is spawned, and mock # find_claude_pid so we don't fail when no Claude is running. with ( _patch("cozempic.guard.subprocess.Popen", side_effect=_fake_popen), _patch("cozempic.guard.find_claude_pid", return_value=12345), _patch("cozempic.guard._cleanup_legacy_pid"), + _patch("cozempic.guard._is_guard_running_for_session", return_value=None), + _patch( + "cozempic.spawn_lock._is_process_alive", side_effect=_fake_is_process_alive + ), ): # Barrier sync: both children pile up here and release at the same instant try: diff --git a/tests/test_spawn_lock.py b/tests/test_spawn_lock.py index cea93eab..67aa7d26 100644 --- a/tests/test_spawn_lock.py +++ b/tests/test_spawn_lock.py @@ -13,6 +13,7 @@ import multiprocessing as mp import os import sys +import tempfile import time import unittest from pathlib import Path @@ -62,10 +63,18 @@ def _fake_popen(cmd_parts, **kwargs): return _DummyProc(900_000 + worker_index) return _real_popen(cmd_parts, **kwargs) + def _fake_is_process_alive(pid: int) -> bool: + """Model the spawned fake guard as alive for the contention check.""" + return pid > 0 + with ( _patch("cozempic.guard.subprocess.Popen", side_effect=_fake_popen), _patch("cozempic.guard.find_claude_pid", return_value=12345), _patch("cozempic.guard._cleanup_legacy_pid"), + _patch("cozempic.guard._is_guard_running_for_session", return_value=None), + _patch( + "cozempic.spawn_lock._is_process_alive", side_effect=_fake_is_process_alive + ), ): try: barrier_handle.wait(timeout=10.0) @@ -433,6 +442,29 @@ def test_env_var_unset_uses_default(self): self.assertEqual(self._reload_with_env(None), _DEFAULT_FRESH) +class TestFreshClaimProtection(unittest.TestCase): + """The race fixtures must not be the only coverage of fresh claims.""" + + def test_fresh_dead_pid_is_not_reclaimed(self): + """A freshly written dead PID is a peer claim, not stale state.""" + from cozempic.spawn_lock import DaemonAlreadyStarting, DaemonSpawnClaim + + with tempfile.TemporaryDirectory() as tmp_dir: + pid_file = Path(tmp_dir) / "guard.pid" + pid_file.write_text("900000\n", encoding="utf-8") + fresh_now = pid_file.stat().st_mtime + 1.0 + + with ( + patch("cozempic.spawn_lock._is_process_alive", return_value=False), + patch("cozempic.spawn_lock.time.time", return_value=fresh_now), + self.assertRaises(DaemonAlreadyStarting) as raised, + ): + with DaemonSpawnClaim("test-session", pid_file): + self.fail("fresh dead PID file was reclaimed") + + self.assertEqual(raised.exception.holder_pid, 900000) + + class TestSymlinkDefense(unittest.TestCase): """Round-3 C1 regression test: the post-Popen ``.pid.tmp`` write must NOT follow symlinks. A local user with write access to ``/tmp`` (a From b6fea7251c191f2c422a794d639f79cbec12a3ce Mon Sep 17 00:00:00 2001 From: Wei Alexander Xin Date: Wed, 15 Jul 2026 02:49:22 -0400 Subject: [PATCH 2/2] Address PR 178 review: honor active freshness window Derive the mocked clock from the configured PID-file freshness window and keep only the claim operation in the expected-exception assertion. --- tests/test_spawn_lock.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/tests/test_spawn_lock.py b/tests/test_spawn_lock.py index 67aa7d26..5fa1b0fd 100644 --- a/tests/test_spawn_lock.py +++ b/tests/test_spawn_lock.py @@ -447,20 +447,22 @@ class TestFreshClaimProtection(unittest.TestCase): def test_fresh_dead_pid_is_not_reclaimed(self): """A freshly written dead PID is a peer claim, not stale state.""" + from cozempic import spawn_lock from cozempic.spawn_lock import DaemonAlreadyStarting, DaemonSpawnClaim with tempfile.TemporaryDirectory() as tmp_dir: pid_file = Path(tmp_dir) / "guard.pid" pid_file.write_text("900000\n", encoding="utf-8") - fresh_now = pid_file.stat().st_mtime + 1.0 + fresh_window = spawn_lock._FRESH_PIDFILE_SECONDS + fresh_now = pid_file.stat().st_mtime + fresh_window / 2 + claim = DaemonSpawnClaim("test-session", pid_file) with ( patch("cozempic.spawn_lock._is_process_alive", return_value=False), patch("cozempic.spawn_lock.time.time", return_value=fresh_now), self.assertRaises(DaemonAlreadyStarting) as raised, ): - with DaemonSpawnClaim("test-session", pid_file): - self.fail("fresh dead PID file was reclaimed") + claim.__enter__() self.assertEqual(raised.exception.holder_pid, 900000)