fix(gateway): post-turn goal/loop state and control snapshots are read off the event loop

Every gateway turn ended with state.db meta reads on the loop: GoalManager
and LoopManager construction in _post_turn_manager (load_goal/load_loop),
then the control-card publish (control_snapshot: goal, loop, heartbeat meta
plus the compression tip). Under load these were a recurring stall.

- _post_turn_manager builds the manager via asyncio.to_thread (the copied
  context carries the turn's profile scope).
- session_control_actions.publish_async reads the snapshot in a worker
  under owner_scope(authority) and publishes on the loop. Publishes keep the
  order they were scheduled in: each takes a ticket on the loop and a
  snapshot whose read finished after a later-scheduled publish already
  painted is dropped. The sync publish takes tickets too, so the two mix.
- the post-turn hook and the loop-wakeup publish use publish_async. The
  per-tool republish stays synchronous: test_session_viewer_automation pins
  it completing within one loop step.

gateway/run_goals.py

@@ -296,7 +296,8 @@ class GatewayGoalsMixin:
         if not sid:
             return None
         await self._warm_goals_session_db(label)
-        return factory(sid)
+        # The manager loads its state from state.db on construction: off the loop, in this scope.
+        return await asyncio.to_thread(factory, sid)
 
     async def _post_turn_goal_continuation(
         self, *, session_entry: Any, source: Any, final_response: str,
@@ -383,16 +384,16 @@ class GatewayGoalsMixin:
                 await hook(session_entry=session_entry, source=source, final_response="" if failed else final_text)
             except Exception as exc:
                 logger.debug("%s hook failed: %s", label, exc)
-        self._publish_local_control(source, only_if_present=True)
+        await self._publish_local_control(source, only_if_present=True)
 
-    def _publish_local_control(self, source: Any, *, only_if_present: bool = False) -> None:
+    async def _publish_local_control(self, source: Any, *, only_if_present: bool = False) -> None:
         """A local (Desktop/Ink) session's goal or loop state just moved here: repaint its viewers' cards,
         whose loop countdown runs off the snapshot's ``next_due_at``."""
         if getattr(source, "platform", None) != Platform.LOCAL:
             return
         from gateway.session_authorities import active_authority
-        from gateway.session_control_actions import publish
-        publish(active_authority(self), source.chat_id, only_if_present=only_if_present)
+        from gateway.session_control_actions import publish_async
+        await publish_async(active_authority(self), source.chat_id, only_if_present=only_if_present)
 
     @staticmethod
     def _turn_failed() -> bool:
@@ -525,7 +526,7 @@ class GatewayGoalsMixin:
             with suppress(Exception):
                 mgr.abandon_tick()
         # Viewers swap the countdown for "running now" (or re-anchor it after a slash tick settled).
-        self._publish_local_control(source)
+        await self._publish_local_control(source)
 
     async def _loop_wakeup_watcher(self, interval: float = 15.0) -> None:
         """Fire due /loop wakeups for idle gateway sessions: a coarse ticker scans persisted loops

gateway/session_authority.py

@@ -91,6 +91,10 @@ class LiveSession:
     # Revision of the last ``session.control.update`` (gateway/session_control_actions.publish): a tool
     # completion republishes only when a tool actually moved the goal/loop/heartbeat state.
     control_revision: str = ""
+    # Tickets ordering control publishes whose snapshot is read off the loop: the last one scheduled
+    # wins, so a slow read cannot land over a newer state (gateway/session_control_actions.publish).
+    control_seq: int = 0
+    control_published_seq: int = 0
 
     def __post_init__(self):
         self.controls = PendingControls(self.event_stream)

gateway/session_control_actions.py

@@ -5,6 +5,7 @@ answered -32601. Each action is the typed slash command the button stands for, r
 ``execute_command`` (the gateway's own /goal /loop /heartbeat /subgoal handlers, with their
 capability, local-source, admission and busy rules), answered with the owner's control snapshot.
 """
+import asyncio
 import logging
 
 from hermes_state_runtime import RuntimeStoreError
@@ -84,15 +85,55 @@ def publish(authority, session_id, *, only_if_present=False, only_if_changed=Fal
     live = authority.sessions.get(session_id) if authority is not None else None
     if live is None:
         return
-    from gateway.session_ancillary import control_snapshot
-    from gateway.session_contract import SessionRef
+    live.control_seq += 1
+    ticket = live.control_seq
     try:
-        snapshot = control_snapshot(authority, SessionRef(authority.profile_id, session_id))
+        snapshot = _read_control_snapshot(authority, session_id)
     except RuntimeStoreError:  # unreadable row: the change itself already landed; the next publish repaints
         logger.debug('session.control.update skipped for %s', session_id, exc_info=True)
         return
+    _publish_snapshot(authority, live, session_id, snapshot, ticket,
+                      only_if_present=only_if_present, only_if_changed=only_if_changed)
+
+
+async def publish_async(authority, session_id, *, only_if_present=False, only_if_changed=False):
+    """``publish`` with the snapshot read in a worker thread: three state.db meta reads per publish
+    (after every tool and every turn) were a measured loop stall under load. Publishes keep the order
+    they were scheduled in: each takes a ticket on the loop, and a snapshot whose read finished after a
+    later-scheduled one already published is dropped (that one read state at least as new)."""
+    live = authority.sessions.get(session_id) if authority is not None else None
+    if live is None:
+        return
+    live.control_seq += 1
+    ticket = live.control_seq
+    try:
+        snapshot = await asyncio.to_thread(_read_control_snapshot, authority, session_id, scoped=True)
+    except RuntimeStoreError:
+        logger.debug('session.control.update skipped for %s', session_id, exc_info=True)
+        return
+    if authority.sessions.get(session_id) is not live:  # closed or re-mounted while the read ran
+        return
+    _publish_snapshot(authority, live, session_id, snapshot, ticket,
+                      only_if_present=only_if_present, only_if_changed=only_if_changed)
+
+
+def _read_control_snapshot(authority, session_id, *, scoped=False):
+    from gateway.session_ancillary import control_snapshot
+    from gateway.session_contract import SessionRef
+    ref = SessionRef(authority.profile_id, session_id)
+    if not scoped:
+        return control_snapshot(authority, ref)
+    from gateway.session_authorities import owner_scope
+    with owner_scope(authority):  # a worker thread inherits no profile scope it was not given
+        return control_snapshot(authority, ref)
+
+
+def _publish_snapshot(authority, live, session_id, snapshot, ticket, *, only_if_present, only_if_changed):
+    if live.control_published_seq > ticket:  # a later-scheduled publish already painted newer state
+        return
     if (only_if_present and not snapshot['revision']) or (only_if_changed and snapshot['revision'] == live.control_revision):
         return
+    live.control_published_seq = ticket
     live.control_revision = snapshot['revision']
     live.event_stream.publish(session_id, {'control': snapshot}, event_type='session.control.update')
 

tests/gateway/test_control_publish_off_loop.py

@@ -0,0 +1,48 @@
+"""Control-card publishes read their snapshot off the loop and still land in the order they were scheduled.
+
+The snapshot is three state.db meta reads; on the loop, the post-turn publish was a measured stall
+with many turns running. Deferring the read must not let a slow, older read repaint the cards over
+a newer state that a later publish already painted.
+"""
+
+import asyncio
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from gateway import session_control_actions as actions
+
+
+def _session(published):
+    live = SimpleNamespace(control_seq=0, control_published_seq=0, control_revision='',
+                           event_stream=SimpleNamespace(
+                               publish=lambda sid, payload, event_type: published.append(payload['control']['revision'])))
+    return SimpleNamespace(sessions={'s1': live}, profile_id='fixture'), live
+
+
+@pytest.mark.asyncio
+async def test_a_slow_snapshot_read_never_paints_over_a_newer_publish(monkeypatch):
+    published = []
+    authority, live = _session(published)
+    old_read_entered, release_old_read = threading.Event(), threading.Event()
+    loop_thread = threading.get_ident()
+    read_threads = []
+
+    def read(_authority, _sid, *, scoped=False):
+        read_threads.append(threading.get_ident())
+        if not old_read_entered.is_set():
+            old_read_entered.set()
+            assert release_old_read.wait(5)
+            return {'revision': 'old'}
+        return {'revision': 'new'}
+
+    monkeypatch.setattr(actions, '_read_control_snapshot', read)
+    slow = asyncio.ensure_future(actions.publish_async(authority, 's1'))
+    assert await asyncio.to_thread(old_read_entered.wait, 5)
+    await actions.publish_async(authority, 's1')  # scheduled later, read finishes first
+    release_old_read.set()
+    await slow
+    assert published == ['new']
+    assert live.control_revision == 'new'
+    assert loop_thread not in read_threads, 'the snapshot was read on the event loop'