fix(agent): fail closed on unexpected compression-lock acquisition errors

Splits the single broad `except Exception` in the compression-lock
acquire path into two handlers: AttributeError/TypeError (version skew —
the lock method is missing, or predates the `ttl_seconds=` kwarg) still
fails OPEN as before, since that's known-safe to proceed without a lock.
Any other exception now fails CLOSED (skips compression this cycle)
instead of treating every failure as "no lock subsystem present" and
letting a second compressor run concurrently and fork the session.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

agent/conversation_compression.py

@@ -551,11 +551,13 @@ def compress_context(
     # prints the error and retries.  Because compression never succeeds,
     # the token count never drops and the loop re-triggers compaction
     # forever (the "API call #47/#48/#49 ... has no attribute
-    # try_acquire_compression_lock" spin).  Fail OPEN here: if the lock
-    # subsystem is missing or broken in any unexpected way, skip locking
-    # and proceed with compression.  Skipping the lock risks a rare
-    # concurrent-compression session fork; an infinite no-progress loop
-    # that never compresses at all is strictly worse.
+    # try_acquire_compression_lock" spin).  Fail OPEN for that structural-
+    # absence case (AttributeError, or TypeError from a stale signature
+    # missing ``ttl_seconds``): skip locking and proceed with compression,
+    # since the alternative is an infinite no-progress loop.  Any OTHER
+    # exception means the lock subsystem exists and ran but genuinely
+    # errored — fail CLOSED there instead (skip compression this cycle)
+    # rather than risk a concurrent second compressor forking the session.
     try:
         _lock_ttl = float(getattr(agent, "_compression_lock_ttl_seconds", 300.0) or 300.0)
     except (TypeError, ValueError):
@@ -568,7 +570,7 @@ def compress_context(
             _lock_acquired = _lock_db.try_acquire_compression_lock(
                 _lock_sid, _lock_holder, ttl_seconds=_lock_ttl
             )
-        except Exception as _lock_err:
+        except (AttributeError, TypeError) as _lock_err:
             # Broken/absent lock subsystem (version skew, etc.).  Log once
             # per session and proceed WITHOUT the lock rather than letting
             # the exception spin the outer loop.
@@ -583,6 +585,21 @@ def compress_context(
                     _lock_sid, type(_lock_err).__name__, _lock_err,
                 )
             _lock_acquired = True  # treat as acquired-but-unlocked; proceed
+        except Exception as _lock_err:
+            # The lock subsystem exists and ran but raised unexpectedly (not
+            # the version-skew AttributeError/TypeError case above) — e.g. a
+            # transient DB error. Unlike the version-skew case, we don't know
+            # this is safe to skip locking for, so fail CLOSED: skip
+            # compression this cycle rather than risk a concurrent second
+            # compressor forking the session. The auto-compress loop will
+            # simply retry next cycle.
+            _lock_holder = None  # we don't own anything to release
+            logger.warning(
+                "compression lock acquisition raised unexpectedly for "
+                "session=%s (%s: %s) — skipping compression this cycle",
+                _lock_sid, type(_lock_err).__name__, _lock_err,
+            )
+            _lock_acquired = False
         if not _lock_acquired:
             try:
                 existing = _lock_db.get_compression_lock_holder(_lock_sid)

tests/agent/test_compression_concurrent_fork.py

@@ -452,6 +452,113 @@ def test_missing_lock_subsystem_fails_open_not_infinite_loop(tmp_path: Path, mon
     assert agent.session_id != parent_sid
 
 
+class _ErroringLockSubsystemDB:
+    """Wraps a real SessionDB but simulates a PRESENT lock subsystem that
+    genuinely errors on acquire (e.g. a transient DB failure) — as opposed to
+    the version-skew AttributeError/TypeError in ``_NoLockSubsystemDB`` /
+    ``_KwargSkewLockDB`` above. Here the method exists and ran; something
+    just went wrong, so this must NOT be treated as safe to skip locking for.
+    """
+
+    def __init__(self, real_db: SessionDB) -> None:
+        self._real = real_db
+
+    def try_acquire_compression_lock(self, *_a, **_k):
+        raise RuntimeError("simulated lock-table corruption")
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+
+def test_lock_acquire_generic_error_fails_closed_skips_compression(tmp_path: Path) -> None:
+    """A present-but-erroring lock subsystem must fail CLOSED, not open.
+
+    Unlike the version-skew AttributeError/TypeError case (known-safe to skip
+    locking for — the method doesn't exist or predates the ``ttl_seconds=``
+    kwarg), a generic exception means the lock subsystem exists and ran but
+    something genuinely went wrong. Failing open there risks a second
+    concurrent compressor forking the session — the exact bug this file
+    guards against. ``_compress_context`` must instead skip compression this
+    cycle and return messages unchanged, same as the "another path is
+    compressing" skip path.
+    """
+    db = SessionDB(db_path=tmp_path / "state.db")
+    parent_sid = "ERRORING_LOCK_TEST"
+    db.create_session(parent_sid, source="discord")
+
+    agent = _build_agent_with_db(db, parent_sid)
+    agent._session_db = _ErroringLockSubsystemDB(db)
+
+    messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
+
+    compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
+
+    # Skipped: messages returned verbatim, no rotation, compressor never ran.
+    assert compressed is messages or compressed == messages
+    assert agent.session_id == parent_sid
+    agent.context_compressor.compress.assert_not_called()
+
+
+class _KwargSkewLockDB:
+    """Simulates the ``ttl_seconds`` kwarg-skew case: a stale SessionDB
+    instance whose ``try_acquire_compression_lock`` predates the
+    ``ttl_seconds=`` kwarg added at the call site, so passing it raises
+    TypeError for the unexpected keyword. This is the same version-skew class
+    as ``_NoLockSubsystemDB``'s AttributeError above and must also fail OPEN.
+    """
+
+    def __init__(self, real_db: SessionDB) -> None:
+        self._real = real_db
+
+    def try_acquire_compression_lock(self, session_id, holder):  # no ttl_seconds
+        raise TypeError(
+            "try_acquire_compression_lock() got an unexpected keyword argument 'ttl_seconds'"
+        )
+
+    def get_compression_lock_holder(self, *_a, **_k):
+        raise TypeError("get_compression_lock_holder() takes no keyword arguments")
+
+    def release_compression_lock(self, *_a, **_k):
+        raise TypeError("release_compression_lock() takes no keyword arguments")
+
+    def __getattr__(self, name):
+        return getattr(self._real, name)
+
+
+def test_kwarg_skew_typeerror_fails_open_not_closed(tmp_path: Path, monkeypatch) -> None:
+    """TypeError from a stale ttl_seconds-less signature must fail OPEN too.
+
+    Same version-skew class as the AttributeError case above: the
+    ``ttl_seconds=`` kwarg was added to the call site after the original
+    fail-open fix, so a stale SessionDB instance now raises TypeError instead
+    of AttributeError. Both must proceed with compression rather than spin
+    the outer retry loop.
+    """
+    db = SessionDB(db_path=tmp_path / "state.db")
+    parent_sid = "KWARG_SKEW_TEST"
+    db.create_session(parent_sid, source="discord")
+
+    agent = _build_agent_with_db(db, parent_sid)
+    agent._session_db = _KwargSkewLockDB(db)
+    monkeypatch.setattr(
+        "agent.conversation_compression._CompressionLockLeaseRefresher",
+        lambda *_a, **_k: (_ for _ in ()).throw(
+            AssertionError("lock refresher should not start on fail-open lock skew")
+        ),
+    )
+
+    messages = [{"role": "user", "content": f"m{i}"} for i in range(20)]
+
+    # MUST NOT raise TypeError. Fails open just like the AttributeError case.
+    compressed, _sp = agent._compress_context(messages, "sys", approx_tokens=120_000)
+
+    agent.context_compressor.compress.assert_called_once()
+    assert len(compressed) < len(messages), (
+        "Compression made no progress despite failing open — loop would still spin."
+    )
+    assert agent.session_id != parent_sid
+
+
 def test_review_fork_disables_compression_to_prevent_stale_parent_fork(tmp_path: Path) -> None:
     """The background-review fork must set ``compression_enabled = False``
     so it can never compress the parent it shares a session_id with