hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 33 Packages Projects Releases Wiki Activity

fix(compression): dedupe current-turn rows when rotation splits the session mid-turn

Browse Source 
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.
This commit is contained in:
kshitijk4poor
2026-08-28 02:39:55 +05:30
parent 28ee6ac043
commit 80ab7d2b1c
5 changed files with 1414 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

192
agent/conversation_compression.py
View File

@@ -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

321
run_agent.py
View File

@@ -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

930
tests/agent/test_compression_rotation_state.py
View File

@@ -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

6
tests/run_agent/test_413_compression.py
View File

@@ -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",
)

102
tests/run_agent/test_compress_context_fallback_shim.py Normal file
View File

@@ -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(),
)