fix(lineage): honor _reset_from, fail closed on untrusted markers, hash full watcher payload

Layer three additive pieces on top of the #7021 / #6931 continuation
predicate without changing its shape or CONTINUATION_STARTED_AT_TOLERANCE_SECONDS:

- _reset_from is a boundary marker alongside _branched_from/_delegate_from
  (Hermes Agent stamps it in gateway/session_recovery.py and filters it in
  _NON_CONTINUATION_CHILD_FILTER_SQL). The same predicate drives the sidebar
  listing, lineage metadata/report and the transcript stitcher.
- _branch_markers() returns ('none' | 'markers' | 'unknown', markers);
  unparsable, non-object, too-deep, or non-string marker payloads are
  'unknown' and fail closed as a lineage boundary instead of degrading to
  "no markers" (fail open).
- gateway_watcher._snapshot_hash() hashes the complete canonicalized published
  payload, and the cheap fingerprint covers model_config, so a projection change
  to title/created_at/metadata with fixed (id, updated_at, message_count) emits
  sessions_changed.

api/agent_sessions.py

@@ -366,28 +366,59 @@ def is_cli_session_row_visible(row: dict) -> bool:
     return _count_user_turns(row) >= CLI_MIN_UNTITLED_USER_MESSAGE_COUNT
 
 
-def _branch_markers(row: dict | None) -> tuple[str | None, str | None]:
-    """Return ``(_branched_from, _delegate_from)`` from a row's ``model_config``.
+# Every model_config marker Hermes Agent binds to ``parent_session_id`` in its
+# non-continuation child predicate (``_NON_CONTINUATION_CHILD_FILTER_SQL``):
+# ``_delegate_from`` (delegate_task), ``_branched_from`` (/branch) and
+# ``_reset_from`` (gateway reset children, stamped at creation or durably by
+# ``reopen_session()`` in ``gateway/session_recovery.py``).
+_MODEL_CONFIG_LINEAGE_KEYS = ('_delegate_from', '_branched_from', '_reset_from')
 
-    Hermes Agent stamps explicit branches and delegate/subagent runs in the
-    ``model_config`` JSON column (``_branched_from`` / ``_delegate_from``);
-    ``session_source='fork'`` alone only covers WebUI-created forks. Missing,
-    empty, or unparsable ``model_config`` degrades to ``(None, None)``.
+
+def _branch_markers(row: dict | None) -> tuple[str, dict[str, str]]:
+    """Return ``(state, markers)`` for a row's ``model_config`` lineage identity.
+
+    Hermes Agent stamps explicit branches, delegate/subagent runs and gateway
+    reset children in the ``model_config`` JSON column (see
+    ``_MODEL_CONFIG_LINEAGE_KEYS``); ``session_source='fork'`` alone only
+    covers WebUI-created forks.
+
+    ``state`` is one of:
+
+    - ``'none'``: no ``model_config``, or an object carrying no lineage marker;
+    - ``'markers'``: every non-null marker is a non-empty string, returned by key;
+    - ``'unknown'``: identity evidence exists but cannot be trusted (payload is
+      not JSON, not a JSON object, too deeply nested to decode, or a marker is
+      not a non-empty string). Callers must fail closed on this state and treat
+      the row as a lineage boundary, never as a continuation.
     """
     if not row:
-        return None, None
+        return 'none', {}
     raw = row.get('model_config')
-    if isinstance(raw, dict):
-        config = raw
-    elif isinstance(raw, str) and raw.strip():
+    if raw is None:
+        return 'none', {}
+    if isinstance(raw, (str, bytes, bytearray)):
+        if not raw.strip():
+            return 'none', {}
         try:
-            parsed = json.loads(raw)
-        except (TypeError, ValueError):
-            parsed = None
-        config = parsed if isinstance(parsed, dict) else {}
+            config = json.loads(raw)
+        except (TypeError, ValueError, RecursionError):
+            # RecursionError: a valid but pathologically deep payload exceeds
+            # the decoder's recursion limit; it is untrusted identity evidence
+            # like any other undecodable payload, never an escaping crash.
+            return 'unknown', {}
     else:
-        config = {}
-    return config.get('_branched_from'), config.get('_delegate_from')
+        config = raw
+    if not isinstance(config, dict):
+        return 'unknown', {}
+    markers: dict[str, str] = {}
+    for key in _MODEL_CONFIG_LINEAGE_KEYS:
+        value = config.get(key)
+        if value is None:
+            continue
+        if not isinstance(value, str) or not value.strip():
+            return 'unknown', {}
+        markers[key] = value.strip()
+    return ('markers', markers) if markers else ('none', {})
 
 
 def _is_continuation_session(parent: dict | None, child: dict | None) -> bool:
@@ -405,27 +436,29 @@ def _is_continuation_session(parent: dict | None, child: dict | None) -> bool:
     conversation; otherwise the tip inherits the root's title/source metadata and
     can disappear under messaging/sidebar policies.
 
-    Explicit branches/delegates never continue: a child whose ``model_config``
-    ``_branched_from`` / ``_delegate_from`` points at this parent is a fork of
-    the conversation and must stay visible. Beyond the bounded tolerance window
-    there is no reliable continuation signal — titles are user-controlled and
-    non-unique, so they are never used to widen the window (#7021 re-gate).
+    Explicit branches/delegates/resets never continue: a child whose
+    ``model_config`` ``_branched_from`` / ``_delegate_from`` / ``_reset_from``
+    points at this parent is a separate conversation and must stay visible.
+    Unparsable or non-string marker evidence fails closed as a boundary.
+    Beyond the bounded tolerance window there is no reliable continuation
+    signal — titles are user-controlled and non-unique, so they are never used
+    to widen the window (#7021 re-gate).
     """
     if not parent or not child:
         return False
     if str(child.get('session_source') or '').strip().lower() == 'fork':
         return False
-    # Real Agent branches/delegates are marked in model_config (not
+    # Real Agent branches/delegates/resets are marked in model_config (not
     # session_source, which only WebUI-created forks carry). The marker must
     # reference THIS parent: compression continuations inherit the rotated
     # agent's model_config verbatim, so presence alone (a delegate's
     # continuation still carries the delegate's own ``_delegate_from``) would
-    # misclassify real continuations.
-    child_branched_from, child_delegate_from = _branch_markers(child)
-    parent_id = parent.get('id')
-    if parent_id and (
-        child_branched_from == parent_id or child_delegate_from == parent_id
-    ):
+    # misclassify real continuations. Untrusted identity evidence fails closed.
+    marker_state, child_markers = _branch_markers(child)
+    if marker_state == 'unknown':
+        return False
+    parent_id = str(parent.get('id') or '').strip()
+    if parent_id and parent_id in child_markers.values():
         return False
     parent_source = str(parent.get('source') or '').strip().lower()
     child_source = str(child.get('source') or '').strip().lower()

api/gateway_watcher.py

@@ -9,6 +9,7 @@ This enables real-time session list updates in the sidebar without
 requiring any changes to hermes-agent.
 """
 import hashlib
+import json
 import logging
 import os
 import queue
@@ -27,12 +28,27 @@ logger = logging.getLogger(__name__)
 # ── State hash tracking ─────────────────────────────────────────────────────
 
 def _snapshot_hash(sessions: list) -> str:
-    """Create a lightweight hash of session IDs and timestamps for change detection."""
-    key = '|'.join(
-        f"{s['session_id']}:{s.get('updated_at', 0)}:{s.get('message_count', 0)}"
-        for s in sorted(sessions, key=lambda x: x['session_id'])
-    )
-    return hashlib.md5(key.encode(), usedforsecurity=False).hexdigest()
+    """Hash the complete published session payload for change detection.
+
+    Every emitted field participates, not only the id / activity timestamp /
+    message count triple. Projection authority (compression collapse,
+    ``model_config`` lineage markers, title promotion) can change a row's
+    ``title``, ``created_at`` or source metadata while that triple stays fixed;
+    hashing only the triple let the projection rerun, keep ``_last_sessions``
+    stale and emit no ``sessions_changed`` event. Entries are canonicalised
+    (sorted keys, stable separators) and ordered by ``session_id`` so the hash
+    is deterministic and independent of sidebar ordering; cost is bounded by
+    the projection's own row limit.
+    """
+    digest = hashlib.md5(usedforsecurity=False)
+    for session in sorted(sessions, key=lambda x: str(x.get('session_id') or '')):
+        digest.update(
+            json.dumps(
+                session, sort_keys=True, separators=(',', ':'), default=str
+            ).encode('utf-8', 'replace')
+        )
+        digest.update(b'\x1e')
+    return digest.hexdigest()
 
 
 # Sources excluded from the WebUI sidebar projection. Must match the default
@@ -70,10 +86,10 @@ def _cheap_change_fingerprint(db_path: Path) -> str | None:
     # are always present (``source`` is required for the projection to run at
     # all); the rest are optional on older agent schemas and filtered below.
     _PROJECTION_SESSION_COLS = (
-        'id', 'source', 'session_source', 'title', 'model', 'message_count',
-        'started_at', 'ended_at', 'end_reason', 'parent_session_id', 'archived',
-        'user_id', 'chat_id', 'chat_type', 'thread_id', 'session_key',
-        'origin_chat_id', 'origin_user_id', 'platform',
+        'id', 'source', 'session_source', 'model_config', 'title', 'model',
+        'message_count', 'started_at', 'ended_at', 'end_reason',
+        'parent_session_id', 'archived', 'user_id', 'chat_id', 'chat_type',
+        'thread_id', 'session_key', 'origin_chat_id', 'origin_user_id', 'platform',
     )
     try:
         with closing(open_state_db_readonly(db_path)) as conn:

tests/test_gateway_sync.py

@@ -18,6 +18,8 @@ import time
 import urllib.error
 import urllib.request
 
+import pytest
+
 REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve()
 from tests._pytest_port import BASE
 
@@ -500,6 +502,178 @@ def test_compression_chain_collapses_to_latest_tip_in_sidebar():
         post('/api/settings', {'show_cli_sessions': False})
 
 
+@pytest.mark.parametrize('reset_marker_origin', ('creation', 'reopen'))
+def test_reopen_then_recompress_reset_child_stays_independent_across_readers(
+    tmp_path, monkeypatch, reset_marker_origin
+):
+    """Real Agent lifecycle: reset child, ``reopen_session()``, later compression.
+
+    Driven through the installed ``SessionDB`` so the rows carry exactly the
+    production shape. ``creation`` mirrors the gateway creating the reset child
+    with ``_reset_from`` in its INSERT; ``reopen`` mirrors a marker-less legacy
+    reset child that ``reopen_session()`` durably stamps. Either way the parent
+    ends up ``end_reason='compression'`` with a live reset child, and every
+    WebUI reader must agree: the reset conversation stays visible and its
+    transcript is never prefixed with the parent's.
+
+    The WebUI readers are the contract under test. The Agent's own
+    ``find_live_compression_child`` / ``get_compression_tip`` verdict is only
+    cross-checked when the installed Agent already binds ``_reset_from`` in
+    its non-continuation child predicate; an older Agent whose predicate
+    predates that marker still has to drive the same fixture without turning
+    this WebUI test red.
+    """
+    hermes_state = pytest.importorskip('hermes_state')
+    # SessionDB lazily imports sibling Agent modules (e.g. hermes_logging);
+    # keep the Agent checkout importable even after the suite's sys.path
+    # isolation guard restored an earlier snapshot.
+    monkeypatch.syspath_prepend(str(pathlib.Path(hermes_state.__file__).resolve().parent))
+    import api.models as models
+    from api.agent_sessions import (
+        read_importable_agent_session_rows,
+        read_session_lineage_metadata,
+        read_session_lineage_report,
+    )
+    from api.gateway_watcher import GatewayWatcher
+    from api.models import get_state_db_session_messages
+
+    parent_id = 'recompressed_reset_parent'
+    child_id = 'reset_child_conversation'
+    routing = {'session_key': 'telegram:reset-peer', 'user_id': 'peer', 'chat_id': 'chat-1'}
+    db_path = tmp_path / 'state.db'
+    db = hermes_state.SessionDB(db_path=db_path)
+    try:
+        db.create_session(parent_id, source='telegram', model='test-model', **routing)
+        db.append_messages_batch(
+            parent_id,
+            [
+                {'role': 'user', 'content': 'before reset', 'timestamp': 100.0},
+                {'role': 'assistant', 'content': 'answer before reset', 'timestamp': 101.0},
+            ],
+        )
+        assert db.promote_to_session_reset(parent_id) is True
+        child_model_config = (
+            {'_reset_from': parent_id} if reset_marker_origin == 'creation' else None
+        )
+        db.create_session(
+            child_id,
+            source='telegram',
+            model='test-model',
+            parent_session_id=parent_id,
+            model_config=child_model_config,
+            **routing,
+        )
+        db.append_messages_batch(
+            child_id,
+            [
+                {'role': 'user', 'content': 'after reset', 'timestamp': 200.0},
+                {'role': 'assistant', 'content': 'answer after reset', 'timestamp': 201.0},
+            ],
+        )
+        if reset_marker_origin == 'reopen':
+            assert db.get_session(child_id)['model_config'] is None
+        # The user resumes the old conversation: reopen_session() durably
+        # stamps _reset_from on the legacy reset child before clearing the
+        # parent's reset boundary. Should an installed Agent predate that
+        # stamping, apply the same schema the upstream Agent writes
+        # (``model_config._reset_from = parent_session_id``) so the rows still
+        # carry the production shape the WebUI readers are contracted against.
+        db.reopen_session(parent_id)
+        expected_child_config = {'_reset_from': parent_id}
+        if json.loads(db.get_session(child_id)['model_config'] or '{}') != expected_child_config:
+            with sqlite3.connect(str(db_path)) as conn:
+                conn.execute(
+                    "UPDATE sessions SET model_config = ? WHERE id = ?",
+                    (json.dumps(expected_child_config), child_id),
+                )
+        assert json.loads(db.get_session(child_id)['model_config']) == expected_child_config
+        reopened = db.get_session(parent_id)
+        assert reopened['ended_at'] is None and reopened['end_reason'] is None
+        db.append_messages_batch(
+            parent_id,
+            [{'role': 'user', 'content': 'resumed parent', 'timestamp': 300.0}],
+        )
+        # ...and that resumed parent is later closed by compression.
+        db.end_session(parent_id, 'compression')
+        assert db.get_session(parent_id)['end_reason'] == 'compression'
+        # Agent cross-check, only when the installed Agent's child predicate
+        # already binds _reset_from: the reset child is not a compression
+        # continuation there either. Older Agents are not this test's subject.
+        agent_filter_sql = str(
+            getattr(hermes_state.SessionDB, '_NON_CONTINUATION_CHILD_FILTER_SQL', '')
+        )
+        if '_reset_from' in agent_filter_sql:
+            assert db.find_live_compression_child(parent_id) is None
+            assert db.get_compression_tip(parent_id) == parent_id
+    finally:
+        db.close()
+
+    with sqlite3.connect(str(db_path)) as conn:
+        conn.execute(
+            "UPDATE sessions SET title = ? WHERE id = ?",
+            ('Reset parent conversation', parent_id),
+        )
+        conn.execute(
+            "UPDATE sessions SET title = ? WHERE id = ?",
+            ('Reset child conversation', child_id),
+        )
+
+    # (a) Sidebar projection: both rows visible; the child keeps its own
+    # title and is a plain child_session, never a hidden continuation.
+    projected = read_importable_agent_session_rows(db_path, limit=None, exclude_sources=None)
+    projected_by_id = {row['id']: row for row in projected}
+    assert set(projected_by_id) == {parent_id, child_id}
+    child_row = projected_by_id[child_id]
+    assert child_row['title'] == 'Reset child conversation'
+    assert child_row['relationship_type'] == 'child_session'
+    assert child_row['parent_session_id'] == parent_id
+    assert '_lineage_root_id' not in child_row
+    parent_row = projected_by_id[parent_id]
+    assert parent_row['title'] == 'Reset parent conversation'
+    assert '_lineage_tip_id' not in parent_row
+    assert '_compression_segment_count' not in parent_row
+
+    # (b) Lineage metadata and report.
+    metadata = read_session_lineage_metadata(db_path, {parent_id, child_id})
+    assert metadata[child_id]['relationship_type'] == 'child_session'
+    assert '_lineage_root_id' not in metadata[child_id]
+    assert '_lineage_tip_id' not in metadata.get(parent_id, {})
+    child_report = read_session_lineage_report(db_path, child_id)
+    assert child_report['lineage_key'] == child_id
+    assert child_report['tip_session_id'] == child_id
+    assert child_report['total_segments'] == 1
+    parent_report = read_session_lineage_report(db_path, parent_id)
+    assert parent_report['tip_session_id'] == parent_id
+    assert parent_report['total_segments'] == 1
+    assert [child['session_id'] for child in parent_report['children']] == [child_id]
+    assert parent_report['children'][0]['role'] == 'child_session'
+
+    # (c) Watcher publication: the SSE payload carries both conversations
+    # with their own titles.
+    watcher = GatewayWatcher(state_db_path=db_path)
+    subscriber = watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    event = subscriber.get_nowait()
+    published = {row['session_id']: row for row in event['sessions']}
+    assert set(published) == {parent_id, child_id}
+    assert published[child_id]['title'] == 'Reset child conversation'
+    assert published[parent_id]['title'] == 'Reset parent conversation'
+
+    # (d) Transcript stitching never prepends the parent's transcript.
+    monkeypatch.setattr(models, '_active_state_db_path', lambda: db_path)
+    stitched = get_state_db_session_messages(child_id, stitch_continuations=True)
+    assert [message['content'] for message in stitched] == [
+        'after reset',
+        'answer after reset',
+    ]
+    parent_messages = get_state_db_session_messages(parent_id, stitch_continuations=True)
+    assert [message['content'] for message in parent_messages] == [
+        'before reset',
+        'answer before reset',
+        'resumed parent',
+    ]
+
+
 def test_compression_lineage_prefers_freshest_descendant_over_newer_direct_sibling():
     """A later-started stale sibling must not hide a deeper active branch."""
     conn = _ensure_state_db()

tests/test_issue3506_memory_and_watcher.py

@@ -159,6 +159,7 @@ def _make_db(tmp_path: Path):
             id TEXT PRIMARY KEY,
             source TEXT NOT NULL,
             session_source TEXT,
+            model_config TEXT,
             model TEXT,
             started_at REAL NOT NULL,
             ended_at REAL,
@@ -355,6 +356,165 @@ def test_periodic_projection_recovers_role_only_sidebar_visibility_change(tmp_pa
     assert subscriber.empty()
 
 
+def test_marker_only_update_invalidates_projection_on_next_poll(tmp_path):
+    """model_config is projection authority, so its mutation cannot wait for parity."""
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    _add_session(conn, "compression-parent", "telegram", mc=1, started=100.0)
+    _add_session(conn, "compression-child", "telegram", mc=1, started=199.0)
+    conn.execute(
+        "UPDATE sessions SET ended_at = 200.0, end_reason = 'compression' "
+        "WHERE id = 'compression-parent'"
+    )
+    conn.execute(
+        "UPDATE sessions SET parent_session_id = 'compression-parent' "
+        "WHERE id = 'compression-child'"
+    )
+    conn.commit()
+
+    watcher = gw.GatewayWatcher(state_db_path=db)
+    subscriber = watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    initial = subscriber.get_nowait()
+    assert [row["session_id"] for row in initial["sessions"]] == [
+        "compression-child"
+    ]
+
+    # No timestamp, count, source, or message row changes. This one field turns
+    # the previously-collapsed child into a direct branch and must be visible on
+    # the ordinary five-second poll rather than the 60-second parity fallback.
+    conn.execute(
+        "UPDATE sessions SET model_config = ? WHERE id = 'compression-child'",
+        ('{"_branched_from":"compression-parent"}',),
+    )
+    conn.commit()
+
+    assert watcher._poll_once(now=2.0) is True
+    event = subscriber.get_nowait()
+    assert {row["session_id"] for row in event["sessions"]} == {
+        "compression-parent",
+        "compression-child",
+    }
+    assert subscriber.empty()
+
+
+def test_snapshot_hash_covers_every_published_field():
+    """Any emitted field change must change the hash; ordering must not."""
+    gw = importlib.import_module("api.gateway_watcher")
+    base = [
+        {
+            "session_id": "b",
+            "title": "Chat B",
+            "model": "m",
+            "message_count": 2,
+            "created_at": 100.0,
+            "updated_at": 150.0,
+            "source": "telegram",
+            "raw_source": "telegram",
+            "session_source": "messaging",
+            "source_label": "Telegram",
+        },
+        {
+            "session_id": "a",
+            "title": "Chat A",
+            "model": "m",
+            "message_count": 1,
+            "created_at": 50.0,
+            "updated_at": 60.0,
+            "source": "cli",
+            "raw_source": "cli",
+            "session_source": "cli",
+            "source_label": "CLI",
+        },
+    ]
+    reference = gw._snapshot_hash(base)
+    assert gw._snapshot_hash(list(reversed(base))) == reference, "order-independent"
+    assert gw._snapshot_hash([dict(row) for row in base]) == reference, "deterministic"
+    assert gw._snapshot_hash([]) != reference
+
+    # Mutating any single published field, including the ones the old
+    # id/updated_at/message_count triple ignored, changes the hash.
+    for field in base[0]:
+        mutated = [dict(row) for row in base]
+        mutated[0][field] = f"changed-{field}"
+        assert gw._snapshot_hash(mutated) != reference, field
+
+
+def test_marker_only_projection_change_emits_sse_event_without_row_churn(tmp_path):
+    """ID / message count / activity stay fixed; title + segments change → event.
+
+    Gate reproduction: a lineage-marker mutation reruns the projection (the
+    cheap fingerprint covers ``model_config``) and changes the published
+    title/segment projection of a row, but the session id set, every
+    ``message_count`` and every ``updated_at`` in the payload are unchanged.
+    The old snapshot hash saw no difference, so ``_last_sessions`` kept the
+    stale projection and no subscriber event was queued.
+    """
+    gw = importlib.import_module("api.gateway_watcher")
+    db, conn = _make_db(tmp_path)
+    # A compression chain whose root segment holds no messages (its
+    # transcript moved to the continuation): the projection publishes ONE
+    # row under the tip's id carrying the root's title and a segment count
+    # of 2. A message-less root is hidden when it stands alone.
+    _add_session(conn, "root-a", "telegram", mc=0, started=100.0, title="Alpha")
+    _add_session(conn, "tip-b", "telegram", mc=1, started=150.0, title="Beta")
+    conn.execute(
+        "UPDATE sessions SET ended_at = 150.0, end_reason = 'compression' WHERE id = 'root-a'"
+    )
+    conn.execute("UPDATE sessions SET parent_session_id = 'root-a' WHERE id = 'tip-b'")
+    conn.commit()
+
+    watcher = gw.GatewayWatcher(state_db_path=db)
+    subscriber = watcher.subscribe()
+    assert watcher._poll_once(now=1.0) is True
+    initial = subscriber.get_nowait()
+    initial_rows = {row["session_id"]: row for row in initial["sessions"]}
+    assert set(initial_rows) == {"tip-b"}
+    assert initial_rows["tip-b"]["title"] == "Alpha"
+    assert initial_rows["tip-b"]["message_count"] == 1
+    assert initial_rows["tip-b"]["updated_at"] == 150.0
+    assert watcher._last_sessions == initial["sessions"]
+
+    # Marker-only flip: stamp _reset_from on the tip. No timestamp, count,
+    # source or message row changes anywhere. The tip is now a reset child
+    # (its own conversation, own title, one segment) and the message-less
+    # root drops out — the published id set, message_count and updated_at
+    # are byte-identical to the previous snapshot.
+    conn.execute(
+        "UPDATE sessions SET model_config = ? WHERE id = 'tip-b'",
+        ('{"_reset_from":"root-a"}',),
+    )
+    conn.commit()
+    assert gw._cheap_change_fingerprint(db) != watcher._last_cheap_fp
+
+    assert watcher._poll_once(now=2.0) is True
+    event = subscriber.get_nowait()
+    rows = {row["session_id"]: row for row in event["sessions"]}
+    assert set(rows) == {"tip-b"}
+    assert rows["tip-b"]["title"] == "Beta"
+    assert rows["tip-b"]["message_count"] == initial_rows["tip-b"]["message_count"]
+    assert rows["tip-b"]["updated_at"] == initial_rows["tip-b"]["updated_at"]
+    assert watcher._last_sessions == event["sessions"]
+    assert subscriber.empty()
+
+    # A title-only projection change on an otherwise identical row is also
+    # delivered on the ordinary poll.
+    conn.execute("UPDATE sessions SET title = 'Beta renamed' WHERE id = 'tip-b'")
+    conn.commit()
+    assert watcher._poll_once(now=3.0) is True
+    event = subscriber.get_nowait()
+    rows = {row["session_id"]: row for row in event["sessions"]}
+    assert set(rows) == {"tip-b"}
+    assert rows["tip-b"]["title"] == "Beta renamed"
+    assert rows["tip-b"]["updated_at"] == initial_rows["tip-b"]["updated_at"]
+    assert watcher._last_sessions == event["sessions"]
+    assert subscriber.empty()
+
+    # Unchanged payload: the parity projection reruns but emits nothing.
+    assert watcher._poll_once(now=3.0 + watcher.PROJECTION_PARITY_INTERVAL) is True
+    assert subscriber.empty()
+
+
 def test_initial_missing_db_does_not_publish_an_empty_snapshot(tmp_path):
     gw = importlib.import_module("api.gateway_watcher")
     watcher = gw.GatewayWatcher(state_db_path=tmp_path / "missing.db")

tests/test_session_lineage_metadata_api.py

@@ -954,8 +954,9 @@ def test_continuation_classification_timestamp_tolerance_and_guards():
             model_config=json.dumps({'_delegate_from': 'some-other-session'}),
         ),
     )
-    # Unparsable model_config degrades to no markers (no crash, no match).
-    assert _is_continuation_session(
+    # Unparsable model_config is untrusted lineage identity: it fails closed
+    # as a boundary (no crash), even inside the tolerance window.
+    assert not _is_continuation_session(
         make_parent(),
         make_child(started_at=999.95, model_config='not-json'),
     )
@@ -1148,3 +1149,197 @@ def test_state_db_stitch_old_schema_without_identity_columns_still_stitches(_iso
         assert "parent turn" in contents
     finally:
         conn.close()
+
+
+def test_branch_markers_report_fail_closed_unknown_state():
+    """Untrusted model_config lineage evidence is 'unknown', never 'no markers'."""
+    from api.agent_sessions import _branch_markers
+
+    assert _branch_markers(None) == ('none', {})
+    assert _branch_markers({'model_config': None}) == ('none', {})
+    assert _branch_markers({'model_config': ''}) == ('none', {})
+    assert _branch_markers({'model_config': '   '}) == ('none', {})
+    assert _branch_markers({'model_config': json.dumps({'model': 'x'})}) == ('none', {})
+    assert _branch_markers({'model_config': json.dumps({'_reset_from': None})}) == ('none', {})
+    assert _branch_markers(
+        {'model_config': json.dumps({'_reset_from': 'p', '_delegate_from': 'a'})}
+    ) == ('markers', {'_reset_from': 'p', '_delegate_from': 'a'})
+    assert _branch_markers({'model_config': {'_branched_from': ' p '}}) == (
+        'markers',
+        {'_branched_from': 'p'},
+    )
+
+    for raw in ('{not-json', '[]', '"a string"', '17', 'null', '{"a":'):
+        assert _branch_markers({'model_config': raw}) == ('unknown', {}), raw
+    assert _branch_markers({'model_config': ['_delegate_from']}) == ('unknown', {})
+    for hostile in (
+        {'_delegate_from': 123},
+        {'_delegate_from': ['p']},
+        {'_delegate_from': {'id': 'p'}},
+        {'_branched_from': True},
+        {'_branched_from': '   '},
+        {'_reset_from': ''},
+        # A usable first marker never hides a malformed second one.
+        {'_delegate_from': 'ancestor', '_branched_from': 123},
+    ):
+        assert _branch_markers({'model_config': json.dumps(hostile)}) == ('unknown', {}), hostile
+
+    # A valid but pathologically deep payload overflows the decoder with
+    # RecursionError (not ValueError); it must degrade to 'unknown', not crash.
+    depth = 12_000
+    deep_payload = '[' * depth + ']' * depth
+    with pytest.raises(RecursionError):
+        json.loads(deep_payload)
+    assert _branch_markers({'model_config': deep_payload}) == ('unknown', {})
+
+
+def test_untrusted_model_config_markers_fail_closed_inside_tolerance():
+    """Master tolerance/predicate shape, but unknown identity is a boundary."""
+    from api.agent_sessions import _is_continuation_session
+
+    parent = {'id': 'fc_parent', 'source': 'webui', 'ended_at': 200.0, 'end_reason': 'compression'}
+    child = {'id': 'fc_child', 'source': 'webui', 'started_at': 199.7, 'parent_session_id': 'fc_parent'}
+
+    # Containment: marker-free and null-marker children still continue.
+    assert _is_continuation_session(parent, child)
+    assert _is_continuation_session(
+        parent, {**child, 'model_config': json.dumps({'_delegate_from': None})}
+    )
+    # Master's shape is kept: a marker naming another session (inherited from
+    # the rotated agent's model_config) does not split a real continuation.
+    assert _is_continuation_session(
+        parent, {**child, 'model_config': json.dumps({'_delegate_from': 'somewhere_else'})}
+    )
+
+    for raw in (
+        '{not-json',
+        '[]',
+        json.dumps({'_delegate_from': 123}),
+        json.dumps({'_branched_from': ''}),
+        json.dumps({'_reset_from': ['fc_parent']}),
+        json.dumps({'_delegate_from': 'somewhere_else', '_branched_from': 123}),
+        '[' * 12_000 + ']' * 12_000,
+    ):
+        assert not _is_continuation_session(parent, {**child, 'model_config': raw}), raw[:40]
+
+    # Legacy rows without ended_at keep continuing, unless identity is untrusted.
+    legacy_parent = {**parent, 'ended_at': None}
+    assert _is_continuation_session(legacy_parent, child)
+    assert not _is_continuation_session(legacy_parent, {**child, 'model_config': '{not-json'})
+
+
+def test_reset_from_marker_naming_parent_is_a_boundary():
+    """Hermes Agent binds ``_reset_from`` like ``_branched_from``/``_delegate_from``.
+
+    ``gateway/session_recovery.py`` stamps ``model_config._reset_from`` on a
+    reset child and ``_NON_CONTINUATION_CHILD_FILTER_SQL`` rejects it as a
+    compression continuation. The WebUI predicate must agree inside the
+    tolerance window and on legacy rows without ``ended_at``.
+    """
+    from api.agent_sessions import _is_continuation_session
+
+    for end_reason in ('compression', 'cli_close'):
+        parent = {'id': 'reset_parent', 'source': 'telegram', 'ended_at': 300.0, 'end_reason': end_reason}
+        child = {'id': 'reset_child', 'source': 'telegram', 'started_at': 299.5, 'parent_session_id': 'reset_parent'}
+        assert _is_continuation_session(parent, child), end_reason
+        assert not _is_continuation_session(
+            parent, {**child, 'model_config': json.dumps({'_reset_from': 'reset_parent'})}
+        ), end_reason
+        assert not _is_continuation_session(
+            parent, {**child, 'model_config': {'max_iterations': 40, '_reset_from': 'reset_parent'}}
+        ), end_reason
+        assert not _is_continuation_session(
+            {**parent, 'ended_at': None},
+            {**child, 'model_config': json.dumps({'_reset_from': 'reset_parent'})},
+        ), end_reason
+        # A direct reset marker is not masked by an inherited delegate marker.
+        assert not _is_continuation_session(
+            parent,
+            {
+                **child,
+                'model_config': json.dumps(
+                    {'_delegate_from': 'ancestor', '_reset_from': 'reset_parent'}
+                ),
+            },
+        ), end_reason
+        # An inherited reset marker naming an ancestor does not split the lineage.
+        assert _is_continuation_session(
+            parent, {**child, 'model_config': json.dumps({'_reset_from': 'ancestor'})}
+        ), end_reason
+
+
+def test_reset_child_inside_tolerance_stays_independent_across_readers(_isolate):
+    """Sidebar listing, lineage metadata/report and transcript stitcher agree.
+
+    A resumed parent closed by compression within the tolerance window of its
+    reset child's creation must not absorb that reset conversation.
+    """
+    from api.agent_sessions import (
+        read_importable_agent_session_rows,
+        read_session_lineage_metadata,
+        read_session_lineage_report,
+    )
+
+    conn = _ensure_state_db(_isolate)
+    _ensure_messages_table(conn)
+    t0 = time.time() - 100
+    parent_id = 'reset_reader_parent'
+    child_id = 'reset_reader_child'
+    try:
+        _insert_state_row(
+            conn,
+            parent_id,
+            title='Reset parent conversation',
+            started_at=t0,
+            ended_at=t0 + 5,
+            end_reason='compression',
+        )
+        _insert_state_message(conn, parent_id, role='user', content='parent turn', timestamp=t0 + 1)
+        _insert_state_row(
+            conn,
+            child_id,
+            title='Reset child conversation',
+            parent=parent_id,
+            started_at=t0 + 5 - 0.5,
+            model_config=json.dumps({'_reset_from': parent_id}),
+        )
+        _insert_state_message(conn, child_id, role='user', content='reset turn', timestamp=t0 + 6)
+
+        projected = {
+            row['id']: row
+            for row in read_importable_agent_session_rows(
+                _isolate, limit=None, exclude_sources=None
+            )
+        }
+        assert {parent_id, child_id} <= set(projected)
+        assert projected[child_id]['title'] == 'Reset child conversation'
+        assert projected[child_id]['relationship_type'] == 'child_session'
+        assert '_lineage_root_id' not in projected[child_id]
+        assert '_compression_segment_count' not in projected[parent_id]
+        assert 'model_config' not in projected[child_id]
+
+        metadata = read_session_lineage_metadata(_isolate, {parent_id, child_id})
+        assert metadata[child_id]['relationship_type'] == 'child_session'
+        assert '_lineage_root_id' not in metadata[child_id]
+
+        report = read_session_lineage_report(_isolate, child_id)
+        assert report['tip_session_id'] == child_id
+        assert report['total_segments'] == 1
+
+        stitched = models.get_state_db_session_messages(child_id, stitch_continuations=True)
+        assert [m['content'] for m in stitched] == ['reset turn']
+
+        # Control: the same row without the reset marker is a continuation.
+        conn.execute('UPDATE sessions SET model_config = NULL WHERE id = ?', (child_id,))
+        conn.commit()
+        stitched = models.get_state_db_session_messages(child_id, stitch_continuations=True)
+        assert [m['content'] for m in stitched] == ['parent turn', 'reset turn']
+        projected_ids = {
+            row['id']
+            for row in read_importable_agent_session_rows(
+                _isolate, limit=None, exclude_sources=None
+            )
+        }
+        assert parent_id not in projected_ids and child_id in projected_ids
+    finally:
+        conn.close()