fix(gateway): a settled native reply keeps its crash marker until the delivery ledger owns it

On main the in-process turn hands its durable active-turn marker to the
adapter's delivery lifecycle, which clears it only after the final reply
is recorded in the delivery ledger (base.py send_final_ledgered ->
_release_turn_marker). A kill between "reply persisted" and "ledgered"
therefore leaves a marked turn, and the next unclean boot ledgers the
persisted reply (_ledger_crash_left_replies) and sends it once without
re-running inference.

Under the canonical FIFO the marker belongs to the drain's restored
event, not the adapter's event, so _handle_message's finally cleared it
as soon as the handler returned, before the drain committed the terminal
outcome and the waiting adapter sent the reply. An owner death in that
window left a terminal admission whose answer sat in state.db with no
marker, no ledger row and no process-local pending delivery: startup
excludes terminal inputs, so the reply was lost (real Telegram polling +
SIGKILL probe: terminal row, zero sends after restart).

execute_admission now marks the restored event as handed off and, once
the handler returns, moves its marker to the adapter event waiting in
admit_message (adopted and released by its ledgered send, exactly as on
main), keeps it with a recovered no-waiter reply until deliver_settled
returns, and otherwise releases it. A turn that dies before its terminal
commit recovers as unknown; startup now clears the marker of a session
whose admission is unknown before the crash-left-reply pass, so an
uncommitted answer is never delivered. Inference never re-runs: the
recovered reply comes from the transcript through the existing ledger.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#issuecomment-6088835949
(cherry picked from commit 8e47cc1cc1734bedf58f451761caf801fb3195c9)

gateway/run_runtime.py

@@ -126,6 +126,27 @@ async def initialize_gateway_runtime(runner):
     # Publish this boot's verdict (an empty map clears a previous run's parked set).
     _record_parked_profiles(parked_profile_map(runner))
     runner.session_ticket_store = TicketStore(instance_id, registry.profile_ids())
+    release_unknown_turn_markers(runner)
+
+
+def release_unknown_turn_markers(runner):
+    """Before the unclean-start pass reads crash-left turn markers: a marked turn whose canonical
+    admission recovered ``unknown`` (the owner died before its terminal commit) owes no reply, so
+    its marker is cleared rather than ledgered as a delivery the FIFO never committed. A terminal
+    admission keeps its marker: its persisted reply is ledgered and sent once, never re-run."""
+    from gateway.session_authorities import owner_scope
+    store = getattr(runner, 'session_store', None)
+    if store is None:
+        return
+    for authority in _authorities(runner):
+        with owner_scope(authority):
+            unknown = {row['target_session_id'] for row in authority.db._read_all(
+                "SELECT DISTINCT target_session_id FROM session_admissions WHERE status='unknown'")}
+            if not unknown:
+                continue
+            for entry in store.list_sessions():
+                if entry.active_turn_token and authority.logical_owner(entry.session_id) in unknown:
+                    store.clear_turn_active(entry.session_key, entry.active_turn_token)
 
 
 def _publish_served_set(runner):

gateway/session_ingress.py

@@ -26,12 +26,63 @@ async def admit_message(authority, event):
     # Only the delivery waiter is process-local; execution reads the committed snapshot.
     authority.native_waiters.add(receipt.admission_id)
     waiter = authority.waiters.setdefault(receipt.admission_id, asyncio.get_running_loop().create_future())
+    adopters = _marker_adopters(authority)
+    if getattr(event, '_turn_marker_handoff', False):
+        # The adapter lifecycle that sends this reply releases the turn's crash marker only once
+        # the reply is in the delivery ledger; the drain hands its marker to this event.
+        adopters[receipt.admission_id] = event
     if accepted is not None and not accepted.done():
         accepted.set_result(None)
     try:
         return await asyncio.shield(waiter)
     except RuntimeStoreError as exc:
         return pause_notice(authority, receipt.ref, exc.reason)
+    finally:
+        if adopters.get(receipt.admission_id) is event:
+            del adopters[receipt.admission_id]
+
+
+def _marker_adopters(authority):
+    adopters = getattr(authority, 'marker_adopters', None)
+    if adopters is None:
+        adopters = authority.marker_adopters = {}
+    return adopters
+
+
+_MARKER_FIELDS = ('_gateway_active_turn_session_key', '_gateway_active_turn_token')
+
+
+async def _hand_over_turn_marker(authority, admission_id, event):
+    """The executed turn's durable active-turn marker outlives the handler, as the in-process
+    turn's does on main, and is released only once its reply is in the delivery ledger (or nothing
+    is owed). A kill between the terminal commit and that ledger write then leaves a marked turn
+    whose persisted reply the next unclean boot ledgers and sends once, without new inference,
+    instead of a settled answer nobody owes. The waiting adapter lifecycle adopts it; a recovered
+    no-waiter reply keeps it until ``deliver_settled``; any other turn releases it here."""
+    adopter = _marker_adopters(authority).get(admission_id)
+    if adopter is not None:
+        for name in _MARKER_FIELDS:
+            if hasattr(event, name):
+                setattr(adopter, name, getattr(event, name))
+                delattr(event, name)
+        return
+    if admission_id in authority.pending_deliveries:
+        # Released by ``deliver_settled``, also when settlement failed and dropped the delivery.
+        _held_markers(authority)[admission_id] = event
+        return
+    await _release_turn_marker(authority, event)
+
+
+def _held_markers(authority):
+    held = getattr(authority, 'held_turn_markers', None)
+    if held is None:
+        held = authority.held_turn_markers = {}
+    return held
+
+
+async def _release_turn_marker(authority, event):
+    if getattr(event, '_gateway_active_turn_token', None):
+        await authority.runner._clear_durable_active_turn(event)
 
 
 def pause_notice(authority, ref, reason):
@@ -107,6 +158,9 @@ async def execute_admission(authority, ref, row):
     captured = {}
     result_token = execution_result.set(captured)
     token = executing_admission.set(True)
+    # This drain, not the handler's unwind, releases the turn's crash marker (see below).
+    event._turn_marker_handoff = True
+    handed = False
     try:
         with scope:
             if is_api:
@@ -144,8 +198,12 @@ async def execute_admission(authority, ref, row):
                     # leaves an externally answered admission started -> unknown.
                     authority.pending_deliveries[row['admission_id']] = (
                         adapter, event, live.route, response, home)
+            handed = True
+            await _hand_over_turn_marker(authority, row['admission_id'], event)
             return response
     finally:
+        if not handed:
+            await _release_turn_marker(authority, event)
         executing_admission.reset(token)
         execution_result.reset(result_token)
         api_execution.reset(api_token)
@@ -168,8 +226,12 @@ def _execution_scope(authority, home):
 async def deliver_settled(authority, admission_id):
     """Send a recovered no-waiter turn's reply after its terminal outcome committed. A send
     failure cannot rewrite that outcome; it is logged, and inference never runs again."""
+    held = _held_markers(authority).pop(admission_id, None)
     delivery = authority.pending_deliveries.get(admission_id)
     if delivery is None:
+        if held is not None:
+            # Fenced unknown: no answer is owed, so no later boot may deliver one either.
+            await _release_turn_marker(authority, held)
         return
     adapter, event, session_key, response, home = delivery
     try:
@@ -180,6 +242,8 @@ async def deliver_settled(authority, admission_id):
     finally:
         # Held until the send returns: an empty map means every settled reply was handed over.
         authority.pending_deliveries.pop(admission_id, None)
+        # The ledgered send already released the marker; an empty or failed send owes no more.
+        await _release_turn_marker(authority, event)
 
 
 async def deliver_response(adapter, event, session_key, response):

tests/gateway/fixtures/native_reply_crash_daemon.py

@@ -0,0 +1,31 @@
+"""Test-only crash barriers around a native reply's terminal commit; every recovery boot is ordinary.
+
+``KILL_BEFORE_LEDGER``: the owner dies after the terminal commit, before the adapter ledgers the reply.
+``KILL_BEFORE_COMMIT``: the owner dies after the reply is in the transcript, before the terminal commit.
+"""
+import os
+import runpy
+import signal
+
+from gateway import session_results
+from gateway.platforms.base import BasePlatformAdapter
+
+record = BasePlatformAdapter._record_delivery_obligation
+finish = session_results.finish_result
+
+
+async def die_before_ledger(self, event, session_key, text_content, *args, **kwargs):
+    if 'KILL_BEFORE_LEDGER' in text_content:
+        os.kill(os.getpid(), signal.SIGKILL)
+    return await record(self, event, session_key, text_content, *args, **kwargs)
+
+
+def die_before_commit(db, *, row, **kwargs):
+    if row['payload'].get('text') == 'KILL_BEFORE_COMMIT':
+        os.kill(os.getpid(), signal.SIGKILL)
+    return finish(db, row=row, **kwargs)
+
+
+BasePlatformAdapter._record_delivery_obligation = die_before_ledger
+session_results.finish_result = die_before_commit
+runpy.run_module('gateway.run', run_name='__main__')

tests/gateway/test_native_reply_crash_recovery.py

@@ -0,0 +1,105 @@
+"""A settled native reply survives an owner death before it reached the delivery ledger; an
+uncommitted one is never delivered. Real Telegram polling ingress, SIGKILL, ordinary restart."""
+from contextlib import closing, contextmanager
+from http.server import ThreadingHTTPServer
+import importlib.machinery
+import json
+from pathlib import Path
+import queue
+import sqlite3
+import threading
+import time
+from types import SimpleNamespace
+
+import pytest
+
+from tests.gateway.fixtures.local_recovery_probe import Model, child_env, daemon
+from tests.gateway.test_native_telegram_startup_recovery import BotAPI, wait_for
+
+pytestmark = [
+    pytest.mark.skipif(importlib.machinery.PathFinder.find_spec("telegram") is None,
+                       reason="python-telegram-bot not installed (on-demand extra)"),
+    pytest.mark.platforms("linux"),
+]
+
+
+@contextmanager
+def _owner(tmp_path):
+    root = Path(__file__).resolve().parents[2]
+    home, user = tmp_path / 'state', tmp_path / 'user'
+    home.mkdir(mode=0o700)
+    user.mkdir()
+    model = ThreadingHTTPServer(('127.0.0.1', 0), Model)
+    model.requests, model.blocked, model.release = [], threading.Event(), threading.Event()
+    bot = ThreadingHTTPServer(('127.0.0.1', 0), BotAPI)
+    bot.calls, bot.sent, bot.updates = [], [], queue.Queue()
+    for peer in (model, bot):
+        threading.Thread(target=peer.serve_forever, daemon=True).start()
+    base = f'http://127.0.0.1:{model.server_port}/v1'
+    cfg = {'model': {'provider': 'custom', 'default': 'local-wire-stub', 'base_url': base},
+           'gateway': {'multiplex_profiles': False},
+           'platforms': {'telegram': {'enabled': True, 'extra': {
+               'base_url': f'http://127.0.0.1:{bot.server_port}/bot', 'dm_policy': 'allowlist'}}},
+           'auxiliary': {'title_generation': {'enabled': False}},
+           'platform_toolsets': {'telegram': []}, 'terminal': {'cwd': str(home)}}
+    (home / 'config.yaml').write_text(json.dumps(cfg))
+    env = child_env()
+    env.update(HOME=str(user), USERPROFILE=str(user), HERMES_HOME=str(home), PYTHONPATH=str(root),
+               OPENAI_API_KEY='loopback-only', OPENAI_BASE_URL=base, PYTHONUNBUFFERED='1',
+               TELEGRAM_BOT_TOKEN='987654321:owned-loopback-fixture', TELEGRAM_ALLOWED_USERS='202')
+
+    def status():
+        with closing(sqlite3.connect(f'file:{home / "state.db"}?mode=ro', uri=True)) as db:
+            return {json.loads(payload)['text']: state for state, payload in db.execute(
+                'SELECT status,payload_json FROM session_admissions')}
+
+    def replies(marker):
+        # MarkdownV2 escapes ``_``: compare the unescaped text.
+        return [body for body in bot.sent if marker in str(body.get('text', '')).replace('\\', '')]
+
+    def diagnostic():
+        return repr((status(), bot.sent)) + '\n' + '\n'.join(p.read_text() for p in (home / 'logs').glob('*.log'))
+
+    def send(update_id, text):
+        bot.updates.put({'update_id': update_id, 'message': {
+            'message_id': update_id, 'date': int(time.time()),
+            'chat': {'id': 202, 'type': 'private', 'first_name': 'Owned'},
+            'from': {'id': 202, 'is_bot': False, 'first_name': 'Owned'}, 'text': text}})
+
+    def crash_then_restart(marker):
+        """Warm the route, send *marker*, let the barrier SIGKILL the owner, restart it plainly.
+        Returns (status at the crash, deliveries of its answer, model calls for it)."""
+        with daemon(root, home, env, barrier=True, fixture='native_reply_crash_daemon.py') as (proc, _):
+            send(1, 'WARM_HISTORY')
+            wait_for(lambda: replies('RECOVERY_ACK_WARM_HISTORY'), diagnostic)
+            send(2, marker)
+            assert proc.wait(timeout=60) == -9, diagnostic()
+        assert not replies(marker)
+        crashed = status()[marker]
+        with daemon(root, home, env, barrier=False):
+            wait_for(lambda: status().get(marker) != 'started', diagnostic)
+            time.sleep(3)  # a (second) delivery would arrive inside this window
+        prompts = [next(m['content'] for m in reversed(r['messages']) if m['role'] == 'user')
+                   for r in model.requests]
+        return crashed, len(replies('RECOVERY_ACK_' + marker)), sum(marker in p for p in prompts)
+
+    try:
+        yield SimpleNamespace(crash_then_restart=crash_then_restart, status=status, diagnostic=diagnostic)
+    finally:
+        for peer in (model, bot):
+            peer.shutdown()
+            peer.server_close()
+
+
+def test_terminal_native_reply_is_delivered_once_after_owner_death_without_new_inference(tmp_path):
+    with _owner(tmp_path) as probe:
+        crashed, delivered, inferences = probe.crash_then_restart('KILL_BEFORE_LEDGER')
+        assert crashed == 'terminal', probe.diagnostic()
+        assert (delivered, inferences) == (1, 1), probe.diagnostic()
+
+
+def test_reply_of_a_turn_lost_before_its_terminal_commit_is_never_delivered(tmp_path):
+    with _owner(tmp_path) as probe:
+        crashed, delivered, inferences = probe.crash_then_restart('KILL_BEFORE_COMMIT')
+        assert crashed == 'started' and probe.status()['KILL_BEFORE_COMMIT'] == 'unknown', probe.diagnostic()
+        assert (delivered, inferences) == (0, 1), probe.diagnostic()