perf(gateway): publish_execution fences on the executing generation, not the ledger

Mechanism: publish_execution (every stream delta, tool and status event of an
in-process turn, and every managed-worker delta frame on the owner loop) ran
check_approval_generation under event_stream.lock: db.get_session plus a
JSON decode of every pending admission, per token. Loop-side attach, receipt
and _publish_pending waited on that SQLite I/O.

Fix: the authority records the executing generation with the claim's stamp
(_stamp_claim, under the same lock) and clears it the moment execution
returns, before settlement, so a late callback still never follows the
completion. publish_execution compares against it; the RPC paths (respond,
interrupt, register_*) keep the exact ledger read.

Test: tests/gateway/test_publish_execution_stamp_fence.py is red on r13-base2
(102 ledger reads for 51 deltas) and pins the stale/late-generation fence.

Review: ahrazzle B1-11

gateway/session_authority.py

@@ -102,6 +102,7 @@ class SessionAuthority:
         self.retiring = False
         self.retired = False
         self._classic_evict_at = {}  # route -> last leaked/reaped-slot eviction check (monotonic)
+        self.executing = {}  # session id -> generation whose execution is in progress (publish fence)
 
     def authorize(self, actor, ref, capability):
         """Every handler calls this first, so a later ``self.sessions[ref.session_id]`` is
@@ -718,9 +719,10 @@ class SessionAuthority:
         """Worker callbacks never outlive their exact running claim."""
         live = self.sessions[session_id]
         with live.event_stream.lock:
-            try:
-                self.check_approval_generation(session_id, generation)
-            except RuntimeStoreError:
+            # The executing generation (set with the claim's stamp, cleared the moment its execution
+            # returns, before settlement): no per-token SQLite read under this lock, and a late
+            # callback (an agent thread outliving a timed-out turn) never follows the completion.
+            if self.executing.get(session_id) != generation:
                 return False
             delivered = live.event_stream.publish(session_id, payload, event_type=event_type)
             from gateway.session_api_turn import publish_api_event
@@ -828,6 +830,7 @@ class SessionAuthority:
                 live.event_stream.execution = {
                     'authority_epoch': self.epoch, 'execution_generation': row['generation'],
                     'admission_id': row['admission_id']}
+                self.executing[ref.session_id] = row['generation']
                 live.event_stream.publish(ref.session_id, {}, event_type='message.start')
                 self._publish_pending(ref)
         return row
@@ -892,8 +895,12 @@ class SessionAuthority:
                 return
             admission_id = row['admission_id']
             try:
-                response = await execute_finite_admission(self, ref, row)
-                outcome = 'completed'
+                try:
+                    response = await execute_finite_admission(self, ref, row)
+                    outcome = 'completed'
+                finally:
+                    with live.event_stream.lock:
+                        self.executing.pop(ref.session_id, None)
             except ManagedExecutionUnknown:
                 with live.event_stream.lock:
                     live.event_stream.execution = {}

tests/gateway/test_managed_worker_tool_events.py

@@ -27,7 +27,7 @@ def test_worker_tool_frames_publish_the_in_process_payload(tmp_path):
         live = authority.sessions['s'] = LiveSession(SimpleNamespace(platform=None, user_id='human'), 'route')
         admit_session_input(db, epoch=epoch, principal_id='human', session_id='s', request_id='r', payload={'text': 'go'})
         row = claim_session_input(db, epoch=epoch, session_id='s')
-        live.event_stream.execution = {'admission_id': row['admission_id']}
+        authority._stamp_claim(SimpleNamespace(session_id='s'), live, row)  # the drain's claim stamp
         frames, observed = [], []
         live.event_stream.observers.add(frames.append)
         authority.api_observers = {row['admission_id']: [{

tests/gateway/test_publish_execution_stamp_fence.py

@@ -0,0 +1,38 @@
+"""publish_execution (per token / tool event, under event_stream.lock) fences on the claim's execution
+stamp, not a per-delta ledger read (get_session + pending-admission decode) (B1-11)."""
+import asyncio
+
+import pytest
+
+from tests.gateway.test_session_authority_cancel_settlement import _authority, _submit
+
+
+@pytest.mark.asyncio
+async def test_publish_execution_reads_no_ledger_per_delta_and_still_fences_stale_generations(tmp_path, monkeypatch):
+    from gateway import session_finite
+
+    db, authority = _authority(tmp_path, monkeypatch)
+    published = {}
+
+    async def execute(authority, ref, row):
+        generation = row['generation']
+        reads = []
+        real_get, real_list = authority.db.get_session, authority._pending_rows
+        authority.db.get_session = lambda *a, **k: reads.append('get_session') or real_get(*a, **k)
+        authority._pending_rows = lambda *a, **k: reads.append('pending') or real_list(*a, **k)
+        try:
+            published['live'] = [authority.publish_execution('s', generation, 'message.delta', {'text': str(i)})
+                                 for i in range(50)]
+            published['stale'] = authority.publish_execution('s', generation - 1, 'message.delta', {'text': 'x'})
+        finally:
+            authority.db.get_session, authority._pending_rows = real_get, real_list
+        published['reads'] = reads
+        return 'done'
+    monkeypatch.setattr(session_finite, 'execute_finite_admission', execute)
+    with db:
+        await _submit(authority, 'turn')
+        await asyncio.wait_for(authority.sessions['s'].task, 5)
+        generation = db.get_session('s')['runtime_generation']
+        after = authority.publish_execution('s', generation, 'message.delta', {'text': 'late'})
+    assert published['reads'] == [], f"{len(published['reads'])} ledger reads for 51 deltas under the stream lock"
+    assert published['stale'] is False and after is False, 'a callback outlived its exact running claim'