fix(gateway): failed-turn boundary keyed on the durable conversation tail; exception path classifies overflow before writing

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

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

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:

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:

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

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

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 = {}