fix(gateway): a busy burst's commit task ends when its row is durable

On Signal and other serial readers: an idle video A with a slow capture,
then queue-mode text B, then /restart. B's burst queued behind A's route
reservation, and _route_turn resolved its acceptance future so the reader
could go on (MSG-26). That also made dispatch_shared_busy return, so the
burst's commit task, the thing /restart's barrier waits on
(_shared_burst_commits), finished before B had a row. /restart closed
admission, and releasing the capture refused both A and B with
runtime_draining; B had no durable row to recover.

"The reader may read on" and "the input is durably accepted" are now
separate signals: admit_message resolves a busy_commit future once
admit_native returns, and the burst's commit task calls
dispatch_shared_busy(durable=True), which waits for it (or for the input
to be refused, fenced or answered). The serial reader still returns as
soon as the input is held (_accepted), so MSG-26 is unchanged; the chat's
next burst, command barriers and /restart now wait for the commit, as
main did.

Test: test_restart_waits_for_a_held_text_queued_behind_a_slow_capture
(red: no rows for V or B after /restart; green: both committed and run).

Review finding by @dokterdok on #140973
Refs #134889 (MSG-23)
Refs #134889 (MSG-26)
Refs #140719

gateway/session_ingress.py

@@ -17,6 +17,10 @@ executing_admission = ContextVar('executing_admission', default=False)
 viewer_turn: ContextVar = ContextVar('viewer_turn', default=None)
 # Set by a busy-path caller that must be released once its input is durably queued.
 busy_acceptance = ContextVar('busy_acceptance', default=None)
+# Set beside it by a caller that must also know when the row is durably committed (a busy burst's
+# commit task, which command barriers and /restart wait on): ``busy_acceptance`` may already resolve
+# once the input's FIFO position is fixed behind an earlier input of its chat still capturing.
+busy_commit = ContextVar('busy_commit', default=None)
 # ``_validate_native`` clauses about the message's own content, safe to name to the sender's chat.
 _CONTENT_CLAUSES = frozenset({
     'text', 'auto_skill', 'channel_prompt', 'channel_context', 'reply_expected',
@@ -30,11 +34,14 @@ QUEUE_FULL_REPLY = ('⏳ Too many messages are already waiting in this conversat
 
 async def admit_message(authority, event):
     from gateway.session_authority import messaging_admission
-    accepted = busy_acceptance.get()
+    accepted, committed = busy_acceptance.get(), busy_commit.get()
     busy_acceptance.set(None)  # the drain task this schedules must not inherit the busy caller's future
+    busy_commit.set(None)
     ingress = messaging_admission.set(True if accepted is None else accepted)
     try:
         receipt = await authority.admit_native(event)
+        if committed is not None and not committed.done():
+            committed.set_result(None)  # the row is durable (any status): a barrier may go on
     except RuntimeStoreError as exc:
         if exc.reason == 'stale_generation' and exc.field == 'route':
             return None  # a later /stop or /new of the chat fenced it, as main cancels a queued row
@@ -392,13 +399,16 @@ async def deliver_response(adapter, event, session_key, response):
                                        record_delivery=results.append)
 
 
-async def dispatch_shared_busy(adapter, event, session_key):
+async def dispatch_shared_busy(adapter, event, session_key, *, durable=False):
     """Serial receive loops (IRC, Signal) await this before reading their next frame, so return
     once the follow-up is committed to the FIFO, or queued behind an earlier input of its chat still
     capturing (its position is then fixed; acceptance still waits for the commit); a tracked
-    observer delivers its one reply."""
+    observer delivers its one reply. *durable*: return only once the row is committed (or the input
+    was refused, fenced or answered), never on the earlier position release: a busy burst's commit
+    task, which a command barrier (``/restart`` closing admission) waits on."""
     delivery_event = replace(event, source=replace(event.source))
     accepted = asyncio.get_running_loop().create_future()
+    committed = asyncio.get_running_loop().create_future() if durable else None
 
     async def observe():
         # Like the first message's lifecycle, this sender adopts the turn's crash marker and
@@ -421,15 +431,18 @@ async def dispatch_shared_busy(adapter, event, session_key):
             await adapter._release_turn_marker(delivery_event)
             await adapter._release_turn_marker(event)
 
-    token = busy_acceptance.set(accepted)
+    token, commit_token = busy_acceptance.set(accepted), busy_commit.set(committed)
     try:
         task = asyncio.create_task(observe())
     finally:
         busy_acceptance.reset(token)
+        busy_commit.reset(commit_token)
     adapter._background_tasks.add(task)
     task.add_done_callback(adapter._background_tasks.discard)
     try:
         await asyncio.wait({accepted, task}, return_when=asyncio.FIRST_COMPLETED)
+        if committed is not None and not task.done():
+            await asyncio.wait({committed, task}, return_when=asyncio.FIRST_COMPLETED)
     except asyncio.CancelledError:
         if not accepted.done():
             task.cancel()  # cancelled before commit: the input was never ACKed

gateway/session_ingress_burst.py

@@ -146,7 +146,9 @@ async def _commit(adapter, session_key, burst, previous):
     if previous is not None:
         await asyncio.wait({previous})
     try:
-        await dispatch_shared_busy(adapter, burst.head, session_key)
+        # Durable: this task is what the chat's next burst, a command barrier and /restart wait on, so
+        # it ends once the row is committed, not when its FIFO position is fixed behind a capture.
+        await dispatch_shared_busy(adapter, burst.head, session_key, durable=True)
     except Exception:
         if not _serial(adapter):
             raise  # every absorbed caller sees it (``_settle``)

tests/gateway/test_busy_burst_serial_reader.py

@@ -174,3 +174,51 @@ async def test_next_burst_waits_while_the_chats_last_one_commits(monkeypatch, tm
         release.set()
     await asyncio.wait_for(second, 10)
     assert await owner.settle(3) == [('', 1), ('', 1)]
+
+
+@pytest.mark.asyncio
+async def test_restart_waits_for_a_held_text_queued_behind_a_slow_capture(monkeypatch, tmp_path):
+    """Idle video V (slow capture), then queue-mode text B (held as a burst, then queued behind V's
+    route reservation, which lets the serial reader go on), then an authorized /restart. The command
+    barrier must wait until B is durably committed, not only until its FIFO position is fixed: admission
+    closes after it, and an input with no row then is lost for good."""
+    adapter, source = _signal(monkeypatch)
+    entered, release = _hold_capture(monkeypatch)
+    runner = GatewayRunner(GatewayConfig())
+
+    async def send(chat_id, content, reply_to=None, metadata=None):
+        return SendResult(success=True, message_id='sent')
+    monkeypatch.setattr(adapter, 'send', send)
+    monkeypatch.setattr(adapter, 'send_typing', lambda *a, **k: asyncio.sleep(0))
+    adapter.gateway_runner = runner
+    runner.adapters = {adapter.platform: adapter}
+    authority = await initialize_session_authority(runner, profile_id='default', instance_id='owner')
+    runner._wire_adapter_handlers(adapter, busy_text_mode='queue')
+    monkeypatch.setattr(runner, 'request_restart', lambda **kw: setattr(runner, '_draining', True))
+
+    async def turn(event, source, quick_key, run_generation):
+        execution_result.get()['result'] = {'final_response': 'ok', 'messages': []}
+        return 'ok'
+    monkeypatch.setattr(runner, '_handle_message_with_agent', turn)
+    clip = tmp_path / 'clip.mp4'
+    clip.write_bytes(b'video-bytes' * 1000)
+    video = MessageEvent(text='', message_type=MessageType.VIDEO, source=source, message_id='V',
+                         media_urls=[str(clip)], media_types=['video/mp4'])
+    restart = None
+    try:
+        await adapter.handle_message(video)
+        assert await asyncio.to_thread(entered.wait, 5)
+        await adapter.handle_message(MessageEvent(text='B', source=source, message_id='B'))
+        restart = asyncio.create_task(adapter.handle_message(
+            MessageEvent(text='/restart', source=source, message_id='R')))
+        await async_wait_until(lambda: any(entry[1] >= 2 for entry in authority._route_turns.values()),
+                               timeout=5, what='B queued behind the capturing V')
+        await asyncio.wait({restart}, timeout=2.0)  # a negative window: the restart must still wait here
+    finally:
+        release.set()
+    await asyncio.wait_for(restart, 15)
+    assert sorted(row['request_id'] for row in _rows(authority)) == ['B', 'V'], _rows(authority)
+    runner._draining = False
+    await async_wait_until(lambda: all(row['status'] == 'terminal' for row in _rows(authority)), timeout=15,
+                           what='both turns to run')
+    await asyncio.to_thread(authority.db.close)