diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index d2df6f25a39f..09df0c66c4d0 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -1402,7 +1402,10 @@ def compress_context( # `name #N` renumber, no contextvar/env/logging re-sync, no memory/context- # engine session-switch. The conversation keeps one durable id for life, # eliminating the session-rotation bug cluster. Default True (2107b86024). - in_place = bool(getattr(agent, "compression_in_place", False)) + # Default True matches DEFAULT_CONFIG / #38763. A missing attribute must + # NOT fall back to rotation mode — that re-enables the pre-lease drift + # path and can wedge busy sessions that never set the flag. + in_place = bool(getattr(agent, "compression_in_place", True)) # Set True once the in-place DB write actually completes (the DB block can # raise and skip it). Surfaced to the gateway via agent._last_compaction_in_place. compacted_in_place = False @@ -1712,6 +1715,13 @@ def compress_context( # non-destructive — pre-compaction rows are soft-archived (active=0, # compacted=1), stay searchable and recoverable, so snapshot/durable # drift cannot lose data there and must not abort compaction. + # + # When durable DID grow, ADOPT it and continue rather than aborting. + # Aborting returned the stale snapshot unchanged, so busy sessions + # (memory review / shared session_id writers) stayed permanently + # behind the DB: every /compress and auto-compress saw + # "changed before lease acquisition", surfaced as the misleading + # "No changes from compression", and never reclaimed tokens. if not in_place and _lock_db is not None and _lock_sid: durable_loader = getattr( type(_lock_db), "get_messages_as_conversation", None @@ -1719,16 +1729,18 @@ def compress_context( if callable(durable_loader): durable_parent = durable_loader(_lock_db, _lock_sid) if isinstance(durable_parent, list) and len(durable_parent) > len(messages): - logger.warning( - "compression aborted: session=%s changed before lease " - "acquisition; preserving newer durable messages", + logger.info( + "compression: session=%s grew before lease " + "(%d → %d msgs); adopting durable snapshot", _lock_sid, + len(messages), + len(durable_parent), ) - _release_lock() - existing_prompt = getattr(agent, "_cached_system_prompt", None) - if not existing_prompt: - existing_prompt = agent._build_system_prompt(system_message) - return messages, existing_prompt + messages = durable_parent + # Token estimate was for the stale snapshot; clear it so + # the compressor re-derives from the adopted transcript + # instead of under-counting the newly visible rows. + approx_tokens = 0 # Notify external memory provider before compression discards context. # The provider's on_pre_compress() may return a string of insights it diff --git a/tests/agent/test_compression_concurrent_fork.py b/tests/agent/test_compression_concurrent_fork.py index b7c696ecb75b..bc20dfcb7afa 100644 --- a/tests/agent/test_compression_concurrent_fork.py +++ b/tests/agent/test_compression_concurrent_fork.py @@ -340,10 +340,17 @@ def test_concurrent_compression_does_not_fork_session(tmp_path: Path) -> None: ) -def test_durable_message_committed_before_lease_aborts_stale_snapshot( +def test_durable_message_committed_before_lease_is_adopted( tmp_path: Path, ) -> None: - """A durable row absent from the caller snapshot must survive in the parent.""" + """A durable row absent from the caller snapshot must still be compressed. + + Previously this path aborted and returned the stale snapshot unchanged, + which permanently wedged busy sessions: every compress attempt saw the + DB ahead of the in-memory list, logged "changed before lease + acquisition", and never called the compressor. Adopting the durable + transcript keeps the late-committed turn and lets compression proceed. + """ db = SessionDB(db_path=tmp_path / "state.db") parent_sid = "PRE_LEASE_DURABLE_RACE" db.create_session(parent_sid, source="webui") @@ -359,15 +366,21 @@ def test_durable_message_committed_before_lease_aborts_stale_snapshot( stale_snapshot, "sys", approx_tokens=120_000 ) - assert returned is stale_snapshot - assert agent.session_id == parent_sid - assert db.find_live_compression_child(parent_sid) is None - assert [m["content"] for m in db.get_messages_as_conversation(parent_sid)] == [ + agent.context_compressor.compress.assert_called_once() + compressed_arg = agent.context_compressor.compress.call_args.args[0] + assert [m["content"] for m in compressed_arg] == [ "old durable", "late committed before lease", ] - agent.context_compressor.compress.assert_not_called() - + # Must not echo the stale snapshot — compression proceeded on the + # adopted durable transcript (rotation publishes a child session). + assert returned is not stale_snapshot + assert returned[0]["content"] == "[CONTEXT COMPACTION] summary" + assert agent.session_id != parent_sid + child = db.find_live_compression_child(parent_sid) + assert child is not None + child_id = child["id"] if isinstance(child, dict) else child + assert child_id == agent.session_id def test_skipped_compression_returns_messages_unchanged(tmp_path: Path) -> None: """The loser of the lock race must return its input messages verbatim.