diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 20be139c3b01..44867e1a5218 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -671,7 +671,17 @@ class _CompressionLockLeaseRefresher: # by the TTL the acquirer set — the lock can never be held past its TTL # by a stuck refresher. consecutive_failures = 0 - while not self._stop.wait(self._refresh_interval_seconds): + # First refresh happens immediately, not one interval late. Everything + # between try_acquire() and start() (the rotation-ownership lookup, the + # durable-breaker re-read, thread startup) is charged against the very + # first lease, so on a short TTL under load the lock could already be + # expired — and reclaimable by a competing path — before tick #1. + first = True + while first or not self._stop.wait(self._refresh_interval_seconds): + if first: + first = False + if self._stop.is_set(): + break try: refreshed = self._db.refresh_compression_lock( self._session_id, diff --git a/hermes_state.py b/hermes_state.py index 2a4af57978b4..0fc1364c2ad8 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -57,12 +57,6 @@ def _compression_lock_holder_process_is_dead(holder: str) -> bool: remain protected until normal TTL expiry (conservative: PID reuse must never steal a live lease, and a wrongly-kept lease self-heals via TTL). """ - # Windows stays TTL-only: stdlib os.kill(pid, 0) is NOT a no-op probe - # there (bpo-14484 — sig=0 maps to CTRL_C_EVENT and can kill the target's - # console group), and PID recycling semantics make liveness a weaker - # deadness signal. The 300s lease TTL remains the recovery path. - if os.name == "nt": - return False match = _COMPRESSION_LOCK_HOLDER_PID_RE.search(holder or "") if match is None: return False @@ -84,10 +78,14 @@ def _compression_lock_holder_process_is_dead(holder: str) -> bool: return not psutil.pid_exists(pid) except Exception: return False # any doubt → keep the lease until TTL expiry - # Scaffold-phase fallback only (psutil missing). POSIX-only by the - # os.name gate above. + # Scaffold-phase fallback only (psutil missing), and POSIX-only: stdlib + # os.kill(pid, 0) is NOT a no-op probe on Windows (bpo-14484 — sig=0 maps + # to CTRL_C_EVENT and can kill the target's console group). Without psutil + # a Windows host stays TTL-only; the lease TTL remains the recovery path. + if os.name == "nt": + return False try: - os.kill(pid, 0) # windows-footgun: ok — function early-returns on nt above + os.kill(pid, 0) # windows-footgun: ok — nt early-returns just above except ProcessLookupError: return True except (PermissionError, OSError, OverflowError): @@ -4592,7 +4590,24 @@ class SessionDB: holder: str, ttl_seconds: float = 300.0, ) -> bool: - """Extend the compression lock lease if ``holder`` still owns it.""" + """Extend the compression lock lease if ``holder`` still owns it. + + Ownership is decided by the ``holder`` column alone, deliberately NOT + by ``expires_at``: a live owner whose refresher thread was starved + (GC pause, loaded CI runner, a slow write escaping ``_execute_write``'s + retry budget) past its own TTL must be able to revive its still-unclaimed + row on the next tick. Requiring ``expires_at >= now`` here made such a + stall permanent — every later refresh matched 0 rows, so the owner kept + compressing and rotating with no lease at all, which is exactly the + unprotected window a competing path can fork the session lineage in. + + This does not resurrect a lock somebody else already took: SQLite + serialises writes, so a reclaim (DELETE-expired + INSERT-or-IGNORE in + :meth:`try_acquire_compression_lock`) and this UPDATE never interleave. + Reclaim-first replaces ``holder``, so this UPDATE matches nothing and + returns False; refresh-first pushes ``expires_at`` into the future, so + the reclaimer's DELETE-expired matches nothing and its acquire fails. + """ if not session_id or not holder: return False now = time.time() @@ -4601,8 +4616,8 @@ class SessionDB: def _do(conn): cur = conn.execute( "UPDATE compression_locks SET expires_at = ? " - "WHERE session_id = ? AND holder = ? AND expires_at >= ?", - (expires_at, session_id, holder, now), + "WHERE session_id = ? AND holder = ?", + (expires_at, session_id, holder), ) return cur.rowcount > 0 diff --git a/tests/agent/test_compression_concurrent_fork.py b/tests/agent/test_compression_concurrent_fork.py index 8af78d5ccdb6..b7c696ecb75b 100644 --- a/tests/agent/test_compression_concurrent_fork.py +++ b/tests/agent/test_compression_concurrent_fork.py @@ -94,6 +94,21 @@ def _count_children(db: SessionDB, parent_sid: str) -> int: return len(rows) +def _live_child_id(db: SessionDB, parent_sid: str) -> str | None: + """The single child id of ``parent_sid``, or None when there is none. + + Fails loudly on more than one child: callers use this to prove the agents + converged on the winner's session, so a multi-child state is a fork and + must not be silently reduced to 'the first row'. + """ + rows = db._conn.execute( + "SELECT id FROM sessions WHERE parent_session_id = ?", + (parent_sid,), + ).fetchall() + assert len(rows) <= 1, f"expected at most one child of {parent_sid}, got {rows!r}" + return rows[0][0] if rows else None + + def _wait_for_touch(touch_calls: list[str], value: str, timeout: float = 1.0) -> None: deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -293,21 +308,30 @@ def test_concurrent_compression_does_not_fork_session(tmp_path: Path) -> None: "the concurrent rotations." ) - # The number of agents that rotated their session_id must match the number - # of children created — and must never exceed one. (Both rotating would be - # the fork; the winner rolling back to parent under contention yields zero, - # which agrees with zero children.) - rotated = sum( - 1 for a in (agent_a, agent_b) if a.session_id != parent_sid + # Every agent that moved off the parent must have landed on the SAME id. + # Counting movers is the wrong contract: the loser can legitimately end up + # on the child too, without rotating anything itself — it takes the lock + # after the winner released it, sees the parent was already rotated, and + # _adopt_live_compression_child() points it at the winner's single child + # (the "compression recovery: stale session=... adopted live child=..." + # path). That convergence is the fix working, not a fork; the fork is two + # DIFFERENT live ids. Asserting ``movers <= 1`` failed on that healthy + # outcome under concurrent load. + moved = {a.session_id for a in (agent_a, agent_b) if a.session_id != parent_sid} + assert len(moved) <= 1, ( + f"Expected at most one post-compression session id, got {sorted(moved)}. " + "Two distinct ids means the lock didn't serialize them (transcript fork)." ) - assert rotated <= 1, ( - f"Expected at most one agent to rotate session_id, got {rotated}. " - "More than one rotating means the lock didn't serialize them." - ) - assert rotated == n_children, ( - f"Inconsistent state: {rotated} agent(s) rotated but {n_children} " + assert len(moved) == n_children, ( + f"Inconsistent state: agents live on {sorted(moved)} but {n_children} " "child session(s) exist — rotation and child creation diverged." ) + if moved: + child = _live_child_id(db, parent_sid) + assert moved == {child}, ( + f"Agents live on {sorted(moved)} but the parent's only child is " + f"{child} — an agent is writing to a session outside the lineage." + ) # The lock must be released after both paths finished, regardless of # whether the winner committed a child or rolled back. @@ -1532,6 +1556,55 @@ def test_lease_refresher_survives_single_transient_failure() -> None: ) +def test_lease_refresher_first_refresh_is_immediate() -> None: + """Tick #1 must land before the first wait, not one interval late. + + The lease clock starts at try_acquire(), but the refresher only starts + after the rotation-ownership lookup, the durable-breaker re-read and thread + startup. Waiting a full interval before the first refresh charges all of + that against the acquirer's first lease, so under load a short TTL can + expire — and be reclaimed by a competitor — before the owner ever renews. + """ + from agent.conversation_compression import _CompressionLockLeaseRefresher + + db = _FlakyRefreshDB([]) # always succeeds + refresher = _CompressionLockLeaseRefresher( + db, "sess", "holder", ttl_seconds=10.0, refresh_interval_seconds=2.0 + ) + + calls_before_first_wait: list[int] = [] + + def _wait(_interval: float) -> bool: + calls_before_first_wait.append(db.calls) + return True # stop after the first wait + + refresher._stop.wait = _wait # type: ignore[assignment] + refresher._run() + + assert calls_before_first_wait and calls_before_first_wait[0] == 1, ( + "Refresher waited a full interval before its first refresh — the lease " + f"is renewed one interval late (calls at first wait: " + f"{calls_before_first_wait!r})." + ) + + +def test_lease_refresher_immediate_tick_still_honors_stop() -> None: + """A refresher stopped before/at startup must not fire the immediate tick.""" + from agent.conversation_compression import _CompressionLockLeaseRefresher + + db = _FlakyRefreshDB([]) + refresher = _CompressionLockLeaseRefresher( + db, "sess", "holder", ttl_seconds=10.0, refresh_interval_seconds=2.0 + ) + refresher._stop.set() # released before the thread got to run + refresher._run() + + assert db.calls == 0, ( + "The immediate first tick must not resurrect a lock whose owner already " + f"released it (refresh calls after stop(): {db.calls})." + ) + + def test_lease_refresher_failure_window_is_bounded_by_ttl() -> None: """Persistent failure stops within one lease's worth of time, not forever. diff --git a/tests/test_hermes_state_compression_locks.py b/tests/test_hermes_state_compression_locks.py index c82b2a706a2d..aeeda8cb70f4 100644 --- a/tests/test_hermes_state_compression_locks.py +++ b/tests/test_hermes_state_compression_locks.py @@ -105,6 +105,11 @@ def test_non_expired_lock_is_held(db: SessionDB) -> None: def test_non_expired_lock_from_dead_pid_is_reclaimed( db: SessionDB, monkeypatch: pytest.MonkeyPatch ) -> None: + # PID probing is POSIX-only by design (see + # test_windows_uses_ttl_only_without_pid_probe). Pin the platform so this + # exercises the probe branch on Windows dev machines too, instead of + # silently asserting the nt early-return. + monkeypatch.setattr(hermes_state.os, "name", "posix") dead_holder = "pid=424242:tid=1:agent=abc:nonce=deadbeef" assert db.try_acquire_compression_lock( "sess1", dead_holder, ttl_seconds=300 @@ -129,7 +134,13 @@ def test_non_expired_lock_from_dead_pid_is_reclaimed( def test_dead_pid_reclaim_via_os_kill_fallback_when_psutil_missing( db: SessionDB, monkeypatch: pytest.MonkeyPatch ) -> None: - """Scaffold-phase installs (no psutil) fall back to os.kill(pid, 0).""" + """Scaffold-phase installs (no psutil) fall back to os.kill(pid, 0). + + POSIX-only path: on Windows ``os.kill(pid, 0)`` is not a probe (sig 0 is + CTRL_C_EVENT, bpo-14484), so the production code early-returns there. Pin + the platform to keep this branch covered on Windows dev machines. + """ + monkeypatch.setattr(hermes_state.os, "name", "posix") dead_holder = "pid=424242:tid=1:agent=abc:nonce=deadbeef" assert db.try_acquire_compression_lock( "sess1", dead_holder, ttl_seconds=300 @@ -236,27 +247,60 @@ def test_unstructured_holder_waits_for_ttl( ) is False -def test_windows_uses_ttl_only_without_pid_probe( +def test_windows_reclaims_dead_holder_via_psutil_probe( db: SessionDB, monkeypatch: pytest.MonkeyPatch ) -> None: + """psutil.pid_exists is safe on Windows, so nt hosts do reclaim dead holders. + + The os.kill(pid, 0) footgun (bpo-14484) only affects the no-psutil + fallback, not this path — see the companion test below. Gating the whole + function on nt used to strand every crashed holder for the full 300s TTL. + """ holder = "pid=424242:tid=1:agent=abc:nonce=windows" assert db.try_acquire_compression_lock( "sess1", holder, ttl_seconds=300 ) is True monkeypatch.setattr(hermes_state.os, "name", "nt") + probed: list[int] = [] + + def _pid_exists(pid: int) -> bool: + probed.append(pid) + return False # holder process is gone + monkeypatch.setattr( - hermes_state, - "psutil", - SimpleNamespace( - pid_exists=lambda _pid: pytest.fail( - "Windows must stay TTL-only — no PID probe" - ) - ), + hermes_state, "psutil", SimpleNamespace(pid_exists=_pid_exists) ) monkeypatch.setattr( hermes_state.os, "kill", - lambda *_args: pytest.fail("Windows must not use os.kill as a PID probe"), + lambda *_args: pytest.fail("Windows must never use os.kill as a PID probe"), + ) + + assert db.try_acquire_compression_lock( + "sess1", "pid=525252:tid=2:agent=def:nonce=other", ttl_seconds=300 + ) is True + assert probed == [424242] + + +def test_windows_without_psutil_stays_ttl_only( + db: SessionDB, monkeypatch: pytest.MonkeyPatch +) -> None: + """Scaffold installs on nt keep the lease: os.kill(pid, 0) is not a probe. + + On Windows signal 0 maps to CTRL_C_EVENT (bpo-14484) and can kill the + target's console group, so with psutil absent the only safe answer is + "assume alive" and let the TTL expire the lease. + """ + holder = "pid=424242:tid=1:agent=abc:nonce=windows" + assert db.try_acquire_compression_lock( + "sess1", holder, ttl_seconds=300 + ) is True + monkeypatch.setattr(hermes_state.os, "name", "nt") + monkeypatch.setattr(hermes_state, "psutil", None) + monkeypatch.setattr( + hermes_state.os, + "kill", + lambda *_args: pytest.fail("Windows must never use os.kill as a PID probe"), ) assert db.try_acquire_compression_lock(