feat(gateway): session.detach end_reason ends the last viewer's idle local session

Main's classic CLI stamped its own row: `end_session(sid, "cli_close")` on quit and
`new_session` / `resumed_other` / `branched` on a switch. On the gateway only ACP rows were
ever ended (the last viewer leaving an idle ACP session), so a CLI quit left every row open.

`session.detach` takes an optional `end_reason` (snake_case, 1-32 chars). When the detaching
viewer was the last subscriber, the owner ends the idle local row with that reason in the same
idle-guarded write ACP uses (`end_idle_local_session`: a queued/started admission or a running
worker keeps it open); a viewer that re-attached meanwhile reopens it. ACP rows keep
`acp_disconnect`; a detach without `end_reason` is unchanged (only a viewer leaving). Generic
name, so the TUI can close with it too. Contracts regenerated.

LEDGER: A-072, DB-WRITE-007, DB-WRITE-011, DB-WRITE-014, SLASH-E2-060 (gateway half).
(cherry picked from commit 8e99979da6ba08027a42ae0b86d095466fd17bc8)

apps/shared/src/gateway-contract.generated.ts

@@ -4586,6 +4586,7 @@ export interface SessionStats {
 export interface SessionDetachParams {
   session_id: string
   subscription_id: string
+  end_reason?: string | null
 }
 export interface SessionDetachResult {
   session_id: string
@@ -6531,7 +6532,7 @@ export interface CanonicalRpcMethods {
   /** Owner identity and the canonical capabilities a client may rely on. */
   'runtime.describe': { params: RuntimeDescribeParams; result: RuntimeDescribeResult }
   'session.create': { params: CanonicalCreateParams; result: CanonicalSnapshot }
-  /** Drop this connection's subscription; a stale subscription id answers detached=false. */
+  /** Drop this connection's subscription; a stale subscription id answers detached=false. end_reason (snake_case): the viewer is closing the conversation; the last viewer leaving an idle local session ends its row with that reason. */
   'session.detach': { params: SessionDetachParams; result: SessionDetachResult }
   'session.events.since': { params: CanonicalEventsParams; result: CanonicalEventsResult }
   /** Frozen launch policy projection of one local session. */

apps/shared/src/gateway-contract.openrpc.json

@@ -35102,6 +35102,18 @@
           "subscription_id": {
             "title": "Subscription Id",
             "type": "string"
+          },
+          "end_reason": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "default": null,
+            "title": "End Reason"
           }
         },
         "required": [
@@ -45758,7 +45770,7 @@
     },
     {
       "name": "session.detach",
-      "summary": "Drop this connection's subscription; a stale subscription id answers detached=false.",
+      "summary": "Drop this connection's subscription; a stale subscription id answers detached=false. end_reason (snake_case): the viewer is closing the conversation; the last viewer leaving an idle local session ends its row with that reason.",
       "params": [
         {
           "name": "params",

gateway/session_acp_lifecycle.py

@@ -32,6 +32,38 @@ def end_idle_acp_session(authority, session_id):
         return False
 
 
+def end_idle_viewer_session(authority, session_id, reason):
+    """``session.detach {end_reason}`` from the last viewer: end *session_id*'s idle local row with the
+    viewer's own reason (main's classic CLI ``end_session(sid, 'cli_close')`` and its switch reasons).
+    ACP rows keep their own reason; a busy FIFO keeps the row open, as for ACP."""
+    live = authority.sessions.get(session_id)
+    if live is not None and live.subscribers:
+        return False
+    try:
+        receipt = local_receipt(authority.db, session_id)
+    except RuntimeStoreError:
+        return False  # not a local session: nothing this verb may end
+    if receipt.get('policy', {}).get('source') == 'acp':
+        return end_idle_acp_session(authority, session_id)
+    try:
+        return bool(end_idle_local_session(authority.db, epoch=authority.epoch, session_id=session_id,
+                                           target_id=receipt['entry']['session_id'], reason=reason))
+    except (RuntimeStoreError, KeyError, TypeError):
+        logging.getLogger(__name__).warning('local session %s not ended', session_id, exc_info=True)
+        return False
+
+
+async def reopen_if_reattached(authority, session_id, ended):
+    """On the owner loop after the close committed: a viewer attached meanwhile, so the session
+    it watches is live again (the stamp would make active work prune/archive eligible)."""
+    live = authority.sessions.get(session_id)
+    if ended and live is not None and live.subscribers:
+        import asyncio
+        target = local_receipt(authority.db, session_id)['entry']['session_id']
+        await asyncio.to_thread(authority.db.reopen_session, target)
+    return ended
+
+
 def end_idle_acp_sessions(authority):
     """Gateway shutdown: every ACP session nobody is viewing, attached this run or not."""
     with authority.db._read_ctx() as conn:

gateway/session_authority.py

@@ -236,7 +236,7 @@ class SessionAuthority:
             if self.events.get(member.transport_id) is transport:
                 del live.subscribers[subscription]
 
-    async def detach(self, actor, subscription_id):
+    async def detach(self, actor, subscription_id, *, end_reason=None):
         for session_id, live in self.sessions.items():
             if subscription_id in live.subscribers:
                 if live.subscribers[subscription_id] != actor:
@@ -246,9 +246,13 @@ class SessionAuthority:
                 if transport is not None:
                     live.event_stream.fanout.detach(transport)
                 if not live.subscribers:
-                    from gateway.session_acp_lifecycle import end_idle_acp_session
+                    from gateway.session_acp_lifecycle import (
+                        end_idle_acp_session, end_idle_viewer_session, reopen_if_reattached)
                     from gateway.session_runtime_workers import tracked_write
-                    await tracked_write(self, partial(end_idle_acp_session, self, session_id))
+                    # A viewer that attached while the close was off the loop reopens it.
+                    close = (partial(end_idle_viewer_session, self, session_id, end_reason) if end_reason
+                             else partial(end_idle_acp_session, self, session_id))
+                    await tracked_write(self, close, then=partial(reopen_if_reattached, self, session_id))
                 return
         raise RuntimeStoreError('not_found')
 

gateway/session_contract.py

@@ -81,7 +81,7 @@ class SessionAuthority(Protocol):
     async def create(self, actor: Principal, request: SessionCreate) -> SessionHandle: ...
     async def resolve(self, actor: Principal, ref: SessionRef) -> SessionHandle: ...
     async def attach(self, actor: Principal, ref: SessionRef) -> SubscriptionSnapshot: ...
-    async def detach(self, actor: Principal, subscription_id: str) -> None: ...
+    async def detach(self, actor: Principal, subscription_id: str, *, end_reason: str | None = None) -> None: ...
     async def submit(self, actor: Principal, request: Submission) -> AdmissionReceipt: ...
     async def receipt(self, actor: Principal, ref: SessionRef, admission_id: str) -> AdmissionReceipt: ...
     async def cancel_queued(self, actor: Principal, ref: SessionRef, admission_id: str) -> AdmissionReceipt: ...

gateway/session_controls.py

@@ -1,5 +1,6 @@
 """Authenticated WS projection of the gateway authority; no TUI execution fallback."""
 import asyncio
+import re
 from dataclasses import asdict
 from functools import partial
 from pathlib import Path
@@ -11,6 +12,8 @@ from gateway.session_contract import CANONICAL_GATEWAY_PROTOCOL, Principal, Sess
 from hermes_state_runtime import RuntimeStoreError
 
 
+# ``session.detach end_reason``: a short snake_case label, stored verbatim as ``sessions.end_reason``.
+_END_REASON = re.compile(r'[a-z][a-z_]{0,31}')
 # Verbs whose param sets are closed and never carried a ``profile`` key.
 _PROFILE_IMPLICIT = frozenset({'session.create', 'session.list', 'session.info', 'session.resume', 'session.detach',
                                'prompt.submit', 'prompt.receipt', 'prompt.cancel', 'prompt.resolve_unknown',
@@ -433,18 +436,24 @@ class AuthorityConnection:
                 'prompts': list(snapshot.prompts), 'info': info}
 
     async def detach(self, ref, params):
-        if set(params) != {'session_id', 'subscription_id'}:
+        if set(params) - {'end_reason'} != {'session_id', 'subscription_id'}:
             raise RuntimeStoreError('invalid_params')
         session_id = params['session_id']
         subscription_id = params['subscription_id']
+        # ``end_reason``: this viewer is CLOSING the conversation (main's classic CLI stamped its row
+        # ``cli_close`` / ``new_session`` / ``resumed_other`` / ``branched`` itself); the owner ends the
+        # idle local row once no viewer is left. Absent, a detach is only a viewer leaving.
+        end_reason = params.get('end_reason')
         if (not isinstance(session_id, str) or not session_id
-                or not isinstance(subscription_id, str) or not subscription_id):
+                or not isinstance(subscription_id, str) or not subscription_id
+                or (end_reason is not None
+                    and not (isinstance(end_reason, str) and _END_REASON.fullmatch(end_reason)))):
             raise RuntimeStoreError('invalid_params')
         result = {'session_id': session_id, 'subscription_id': subscription_id,
                   'detached': False}
         if self.subscriptions.get(session_id) != subscription_id:
             return result
-        await self.authority.detach(self.actor, subscription_id)
+        await self.authority.detach(self.actor, subscription_id, end_reason=end_reason)
         if self.subscriptions.get(session_id) == subscription_id:
             del self.subscriptions[session_id]
         result['detached'] = True

tests/gateway/test_session_detach_end_reason.py

@@ -0,0 +1,62 @@
+"""``session.detach {end_reason}``: the last viewer leaving an idle local session ends its row with the
+reason main's classic CLI wrote itself (``cli_close`` on quit, ``new_session`` / ``resumed_other`` /
+``branched`` on a switch). Real authority + SQLite."""
+from types import SimpleNamespace
+
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_detach_with_end_reason_ends_the_idle_local_row(tmp_path, monkeypatch):
+    from gateway import run
+    from gateway.config import GatewayConfig
+    from gateway.session import SessionStore
+    from gateway.session_authority import initialize_session_authority
+    from gateway.session_controls import AuthorityConnection
+    from gateway.session_local import create_local_session
+    from hermes_state_local import local_receipt
+
+    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'platform_toolsets': {'cli': []}})
+    store = SessionStore(tmp_path / 'sessions', GatewayConfig())
+    runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False,
+                             _cached_agent_for=lambda route: None)
+    runner._adapter_for_source = lambda source: runner.adapters.get(source.platform)
+    authority = await initialize_session_authority(runner, profile_id='fixture', instance_id='owner')
+    db = authority.db
+
+    def viewer():
+        return AuthorityConnection(authority, object(), {'user_id': 'human', 'profile_id': 'fixture'})
+
+    try:
+        first, second = viewer(), viewer()
+        refs = {}
+        for name in ('quit', 'shared', 'plain'):
+            ref = create_local_session(authority, first.actor, {'request_id': name, 'source': 'cli',
+                                                                'cwd': str(tmp_path), 'model': 'm', 'toolsets': []})
+            refs[name] = ref, local_receipt(db, ref.session_id)['entry']['session_id']
+        subs = {}
+        for name, (ref, _) in refs.items():
+            reply = await first.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': ref.session_id}})
+            subs[name] = reply['result']['subscription_id']
+        await second.dispatch({'id': 1, 'method': 'session.resume', 'params': {'session_id': refs['shared'][0].session_id}})
+
+        async def detach(name, **extra):
+            return await first.dispatch({'id': 2, 'method': 'session.detach', 'params': {
+                'session_id': refs[name][0].session_id, 'subscription_id': subs[name], **extra}})
+
+        def ended(name):
+            row = db.get_session(refs[name][1])
+            return row['ended_at'] is not None, row['end_reason']
+
+        reply = await detach('quit', end_reason='cli_close')
+        assert reply['result']['detached'] is True, reply
+        assert ended('quit') == (True, 'cli_close')
+        await detach('shared', end_reason='new_session')
+        assert ended('shared') == (False, None), 'another viewer still holds the session'
+        await detach('plain')
+        assert ended('plain') == (False, None), 'a plain detach is not an end'
+        bad = await first.dispatch({'id': 3, 'method': 'session.detach', 'params': {
+            'session_id': refs['plain'][0].session_id, 'subscription_id': 'x', 'end_reason': 7}})
+        assert bad['error']['message'] == 'invalid_params'
+    finally:
+        store.close_all_db_handles()

tui_gateway/contracts/canonical.py

@@ -222,6 +222,7 @@ canonical_method("session.stats", params=SessionStatsParams, result=SessionStats
 class SessionDetachParams(Params):
     session_id: str
     subscription_id: str
+    end_reason: str | None = None
 
 
 class SessionDetachResult(Result):
@@ -231,7 +232,9 @@ class SessionDetachResult(Result):
 
 
 canonical_method("session.detach", params=SessionDetachParams, result=SessionDetachResult,
-                 doc="Drop this connection's subscription; a stale subscription id answers detached=false.")
+                 doc="Drop this connection's subscription; a stale subscription id answers detached=false. "
+                     "end_reason (snake_case): the viewer is closing the conversation; the last viewer "
+                     "leaving an idle local session ends its row with that reason.")
 
 
 class RuntimeDescribeParams(Params):