fix(state): list, delete and retire a local reset lineage as one conversation

A canonical local reset keeps the creation id S0 as the logical owner (policy,
FIFO, generation) and moves the transcript to a physical child S1. The paths
disagreed about that:
- list_sessions_rich (Desktop/dashboard /api/sessions, TUI picker) showed S0
  and S1 as two conversations;
- deleting the current S1 through the gateway was refused storage_unavailable
  (restore_local_session(S1) finds no receipt), while deleting the old S0
  silently took the current conversation too;
- `hermes sessions delete <S0>` / dashboard bulk delete removed only S0 and
  left S1 readable but never continuable, restorable or deletable.

Project the lineage as one conversation everywhere, like
acp_adapter/catalog.py:_local_owners:
- listing: earlier reset segments are filtered in SQL before LIMIT/OFFSET (no
  short or duplicate pages); the representative row carries the creation id as
  _lineage_root_id and every segment in _lineage_ids, which the Desktop already
  uses for delete/pin/tombstone matching;
- gateway delete of any segment authorizes and gates on the owner and removes
  owner + every segment (delete_targets resolves the owner first);
- legacy delete of the creation id takes its segments with it (its policy is
  retired); bulk/chain delete expands to the whole local conversation. Deleting
  only a physical child through the legacy single-row path keeps its historic
  contract (test_terminal_retirement D7).

Keeps test_delete_after_reset_covers_physical_route_and_cache_generation.

Review: https://github.com/NousResearch/hermes-agent/pull/106742#pullrequestreview-5475590057

gateway/session_mutations.py

@@ -37,6 +37,7 @@ def _authorize_control(authority, actor, ref, params):
     # A receipt authorizes only its original principal's exact retry; the
     # transaction still verifies the entire digest and current epoch.
     if not has_mutation_receipt(authority.db, actor.subject, ref.session_id, params['request_id']):
+        ref = _owner_ref(authority, ref, params)
         if ref.session_id not in authority.sessions:
             from hermes_state_mutation_binding import authorize_history
             with authority.db._read_ctx() as conn:
@@ -48,6 +49,19 @@ def _authorize_control(authority, actor, ref, params):
     return cold_history
 
 
+def _owner_ref(authority, ref, params):
+    """A delete names the listed row, which may be a later reset segment of a local conversation;
+    the conversation's creation id owns its policy, FIFO and live session, so control and the
+    session gate are checked on that owner (the store deletes the whole lineage either way)."""
+    if params.get('operation') != 'delete':
+        return ref
+    from gateway.session_contract import SessionRef
+    from hermes_state_local import local_lineage_owner
+    with authority.db._read_ctx() as conn:
+        owner = local_lineage_owner(conn, ref.session_id)
+    return ref if owner == ref.session_id else SessionRef(ref.profile_id, owner)
+
+
 async def mutate_session(authority, actor, ref, params):
     from gateway.session_runtime_workers import track_mutation
     authority._require_admission_open()
@@ -61,7 +75,7 @@ async def _owned_mutation(authority, actor, ref, params):
     if (params.get('operation') in {'model', 'compress', 'reset', 'rewind'}
             and ref.session_id not in authority.sessions and 'request_id' in params):
         _authorize_control(authority, actor, ref, params)
-    live = authority.sessions.get(ref.session_id)
+    live = authority.sessions.get(_owner_ref(authority, ref, params).session_id)
     if live is None:
         return await _mutate_session(authority, actor, ref, params)
     # Prepared policy/route publication is inseparable from its durable commit

hermes_state_local.py

@@ -88,6 +88,79 @@ def retire_local_receipts(conn, session_ids):
     conn.executemany('DELETE FROM state_meta WHERE key=?', [(key,) for key in gone])
 
 
+def local_lineage_owner(conn, session_id):
+    """The creation id whose receipt lineage holds physical segment *session_id* (a reset or
+    compression child, which carries the owner as ``chat_id``); *session_id* itself otherwise."""
+    row = conn.execute('SELECT chat_id FROM sessions WHERE id=?', (session_id,)).fetchone()
+    owner = row[0] if row else None
+    if not owner or owner == session_id:
+        return session_id
+    saved = conn.execute('SELECT value FROM state_meta WHERE key=?', (POLICY_PREFIX + owner,)).fetchone()
+    try:
+        lineage = json.loads(saved[0]).get('lineage') if saved else None
+    except (TypeError, ValueError, AttributeError):
+        return session_id
+    return owner if isinstance(lineage, list) and session_id in lineage else session_id
+
+
+def owned_lineage_ids(conn, session_id):
+    """[session_id] for any row, or the whole conversation when *session_id* IS a local creation id
+    whose receipt has moved its transcript to later segments: deleting the owner retires its policy,
+    so its segments must go with it or they are left readable but never continuable."""
+    if local_lineage_owner(conn, session_id) != session_id:
+        return [session_id]
+    return list(dict.fromkeys([session_id, *local_conversation_ids(conn, session_id)]))
+
+
+def local_conversation_ids(conn, session_id):
+    """Every row of the local conversation *session_id* belongs to: its creation id first, then
+    each segment of the receipt lineage. A canonical reset never forks the logical session (one
+    policy, FIFO and generation), so listing, delete and retirement treat these ids as one."""
+    owner = local_lineage_owner(conn, session_id)
+    saved = conn.execute('SELECT value FROM state_meta WHERE key=?', (POLICY_PREFIX + owner,)).fetchone()
+    try:
+        lineage = json.loads(saved[0]).get('lineage') if saved else None
+    except (TypeError, ValueError, AttributeError):
+        lineage = None
+    if not isinstance(lineage, list):
+        return [session_id]
+    return list(dict.fromkeys([owner, *(sid for sid in lineage if isinstance(sid, str))]))
+
+
+def local_lineage_index(conn):
+    """``(superseded, heads)`` for listing a local conversation as one row, as
+    ``acp_adapter.catalog`` does: a canonical reset keeps the creation id as the owner and moves the
+    transcript to a child, so every segment is one conversation. ``superseded`` holds the listable
+    roots of earlier reset segments (filtered in SQL, before paging); ``heads`` maps the latest
+    reset segment to ``(owner, lineage)`` for the row that represents the conversation."""
+    superseded, heads = set(), {}
+    rows = conn.execute("SELECT value FROM state_meta WHERE key GLOB ? AND json_valid(value) "
+                        "AND json_array_length(value, '$.lineage') > 1", (POLICY_PREFIX + '*',)).fetchall()
+    for (raw,) in rows:
+        receipt = json.loads(raw)
+        owner, lineage = receipt.get('session_id'), receipt.get('lineage')
+        if not isinstance(owner, str) or not all(isinstance(sid, str) for sid in lineage):
+            continue
+        marks = ','.join('?' * len(lineage))
+        ended = dict(conn.execute(f'SELECT id,end_reason FROM sessions WHERE id IN ({marks})', lineage).fetchall())
+        starts = [sid for i, sid in enumerate(lineage) if i == 0 or ended.get(lineage[i - 1]) == 'session_reset']
+        superseded.update(starts[:-1])
+        heads[starts[-1]] = (owner, lineage)
+    return superseded, heads
+
+
+def annotate_local_lineages(sessions, heads):
+    """Name the whole conversation on its representative row: ``_lineage_root_id`` is the creation
+    id (the id every surface resolves, deletes and pins by), ``_lineage_ids`` every segment."""
+    for row in sessions:
+        found = heads.get(row.get('_lineage_root_id') or row['id'])
+        if found is not None:
+            owner, lineage = found
+            row['_lineage_ids'] = list(lineage)
+            row['_lineage_root_id'] = owner if row['id'] != owner else None
+    return sessions
+
+
 def local_receipt(db, session_id):
     with db._read_ctx() as conn:
         row = conn.execute('SELECT value FROM state_meta WHERE key=?', (POLICY_PREFIX + session_id,)).fetchone()

hermes_state_mutation_guards.py

@@ -77,9 +77,14 @@ def delete_targets(conn, session_id):
     from hermes_state_compression import _CHAIN_CAP
     from hermes_state_local import POLICY_PREFIX
     from hermes_state_local_lineage import validate_local_lineage
+    from hermes_state_local import local_lineage_owner
     targets = {session_id}
+    # A reset/compression segment of a local conversation is that conversation: the listing
+    # shows one row for it, so its delete takes the creation id (policy, FIFO) and every segment.
+    owner = local_lineage_owner(conn, session_id)
+    targets.add(owner)
     saved = conn.execute('SELECT value FROM state_meta WHERE key=?',
-                         (POLICY_PREFIX + session_id,)).fetchone()
+                         (POLICY_PREFIX + owner,)).fetchone()
     if saved is not None:
         receipt = json.loads(saved[0])
         validate_local_lineage(conn, receipt)

hermes_state_sessions.py

@@ -1333,6 +1333,16 @@ class SessionSessionsMixin:
         # it back (#90946).
         if not include_hidden and not archived_only:
             where_clauses.append("s.hidden = 0")
+        heads = {}
+        if not include_children:
+            # A local reset lineage is one conversation (hermes_state_local.local_lineage_index):
+            # earlier segments leave the page in SQL, before LIMIT/OFFSET, not after it.
+            from hermes_state_local import local_lineage_index
+            with self._read_ctx() as conn:
+                superseded, heads = local_lineage_index(conn)
+            if superseded:
+                where_clauses.append("s.id NOT IN (SELECT value FROM json_each(?))")
+                params.append(json.dumps(sorted(superseded)))
         where_sql = _where_sql(where_clauses)
         # Shared projection head of the three list queries (whitespace is part of the SQL text).
         select_head = (
@@ -1407,6 +1417,9 @@ class SessionSessionsMixin:
             )
             if not include_hidden and not archived_only:
                 pinned_clauses.append("s.hidden = 0")
+            if heads:
+                pinned_clauses.append("s.id NOT IN (SELECT value FROM json_each(?))")
+                pinned_params.append(json.dumps(sorted(superseded)))
             pinned_clauses.append("s.pinned = 1")
             pinned_where = _where_sql(pinned_clauses)
             pinned_query = f"""
@@ -1422,6 +1435,9 @@ class SessionSessionsMixin:
                     sessions.append(s)
         if project_compression_tips and not include_children:
             sessions = self._project_compression_tips(sessions, compact_rows)
+        if heads:
+            from hermes_state_local import annotate_local_lineages
+            sessions = annotate_local_lineages(sessions, heads)
         # last_read_at is lineage-stamped, so root and tip watermarks agree.
         for s in sessions:
             s["unread"] = self.session_unread(s)
@@ -1614,8 +1630,10 @@ class SessionSessionsMixin:
             # Use the borrowed read connection, never self._conn: handing the shared writer connection to a
             # helper here executes on it without self._lock — the same unsynchronized-read class as
             # #99349/#90734.
-            delegate_ids = _collect_delegate_child_ids(conn, [session_id])
-        return [session_id, *sorted(delegate_ids)]
+            from hermes_state_local import owned_lineage_ids
+            scope = owned_lineage_ids(conn, session_id)
+            delegate_ids = _collect_delegate_child_ids(conn, scope)
+        return [session_id, *sorted(set(scope) - {session_id}), *sorted(delegate_ids)]
 
     def _expand_compression_lineage(self, conn, session_ids: list[str]) -> list[str]:
         """Expand *session_ids* to every member of their compression chains.
@@ -1647,6 +1665,11 @@ class SessionSessionsMixin:
             targets = _expand_compression_lineage_ids(conn, frontier)
             frontier = [sid for sid in targets if sid not in seen]
             seen.update(frontier)
+        # A local reset lineage (creation id + every segment, compression children included) is one
+        # listed conversation: deleting any segment must not strand the others without an owner.
+        from hermes_state_local import local_conversation_ids
+        for sid in list(seen):
+            seen.update(local_conversation_ids(conn, sid))
         return list(seen)
 
     def delete_session(
@@ -1681,8 +1704,12 @@ class SessionSessionsMixin:
         def _do(conn):
             if conn.execute("SELECT 1 FROM sessions WHERE id = ? LIMIT 1", (session_id,)).fetchone() is None:
                 return False
+            # Deleting a local creation id takes its reset/compression segments with it (its policy
+            # goes); any other row, a single segment included, is just itself.
+            from hermes_state_local import owned_lineage_ids
+            scope = owned_lineage_ids(conn, session_id)
             target_ids = (
-                [session_id, *_collect_delegate_child_ids(conn, [session_id])]
+                [*scope, *_collect_delegate_child_ids(conn, scope)]
                 if exclude_active_write_guards or expected_ids is not None else None
             )
             if exclude_active_write_guards and self._guarded_ids(conn, target_ids):
@@ -1698,15 +1725,14 @@ class SessionSessionsMixin:
             ):
                 return False
             from hermes_state_mutation_retirement import retire_sessions
-            retire_sessions(conn, [session_id])
-            removed_ids.extend(_delete_delegate_children(conn, [session_id]))
-            conn.execute(  # orphan remaining children (branches) so FK is satisfied
-                "UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id = ?", (session_id,),
-            )
-            conn.execute("DELETE FROM messages WHERE session_id = ?", (session_id,))
-            conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
+            retire_sessions(conn, scope)
+            removed_ids.extend(_delete_delegate_children(conn, scope))
+            for sid in scope:  # orphan remaining children (branches) so FK is satisfied
+                conn.execute("UPDATE sessions SET parent_session_id = NULL WHERE parent_session_id = ?", (sid,))
+                conn.execute("DELETE FROM messages WHERE session_id = ?", (sid,))
+            conn.executemany("DELETE FROM sessions WHERE id = ?", [(sid,) for sid in scope])
             self._delete_unreferenced_system_prompts(conn)
-            removed_ids.append(session_id)
+            removed_ids.extend(scope)
             return True
         deleted = self._execute_write(_do)
         from hermes_state_media import collect_retired_media

tests/gateway/test_local_reset_lineage_one_conversation.py

@@ -0,0 +1,68 @@
+"""N18: a local reset lineage is ONE conversation in the listing, the gateway delete and the
+legacy store delete (creation id S0 owns policy/FIFO; the transcript moved to reset child S1)."""
+from types import SimpleNamespace
+
+import pytest
+import pytest_asyncio
+
+
+@pytest_asyncio.fixture
+async def reset_lineage(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_contract import Principal
+    from gateway.session_local import create_local_session
+    from gateway.session_mutations import mutate_session
+    from hermes_state import SessionDB
+
+    monkeypatch.setenv('HERMES_HOME', str(tmp_path))
+    monkeypatch.setattr(run, '_load_gateway_config', lambda: {'platform_toolsets': {'cli': []}})
+    store = SessionStore(tmp_path / 'sessions', GatewayConfig())
+    store._db = SessionDB(db_path=tmp_path / 'state.db')
+    runner = SimpleNamespace(session_store=store, _session_db=store._db, adapters={}, _draining=False,
+                             _evict_cached_agent=lambda route: None)
+    runner._adapter_for_source = lambda source: runner.adapters.get(source.platform)
+    authority = await initialize_session_authority(runner, profile_id='default', instance_id='fixture')
+    authority._schedule = lambda ref: None
+    owner = Principal('uid:1000', 'default', frozenset({'session:create', 'session:read', 'session:submit',
+                                                        'session:control'}), 'native')
+    ref = create_local_session(authority, owner, {'request_id': 'r', 'source': 'gui', 'cwd': str(tmp_path),
+                                                  'model': 'fixture', 'toolsets': []})
+    authority.db.append_message(ref.session_id, 'user', 'before reset')
+    row = authority.db.get_session(ref.session_id)
+    reset = await mutate_session(authority, owner, ref, dict(
+        session_id=ref.session_id, request_id='reset', expected_revision=row['runtime_revision'],
+        expected_generation=row['runtime_generation'], operation='reset', payload={}))
+    authority.db.append_message(reset['target_session_id'], 'user', 'after reset')
+    yield SimpleNamespace(authority=authority, owner=owner, s0=ref.session_id, s1=reset['target_session_id'])
+    store._db.close()
+
+
+@pytest.mark.asyncio
+async def test_reset_lineage_lists_once_and_deleting_the_listed_row_deletes_it(reset_lineage):
+    from gateway.session_contract import SessionRef
+    from gateway.session_mutations import mutate_session
+    t = reset_lineage
+    rows = t.authority.db.list_sessions_rich(order_by_last_active=True, limit=50)
+    assert [(r['id'], r.get('_lineage_root_id')) for r in rows] == [(t.s1, t.s0)], 'listed as two conversations'
+    # Pages are filtered before LIMIT/OFFSET: the superseded segment never fills a later page.
+    assert t.authority.db.list_sessions_rich(limit=1, offset=1) == []
+    row = t.authority.db.get_session(t.s1)
+    result = await mutate_session(t.authority, t.owner, SessionRef('default', t.s1), dict(
+        session_id=t.s1, request_id='delete-current', expected_revision=row['runtime_revision'],
+        expected_generation=row['runtime_generation'], operation='delete', payload={}))
+    assert set(result['deleted_ids']) == {t.s0, t.s1}
+    assert t.authority.db.get_session(t.s0) is None and t.s0 not in t.authority.sessions
+
+
+@pytest.mark.asyncio
+async def test_legacy_delete_of_either_segment_retires_the_whole_lineage(reset_lineage):
+    from hermes_state_local import POLICY_PREFIX
+    t = reset_lineage
+    db = t.authority.db
+    assert db.delete_session(t.s0, exclude_active_write_guards=True)
+    assert db.get_session(t.s1) is None, 'the current segment survived without its owner'
+    with db._read_ctx() as conn:
+        assert conn.execute('SELECT 1 FROM state_meta WHERE key=?', (POLICY_PREFIX + t.s0,)).fetchone() is None