From 044a77b3b6af4ce16138d42762f812a20b9f7a89 Mon Sep 17 00:00:00 2001 From: kshitij <82637225+kshitijk4poor@users.noreply.github.com> Date: Sat, 12 Sep 2026 12:40:17 +0530 Subject: [PATCH] fix(gateway): failed-turn boundary keyed on the durable conversation tail; exception path classifies overflow before writing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review findings on the salvage (all reproduced with a real SessionStore): 1. Primary persisted-agent path skipped the boundary. The agent's turn-start flush already persists the user row stamped with the inbound platform id, so `has_platform_message_id` saw THIS turn's own row, took the "duplicate" branch and skipped the whole block — including the new assistant boundary. The transcript stayed `[..., 'user']`, exactly the open tail #107070 is about. Fresh sessions hid it a second way: `session_meta` is appended after the agent-flushed user row, so a naive "newest row" tail read sees `session_meta`. 2. The exception fallback appended the boundary unconditionally; a redelivery of an already-closed turn produced `['user', 'assistant', 'assistant']`. 3. The exception fallback wrote the user row + boundary before classifying a 400/500-on-long-session as overflow, growing a session that is already too large (the #1630 no-grow rule the persist path honours). Fix: `SessionDB.latest_conversation_role()` (newest active row excluding the `session_meta`/`system` bookkeeping rows the model never sees) behind `SessionStore.transcript_tail_role()`, which resolves the same route `load_transcript` reads via the existing `_compression_tip_for_session_id`. One `_hmwa_close_failed_turn()` appends the boundary iff that tail is an open user row; both the persist path and the exception fallback call it, so the user-row dedupe no longer gates the boundary and a redelivery never stacks. The overflow verdict in `_hmwa_agent_error_reply` is an early return ahead of every transcript write. The `failed_turn_notice` kwarg and its dead `or _hmwa_failed_turn_notice(...)` fallback are gone; the notice is derived where each consumer needs it. Tests (each red with the production change reverted, green here): boundary keyed on the durable tail with the user write deduped (agent-flushed row → closed; redelivery → nothing); fresh-session agent-flushed failed first turn closed despite `session_meta` (real store); exception-path redelivery adds no second boundary (real store, every lineage location, contract asserted from store state); exception-path overflow persists nothing. Live E2E: `evals/gateway_failure_ownership/probe.py` (real AIAgent + fixture provider) 20/20; the two `failed provider input` turns that previously left an open user tail now close with the "not processed" row. --- gateway/run_turn.py | 66 ++++++------ gateway/session_transcript.py | 13 +++ hermes_state_messages.py | 9 ++ .../test_42039_duplicate_user_message.py | 25 +++-- .../gateway/test_failure_writer_ownership.py | 102 ++++++++++++++++-- .../gateway/test_incomplete_gateway_turns.py | 1 + 6 files changed, 171 insertions(+), 45 deletions(-) diff --git a/gateway/run_turn.py b/gateway/run_turn.py index ca04c21e2e..ff05205106 100644 --- a/gateway/run_turn.py +++ b/gateway/run_turn.py @@ -1548,6 +1548,19 @@ class GatewayTurnMixin: return self._PARTIAL_FAILED_TURN_NOTICE return self._FAILED_TURN_NOTICE + async def _hmwa_close_failed_turn(self, session_id, notice): + """Append the gateway-owned assistant boundary iff the durable tail is an open user row. + + The tail, not "did the gateway write the user row", is the key: on the primary path the + agent's turn-start flush already persisted the row (so the platform-id dedupe skips the + gateway write), and a platform redelivery of an already-closed turn must not stack a + second assistant row.""" + if await self.async_session_store.transcript_tail_role(session_id) != "user": + return + await self.async_session_store.append_to_transcript(session_id, { + "role": "assistant", "content": notice, "timestamp": time.time(), + }) + def _hmwa_classify_turn_failure(self, agent_result, history, session_entry): """Classify a finished turn for transcript persistence. Returns ``(agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure)``. @@ -1654,12 +1667,10 @@ class GatewayTurnMixin: async def _hmwa_persist_turn_transcript( self, *, event, source, session_entry, session_key, agent_result, agent_messages, prepared, response, agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure, - failed_turn_notice=None, ): """Persist this turn to the transcript (session_meta on first turn, closed failed turn on transient failure, nothing on context overflow), update last_prompt_tokens, and re-baseline the - cached agent's message count. *failed_turn_notice* is the SAME string the user was shown; it - closes the failed turn as a gateway-owned assistant row.""" + cached agent's message count.""" from gateway.run import _resolve_gateway_model ts = time.time() # Unix epoch float — consistent with DB storage store = self.async_session_store @@ -1701,13 +1712,9 @@ class GatewayTurnMixin: ) else: await store.append_to_transcript(sid, _user_row, skip_db=agent_persisted) - # Gateway-owned assistant row closing the failed turn: a user-only tail lets - # alternation repair merge this request into an unrelated future message and replay - # stale side effects. Belongs to the user row it closes — a deduped retry has one. - await store.append_to_transcript(sid, { - "role": "assistant", "timestamp": ts, - "content": failed_turn_notice or self._hmwa_failed_turn_notice(agent_result), - }) + # Close the failed turn: a user-only tail lets alternation repair merge this request + # into an unrelated future message and replay stale side effects (#107070). + await self._hmwa_close_failed_turn(sid, self._hmwa_failed_turn_notice(agent_result)) else: # Only the NEW messages: history_offset (what the agent saw), not len(history), which # counts session_meta entries stripped before the agent saw them. @@ -1798,10 +1805,18 @@ class GatewayTurnMixin: async def _hmwa_agent_error_reply(self, e, event, source, session_entry, session_key, prepared): """``except Exception`` body of the agent turn: stop typing, log, persist the inbound user - turn once, and build the sanitized user-facing error reply.""" + turn once and close it, and build the sanitized user-facing error reply.""" # Retain Slack thread/workspace routing so a failed turn cannot leave its status visible. await self._hmwa_stop_typing_for_turn(event, source) logger.exception("Agent error in session %s", session_key) + status_code = getattr(e, "status_code", None) + if status_code in {400, 500} and len(prepared.history) > 50: + # Context overflow / payload too large: a deterministic rejection (#107567), and the same + # no-grow rule as the persist path (#1630) — nothing is written into an oversized session. + return ( + "⚠️ Session too large for the model's context window.\nUse /compact to " + "compress the conversation, or /reset to start fresh." + ) # Replay can coalesce inputs; only this input's durable marker establishes ownership. try: if prepared.message_text is not None and session_entry is not None: @@ -1812,14 +1827,11 @@ class GatewayTurnMixin: await self.async_session_store.append_to_transcript( session_entry.session_id, self._hmwa_user_transcript_entry(event, prepared, time.time()), ) - # Same boundary row as the persist path: tool effects are unknown after an exception. - await self.async_session_store.append_to_transcript(session_entry.session_id, { - "role": "assistant", "content": self._PARTIAL_FAILED_TURN_NOTICE, "timestamp": time.time(), - }) + # Tool effects are unknown after an exception. + await self._hmwa_close_failed_turn(session_entry.session_id, self._PARTIAL_FAILED_TURN_NOTICE) except Exception: logger.debug("Failed to persist inbound user message after agent exception", exc_info=True) # Never expose raw exception types/messages to end users (info-leakage risk). - status_code = getattr(e, "status_code", None) status_hint = self._STATUS_HINTS.get(status_code, "") if status_code == 429: # Plan usage limit (resets on a schedule) vs a transient rate limit @@ -1836,17 +1848,8 @@ class GatewayTurnMixin: status_hint = f" Your plan's usage limit has been reached. It resets in ~{math.ceil(_resets_in / 3600)}h." else: status_hint = " Your plan's usage limit has been reached. Please wait until it resets." - elif status_code in {400, 500}: - # 400/500 on a large session: context overflow / payload too large. - if len(prepared.history) > 50: - # Overflow is a deterministic request rejection, not an indeterminate-effect - # failure (#107567): keep the reply to the /compact / /reset guidance only. - return ( - "⚠️ Session too large for the model's context window.\nUse /compact to " - "compress the conversation, or /reset to start fresh." - ) - elif status_code == 400: - status_hint = " The request was rejected by the API." + elif status_code == 400: + status_hint = " The request was rejected by the API." return self._hmwa_add_failed_turn_notice( f"Sorry, I encountered an unexpected error.{status_hint}\n" "Try again or use /reset to start a fresh session.", @@ -2054,11 +2057,8 @@ class GatewayTurnMixin: agent_failed_early, hidden_reasoning_incomplete, is_context_overflow_failure = ( self._hmwa_classify_turn_failure(agent_result, history, session_entry) ) - failed_turn_notice = None - if (agent_failed_early or hidden_reasoning_incomplete) and not is_context_overflow_failure: - failed_turn_notice = self._hmwa_failed_turn_notice(agent_result) - if agent_failed_early and failed_turn_notice: - response = self._hmwa_add_failed_turn_notice(response, failed_turn_notice) + if agent_failed_early and not is_context_overflow_failure: + response = self._hmwa_add_failed_turn_notice(response, self._hmwa_failed_turn_notice(agent_result)) response, session_entry = await self._hmwa_compression_exhaustion_reset( agent_result, response, session_entry, session_key, source, ) @@ -2067,7 +2067,7 @@ class GatewayTurnMixin: agent_result=agent_result, agent_messages=agent_messages, prepared=prepared, response=response, agent_failed_early=agent_failed_early, hidden_reasoning_incomplete=hidden_reasoning_incomplete, - is_context_overflow_failure=is_context_overflow_failure, failed_turn_notice=failed_turn_notice, + is_context_overflow_failure=is_context_overflow_failure, ) return await self._hmwa_deliver_turn_response( event, source, session_entry, session_key, run_generation, diff --git a/gateway/session_transcript.py b/gateway/session_transcript.py index 50ade91a3a..ba4153d4e0 100644 --- a/gateway/session_transcript.py +++ b/gateway/session_transcript.py @@ -419,6 +419,19 @@ class SessionTranscriptMixin: logger.debug("has_platform_message_id lookup failed", exc_info=True) return False + def transcript_tail_role(self, session_id: str) -> Optional[str]: + """Role of the newest live conversation row on the route ``load_transcript`` reads (``None`` + when empty, no DB, or the read fails — the boundary write would fail the same way).""" + session_id = self._compression_tip_for_session_id(self._follow_reroutes(session_id)) + db = self._db_for_session_id(session_id) + if not db: + return None + try: + return db.latest_conversation_role(session_id) + except Exception: + logger.debug("transcript tail lookup failed for %s", session_id, exc_info=True) + return None + def rewrite_transcript( self, session_id: str, messages: List[Dict[str, Any]], active_only: bool = False, reject_active_turn_lease: bool = False) -> bool: diff --git a/hermes_state_messages.py b/hermes_state_messages.py index c587bd87ae..4e7b96faa7 100644 --- a/hermes_state_messages.py +++ b/hermes_state_messages.py @@ -458,6 +458,15 @@ class SessionMessagesMixin: (session_id, role, int(offset))) return row[0] if row else None + def latest_conversation_role(self, session_id: str) -> Optional[str]: + """Role of the newest active user/assistant/tool row, or ``None``. ``session_meta`` / + ``system`` rows are transcript bookkeeping stripped before the model sees history, so + they must not hide an open user tail from the failed-turn boundary check.""" + row = self._read_one( + "SELECT role FROM messages WHERE session_id = ? AND active = 1 " + "AND role NOT IN ('session_meta', 'system') ORDER BY id DESC LIMIT 1", (session_id,)) + return row[0] if row else None + def get_message_role(self, session_id: str, row_id: int) -> Optional[str]: """Role of the active message at *row_id* in *session_id*, or ``None``.""" if not session_id: diff --git a/tests/gateway/test_42039_duplicate_user_message.py b/tests/gateway/test_42039_duplicate_user_message.py index 38129eb7fb..3d7d00a2e4 100644 --- a/tests/gateway/test_42039_duplicate_user_message.py +++ b/tests/gateway/test_42039_duplicate_user_message.py @@ -69,6 +69,8 @@ def _bootstrap(monkeypatch, tmp_path): # Mock has_platform_message_id to return False so the dedupe guard # (#47237) in gateway/run.py does not skip the append_to_transcript call. runner.session_store.has_platform_message_id.return_value = False + # The durable tail after the user row landed (gateway write or agent flush) is that user row. + runner.session_store.transcript_tail_role.return_value = "user" runner.session_store.update_session = MagicMock() monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path) @@ -173,11 +175,20 @@ async def test_agent_failed_early_skip_db_when_agent_has_session_db( @pytest.mark.asyncio -async def test_deduped_retry_of_failed_turn_adds_no_second_boundary(monkeypatch, tmp_path): - """A platform retry of the same failed message (dedupe skips the user row) must not stack a - second boundary behind the first — that would be two consecutive assistant rows.""" +@pytest.mark.parametrize( + "tail_role, expected_roles", + [("user", ["assistant"]), ("assistant", [])], + ids=["agent-flushed-user-row-still-closed", "redelivery-of-closed-turn-adds-nothing"], +) +async def test_boundary_keyed_on_durable_tail_when_user_row_is_deduped( + monkeypatch, tmp_path, tail_role, expected_roles +): + """The platform-id dedupe skips the gateway's user write in two production shapes: the agent's + own turn-start flush already persisted THIS turn's row (tail = user → boundary must still land), + and a platform redelivery of an already-closed turn (tail = boundary → nothing may stack).""" runner = _bootstrap(monkeypatch, tmp_path) runner.session_store.has_platform_message_id.return_value = True + runner.session_store.transcript_tail_role.return_value = tail_role runner._run_agent = AsyncMock( return_value={ "failed": True, @@ -191,10 +202,12 @@ async def test_deduped_retry_of_failed_turn_adds_no_second_boundary(monkeypatch, await runner._handle_message_with_agent(_event(), _source(), "agent:main:telegram:group:-1001:12345", 1) - assert [ - call.args[1]["role"] for call in runner.session_store.append_to_transcript.call_args_list + rows = [ + call.args[1] for call in runner.session_store.append_to_transcript.call_args_list if len(call.args) >= 2 and call.args[1].get("role") in {"user", "assistant"} - ] == [] + ] + assert [row["role"] for row in rows] == expected_roles + assert all(row["content"] == runner._FAILED_TURN_NOTICE for row in rows) @pytest.mark.asyncio diff --git a/tests/gateway/test_failure_writer_ownership.py b/tests/gateway/test_failure_writer_ownership.py index d93aac7b5b..51ce576323 100644 --- a/tests/gateway/test_failure_writer_ownership.py +++ b/tests/gateway/test_failure_writer_ownership.py @@ -93,23 +93,69 @@ def test_failure_owner_follows_only_live_lineage_markers(tmp_path): store._transcript_reroutes.clear() assert store.has_input_owner(sid, owner) is owned, location before = db.message_count() + open_tail = store.transcript_tail_role(sid) == "user" reply = await runner._hmwa_agent_error_reply( RuntimeError("controlled post-compaction failure"), MessageEvent(text="same", source=source, message_id=pid), source, entry, entry.session_key, prepared, ) - # Exception fallback adds the missing user only when unowned, then always - # closes the failed turn with a durable assistant safety boundary. - assert db.message_count() == before + (not owned) + 1, location + # Exception fallback adds the missing user only when unowned, and a boundary iff that + # leaves an open user tail on the live route — never anything else. + assert db.message_count() == before + (not owned) + ((not owned) or open_tail), location assert store.has_input_owner(sid, owner), location - live_messages = db.get_messages(child) - assert live_messages[-1]["role"] == "assistant" - assert live_messages[-1]["content"] == runner._PARTIAL_FAILED_TURN_NOTICE assert runner._PARTIAL_FAILED_TURN_NOTICE in reply + live_messages = db.get_messages(child) + assert not live_messages or live_messages[-1]["role"] != "user", location + assert sum(m["role"] == "assistant" for m in live_messages) <= 1, location if not owned: + assert live_messages[-1]["content"] == runner._PARTIAL_FAILED_TURN_NOTICE persisted_user = live_messages[-2] assert persisted_user["content"] == prepared.persist_user_message assert persisted_user["display_metadata"]["gateway_input_owner"] == owner + # A repeated delivery of the same closed turn adds no second boundary. + closed = db.message_count() + await runner._hmwa_agent_error_reply( + RuntimeError("redelivered"), MessageEvent(text="same", source=source, message_id=pid), + source, entry, entry.session_key, prepared, + ) + assert db.message_count() == closed, location + db.close() + + asyncio.run(check()) + + +def test_context_overflow_exception_persists_nothing(tmp_path): + """Exception-path overflow (400/500 on a long session) must not write the user row or a + boundary — the same no-grow rule as the persist path (#1630).""" + import asyncio + from gateway.config import GatewayConfig, Platform + from gateway.platforms.event import MessageEvent + from gateway.run import GatewayRunner + from gateway.session import SessionSource, SessionStore + + async def check(): + store = SessionStore(tmp_path / "sessions", GatewayConfig()) + runner = object.__new__(GatewayRunner) + runner.session_store = store + + async def stop_typing(event, source): + return None + + runner._hmwa_stop_typing_for_turn = stop_typing + source = SessionSource(platform=Platform.TELEGRAM, chat_id="overflow", user_id="u") + entry = store.get_or_create_session(source) + db = store._db_for_session_id(entry.session_id) + prepared = runner._PreparedTurn( + [{"role": "user", "content": "x"}] * 51, "", "x", "x", None, None, entry.session_id, "owner-overflow", + ) + err = RuntimeError("payload too large") + err.status_code = 400 + before = db.message_count() + reply = await runner._hmwa_agent_error_reply( + err, MessageEvent(text="x", source=source, message_id="m-overflow"), source, entry, entry.session_key, prepared, + ) + assert db.message_count() == before + assert reply.startswith("⚠️ Session too large for the model's context window.") db.close() asyncio.run(check()) @@ -142,3 +188,47 @@ def test_context_overflow_error_reply_carries_no_partial_effect_notice(): assert reply.startswith("⚠️ Session too large for the model's context window.") assert reply.endswith("or /reset to start fresh.") assert runner._PARTIAL_FAILED_TURN_NOTICE not in reply + + +def test_fresh_session_agent_flushed_failed_turn_is_closed(tmp_path): + """First turn of a new session, agent-persisted runtime: the agent's turn-start flush wrote the + user row (platform id stamped) before the gateway appends ``session_meta``. The boundary must + still land — ``session_meta`` is stripped before the model sees history, so an open user row + behind it is exactly the #107070 replay shape.""" + import asyncio + from gateway.config import GatewayConfig, Platform + from gateway.platforms.event import MessageEvent + from gateway.run import GatewayRunner + from gateway.session import SessionSource, SessionStore + + async def check(): + store = SessionStore(tmp_path / "sessions", GatewayConfig()) + runner = object.__new__(GatewayRunner) + runner.session_store = store + runner._session_db = object() # agent_persisted defaults True + + async def noop(*args, **kwargs): + return None + + runner._refresh_agent_cache_message_count = noop + source = SessionSource(platform=Platform.TELEGRAM, chat_id="fresh", user_id="u") + entry = store.get_or_create_session(source) + sid = entry.session_id + db = store._db_for_session_id(sid) + db.append_message(sid, "user", "reset the password", platform_message_id="m-1") + failed = {"failed": True, "final_response": "429", "error": "429", "messages": [], + "history_offset": 0, "last_prompt_tokens": 0} + for _ in range(2): # second pass = platform redelivery of the same failed message + await runner._hmwa_persist_turn_transcript( + event=MessageEvent(text="reset the password", source=source, message_id="m-1"), + source=source, session_entry=entry, session_key=entry.session_key, agent_result=failed, + agent_messages=[], prepared=runner._PreparedTurn([], "", "reset the password", None, None, None, sid, "o"), + response="x", agent_failed_early=True, hidden_reasoning_incomplete=False, + is_context_overflow_failure=False, + ) + roles = [m["role"] for m in db.get_messages(sid) if m["role"] != "session_meta"] + assert roles == ["user", "assistant"] + assert store.transcript_tail_role(sid) == "assistant" + db.close() + + asyncio.run(check()) diff --git a/tests/gateway/test_incomplete_gateway_turns.py b/tests/gateway/test_incomplete_gateway_turns.py index 51783948a0..311de51421 100644 --- a/tests/gateway/test_incomplete_gateway_turns.py +++ b/tests/gateway/test_incomplete_gateway_turns.py @@ -98,6 +98,7 @@ def _make_runner(adapter: CaptureSlackAdapter) -> gateway_run.GatewayRunner: # (#47237). A bare MagicMock returns a truthy mock, which would wrongly # mark the user turn as a duplicate and skip persisting it. runner.session_store.has_platform_message_id = MagicMock(return_value=False) + runner.session_store.transcript_tail_role = MagicMock(return_value="user") runner._running_agents = {} runner._pending_messages = {} runner._pending_approvals = {}