From 85e08110fbc623d967745e2231b84e3b8682d65c Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 13 Aug 2026 10:20:58 -0700 Subject: [PATCH] fix(relay): defer rotating-compaction session close while a turn is live MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit notify_session_compacted closed the old session scope immediately on a legacy rotating compaction. A compaction can complete while a turn is still live on the old session; closing then pops the session scope under the live turn scope, violating the stack's LIFO order — the exact invariant the rest of the segmentation feature protects. Now: when the old session has an active turn, set close_pending instead; that turn's end_turn consumes the flag after its own turn scope pops and it unregisters from the active-turn table. Sabotage-verified: the new test fails without the fix. --- agent/relay_runtime.py | 50 +++++++++++++++++++++- tests/agent/test_relay_session_segments.py | 33 ++++++++++++++ 2 files changed, 82 insertions(+), 1 deletion(-) diff --git a/agent/relay_runtime.py b/agent/relay_runtime.py index 0e0528c0af..a1af9385ae 100644 --- a/agent/relay_runtime.py +++ b/agent/relay_runtime.py @@ -112,6 +112,10 @@ class RelaySession: segment_turns: int = 0 # Set by compaction notification; consumed at the next begin_turn. rotate_pending: bool = False + # Rotating compaction landed while a turn was live on THIS session: + # closing now would pop the session scope under a live turn scope + # (LIFO violation). end_turn consumes this and closes the session. + close_pending: bool = False # --------------------------------------------------------------------------- @@ -1007,6 +1011,38 @@ class RelaySessionCoordinator: finally: self._unregister_active_turn(turn) self._reset_turn_context(turn) + self._consume_deferred_close(lease) + + def _consume_deferred_close(self, lease: Any) -> None: + """Close a session whose rotating-compaction close was deferred. + + ``notify_session_compacted`` sets ``close_pending`` instead of + closing when the old session still has a live turn (closing then + would pop the session scope under the live turn scope — LIFO + violation). The turn that was live consumes the flag here, after + its own turn scope popped and it unregistered from the + active-turn table. Skips when another turn is still live on the + same session; that turn's end_turn will consume it instead. + """ + try: + if not ( + isinstance(lease.host, RelayRuntime) and lease.session is not None + ): + return + session = lease.session + with session.lock: + pending = session.close_pending and not session.closing + if not pending: + return + if self.has_active_turn( + profile_key=lease.profile_key, session_id=lease.session_id + ): + return + lease.host.close_session({"session_id": lease.session_id}) + except Exception: # noqa: BLE001 - telemetry must never block end_turn + logger.warning( + "Hermes Relay deferred session close failed", exc_info=True + ) def notify_session_compacted( self, @@ -1042,7 +1078,19 @@ class RelaySessionCoordinator: return if old_session_id and old_session_id != session_id: # Rotating compaction: export the orphaned pre-compaction - # session scope (close_session is already bounded). + # session scope (close_session is already bounded). If a + # turn is still LIVE on the old session, closing now would + # pop the session scope under the live turn scope (LIFO + # violation) — defer to that turn's end_turn instead. + with host._sessions_lock: + old_session = host._sessions.get(old_session_id) + if old_session is not None and self.has_active_turn( + profile_key=profile_key, session_id=old_session_id + ): + with old_session.lock: + if not old_session.closing: + old_session.close_pending = True + return host.close_session({"session_id": old_session_id}) return with host._sessions_lock: diff --git a/tests/agent/test_relay_session_segments.py b/tests/agent/test_relay_session_segments.py index c9e522e9a5..8a228891ca 100644 --- a/tests/agent/test_relay_session_segments.py +++ b/tests/agent/test_relay_session_segments.py @@ -261,6 +261,39 @@ class TestCompactionRotation: ) assert fake.subscribers.flushed >= 1 + def test_rotating_compaction_mid_turn_defers_close_to_end_turn( + self, coordinator, monkeypatch + ): + """A rotating compaction completing while a turn is LIVE on the old + session must NOT close the session scope immediately — that would pop + it under the live turn scope (LIFO violation). The close defers to + that turn's end_turn.""" + _set_segments(monkeypatch, on_compaction=True) + fake = _FakeRelay() + runtime = _make_runtime(fake) + lease = _acquire(coordinator, runtime, session_id="parent-1") + + turn = coordinator.begin_turn(lease, turn_id="t1", task_id="task1") + coordinator.notify_session_compacted( + profile_key=runtime.profile_key, + session_id="child-1", + old_session_id="parent-1", + ) + # No pops yet: neither the turn scope nor the session scope closed. + assert not fake.scope.pops, ( + "old-session close must defer while its turn is live" + ) + + coordinator.end_turn(turn, outcome="success") + # Turn scope popped first, then the deferred session close popped + # the session scope — LIFO order preserved. + assert len(fake.scope.pops) == 2, "end_turn must consume deferred close" + assert fake.scope.pops[0].name == relay_runtime.TURN_SCOPE, ( + "turn scope must pop before the session scope" + ) + assert fake.scope.pops[-1].name == relay_runtime.SESSION_SCOPE + assert runtime.get_session("parent-1") is None + def test_rotating_compaction_noop_when_disabled(self, coordinator, monkeypatch): fake = _FakeRelay() runtime = _make_runtime(fake)