fix(gateway): a note into an old conversation leaves the row the chat's /new ended closed

Re-review N2a of #144840. A reminder or background result written into a conversation its chat moved
past as a note (no viewer, no turn) reopened the row the chat's /new had ended. The local path already
skips the reopen for notes; the detached branch now does the same, and a result that runs as a turn
(a viewer is open) reopens it as any first turn does.

gateway/session_authority.py

@@ -1097,10 +1097,11 @@ class SessionAuthority:
                 check_api_turn(self, ref, first['payload'])
                 await tracked_write(self, partial(reopen_api_session, self, ref))
             elif first is not None and live.detached:
-                # The chat's /new ended its row; the app's turn makes it live again, as an API row's does.
-                from gateway.session_api import reopen_api_session
-                from gateway.session_runtime_workers import tracked_write
-                await tracked_write(self, partial(reopen_api_session, self, ref))
+                # The chat's /new ended its row; the app's turn makes it live again, as an API row's
+                # does. A note leaves it closed, as a local one's does (session_completion_notes).
+                from gateway.session_completion_notes import noted_completion, reopen_for_turn
+                if not noted_completion(first):
+                    await reopen_for_turn(self, ref)
             if first is not None and self._classic_turn_holds(live.route):
                 # A classic turn (fallback wake, CLI handoff, legacy resume) holds the route's slot:
                 # a claim now would hand this row to its busy path and settle it unrun. The head

gateway/session_completion_notes.py

@@ -36,6 +36,19 @@ def noted_completion(row):
     return descriptor is not None and _producer(descriptor['identity']) in _NOTED_PRODUCERS | _ALWAYS_NOTED
 
 
+async def reopen_for_turn(authority, ref):
+    """A turn (never a note) makes *ref*'s finalized row live again: a local session's receipt tip, or
+    the row of a conversation its chat moved past (gateway/session_detached.py), which the chat's /new
+    ended."""
+    from gateway.session_runtime_workers import tracked_write
+    live = authority.sessions.get(ref.session_id)
+    if live is not None and live.detached:
+        from gateway.session_api import reopen_api_session as reopen
+    else:
+        from gateway.session_local_recovery import reopen_local_session as reopen
+    await tracked_write(authority, partial(reopen, authority, ref))
+
+
 async def note_if_unviewed(authority, ref, row):
     """The claimed *row*'s response when it was written as a note instead of running, else None.
     It is the FIFO head, so the note lands after every input committed before it and before every
@@ -47,8 +60,7 @@ async def note_if_unviewed(authority, ref, row):
     descriptor = row['payload']['local_automation_v1']
     if live is None or (live.subscribers and _producer(descriptor['identity']) not in _ALWAYS_NOTED):
         # The claim left a closed row closed for a note; the turn reopens it as any first turn does.
-        from gateway.session_local_recovery import reopen_local_session
-        await tracked_write(authority, partial(reopen_local_session, authority, ref))
+        await reopen_for_turn(authority, ref)
         return None
     from gateway.session_viewer_close import _physical_target
     metadata = {'completion_note': sorted(set(descriptor.get('identities') or ()) | {descriptor['identity']})}

tests/gateway/test_detached_conversation_results.py

@@ -40,3 +40,18 @@ async def test_a_reminder_the_app_set_on_the_old_conversation_fires_after_a_resu
     assert await asyncio.to_thread(deliver_to_local_chat, home, a, 'fire after restart') is None
     await async_wait_until(lambda: _noted(authority, a, 'fire after restart'), timeout=10, what='the note in A')
     assert authority.sessions[a].detached and sent[chat_before:] == []
+
+
+@pytest.mark.asyncio
+async def test_a_note_into_the_old_conversation_leaves_the_row_the_chats_new_ended_closed(monkeypatch):
+    """N2a: a reminder written into A as a note (no viewer, no turn) keeps the ``/new``'s end stamp, as
+    a note into a closed local session does; only a turn reopens the row."""
+    from gateway.session_cron import deliver_to_local_chat
+    _runner, _adapter, authority, _source, _sent, ran, _gates, _lane, ref_a = await _moved_on(monkeypatch)
+    a = ref_a.session_id
+    ran_before = len(ran)
+    assert await asyncio.to_thread(deliver_to_local_chat, Path(authority.db.db_path).parent, a, 'a reminder') is None
+    await async_wait_until(lambda: _noted(authority, a, 'a reminder'), timeout=10, what='the note in A')
+    await asyncio.wait_for(authority.sessions[a].task, 10)
+    assert ran[ran_before:] == []
+    assert authority.db.get_session(a)['end_reason'] == 'session_reset'