From 80ab7d2b1cdbe0baed0bb967fefb1a3679ae2fff Mon Sep 17 00:00:00 2001 From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com> Date: Fri, 28 Aug 2026 02:39:55 +0530 Subject: [PATCH] fix(compression): dedupe current-turn rows when rotation splits the session mid-turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When context-compression rotation fires mid-turn, the current user message was persisted twice into the child session. Root cause: dedup used id()-seeded sets of copies instead of markers on the live objects. Replace with _DB_PERSISTED_MARKER-based dedup as the sole authority: - _ensure_compressed_has_user_turn returns CompressedUserTurnOutcome - After publish_compression_child succeeds, stamp the live anchor-source row (not a drifted index) with _DB_PERSISTED_MARKER - _sync_persisted_markers mirrors stamps from result to live lists by scoped identity (handles direct-path, adoption divergence, _session_messages) - Remove _flushed_db_message_ids from rotation commit path (markers replace it) - Unconditional (loud) imports — no silent fallback Salvage of #94996 by @fedosis, rebased on top of #95433 (stall-fallback, already merged). Both conversation_compression.py and run_agent.py are built from origin/main + #94996's diff applied on top, preserving the force_terminal refactor and _publish_new_fence from #95433. Credit: @fedosis original PR #94996. --- agent/conversation_compression.py | 192 +++- run_agent.py | 321 +++--- .../agent/test_compression_rotation_state.py | 930 +++++++++++++++++- tests/run_agent/test_413_compression.py | 6 +- .../test_compress_context_fallback_shim.py | 102 ++ 5 files changed, 1414 insertions(+), 137 deletions(-) create mode 100644 tests/run_agent/test_compress_context_fallback_shim.py diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 6c25c05dd0..c9a0979f48 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -64,7 +64,7 @@ import uuid import threading from datetime import datetime from pathlib import Path -from typing import Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Literal, Optional, Tuple from agent.auxiliary_client import AuxiliaryExplicitCancellation from agent.context_engine import ( @@ -2351,8 +2351,17 @@ def _merge_anchor_into_user_message(target: dict, anchor: dict) -> None: target.pop(flag, None) -def _insert_real_user_anchor(messages: list, anchor: dict) -> None: +CompressedUserTurnOutcome = Literal[ + "inserted", + "merged", + "already_present", + "placeholder_appended", +] + + +def _insert_real_user_anchor(messages: list, anchor: dict) -> CompressedUserTurnOutcome: """Insert the latest human turn without breaking role alternation.""" + from agent.context_compressor import _DB_PERSISTED_MARKER def _role(msg: Any) -> Optional[str]: return msg.get("role") if isinstance(msg, dict) else None @@ -2365,13 +2374,15 @@ def _insert_real_user_anchor(messages: list, anchor: dict) -> None: continue previous_role = _role(messages[index - 1]) if index > 0 else None if previous_role != "user": + anchor[_DB_PERSISTED_MARKER] = True messages.insert(index, anchor) - return + return "inserted" # Every assistant is user-preceded (or there are none). Appending is # safe whenever the transcript does not already end with a user turn. if not messages or _role(messages[-1]) != "user": + anchor[_DB_PERSISTED_MARKER] = True messages.append(anchor) - return + return "inserted" # The transcript ends with a user-role message and no slot avoids # user/user adjacency. from agent.context_compressor import ContextCompressor @@ -2385,17 +2396,22 @@ def _insert_real_user_anchor(messages: list, anchor: dict) -> None: # the summary" — exactly what the handoff prefix instructs — and the # adjacent user turns are merged summary-first by # repair_message_sequence before the next API call. + anchor[_DB_PERSISTED_MARKER] = True messages.append(anchor) - return + return "inserted" # Trailing user-role scaffolding (e.g. the todo snapshot): merge instead # of inserting a consecutive same-role message (#55677 strict templates). _merge_anchor_into_user_message(messages[-1], anchor) + messages[-1][_DB_PERSISTED_MARKER] = True + return "merged" -def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) -> None: +def _ensure_compressed_has_user_turn( + original_messages: list, compressed: list +) -> CompressedUserTurnOutcome: """Preserve human intent, not merely a synthetic user-role placeholder.""" if any(_is_real_user_message(message) for message in compressed): - return + return "already_present" from agent.context_compressor import ( COMPRESSION_CONTINUATION_USER_CONTENT, _fresh_compaction_message_copy, @@ -2403,11 +2419,10 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) for message in reversed(original_messages): if _is_real_user_message(message): - _insert_real_user_anchor( + return _insert_real_user_anchor( compressed, _fresh_compaction_message_copy(message), ) - return from agent.message_metadata import append_message append_message( @@ -2417,6 +2432,22 @@ def _ensure_compressed_has_user_turn(original_messages: list, compressed: list) "content": COMPRESSION_CONTINUATION_USER_CONTENT, }, ) + return "placeholder_appended" + + +def _messages_match_scoped_identity(left: Any, right: Any) -> bool: + """Compare the live turn identity we care about for rotation stamping.""" + if not isinstance(left, dict) or not isinstance(right, dict): + return False + if left.get("role") != right.get("role"): + return False + if left.get("content") != right.get("content"): + return False + left_timestamp = left.get("timestamp") + right_timestamp = right.get("timestamp") + if left_timestamp is not None and right_timestamp is not None: + return left_timestamp == right_timestamp + return True _PENDING_CONTEXT_ENGINE_NOTIFICATION = ( @@ -3219,6 +3250,11 @@ def compress_context( # after compression skips the adopted rows by identity # (conversation_history=messages[:idx]) instead of # re-appending the concurrent rows and the live tail. + # This rebind is the concrete divergence path that can + # leave `agent._session_messages` pointing at the old + # live list while `messages` now points at the adopted + # durable snapshot; the post-publish marker sync in + # run_agent.py keeps both views aligned. agent._persist_user_message_idx = len(messages) # Notify external memory provider before compression discards context. @@ -3664,7 +3700,9 @@ def compress_context( "content": todo_snapshot, "_todo_snapshot_synthetic": True, }) - _ensure_compressed_has_user_turn(messages, compressed) + compressed_user_turn_outcome = _ensure_compressed_has_user_turn( + messages, compressed + ) cached_system_prompt = agent._cached_system_prompt agent._invalidate_system_prompt() @@ -3978,6 +4016,7 @@ def compress_context( f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_" f"{uuid.uuid4().hex[:6]}" ) + from agent.context_compressor import _DB_PERSISTED_MARKER agent._session_db.publish_compression_child( parent_session_id=old_session_id, child_session_id=new_session_id, @@ -3998,7 +4037,127 @@ def compress_context( ), watermark_ceiling=_foreign_tail_ceiling, ) + # For the `already_present` outcome the live-dict stamping is + # handled by the run_agent _compress_context wrapper's + # _sync_persisted_markers (it mirrors the handoff stamps back + # to the live lists by scoped identity). This branch only + # covers inserted/merged; compress_context is intentionally + # NOT self-contained for already_present — a future direct + # caller of compress_context must go through that wrapper. + if compressed_user_turn_outcome in {"inserted", "merged"}: + # The published child represents exactly one live row: + # the anchor source _ensure_compressed_has_user_turn + # inserted (verbatim) or merged (as the leading + # content) into the handoff. Stamp THAT row, not an + # index that may have drifted (adoption rebind at + # :3028/:3045 rebinds _persist_user_message_idx to + # len(messages) — deliberately OUT OF RANGE — or + # reanchor_current_turn_user_idx falling back to a + # user-role neighbor). The persist index is + # deliberately NOT read here: trusting it is the + # drift vector. _messages_match_scoped_identity is + # intentionally NOT used against the HANDOFF row: for + # `merged` the handoff is a superset (anchor + + # scaffolding), so a verbatim match would break the + # legitimate merged stamp. Live views are compared + # against the ANCHOR SOURCE only. + _compressed_anchor_source = None + for _reversed_message in reversed(messages): + if _is_real_user_message(_reversed_message): + _compressed_anchor_source = _reversed_message + break + if isinstance(_compressed_anchor_source, dict): + _compressed_anchor_source[_DB_PERSISTED_MARKER] = True + _session_messages = getattr( + agent, "_session_messages", None + ) + if ( + isinstance(_session_messages, list) + and _session_messages is not messages + ): + # Adoption divergence: `messages` now points + # at the adopted durable snapshot while + # agent._session_messages may still point at + # the pre-adoption live list (comment + # :3040-3044), and the persist index is OUT + # OF RANGE for the twin by design. The + # post-publish _sync_persisted_markers cannot + # mirror a MERGED handoff either (anchor + + # scaffolding superset has no scoped match + # with the standalone live row). Locate the + # twin's corresponding live row(s) by scoped + # identity AGAINST THE ANCHOR SOURCE and + # stamp them — mirroring the wrapper's "stamp + # every scoped match" pattern for + # timestamp-less ambiguity + # (run_agent.py:8237-8281). + _anchor_timestamp = _compressed_anchor_source.get( + "timestamp" + ) + _found_exact_timestamp_candidate = False + if _anchor_timestamp is not None: + for _twin_message in _session_messages: + if ( + isinstance(_twin_message, dict) + and _twin_message.get("timestamp") + == _anchor_timestamp + and _messages_match_scoped_identity( + _twin_message, + _compressed_anchor_source, + ) + ): + # An exact scoped twin EXISTS — + # count the hit REGARDLESS of the + # marker. An already-stamped exact + # twin (stamped by a prior + # flush/rotation) must still + # suppress the broad fallback: + # otherwise a timestamp-less + # historical duplicate with the + # same content would be stamped as + # a "twin" of the anchor — the + # exact wrong-row path the + # reviewer flagged. The stamp + # itself stays idempotent (skip + # rows already carrying the + # marker); only the HIT is + # marker-independent. + _found_exact_timestamp_candidate = True + if not _twin_message.get( + _DB_PERSISTED_MARKER + ): + _twin_message[ + _DB_PERSISTED_MARKER + ] = True + if not _found_exact_timestamp_candidate: + # No exact scoped twin exists anywhere (or + # the anchor itself is timestamp-less): + # stamp every scoped match — the wrapper's + # documented ambiguity policy for + # timestamp-less anchors. This branch is + # reached ONLY when an exact hit is + # provably absent; an exact hit that is + # already stamped does NOT open this + # branch. + for _twin_message in _session_messages: + if ( + isinstance(_twin_message, dict) + and not _twin_message.get( + _DB_PERSISTED_MARKER + ) + and _messages_match_scoped_identity( + _twin_message, + _compressed_anchor_source, + ) + ): + _twin_message[ + _DB_PERSISTED_MARKER + ] = True + for _handoff_message in compressed: + if isinstance(_handoff_message, dict): + _handoff_message[_DB_PERSISTED_MARKER] = True agent.session_id = new_session_id + agent._db_flush_scan_prefix = None try: from gateway.session_context import set_current_session_id @@ -4093,11 +4252,6 @@ def compress_context( else: agent._last_flushed_db_idx = len(compressed) agent._flushed_db_message_session_id = agent.session_id - agent._flushed_db_message_ids = { - id(message) - for message in compressed - if isinstance(message, dict) - } _session_commit_succeeded = True except Exception as e: if ( @@ -4108,6 +4262,14 @@ def compress_context( # Atomic publication failed (including lease loss): keep the # parent live and discard the stale compacted snapshot. old_session_id = None + # NOTE: _db_flush_scan_prefix is intentionally NOT cleared + # here. The flush's bounded scan is identity-based + # (messages[i] is prefix[i]); the deepcopy rollback below + # replaces every row, so a stale prefix can never + # identity-match again. A failed parent flush clears its own + # prefix, and on the snapshot path the live list is + # untouched — do not "restore" a clear here without + # re-checking those invariants. messages[:] = copy.deepcopy(messages_before_compression) compressed = messages _compression_made_progress = False diff --git a/run_agent.py b/run_agent.py index 9c38ced8eb..d9707a3e0a 100644 --- a/run_agent.py +++ b/run_agent.py @@ -8084,141 +8084,228 @@ class AIAgent: # Callers that already own a progress-aware wait (gateway session # hygiene) pass commit_fence and must not be double-wrapped. - if commit_fence is not None: - return _run(active_fence) + direct_path = commit_fence is not None + idle_timeout = total_ceiling = None + if not direct_path: + idle_timeout, total_ceiling = resolve_context_compression_timeouts() + if idle_timeout <= 0: + direct_path = True - idle_timeout, total_ceiling = resolve_context_compression_timeouts() - if idle_timeout <= 0: - return _run(active_fence) - - def _snapshot_worker(fence=None): - # #76354 review F3: the pooled worker must NEVER share the - # caller's live transcript. Plugin/legacy context engines are - # allowed to mutate their input list in place; after a host - # timeout the worker stays alive, so a shared list would let - # a late engine rewrite the live conversation (roles, - # ordering, persisted content) behind the caller's back. - # Deep-snapshot here, on the worker thread, so the caller's - # list object is never touched by pooled code. Results are - # published to caller-visible state only via the returned - # value of an ADMITTED commit (the host discards results on - # timeout/cancel); durable SessionDB mutation is already - # gated behind the commit fence inside compress_context. - snapshot = copy.deepcopy(messages) - result_msgs, result_prompt = _run( - fence, target_messages=snapshot - ) - if result_msgs is snapshot: - # No-op/abort path returned the snapshot unchanged: hand - # back the caller's ORIGINAL list so identity-based - # semantics (len/identity no-op detection, flush dedup - # by id()) keep working. - return messages, result_prompt - return result_msgs, result_prompt - - # Resolve the fallback prompt lazily on timeout only. Eager - # rebuild here would raise before compress_context runs whenever - # _cached_system_prompt is unset and _build_system_prompt fails - # (lock-refresher / noop-exception tests rely on that path). - def _fallback_prompt(): - cached = getattr(self, "_cached_system_prompt", None) - if cached: - return cached - try: - return self._build_system_prompt(system_message) - except Exception: - logger.debug( - "compress_context timeout fallback prompt rebuild " - "failed; using raw system_message", - exc_info=True, + if direct_path: + result = _run(active_fence) + else: + def _snapshot_worker(fence=None): + # #76354 review F3: the pooled worker must NEVER share the + # caller's live transcript. Plugin/legacy context engines are + # allowed to mutate their input list in place; after a host + # timeout the worker stays alive, so a shared list would let + # a late engine rewrite the live conversation (roles, + # ordering, persisted content) behind the caller's back. + # Deep-snapshot here, on the worker thread, so the caller's + # list object is never touched by pooled code. Results are + # published to caller-visible state only via the returned + # value of an ADMITTED commit (the host discards results on + # timeout/cancel); durable SessionDB mutation is already + # gated behind the commit fence inside compress_context. + snapshot = copy.deepcopy(messages) + result_msgs, result_prompt = _run( + fence, target_messages=snapshot ) - return system_message or "" + if result_msgs is snapshot: + # No-op/abort path returned the snapshot unchanged: hand + # back the caller's ORIGINAL list so identity-based + # semantics (len/identity no-op detection, flush dedup + # by id()) keep working. + return messages, result_prompt + return result_msgs, result_prompt - def _on_timeout(idle, waited, since_progress): - logger.warning( - "Context compression made no progress for %.1fs " - "(total wait %.1fs, ceiling %.1fs); continuing without " - "compression", - since_progress, - waited, - total_ceiling, - ) - touch = getattr(self, "_touch_activity", None) - if callable(touch): + # Resolve the fallback prompt lazily on timeout only. Eager + # rebuild here would raise before compress_context runs whenever + # _cached_system_prompt is unset and _build_system_prompt fails + # (lock-refresher / noop-exception tests rely on that path). + def _fallback_prompt(): + cached = getattr(self, "_cached_system_prompt", None) + if cached: + return cached try: - touch( - "context compression timed out", - provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT, - ) + return self._build_system_prompt(system_message) except Exception: logger.debug( - "compress_context timeout activity touch failed", + "compress_context timeout fallback prompt rebuild " + "failed; using raw system_message", exc_info=True, ) - # Same timeout cooldown ladder as summary-LLM timeouts - # (#62452): avoid re-burning the full idle budget every turn. - compressor = getattr(self, "context_compressor", None) - if compressor is not None: - record = getattr(compressor, "record_timeout_failure", None) - if callable(record): + return system_message or "" + + def _on_timeout(idle, waited, since_progress): + logger.warning( + "Context compression made no progress for %.1fs " + "(total wait %.1fs, ceiling %.1fs); continuing without " + "compression", + since_progress, + waited, + total_ceiling, + ) + touch = getattr(self, "_touch_activity", None) + if callable(touch): try: - record( - "host compress_context timeout " - "(no summary progress)" + touch( + "context compression timed out", + provenance=ActivityProvenance.AGENT_COMPRESSION_TIMEOUT, ) except Exception: logger.debug( - "failed to record compress_context timeout " - "cooldown", + "compress_context timeout activity touch failed", exc_info=True, ) - emit = getattr(self, "_emit_warning", None) - if callable(emit): - emit( - "⚠ Context compression timed out " - f"after {idle:.1f}s with no output from the summary " - "model. No messages were dropped — continuing without " - "compression. Run /compress to retry, /new for a clean " - "session, or check auxiliary.compression." - ) + # Same timeout cooldown ladder as summary-LLM timeouts + # (#62452): avoid re-burning the full idle budget every turn. + compressor = getattr(self, "context_compressor", None) + if compressor is not None: + record = getattr(compressor, "record_timeout_failure", None) + if callable(record): + try: + record( + "host compress_context timeout " + "(no summary progress)" + ) + except Exception: + logger.debug( + "failed to record compress_context timeout " + "cooldown", + exc_info=True, + ) + emit = getattr(self, "_emit_warning", None) + if callable(emit): + emit( + "⚠ Context compression timed out " + f"after {idle:.1f}s with no output from the summary " + "model. No messages were dropped — continuing without " + "compression. Run /compress to retry, /new for a clean " + "session, or check auxiliary.compression." + ) - def _on_commit_overrun(waited, ceiling): - # Commit-phase ceiling breach: the SessionDB mutation is in - # flight and must complete (abandoning it mid-commit would - # diverge live messages from durable session state), so this - # only surfaces the overrun — it never cancels the commit. - emit = getattr(self, "_emit_warning", None) - if callable(emit): - emit( - "⚠ Context compression commit is taking unusually " - f"long ({waited:.0f}s, ceiling {ceiling:.0f}s). " - "Waiting for it to finish safely — if this persists, " - "check SessionDB health (disk / lock contention)." - ) + def _on_commit_overrun(waited, ceiling): + # Commit-phase ceiling breach: the SessionDB mutation is in + # flight and must complete (abandoning it mid-commit would + # diverge live messages from durable session state), so this + # only surfaces the overrun — it never cancels the commit. + emit = getattr(self, "_emit_warning", None) + if callable(emit): + emit( + "⚠ Context compression commit is taking unusually " + f"long ({waited:.0f}s, ceiling {ceiling:.0f}s). " + "Waiting for it to finish safely — if this persists, " + "check SessionDB health (disk / lock contention)." + ) - def _publish_new_fence(): - # The stall-fallback retry (#78981) needs a fence the aborted - # attempt cannot veto. Publish it on the same serialized slot - # hard_interrupt() reads, so a /stop during the retry admits - # against the attempt that is actually running. The finally - # below restores whatever the caller had either way. - retry_fence = CompressionCommitFence() - with fence_registration_lock: - self._active_compression_commit_fence = retry_fence - return retry_fence + def _publish_new_fence(): + # The stall-fallback retry (#78981) needs a fence the aborted + # attempt cannot veto. Publish it on the same serialized slot + # hard_interrupt() reads, so a /stop during the retry admits + # against the attempt that is actually running. The finally + # below restores whatever the caller had either way. + retry_fence = CompressionCommitFence() + with fence_registration_lock: + self._active_compression_commit_fence = retry_fence + return retry_fence + + result = run_compress_context_with_progress_timeout( + worker=_snapshot_worker, + messages=messages, + system_prompt_fallback=_fallback_prompt, + idle_timeout_seconds=idle_timeout, + total_ceiling_seconds=total_ceiling, + on_timeout=_on_timeout, + on_commit_overrun=_on_commit_overrun, + fence=active_fence, + telemetry_agent=self, + new_fence=_publish_new_fence, + ) + # _DB_PERSISTED_MARKER lives at module level in + # agent.context_compressor; conversation_compression only + # imports it locally (cannot be imported from there). Imported + # UNCONDITIONALLY (no fallback): both modules are already + # hard dependencies at this point — agent.context_compressor is + # imported at the top of this module (line ~162), and + # agent.conversation_compression is imported at the top of this + # very method and its compress_context is invoked below. The + # only way these imports can fail while the wrapper is + # functional is a renamed/removed symbol, and that must fail + # LOUDLY: a silent fallback literal ("_db_persisted") would + # split the stamping key from the flush's and quietly resurrect + # the duplicate-row bug this fix removed. + from agent.context_compressor import _DB_PERSISTED_MARKER + from agent.conversation_compression import ( + _messages_match_scoped_identity, - result = run_compress_context_with_progress_timeout( - worker=_snapshot_worker, - messages=messages, - system_prompt_fallback=_fallback_prompt, - idle_timeout_seconds=idle_timeout, - total_ceiling_seconds=total_ceiling, - on_timeout=_on_timeout, - on_commit_overrun=_on_commit_overrun, - fence=active_fence, - telemetry_agent=self, - new_fence=_publish_new_fence, ) + + def _sync_persisted_markers(target_messages, source_messages): + if not isinstance(target_messages, list) or not isinstance( + source_messages, list + ): + return + # Compression runs against a deepcopy snapshot on the pooled + # worker path, so publish stamps land on the result list first. + # Mirror them back onto the live caller lists by scoped + # identity after publish succeeds; timestamp-less repeated + # content is ambiguous, so we stamp every scoped match instead + # of stopping at the first one. + for source_message in source_messages: + if not ( + isinstance(source_message, dict) + and source_message.get(_DB_PERSISTED_MARKER) + ): + continue + source_timestamp = source_message.get("timestamp") + matched_exact_timestamp = False + if source_timestamp is not None: + for target_message in target_messages: + if not isinstance(target_message, dict): + continue + if target_message.get(_DB_PERSISTED_MARKER): + continue + if not _messages_match_scoped_identity( + target_message, source_message + ): + continue + if target_message.get("timestamp") != source_timestamp: + continue + target_message[_DB_PERSISTED_MARKER] = True + matched_exact_timestamp = True + if matched_exact_timestamp: + continue + for target_message in target_messages: + if not isinstance(target_message, dict): + continue + if target_message.get(_DB_PERSISTED_MARKER): + continue + if not _messages_match_scoped_identity( + target_message, source_message + ): + continue + target_message[_DB_PERSISTED_MARKER] = True + + if isinstance(result, tuple) and result: + result_messages = result[0] + if isinstance(result_messages, list): + # Direct-path callers bypass the snapshot worker, so they + # still need the same post-publish mirror onto the live + # caller list even when the returned list already points at + # the active transcript. + if direct_path or result_messages is not messages: + _sync_persisted_markers(messages, result_messages) + session_messages = getattr(self, "_session_messages", None) + if ( + isinstance(session_messages, list) + and session_messages is not messages + ): + # Intentional: durable-parent adoption can leave + # `_session_messages` on the pre-adoption live list + # while `messages` now points at the adopted snapshot, + # so both lists need the post-publish marker sync. + _sync_persisted_markers(session_messages, result_messages) # compress_context ran on a daemon pool worker thread; the session # id rotation updated hermes_logging._session_context (a # threading.local) on the WORKER thread, not this one. Propagate diff --git a/tests/agent/test_compression_rotation_state.py b/tests/agent/test_compression_rotation_state.py index 490eee2483..6e803dba7e 100644 --- a/tests/agent/test_compression_rotation_state.py +++ b/tests/agent/test_compression_rotation_state.py @@ -17,14 +17,20 @@ These tests drive the real ``compress_context`` path against a real SessionDB. from __future__ import annotations +import copy import os import time from pathlib import Path +from typing import Any from unittest.mock import MagicMock, patch import pytest -from agent.context_compressor import ContextCompressor +from agent.context_compressor import ContextCompressor, _DB_PERSISTED_MARKER +from agent.conversation_compression import ( + CompressionCommitFence, + _is_real_user_message, +) from hermes_state import SessionDB @@ -68,6 +74,15 @@ def _msgs(n=20): return [{"role": "user", "content": f"m{i}"} for i in range(n)] +def _count_rows(rows, *, content: Any = None, role: str | None = None): + return sum( + 1 + for row in rows + if (content is None or row.get("content") == content) + and (role is None or row.get("role") == role) + ) + + def _bound_context_compressor(db: SessionDB, session_id: str) -> ContextCompressor: with patch( "agent.context_compressor.get_model_context_length", @@ -197,6 +212,910 @@ class TestWorkspaceMetadataFollowsRotation: assert row["user_id"] == "u1" +class TestRotationChildFlushDedup: + def test_summary_handoff_row_is_persisted_once_in_child( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_LIVE_USER" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [*loaded, {"role": "user", "content": "live question"}] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + ] + + returned, _ = agent._compress_context(messages, "sys", approx_tokens=120_000) + assert any( + isinstance(msg, dict) + and msg.get("content") == "[CONTEXT COMPACTION] summary" + and msg.get(_DB_PERSISTED_MARKER) + for msg in returned + ) + assert any( + isinstance(msg, dict) + and msg.get("content") == "live question" + and msg.get(_DB_PERSISTED_MARKER) + for msg in returned + ) + + def test_rotation_flush_of_original_live_list_keeps_user_once_when_handoff_already_contains_user( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_ORIGINAL_LIVE" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + live_user = { + "role": "user", + "content": "live question", + "timestamp": 1234.5, + } + messages = [*loaded, live_user] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + copy.deepcopy(live_user), + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + returned, _ = agent._compress_context( + messages, "sys", approx_tokens=120_000 + ) + + assert agent.session_id != parent + assert _DB_PERSISTED_MARKER in live_user + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows(child_rows, content="live question", role="user") == 1 + assert _count_rows( + child_rows, content="[CONTEXT COMPACTION] summary", role="assistant" + ) == 1 + + def test_failed_publish_leaves_live_user_unmarked_for_later_flush( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_PUBLISH_FAIL" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [*loaded, {"role": "user", "content": "live question"}] + live_user = messages[-1] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + {"role": "user", "content": "[CONTEXT COMPACTION] summary"}, + {"role": "assistant", "content": "tail"}, + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + db, + "publish_compression_child", + side_effect=RuntimeError("simulated publish failure"), + ): + returned, _ = agent._compress_context( + messages, "sys", approx_tokens=120_000 + ) + + assert _DB_PERSISTED_MARKER not in live_user + + retry_session = "PARENT_ROT_PUBLISH_RETRY" + db.create_session(retry_session, source="cli") + agent.session_id = retry_session + real_flush([live_user]) + + retry_rows = db.get_messages_as_conversation( + retry_session, include_inactive=True + ) + assert _count_rows(retry_rows, content="live question", role="user") == 1 + + def test_mid_tool_loop_rows_do_not_duplicate_after_failed_parent_flush( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_TOOL_LOOP" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + assistant_turn = { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call-1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + } + tool_turn = { + "role": "tool", + "tool_call_id": "call-1", + "content": "tool result", + } + messages = [ + *loaded, + {"role": "user", "content": "live tool question"}, + assistant_turn, + tool_turn, + ] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(loaded) + agent.context_compressor.compress.return_value = [ + copy.deepcopy(assistant_turn), + copy.deepcopy(tool_turn), + ] + + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + returned, _ = agent._compress_context( + messages, "sys", approx_tokens=120_000 + ) + + agent._flush_messages_to_session_db(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows( + child_rows, content="live tool question", role="user" + ) == 1 + assert _count_rows(child_rows, content="tool result", role="tool") == 1 + + def test_mid_tool_loop_rows_do_not_duplicate_after_failed_parent_flush_direct_path( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_TOOL_LOOP_DIRECT" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + assistant_turn = { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call-1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + } + tool_turn = { + "role": "tool", + "tool_call_id": "call-1", + "content": "tool result", + } + messages = [ + *loaded, + {"role": "user", "content": "live tool question"}, + assistant_turn, + tool_turn, + ] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(loaded) + agent.context_compressor.compress.return_value = [ + copy.deepcopy(assistant_turn), + copy.deepcopy(tool_turn), + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + assert agent.session_id != parent + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows( + child_rows, content="live tool question", role="user" + ) == 1 + assert _count_rows(child_rows, content="", role="assistant") == 1 + assert _count_rows(child_rows, content="tool result", role="tool") == 1 + + def test_timestampless_duplicate_content_rows_are_all_stamped( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_DUPLICATE_CONTENT" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + assistant_turn = { + "role": "assistant", + "content": "", + "tool_calls": [ + { + "id": "call-1", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ], + } + tool_turn = { + "role": "tool", + "tool_call_id": "call-1", + "content": "tool result", + } + messages = [ + *loaded, + {"role": "user", "content": "live tool question"}, + assistant_turn, + copy.deepcopy(assistant_turn), + tool_turn, + copy.deepcopy(tool_turn), + ] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(loaded) + agent.context_compressor.compress.return_value = [ + copy.deepcopy(assistant_turn), + copy.deepcopy(tool_turn), + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows( + child_rows, content="live tool question", role="user" + ) == 1 + assert _count_rows(child_rows, content="", role="assistant") == 1 + assert _count_rows(child_rows, content="tool result", role="tool") == 1 + + def test_rotation_stamps_diverged_session_messages_entry_only_when_it_matches( + self, tmp_path: Path + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_SESSION_MESSAGES_DIVERGE" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [*loaded, {"role": "user", "content": "live question"}] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent._session_messages = [ + *loaded, + {"role": "user", "content": "different live question"}, + ] + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + ] + + returned, _ = agent._compress_context(messages, "sys", approx_tokens=120_000) + agent._flush_messages_to_session_db(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows(child_rows, content="live question", role="user") == 1 + assert _DB_PERSISTED_MARKER in messages[-1] + assert _DB_PERSISTED_MARKER not in agent._session_messages[-1] + + # ------------------------------------------------------------------ + # Item 2 review fixes — symmetric identity validation on the primary + # stamp. The guard stamps the anchor-source row (the last real user + # message in `messages`, the row the published child actually + # represents), NEVER an index that may have drifted, and mirrors the + # twin (`_session_messages`) by scoped identity against that anchor + # source with a marker-independent exact-hit two-phase scan. + # ------------------------------------------------------------------ + + def test_rotation_never_stamps_drifted_user_role_neighbor( + self, tmp_path: Path + ): + """A user-role neighbor at a drifted index must not be stamped.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_DRIFTED_NEIGHBOR" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [ + *loaded, + {"role": "user", "content": "live question"}, + { + "role": "user", + "content": "drifted neighbor", + "_todo_snapshot_synthetic": True, + }, + ] + + agent = _build_agent_with_db(db, parent) + # Index drifted onto the synthetic user-role neighbor (the reanchor + # fallback / stale-index failure shape the guard must not trust). + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + ] + + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + # The drifted neighbor is not the row the child represents. + assert _DB_PERSISTED_MARKER not in messages[-1] + # The anchor source (the real live question) is stamped. + assert _DB_PERSISTED_MARKER in messages[-2] + + def test_rotation_drifted_index_does_not_duplicate_live_question_in_child( + self, tmp_path: Path + ): + """Merged outcome + drifted index: real flush must not re-append the + live question standalone (the duplicate the PR eliminates).""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_DRIFTED_MERGED" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [ + *loaded, + {"role": "user", "content": "live question"}, + { + "role": "user", + "content": "drifted neighbor", + "_todo_snapshot_synthetic": True, + }, + ] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": "handoff scaffolding", + "_todo_snapshot_synthetic": True, + }, + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + # No standalone "live question" row — the merged handoff already + # represents it. + assert _count_rows(child_rows, content="live question") == 0 + assert ( + _count_rows( + child_rows, content="live question\n\nhandoff scaffolding" + ) + == 1 + ) + + def test_merged_outcome_still_stamps_live_question(self, tmp_path: Path): + """Constraint: the guard must not break the legitimate merged stamp.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_MERGED_LIVE" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [*loaded, {"role": "user", "content": "live question"}] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": "scaffolding", + "_todo_snapshot_synthetic": True, + }, + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + assert _DB_PERSISTED_MARKER in messages[-1] + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert _count_rows( + child_rows, content="live question", role="user" + ) == 0 + assert ( + _count_rows(child_rows, content="live question\n\nscaffolding") + == 1 + ) + + def test_adoption_divergence_merged_stamps_both_views_and_no_duplicate( + self, tmp_path: Path + ): + """REAL adoption divergence: durable parent grows under the lease, + `messages` rebinds to the adopted snapshot while + `agent._session_messages` stays on the old live list, and the merged + handoff cannot be mirrored by the wrapper's scoped sync. Both views + must carry the marker and the real post-rotation flush over the old + live view must not re-append the live question standalone.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_ADOPT_DIVERGE" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + # (b) Old live list object kept alive; divergence set so the twin is + # the SAME object the guard scans (not a fresh copy). + loaded = db.get_messages_as_conversation(parent, include_inactive=True) + old_live_list = [ + *loaded, + {"role": "user", "content": "live question", "timestamp": 1234.5}, + ] + + agent = _build_agent_with_db(db, parent) + agent._session_messages = old_live_list + assert agent._session_messages is old_live_list + + # (c) The stale snapshot passed to _compress_context is a separate + # object (the production frontend-snapshot shape). + stale_snapshot = [ + {"role": "user", "content": "persisted question"}, + {"role": "assistant", "content": "persisted answer"}, + ] + assert stale_snapshot is not agent._session_messages + + # (d) Pin the initial persist-index state: production "no known + # un-persisted tail" shape, so the real code takes the adopt-directly + # branch (:2994-3001) and the pre-adoption flush (:2988) is provably + # never attempted (no fixture flush can mask the divergence). + agent._persist_user_message_idx = None + assert agent._persist_user_message_idx is None + + # Grow the DB AFTER the snapshot is taken so the REAL adoption + # condition (durable parent longer than the caller snapshot) fires. + db.append_message(parent, "user", "live question") + durable_check = db.get_messages_as_conversation(parent) + assert len(durable_check) == 3 > len(stale_snapshot) == 2 + # Sync the twin's timestamp to the committed row so the guard's + # exact-timestamp twin scan matches the adopted anchor. + old_live_list[-1]["timestamp"] = durable_check[-1]["timestamp"] + + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": "handoff scaffolding", + "_todo_snapshot_synthetic": True, + }, + ] + + # Phase-keyed flush failure: fail ONLY the pre-publish flush (:3780); + # a blanket failure would not distinguish the phases and a masked + # pre-adoption flush would hide the divergence. + flush_attempts = [] + + def _fail_only_prepublish_flush(messages_arg, **kwargs): + flush_attempts.append((messages_arg, kwargs)) + raise RuntimeError("simulated pre-publish flush failure") + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=_fail_only_prepublish_flush, + ): + _returned, _ = agent._compress_context( + stale_snapshot, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + # The ONLY internal flush was the single pre-publish one. + assert len(flush_attempts) == 1 + + # (e) Identity and shape asserts BEFORE markers: adoption fired, the + # divergence is preserved, the persist index was rebound out of range. + adopted = agent.context_compressor.compress.call_args.args[0] + assert adopted is not stale_snapshot + assert adopted is not agent._session_messages + assert agent._session_messages is old_live_list + assert agent._persist_user_message_idx == len(adopted) + assert adopted[-1]["role"] == "user" + assert adopted[-1]["content"] == "live question" + assert adopted[-1].get("timestamp") is not None + assert old_live_list[-1]["content"] == "live question" + assert ( + old_live_list[-1].get("timestamp") == adopted[-1].get("timestamp") + ) + + # (f) Markers on BOTH views. Note: the adopted view's rows are + # "born durable" (hermes_state stamps _DB_PERSISTED_MARKER on rows + # materialized from the DB), so the adopted assert holds even + # pre-fix; the DISCRIMINATING assert is the twin's — the old live + # view is a constructed list the production code only stamps via + # the guard's twin scan (pre-fix it stays unstamped → FAIL). + assert _DB_PERSISTED_MARKER in adopted[-1] + assert _DB_PERSISTED_MARKER in old_live_list[-1] + real_flush(agent._session_messages, conversation_history=loaded) + + # (g) No standalone duplicate by EXACT SCOPED IDENTITY (content + + # timestamp), and exactly one merged row. + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert not any( + row.get("content") == "live question" + and row.get("timestamp") == adopted[-1].get("timestamp") + for row in child_rows + ) + assert ( + _count_rows( + child_rows, content="live question\n\nhandoff scaffolding" + ) + == 1 + ) + + def test_rotation_stamps_anchor_source_when_reanchor_fallback_rewrote_turn( + self, tmp_path: Path + ): + """REAL reanchor drift: reanchor_current_turn_user_idx's last-user + fallback lands on a trailing production-shaped todo-snapshot row + (index 2) while the anchor-source scan selects the rewritten carrier + (index 1). The stamp must land on the carrier, not the todo row.""" + from agent.turn_context import reanchor_current_turn_user_idx + + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_REANCHOR_DRIFT" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "old durable") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + # Pinned production-shaped fixture (plan §3.5): the trailing row is + # the todo-snapshot shape compress_context appends at :3484-3489. + # The reanchor helper's last-user-originated fallback lands on it + # (index 2) while the anchor-source scan skips it (synthetic flag) + # and selects the rewritten carrier (index 1) — the drift is real. + # Deliberately a 3-row fixture (NOT prefixed with `loaded`): the + # pinned drift values below were verified against the real + # reanchor_current_turn_user_idx on this shape. + messages = [ + {"role": "user", "content": "old durable"}, + {"role": "user", "content": "current ask\n\n[merged summary]"}, + { + "role": "user", + "content": "Current todos:\n- [ ] leftover", + "_todo_snapshot_synthetic": True, + }, + ] + + # Drift-first assertions: prove the reanchor index and the anchor + # source diverge BEFORE any stamp behavior is checked. + drifted = reanchor_current_turn_user_idx(messages, "current ask") + anchor_source = max( + i for i, m in enumerate(messages) if _is_real_user_message(m) + ) + assert drifted == 2 + assert anchor_source == 1 + assert drifted != anchor_source + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = drifted + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": "handoff scaffolding", + "_todo_snapshot_synthetic": True, + }, + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + # The carrier (anchor source) is stamped; the todo-snapshot row the + # drifted index points at is NOT. + assert _DB_PERSISTED_MARKER in messages[1] + assert _DB_PERSISTED_MARKER not in messages[2] + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + assert ( + _count_rows( + child_rows, + content="current ask\n\n[merged summary]", + role="user", + ) + == 0 + ) + assert ( + _count_rows( + child_rows, + content=( + "current ask\n\n[merged summary]\n\nhandoff scaffolding" + ), + ) + == 1 + ) + + def test_no_real_user_anchor_guard_not_entered(self, tmp_path: Path): + """Negative regression: placeholder_appended/already_present must not + enter the anchor-source guard branch — no exception, rotation happens, + no live row outside the handoff carries the marker.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_NO_REAL_ANCHOR" + db.create_session(parent, source="cli") + + # All-user-synthetic transcript with NO real user. The rows carry + # enough content that compression shrinks the transcript (a single + # short synthetic row trips the would-grow gate and aborts rotation, + # which would make this a fixture failure, not a regression). + messages = [ + { + "role": "user", + "content": f"synthetic scaffolding block {i} with enough " + f"content to keep the compressed transcript smaller", + "_todo_snapshot_synthetic": True, + } + for i in range(6) + ] + + agent = _build_agent_with_db(db, parent) + agent.context_compressor.compress.return_value = [ + {"role": "assistant", "content": "[CONTEXT COMPACTION] summary"}, + ] + + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + # Rotation happened; no live row carries the marker. + assert agent.session_id != parent + assert _DB_PERSISTED_MARKER not in messages[0] + + def test_list_content_merged_outcome_still_stamps_live_question( + self, tmp_path: Path + ): + """Constraint (reviewer list-content requirement): list-content anchor + merged via the list branch (anchor_parts + target_parts) must still + stamp the live row and not duplicate it.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_LIST_MERGED" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [ + *loaded, + { + "role": "user", + "content": [{"type": "text", "text": "live question"}], + }, + ] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": [{"type": "text", "text": "scaffolding"}], + "_todo_snapshot_synthetic": True, + }, + ] + + real_flush = agent._flush_messages_to_session_db + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + assert _DB_PERSISTED_MARKER in messages[-1] + real_flush(messages, conversation_history=loaded) + + child_rows = db.get_messages_as_conversation( + agent.session_id, include_inactive=True + ) + # No standalone live-question row (the flush stores list content + # flattened to its text join, so match the flattened string too). + assert _count_rows(child_rows, content="live question") == 0 + assert ( + _count_rows( + child_rows, + content=[{"type": "text", "text": "live question"}], + ) + == 0 + ) + # Exactly one merged row with the concatenated parts list. + assert ( + _count_rows( + child_rows, + content=[ + {"type": "text", "text": "live question"}, + {"type": "text", "text": "scaffolding"}, + ], + ) + == 1 + ) + + def test_already_stamped_exact_twin_suppresses_broad_fallback( + self, tmp_path: Path + ): + """Two-phase edge (reviewer scenario): an already-stamped exact twin + must still count as an exact hit (marker-INDEPENDENT), suppressing the + broad fallback so a timestamp-less same-content historical row is NOT + stamped as the anchor's twin.""" + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_STAMPED_TWIN" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [ + *loaded, + { + "role": "user", + "content": "live question", + "timestamp": "2026-01-01T00:00:00Z", + }, + ] + + agent = _build_agent_with_db(db, parent) + # Deliberately do NOT set _persist_user_message_idx: the guard must + # not trust a persist index at all. + agent._session_messages = [ + { + "role": "user", + "content": "live question", + "timestamp": "2026-01-01T00:00:00Z", + _DB_PERSISTED_MARKER: True, + }, + {"role": "user", "content": "live question"}, + ] + agent.context_compressor.compress.return_value = [ + { + "role": "user", + "content": "handoff scaffolding", + "_todo_snapshot_synthetic": True, + }, + ] + + with patch.object( + agent, + "_flush_messages_to_session_db", + side_effect=RuntimeError("simulated parent flush failure"), + ): + _returned, _ = agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + ) + + # The timestamp-less ambiguous row must NOT be stamped as a twin. + assert _DB_PERSISTED_MARKER not in agent._session_messages[1] + # The already-stamped exact twin keeps its marker (idempotent). + assert _DB_PERSISTED_MARKER in agent._session_messages[0] + # The primary anchor is still stamped. + assert _DB_PERSISTED_MARKER in messages[-1] + + class TestPlatformForwardedAtBoundary: def test_on_session_start_receives_platform(self, tmp_path: Path): db = SessionDB(db_path=tmp_path / "state.db") @@ -688,7 +1607,14 @@ class TestTodoSnapshotScaffoldingTails: _msgs(), "sys", approx_tokens=120_000 ) - assert [{k: v for k, v in m.items() if k != "_row_id"} for m in compressed] == expected + assert [ + { + k: v + for k, v in m.items() + if k not in {"_row_id", _DB_PERSISTED_MARKER} + } + for m in compressed + ] == expected assert not any( TODO_INJECTION_HEADER in str(message.get("content") or "") for message in compressed diff --git a/tests/run_agent/test_413_compression.py b/tests/run_agent/test_413_compression.py index ad9bfcfca4..9800c44cef 100644 --- a/tests/run_agent/test_413_compression.py +++ b/tests/run_agent/test_413_compression.py @@ -15,7 +15,7 @@ from types import SimpleNamespace from unittest.mock import MagicMock, patch -from agent.context_compressor import SUMMARY_PREFIX +from agent.context_compressor import SUMMARY_PREFIX, _DB_PERSISTED_MARKER from agent.conversation_compression import COMPACTION_DONE_STATUS, COMPACTION_STATUS from run_agent import AIAgent import run_agent @@ -479,7 +479,7 @@ class TestPreflightCompression: # summary-first before the next API call). assert compressed == [ {"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}, - {"role": "user", "content": "hello"}, + {"role": "user", "content": "hello", _DB_PERSISTED_MARKER: True}, ] assert new_system_prompt == "You are helpful." build_prompt.assert_not_called() @@ -675,7 +675,7 @@ class TestPreflightCompression: mock_compress.return_value = ( [ {"role": "user", "content": f"{SUMMARY_PREFIX}\nPrevious conversation"}, - {"role": "user", "content": "hello"}, + {"role": "user", "content": "hello", _DB_PERSISTED_MARKER: True}, ], "new system prompt", ) diff --git a/tests/run_agent/test_compress_context_fallback_shim.py b/tests/run_agent/test_compress_context_fallback_shim.py new file mode 100644 index 0000000000..a121934995 --- /dev/null +++ b/tests/run_agent/test_compress_context_fallback_shim.py @@ -0,0 +1,102 @@ +"""Item 1 regression — the run_agent._compress_context fallback shim must be loud. + +Before the fix, _compress_context wrapped the imports of _DB_PERSISTED_MARKER +(agent.context_compressor) and _messages_match_scoped_identity +(agent.conversation_compression) in a try/except that silently defined local +fallbacks (a hard-coded ``"_db_persisted"`` literal and a local copy of the +identity helper) with NO logging. If the canonical constant/helper is renamed +or removed upstream, the import raises, the fallback silently keeps stamping +with the stale literal, and the stamping key splits from the flush's — the +duplicate-row bug this PR fixes returns with no error anywhere. + +The fix imports both symbols UNCONDITIONALLY (no fallback), so a +renamed/removed symbol must fail the wrapper loudly with ImportError before +any stamping happens. + +The ``already_present`` outcome is load-bearing: it keeps compress_context +from touching the deleted module-global name (which would raise NameError on +BOTH pre- and post-fix code and make the test non-discriminating), because the +stamp block at conversation_compression.py:3834-3859 — the ONLY in-module use +of _messages_match_scoped_identity — is skipped for already_present. +""" + +import os +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +import agent.conversation_compression as conversation_compression +from agent.conversation_compression import CompressionCommitFence +from hermes_state import SessionDB + + +def _build_agent_with_db(db: SessionDB, session_id: str, platform: str = "telegram"): + with patch.dict(os.environ, {"OPENROUTER_API_KEY": "test-key"}): + from run_agent import AIAgent + + agent = AIAgent( + api_key="test-key", + base_url="https://openrouter.ai/api/v1", + model="test/model", + platform=platform, + quiet_mode=True, + session_db=db, + session_id=session_id, + skip_context_files=True, + skip_memory=True, + ) + + compressor = MagicMock() + # A real user row in the stub return makes _ensure_compressed_has_user_turn + # return `already_present`, so the in-module stamp block (the only user of + # _messages_match_scoped_identity inside compress_context) is skipped and + # the deleted name is referenced ONLY by the run_agent shim import. + compressor.compress.return_value = [ + {"role": "user", "content": "real user row"}, + ] + compressor.compression_count = 1 + compressor.last_prompt_tokens = 0 + compressor.last_completion_tokens = 0 + compressor._last_summary_error = None + compressor._last_compress_aborted = False + compressor._last_summary_auth_failure = False + compressor._last_aux_model_failure_model = None + compressor._last_aux_model_failure_error = None + agent.context_compressor = compressor + # ROTATION fallback path — pin in_place=False so the fork-rotation path is + # exercised regardless of the global default (flipped to True in #38763). + agent.compression_in_place = False + return agent + + +class TestCompressContextFallbackShim: + def test_compress_context_shim_import_failure_is_loud( + self, tmp_path: Path, monkeypatch + ): + db = SessionDB(db_path=tmp_path / "state.db") + parent = "PARENT_ROT_SHIM_LOUD" + db.create_session(parent, source="cli") + db.append_message(parent, "user", "persisted question") + db.append_message(parent, "assistant", "persisted answer") + + loaded = db.get_messages_as_conversation(parent) + messages = [*loaded, {"role": "user", "content": "live question"}] + + agent = _build_agent_with_db(db, parent) + agent._persist_user_message_idx = len(messages) - 1 + + # Delete the canonical helper from its defining module: only the shim + # import can still reference it (already_present skips the in-module + # stamp block). Pre-fix the except branch silently defines a fallback; + # post-fix the unconditional import must raise ImportError. + monkeypatch.delattr( + conversation_compression, "_messages_match_scoped_identity" + ) + with pytest.raises(ImportError): + agent._compress_context( + messages, + "sys", + approx_tokens=120_000, + commit_fence=CompressionCommitFence(), + )