fix(compression): recover live tip across rotation chains - #71486
Open
ruizanthony wants to merge 1 commit into
Open
fix(compression): recover live tip across rotation chains#71486ruizanthony wants to merge 1 commit into
ruizanthony wants to merge 1 commit into
Conversation
alt-glitch
added
type/bug
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
3 times, most recently
from
July 29, 2026 18:13
f9810d6 to
7761d9a
Compare
Contributor
Author
|
Rebased cleanly onto current |
Contributor
Author
|
Friendly nudge for merge order: this PR is gate-green (38/38 checks) at exact head |
Independent production reproduction confirms this bug and the fix:
This is not theoretical depth hardening: repeated compression on a long-lived WebUI session reproduced it in production. The multi-hop traversal and gateway rerouting here address the observed failure class. |
|
Final correction for exact PR head The original 57-hop production reproduction remains valid, but the current PR head should not be merged unchanged. The earlier strict follow-up still had a shallow 100-hop cap and did not cover two caller-side FTS recovery errors. I replaced it with a complete follow-up against the exact PR head. The final follow-up:
Verification on the exact PR head plus this patch: Patch SHA-256: Complete follow-up patch for exact head 7761d9ad…diff --git a/gateway/session.py b/gateway/session.py
index 89f418866..d7c4ab3c9 100644
--- a/gateway/session.py
+++ b/gateway/session.py
@@ -2978,11 +2978,24 @@ class SessionStore:
child = self._db.find_live_compression_child(session_id)
child_id = str(child["id"]) if child and child.get("id") else ""
if child_id:
+ reroute_succeeded = False
try:
self._append_transcript_message(child_id, msg)
except Exception as reroute_exc:
exc = reroute_exc
+ if (
+ self._is_fts_corruption_error(reroute_exc)
+ and self._rebuild_fts_once()
+ ):
+ try:
+ self._append_transcript_message(child_id, msg)
+ except Exception as retry_exc:
+ exc = retry_exc
+ else:
+ reroute_succeeded = True
else:
+ reroute_succeeded = True
+ if reroute_succeeded:
with self._transcript_retry_lock:
if pending and pending[0] is msg:
pending.pop(0)
@@ -3047,6 +3060,8 @@ class SessionStore:
if not pending:
self._dirty_transcripts.pop(queue_session_id, None)
self._transcript_append_failures.pop(session_id, None)
+ return
+ msg = pending[0]
continue
with self._transcript_retry_lock:
failures = self._transcript_append_failures.get(session_id, 0) + 1
diff --git a/hermes_state.py b/hermes_state.py
index 51768174e..0221b7e84 100644
--- a/hermes_state.py
+++ b/hermes_state.py
@@ -2760,59 +2760,119 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin)
def find_live_compression_child(
self, parent_session_id: str
) -> Optional[Dict[str, Any]]:
- """Follow a unique compression-continuation chain to its live tip.
-
- A stale agent may observe that another compression path already rotated
- its parent more than once. Recovery is safe only when every hop in the
- durable lineage identifies exactly one continuation. Multiple eligible
- children are treated as ambiguous and fail closed rather than guessing
- which transcript owns subsequent messages.
+ """Return the unique live tip of a compression continuation chain.
+
+ A stale agent may still address any archived segment after one or more
+ compression rotations. Recovery is safe only while every compression
+ segment has exactly one canonical continuation. Branches, delegates,
+ tool children, malformed metadata, invalid lifecycle states, and
+ lineage cycles all fail closed rather than guessing which transcript
+ owns subsequent messages.
"""
if not parent_session_id:
return None
- current_session_id = parent_session_id
- seen: set[str] = set()
with self._lock:
conn = self._conn
- if conn is None:
+ if conn is None or conn.in_transaction:
return None
- for _ in range(100):
- if current_session_id in seen:
- return None
- seen.add(current_session_id)
- parent = conn.execute(
- "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
- (current_session_id,),
- ).fetchone()
- if (
- parent is None
- or parent["ended_at"] is None
- or parent["end_reason"] != "compression"
- ):
- return None
+
+ def _canonical_children(session_id: str):
rows = conn.execute(
"""
- SELECT * FROM sessions
+ SELECT id, ended_at, end_reason, model_config, source
+ FROM sessions
WHERE parent_session_id = ?
- AND (ended_at IS NULL OR end_reason = 'compression')
- AND json_extract(COALESCE(model_config, '{}'), '$._branched_from') IS NULL
- AND json_extract(COALESCE(model_config, '{}'), '$._delegate_from') IS NULL
- AND COALESCE(source, '') != 'tool'
- ORDER BY started_at ASC
- LIMIT 2
+ ORDER BY started_at ASC, id ASC
""",
- (current_session_id,),
+ (session_id,),
).fetchall()
- if len(rows) != 1:
- return None
- child = rows[0]
- child_session_id = str(child["id"] or "")
- if not child_session_id or child_session_id in seen:
- return None
- if child["ended_at"] is None:
- return dict(child)
- current_session_id = child_session_id
- return None
+ canonical = []
+ for row in rows:
+ source = row["source"]
+ if not isinstance(source, str) or not source:
+ return None
+ if source.lower() == "tool":
+ continue
+ raw_config = row["model_config"]
+ if raw_config is None:
+ model_config = {}
+ elif not isinstance(raw_config, str):
+ return None
+ else:
+ try:
+ model_config = json.loads(raw_config)
+ except (TypeError, ValueError, json.JSONDecodeError):
+ return None
+ if not isinstance(model_config, dict):
+ return None
+ marker_names = [
+ marker
+ for marker in ("_branched_from", "_delegate_from")
+ if marker in model_config
+ ]
+ if len(marker_names) > 1:
+ return None
+ if marker_names:
+ marker_value = model_config[marker_names[0]]
+ if (
+ not isinstance(marker_value, str)
+ or not marker_value
+ or marker_value != session_id
+ ):
+ return None
+ continue
+ canonical.append(row)
+ return canonical
+
+ conn.execute("BEGIN")
+ try:
+ current_session_id = parent_session_id
+ seen: set[str] = set()
+ while current_session_id not in seen:
+ seen.add(current_session_id)
+ parent = conn.execute(
+ "SELECT ended_at, end_reason FROM sessions WHERE id = ?",
+ (current_session_id,),
+ ).fetchone()
+ if (
+ parent is None
+ or parent["ended_at"] is None
+ or parent["end_reason"] != "compression"
+ ):
+ return None
+
+ children = _canonical_children(current_session_id)
+ if children is None or len(children) != 1:
+ return None
+ child = children[0]
+ child_session_id = child["id"]
+ if (
+ not isinstance(child_session_id, str)
+ or not child_session_id
+ or child_session_id in seen
+ ):
+ return None
+
+ child_ended_at = child["ended_at"]
+ child_end_reason = child["end_reason"]
+ if child_ended_at is None:
+ if child_end_reason is not None:
+ return None
+ descendants = _canonical_children(child_session_id)
+ if descendants is None or descendants:
+ return None
+ live_child = conn.execute(
+ "SELECT * FROM sessions WHERE id = ?",
+ (child_session_id,),
+ ).fetchone()
+ return dict(live_child) if live_child is not None else None
+ if child_end_reason != "compression":
+ return None
+ current_session_id = child_session_id
+ return None
+ finally:
+ if conn.in_transaction:
+ conn.rollback()
def publish_compression_child(
self,
diff --git a/tests/gateway/test_session.py b/tests/gateway/test_session.py
index f6f7b3703..7fb3c3d3c 100644
--- a/tests/gateway/test_session.py
+++ b/tests/gateway/test_session.py
@@ -2515,6 +2515,114 @@ class TestGatewaySessionDbRecovery:
store.rewind_session("s1", 1)
assert "s1" not in store._dirty_transcripts
+ def test_fts_retry_success_empty_queue_does_not_duplicate(self):
+ import threading
+
+ store = object.__new__(SessionStore)
+ store._transcript_retry_lock = threading.Lock()
+ store._dirty_transcripts = {}
+ store._transcript_append_failures = {}
+ attempts = []
+ first_attempt = True
+
+ def _append(session_id, message):
+ nonlocal first_attempt
+ attempts.append((session_id, message["content"]))
+ if first_attempt:
+ first_attempt = False
+ raise RuntimeError("database disk image is malformed")
+
+ store._append_transcript_message = _append
+ store._rebuild_fts_once = lambda: True
+
+ store._append_to_transcript_serialized(
+ "session", {"role": "user", "content": "only"}
+ )
+
+ assert attempts == [("session", "only"), ("session", "only")]
+ assert store._dirty_transcripts == {}
+
+ def test_fts_retry_success_advances_to_remaining_message(self):
+ import threading
+
+ store = object.__new__(SessionStore)
+ store._transcript_retry_lock = threading.Lock()
+ store._dirty_transcripts = {
+ "session": [{"role": "user", "content": "older"}]
+ }
+ store._transcript_append_failures = {}
+ attempts = []
+ first_attempt = True
+
+ def _append(session_id, message):
+ nonlocal first_attempt
+ attempts.append((session_id, message["content"]))
+ if first_attempt:
+ first_attempt = False
+ raise RuntimeError("database disk image is malformed")
+
+ store._append_transcript_message = _append
+ store._rebuild_fts_once = lambda: True
+
+ store._append_to_transcript_serialized(
+ "session", {"role": "assistant", "content": "newer"}
+ )
+
+ assert attempts == [
+ ("session", "older"),
+ ("session", "older"),
+ ("session", "newer"),
+ ]
+ assert store._dirty_transcripts == {}
+
+ def test_fts_retry_during_reroute_targets_child(self):
+ import threading
+ from types import SimpleNamespace
+ from typing import Any
+
+ from hermes_state import CompressionSessionClosedError
+
+ class FakeDb:
+ def find_live_compression_child(self, session_id):
+ assert session_id == "parent"
+ return {"id": "child"}
+
+ store: Any = object.__new__(SessionStore)
+ store._db = FakeDb()
+ store._lock = threading.RLock()
+ store._entries = {"route": SimpleNamespace(session_id="parent")}
+ store._save = lambda: None
+ store._transcript_retry_lock = threading.Lock()
+ store._dirty_transcripts = {}
+ store._transcript_append_failures = {}
+ store._transcript_reroutes = {}
+ attempts = []
+ child_first_attempt = True
+
+ def _append(session_id, message):
+ nonlocal child_first_attempt
+ attempts.append((session_id, message["content"]))
+ if session_id == "parent":
+ raise CompressionSessionClosedError("parent")
+ if child_first_attempt:
+ child_first_attempt = False
+ raise RuntimeError("database disk image is malformed")
+
+ store._append_transcript_message = _append
+ store._rebuild_fts_once = lambda: True
+
+ store._append_to_transcript_serialized(
+ "parent", {"role": "assistant", "content": "rerouted"}
+ )
+
+ assert attempts == [
+ ("parent", "rerouted"),
+ ("child", "rerouted"),
+ ("child", "rerouted"),
+ ]
+ assert store._entries["route"].session_id == "child"
+ assert store._dirty_transcripts == {}
+
def test_fts_corruption_error_does_not_match_false_positives(self):
"""_is_fts_corruption_error must not match unrelated error strings
containing 'fts' as a substring (e.g. 'shifts', 'gifts')."""
diff --git a/tests/state/test_compression_lineage_guard.py b/tests/state/test_compression_lineage_guard.py
index 7a4869e4a..072cdaf84 100644
--- a/tests/state/test_compression_lineage_guard.py
+++ b/tests/state/test_compression_lineage_guard.py
@@ -126,12 +126,10 @@ def test_find_live_compression_child_rejects_cycle(db: SessionDB) -> None:
assert db.find_live_compression_child("parent") is None
-def test_find_live_compression_child_accepts_maximum_supported_depth(
- db: SessionDB,
-) -> None:
+def test_find_live_compression_child_has_no_shallow_depth_cap(db: SessionDB) -> None:
_compression_parent(db)
parent = "parent"
- for index in range(99):
+ for index in range(128):
child = f"compressed-{index}"
db.create_session(child, source="webui", parent_session_id=parent)
db.end_session(child, "compression")
@@ -144,15 +142,92 @@ def test_find_live_compression_child_accepts_maximum_supported_depth(
assert child["id"] == "deep-live-tip"
-def test_find_live_compression_child_bounds_pathological_depth(db: SessionDB) -> None:
+def test_find_live_compression_child_rejects_live_intermediate(
+ db: SessionDB,
+) -> None:
_compression_parent(db)
- parent = "parent"
- for index in range(100):
- child = f"compressed-{index}"
- db.create_session(child, source="webui", parent_session_id=parent)
- db.end_session(child, "compression")
- parent = child
- db.create_session("too-deep-tip", source="webui", parent_session_id=parent)
+ db.create_session("live-intermediate", source="webui", parent_session_id="parent")
+ db.create_session(
+ "live-grandchild",
+ source="webui",
+ parent_session_id="live-intermediate",
+ )
+
+ assert db.find_live_compression_child("parent") is None
+
+
+def test_find_live_compression_child_counts_closed_noncompression_sibling(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("closed-sibling", source="webui", parent_session_id="parent")
+ db.end_session("closed-sibling", "agent_close")
+ db.create_session("live-child", source="webui", parent_session_id="parent")
+
+ assert db.find_live_compression_child("parent") is None
+
+
+def test_find_live_compression_child_rejects_live_row_with_end_reason(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("inconsistent", source="webui", parent_session_id="parent")
+ with db._lock:
+ assert db._conn is not None
+ db._conn.execute(
+ "UPDATE sessions SET end_reason = 'compression' WHERE id = ?",
+ ("inconsistent",),
+ )
+ db._conn.commit()
+
+ assert db.find_live_compression_child("parent") is None
+
+
+def test_find_live_compression_child_rejects_malformed_decoy_marker(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("live-child", source="webui", parent_session_id="parent")
+ db.create_session(
+ "malformed-decoy",
+ source="webui",
+ parent_session_id="parent",
+ model_config={"_branched_from": None},
+ )
+
+ assert db.find_live_compression_child("parent") is None
+
+
+def test_find_live_compression_child_rejects_nontext_child_metadata(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("blob-config", source="webui", parent_session_id="parent")
+ with db._lock:
+ assert db._conn is not None
+ db._conn.execute(
+ "UPDATE sessions SET model_config = ? WHERE id = ?",
+ (b"{}", "blob-config"),
+ )
+ db._conn.commit()
+
+ assert db.find_live_compression_child("parent") is None
+
+
+@pytest.mark.parametrize("raw_config", ["{not-json", "[]", "null"])
+def test_find_live_compression_child_rejects_malformed_child_metadata(
+ db: SessionDB,
+ raw_config: str,
+) -> None:
+ _compression_parent(db)
+ db.create_session("malformed", source="webui", parent_session_id="parent")
+ with db._lock:
+ assert db._conn is not None
+ db._conn.execute(
+ "UPDATE sessions SET model_config = ? WHERE id = ?",
+ (raw_config, "malformed"),
+ )
+ db._conn.commit()
assert db.find_live_compression_child("parent") is None
@@ -198,6 +273,68 @@ def test_find_live_compression_child_ignores_non_continuation_children(
assert child["id"] == "canonical"
+@pytest.mark.parametrize("marker", ["_branched_from", "_delegate_from"])
+def test_find_live_compression_child_rejects_decoy_marker_for_different_parent(
+ db: SessionDB,
+ marker: str,
+) -> None:
+ _compression_parent(db)
+ db.create_session("canonical", source="webui", parent_session_id="parent")
+ db.create_session(
+ "false-decoy",
+ source="webui",
+ parent_session_id="parent",
+ model_config={marker: "different-parent"},
+ )
+
+ assert db.find_live_compression_child("parent") is None
+
+
+def test_find_live_compression_child_ignores_malformed_tool_decoy(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("live-tip", source="webui", parent_session_id="parent")
+ db.create_session("tool-child", source="tool", parent_session_id="parent")
+ with db._lock:
+ assert db._conn is not None
+ db._conn.execute(
+ "UPDATE sessions SET model_config = ? WHERE id = ?",
+ ("{not-json", "tool-child"),
+ )
+ db._conn.commit()
+
+ child = db.find_live_compression_child("parent")
+
+ assert child is not None
+ assert child["id"] == "live-tip"
+
+
+def test_find_live_compression_child_ignores_decoys_below_live_tip(
+ db: SessionDB,
+) -> None:
+ _compression_parent(db)
+ db.create_session("live-tip", source="webui", parent_session_id="parent")
+ db.create_session(
+ "branch",
+ source="webui",
+ parent_session_id="live-tip",
+ model_config={"_branched_from": "live-tip"},
+ )
+ db.create_session(
+ "delegate",
+ source="webui",
+ parent_session_id="live-tip",
+ model_config={"_delegate_from": "live-tip"},
+ )
+ db.create_session("tool-child", source="tool", parent_session_id="live-tip")
+
+ child = db.find_live_compression_child("parent")
+
+ assert child is not None
+ assert child["id"] == "live-tip"
+
+
def test_append_message_rejects_compression_ended_parent_atomically(db: SessionDB) -> None:
_compression_parent(db)
before = db.get_session("parent")["message_count"] |
teknium1
reviewed
Jul 30, 2026
teknium1
left a comment
teknium1
left a comment
Contributor
There was a problem hiding this comment.
Thanks for addressing the real multi-hop recovery gap: current main still has a direct-child-only resolver at hermes_state.py:3076-3113.
Problems
hermes_state.py:2812at PR head returns a child solely becauseended_at IS NULL; it does not prove the row is a leaf. A live intermediate with a canonical descendant would be adopted instead of failing closed.hermes_state.py:2797filters out closed non-compression children beforelen(rows)is checked, so a canonicalagent_closesibling does not make the lineage ambiguous.hermes_state.py:2798-2799use SQLitejson_extract()on arbitrary metadata. Malformed values can raise instead of failing closed, and decoy markers are not validated before excluding a child.
Suggested changes
- Read all direct children in one snapshot, defensively validate TEXT metadata, remove only validated decoys, then require exactly one canonical child before lifecycle classification.
- Return a live child only if both lifecycle fields are NULL and it has no canonical descendants; continue only through a child closed for compression. Add the corresponding malformed-metadata and ambiguity regression coverage.
Automated hermes-sweeper review.
teknium1
added
sweeper:risk-message-delivery
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
from
August 3, 2026 07:12
7761d9a to
6e71486
Compare
Contributor
Author
|
Addressed all three fail-closed review findings at exact head |
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
from
August 3, 2026 13:02
6e71486 to
5d4539d
Compare
Contributor
Author
|
Rebased on current |
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
from
August 4, 2026 00:51
5d4539d to
9a97f2c
Compare
Contributor
Author
|
Published exact candidate Certification evidence:
Please re-certify and review the exact head |
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
from
August 4, 2026 01:28
9a97f2c to
08f1720
Compare
Contributor
Author
|
Published exact candidate This candidate closes the three failures from CI run
Certification evidence:
Please re-certify and review the exact head |
ruizanthony
force-pushed
the
fix/compression-chain-tip-recovery
branch
from
August 20, 2026 22:53
08f1720 to
1d1edf7
Compare
Contributor
Author
|
Final gate remediation is now at What changed:
Regression coverage includes depth 128, cycle, fork, closed sibling ambiguity, malformed/non-TEXT metadata, invalid/duplicate/wrong-parent decoy markers, lifecycle corruption, live intermediate, valid decoys/tool children, one-snapshot concurrency, FTS empty/backlogged queues, and rerouted child FTS retry. Validation:
No compression rearm/headroom, active-user-tail, |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fix compression continuation recovery when a session has rotated through more than one compression child, without allowing a new turn to continue on a compression-ended parent or with mixed lifecycle bindings.
The resolver now:
Falseor exception from context lifecycle hooks as terminal, restores the rollback-capable context binding directly, and aborts before core commit;CompressionRecoveryUnavailableErrorfields;patience_sandraise_on_errorcontrols rather than merely accepting**kwargs, and composes each write against one recovery deadline;__getattr__;on_session_switchthrough__getattr__, keeping the pending gate active without invoking the dynamic hook;build_turn_context()before persistence, model, or tool execution whenever recovery or pending memory binding cannot complete safely;TurnContext, so a concurrent parent rotation cannot be downgraded to a warning before the first model call;mainsystem-prompt deduplication contract when returning the recovered child.Rebase / conflict resolution
08f17204adfd272e610f350ec5261435feb6f2f8mainparent:91937a6dc3ffbbe2f3be91a500f0ecf962c4cf539a97f2cb7917a500b8496e8ac70210446b0492ff.SessionDB.find_live_compression_child()inhermes_state.py.system_promptsjoin required by currentmain.Review feedback addressed
The follow-up covers the sweeper findings and successive exact-SHA counter-review rounds:
Falseand exceptions from both context and memory lifecycle hooks are propagated;unsupported_dbrather than inheriting the ordinary 20-second SQLite budget;**kwargs-only lock adapters fail the same declared-contract check instead of silently accepting undeclared budget controls;**kwargsto retain legacy 20-second/default-no-raise behavior;Falsefrom lease release is handled as a release failure and preserves the committed holder identity;TurnContext, model, tools, or downstream persistence;on_session_startcannot fall through to an otherwise declared rollback binder and complete adoption;finallyon every exit.TDD and validation
RED before the fixes:
None, allowing turn setup to continue on the compression-ended parent;Falselifecycle returns were ignored;**kwargsbypassed the bounded SQLite budget contract;release()returningFalsewas treated as success and discarded the durable holder identity;on_session_switchvia__getattr__, return implicit success, and clear the pending gate without a declared lifecycle contract.on_session_startvia__getattr__, run the callback, and complete adoption without a declared lifecycle contract.**kwargs-only lock adapter was treated as if it explicitly declared the bounded SQLite recovery contract.TurnContextinstead of reloading the durable child transcript.on_session_startcould be rejected but then bypassed through a declaredbind_session_statefallback, completing adoption.MagicMockSessionDB fixtures that no longer declaredget_session, plus one idle-lock test that still expected continuation after fenced turn-start persistence failed.GREEN on the exact committed candidate:
git diff --checkpassed;d0ceb64ccc5944642e7c47e0b147aaacb474d46dis verified read-only.Scope
Seventeen files only: lineage resolver/caller behavior, adoption/recovery and memory lifecycle contracts, SessionDB lock API controls, transcript-holder validation, turn-start persistence gating, and regression coverage. No dependency, schema, configuration, or unrelated user-facing workflow changes.