Merge PR #6565 into stage/0924-g
This commit is contained in:
commit
30f12dc526
6 files changed
+694
-40
No files matched your search
@@ -278,6 +278,26 @@ visibility stage decides whether recovered background rows are shown. In
|
||||
are merged; cross-profile scoping, visibility, deduplication, and final route
|
||||
limits remain downstream responsibilities.
|
||||
|
||||
#### Compression lineage and session-list invalidation
|
||||
|
||||
`api.agent_sessions._is_continuation_session()` is the shared classifier for
|
||||
sidebar projection, lineage metadata/reporting, and `state.db` transcript
|
||||
stitching. It uses the direct parent link, no conflicting non-empty source,
|
||||
`compression` or `cli_close` parent end reason, and the existing two-second
|
||||
`started_at` overlap allowance. A `source="tool"` child is always a separate
|
||||
conversation, even when its parent is also a tool session or has no source.
|
||||
A direct `_branched_from`, `_delegate_from`, or `_reset_from` marker in the
|
||||
child's `model_config` also makes a boundary; inherited ancestor markers do
|
||||
not. Malformed or unverifiable marker evidence fails closed as a boundary.
|
||||
The overlap allowance is the existing master policy, not a new window set by
|
||||
this change.
|
||||
|
||||
The gateway watcher's cheap database fingerprint includes `model_config`, so
|
||||
a marker-only update causes a fresh projection. Its published-payload hash
|
||||
covers every emitted session field, not just ID, activity time and message
|
||||
count; a changed projected title or lineage field can therefore emit
|
||||
`sessions_changed` even without message-row churn.
|
||||
|
||||
### 4.3 SSE Streaming Engine
|
||||
|
||||
This is the most architecturally interesting part. Two endpoints cooperate:
|
||||
|
||||
+65
-28
@@ -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,30 +436,36 @@ 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()
|
||||
# Agent lineage excludes tool children even when the parent has the same
|
||||
# source (or no source); a timestamp overlap cannot override that boundary.
|
||||
if child_source == 'tool':
|
||||
return False
|
||||
if parent_source and child_source and parent_source != child_source:
|
||||
return False
|
||||
if parent.get('end_reason') not in {'compression', 'cli_close'}:
|
||||
|
||||
+26
-10
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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'),
|
||||
)
|
||||
@@ -980,6 +981,58 @@ def test_continuation_classification_timestamp_tolerance_and_guards():
|
||||
assert not _is_continuation_session(make_parent(), make_child(started_at='not-a-number'))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parent_source", ["tool", ""])
|
||||
def test_tool_child_is_not_stitched_into_parent_lineage(_isolate, parent_source):
|
||||
"""A marker-less tool child stays independent even with a matching/empty source."""
|
||||
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, child_id = "tool_boundary_parent", "tool_boundary_child"
|
||||
try:
|
||||
_insert_state_row(
|
||||
conn, parent_id, source=parent_source, title="Parent conversation",
|
||||
started_at=t0, ended_at=t0 + 5, end_reason="compression",
|
||||
)
|
||||
_insert_state_row(
|
||||
conn, child_id, source="tool", title="Tool conversation",
|
||||
parent=parent_id, started_at=t0 + 4.5,
|
||||
)
|
||||
_insert_state_message(
|
||||
conn, parent_id, role="user", content="parent request", timestamp=t0 + 1,
|
||||
)
|
||||
_insert_state_message(
|
||||
conn, child_id, role="user", content="tool request", timestamp=t0 + 6,
|
||||
)
|
||||
|
||||
projected = read_importable_agent_session_rows(
|
||||
_isolate, limit=None, exclude_sources=None,
|
||||
)
|
||||
rows = {row["id"]: row for row in projected}
|
||||
assert set(rows) == {parent_id, child_id}
|
||||
assert rows[child_id]["relationship_type"] == "child_session"
|
||||
assert "_lineage_root_id" not in rows[child_id]
|
||||
assert "_lineage_tip_id" not in rows[parent_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]
|
||||
assert read_session_lineage_report(_isolate, parent_id)["total_segments"] == 1
|
||||
assert read_session_lineage_report(_isolate, child_id)["total_segments"] == 1
|
||||
|
||||
messages = models.get_state_db_session_messages(
|
||||
child_id, stitch_continuations=True,
|
||||
)
|
||||
assert [message["content"] for message in messages] == ["tool request"]
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_state_db_stitch_keeps_branched_child_out_of_parent_transcript(_isolate):
|
||||
"""#7021 r2: the open/import transcript stitcher (get_state_db_session_messages
|
||||
with stitch_continuations=True) must apply the same model_config branch-marker
|
||||
@@ -1148,3 +1201,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()
|
||||
Reference in new issue
Block a user