288 lines
14 KiB
Python
288 lines
14 KiB
Python
"""Project-scoped session_search must fence every public read shape."""
|
|
import json
|
|
import hashlib
|
|
import inspect
|
|
from types import SimpleNamespace
|
|
|
|
from hermes_state import SessionDB
|
|
from tools.session_search_tool import session_search
|
|
|
|
|
|
def test_discovery_continues_past_filtered_candidates(tmp_path, monkeypatch):
|
|
from tools.registry import registry
|
|
|
|
root = tmp_path / 'project'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'noise', 'own', 'title']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
db._conn.execute("UPDATE sessions SET title='paginationneedle' WHERE id='title'")
|
|
db._conn.commit()
|
|
db.append_message('title', role='user', content='title source')
|
|
for _ in range(1100):
|
|
db.append_message('noise', role='user', content='[System: paginationneedle]')
|
|
target = db.append_message('own', role='user', content='paginationneedle visible')
|
|
calls = []
|
|
original = db.search_messages
|
|
def capture(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
return original(*args, **kwargs)
|
|
monkeypatch.setattr(db, 'search_messages', capture)
|
|
args = {'query': 'paginationneedle', 'sort': 'oldest', 'limit': 3,
|
|
'scope': 'legacy_profile'}
|
|
first = json.loads(registry.dispatch('session_search', args, db=db, session_id='current'))
|
|
assert first['truncated']
|
|
assert first.get('next_search_cursor'), 'filtered candidates need a continuation'
|
|
assert 300 < sum(c['limit'] for c in calls) <= 1000
|
|
assert all(c['exclude_sources'] and c['role_filter'] == ['user', 'assistant'] for c in calls)
|
|
cursor = first['next_search_cursor']
|
|
second = json.loads(registry.dispatch('session_search', {**args, 'search_cursor': cursor},
|
|
db=db, session_id='current'))
|
|
assert any(r['match_message_id'] == target for r in second['results'])
|
|
assert not any(r['session_id'] == 'title' for r in second['results'])
|
|
assert second['count'] == len(second['results'])
|
|
bad = json.loads(registry.dispatch('session_search', {**args, 'query': 'different',
|
|
'search_cursor': cursor}, db=db, session_id='current'))
|
|
assert bad['status'] == 'invalid_cursor'
|
|
db._conn.execute("UPDATE sessions SET archived=1 WHERE id='own'")
|
|
db._conn.commit()
|
|
stale = json.loads(registry.dispatch('session_search', {**args, 'search_cursor': cursor},
|
|
db=db, session_id='current'))
|
|
assert stale['status'] == 'invalid_cursor'
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_long_text_anchor_slices_and_safe_hit_snippet(tmp_path):
|
|
from agent.inline_tool_executors import InlineToolContext, _session_search
|
|
from tools.session_search_project import ProjectRecall
|
|
from tools.session_search_tool import SESSION_SEARCH_SCHEMA
|
|
|
|
root = tmp_path / 'project'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'own']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
text = 'prefix ' * 900 + 'lateanchorneedle original ending'
|
|
mid = db.append_message('own', role='user', content=[
|
|
{'type': 'text', 'text': text},
|
|
{'type': 'image_url', 'image_url': {'url': 'PRIVATE_BINARY'}}])
|
|
args = dict(db=db, current_session_id='current', scope='current_project')
|
|
discovered = json.loads(session_search(query='lateanchorneedle', **args))
|
|
assert 'lateanchorneedle' in discovered['results'][0]['snippet']
|
|
assert 'PRIVATE_BINARY' not in json.dumps(discovered)
|
|
agent = SimpleNamespace(session_id='current', _get_session_db_for_recall=lambda: db)
|
|
offset = text.index('lateanchorneedle')
|
|
result = json.loads(_session_search(agent, dict(session_id='own', around_message_id=mid,
|
|
content_offset=offset, content_length=12, scope='legacy_profile'),
|
|
InlineToolContext(effective_task_id='task')))
|
|
anchor = next(m for m in result['messages'] if m['id'] == mid)
|
|
assert anchor['content'] == text[offset:offset + 12]
|
|
assert anchor['content_offset'] == offset
|
|
assert anchor['next_content_offset'] == offset + 12
|
|
# Existing text projection joins all parts, including the empty image projection.
|
|
assert anchor['content_hash'] == hashlib.sha256((text + '\n').encode()).hexdigest()
|
|
assert anchor['content_hash'] == discovered['results'][0]['messages'][0]['content_hash']
|
|
recall = ProjectRecall(db, 'current')
|
|
rows = recall.messages('own', mid, content_offset=offset, content_length=12)
|
|
db._conn.execute('UPDATE messages SET content=? WHERE id=?', (text + ' changed', mid))
|
|
db._conn.commit()
|
|
assert recall.response(messages=rows)['status'] == 'source_revoked'
|
|
assert json.loads(session_search(role_filter='tool', **args))['status'] == 'unsupported'
|
|
properties = SESSION_SEARCH_SCHEMA['parameters']['properties']
|
|
assert 'scope' not in properties
|
|
for name in ['content_offset', 'content_length', 'search_cursor', 'before_message_id']:
|
|
assert name in properties
|
|
assert inspect.signature(session_search).parameters[name].kind == inspect.Parameter.KEYWORD_ONLY
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_scroll_fills_safe_neighbors_and_continues_scan_frontiers(tmp_path):
|
|
from agent.inline_tool_executors import InlineToolContext, _session_search
|
|
from tools.registry import registry
|
|
|
|
root = tmp_path / 'project'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'own']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
left = db.append_message('own', role='user', content='left safe')
|
|
for _ in range(1100):
|
|
db.append_message('own', role='user', content='[System: hidden before]')
|
|
anchor = db.append_message('own', role='user', content='center safe')
|
|
for _ in range(1100):
|
|
db.append_message('own', role='user', content='[System: hidden after]')
|
|
right = db.append_message('own', role='user', content='right safe')
|
|
args = dict(session_id='own', around_message_id=anchor, window=1)
|
|
first = json.loads(registry.dispatch('session_search', args, db=db, session_id='current'))
|
|
assert first['truncated'], 'budget exhaustion must be visible'
|
|
assert left < first['next_before_message_id'] < anchor
|
|
assert anchor < first['next_after_message_id'] < right
|
|
agent = SimpleNamespace(session_id='current', _get_session_db_for_recall=lambda: db)
|
|
second = json.loads(_session_search(agent, {**args,
|
|
'before_message_id': first['next_before_message_id'],
|
|
'after_message_id': first['next_after_message_id']},
|
|
InlineToolContext(effective_task_id='task')))
|
|
assert [m['id'] for m in second['messages']] == [left, anchor, right]
|
|
assert not second['truncated']
|
|
assert second['next_before_message_id'] is None
|
|
assert second['next_after_message_id'] is None
|
|
read = json.loads(session_search(db=db, current_session_id='current', scope='current_project',
|
|
session_id='own', after_message_id=left))
|
|
assert read['truncated'] and read['next_after_message_id'] > left
|
|
continued = json.loads(session_search(db=db, current_session_id='current', scope='current_project',
|
|
session_id='own', after_message_id=read['next_after_message_id']))
|
|
assert any(m['id'] == anchor for m in continued['messages'])
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_project_read_shapes_do_not_fall_back_to_other_workspaces(tmp_path):
|
|
p = tmp_path / 'p'
|
|
q = tmp_path / 'q'
|
|
p.mkdir()
|
|
q.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid, cwd in [('current', p), ('own', p), ('foreign', q)]:
|
|
db.create_session(sid, source='desktop', cwd=str(cwd))
|
|
own = db.append_message('own', role='user', content='recall_canary own history')
|
|
other = db.append_message('foreign', role='user', content='recall_canary FOREIGN_PRIVATE')
|
|
for args in [dict(query='recall_canary'), {}, dict(session_id='foreign'),
|
|
dict(session_id='foreign', around_message_id=other)]:
|
|
out = session_search(db=db, current_session_id='current', scope='current_project', **args)
|
|
assert 'FOREIGN_PRIVATE' not in out
|
|
result = json.loads(out)
|
|
if args.get('session_id'):
|
|
assert not result['success']
|
|
own_read = json.loads(session_search(db=db, current_session_id='current',
|
|
scope='current_project', session_id='own'))
|
|
assert own_read['success']
|
|
assert any(m['id'] == own for m in own_read['messages'])
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_registry_preserves_current_session_identity(monkeypatch):
|
|
from tools import session_search_tool as tool
|
|
from tools.registry import registry
|
|
|
|
seen = {}
|
|
def capture(**kwargs):
|
|
seen.update(kwargs)
|
|
return json.dumps({'success': True})
|
|
monkeypatch.setattr(tool, 'session_search', capture)
|
|
registry.dispatch('session_search', {'query': 'history'}, session_id='caller')
|
|
assert seen['current_session_id'] == 'caller'
|
|
assert seen['scope'] == 'current_project'
|
|
|
|
|
|
def test_project_scroll_excludes_rewind_and_keeps_compaction(tmp_path):
|
|
root = tmp_path / 'project'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'own']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
anchor = db.append_message('own', role='user', content='retained_original')
|
|
removed = db.append_message('own', role='assistant', content='REWOUND_PRIVATE')
|
|
db.append_message('own', role='user', content='following_message')
|
|
db._conn.execute('UPDATE messages SET active=0, compacted=1 WHERE id=?', (anchor,))
|
|
db._conn.execute('UPDATE messages SET active=0, compacted=0 WHERE id=?', (removed,))
|
|
db._conn.commit()
|
|
out = session_search(db=db, current_session_id='current', scope='current_project',
|
|
session_id='own', around_message_id=anchor)
|
|
assert 'retained_original' in out
|
|
assert 'following_message' in out
|
|
assert 'REWOUND_PRIVATE' not in out
|
|
denied = json.loads(session_search(db=db, current_session_id='current', scope='current_project',
|
|
session_id='own', profile='foreign'))
|
|
assert denied['status'] == 'access_denied'
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_revoked_source_is_rechecked_before_response(tmp_path):
|
|
from tools.session_search_project import ProjectRecall
|
|
|
|
root = tmp_path / 'p'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'own']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
mid = db.append_message('own', role='user', content='old_source_value')
|
|
recall = ProjectRecall(db, 'current')
|
|
rows = recall.messages('own', mid)
|
|
db._conn.execute('UPDATE messages SET content=? WHERE id=?', ('new_source_value', mid))
|
|
db._conn.commit()
|
|
result = recall.response(messages=rows)
|
|
assert result['status'] == 'source_revoked'
|
|
assert 'old_source_value' not in json.dumps(result)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_project_read_skips_control_messages_and_projects_multimodal_text(tmp_path):
|
|
root = tmp_path / 'p'
|
|
root.mkdir()
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
for sid in ['current', 'own']:
|
|
db.create_session(sid, source='desktop', cwd=str(root))
|
|
for _ in range(40):
|
|
db.append_message('own', role='user', content='[System: INTERNAL_CONTROL]')
|
|
db.append_message('own', role='user', content='AUTO_CONTROL', display_kind='auto_continue')
|
|
db.append_message('own', role='user', content=[
|
|
{'type':'image_url','image_url':{'url':'data:image/png;base64,PRIVATE_BINARY'}},
|
|
{'type':'text','text':'visible_multimodal_text'}])
|
|
for i in range(40):
|
|
db.append_message('own', role='user', content=f'visible_history_{i}')
|
|
result = json.loads(session_search(db=db, current_session_id='current', scope='current_project', session_id='own'))
|
|
text = json.dumps(result)
|
|
assert 'INTERNAL_CONTROL' not in text
|
|
assert 'AUTO_CONTROL' not in text
|
|
assert 'PRIVATE_BINARY' not in text
|
|
assert 'visible_multimodal_text' in text
|
|
assert result['truncated']
|
|
assert result['next_after_message_id']
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_home_reads_only_its_own_compacted_history(tmp_path):
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
db.create_session('home', source='desktop')
|
|
db.create_session('other_home', source='desktop')
|
|
old = db.append_message('home', role='user', content='HOME_ARCHIVED')
|
|
db.append_message('home', role='user', content='HOME_LIVE')
|
|
db.append_message('other_home', role='user', content='OTHER_HOME_SECRET')
|
|
db._conn.execute('UPDATE messages SET active=0,compacted=1 WHERE id=?', (old,))
|
|
db._conn.commit()
|
|
result = session_search(db=db, current_session_id='home', scope='current_project',
|
|
session_id='home', around_message_id=old)
|
|
assert json.loads(result)['success']
|
|
assert 'HOME_ARCHIVED' in result
|
|
assert 'HOME_LIVE' not in result
|
|
assert 'OTHER_HOME_SECRET' not in result
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_unresolved_project_cannot_silently_browse_profile(tmp_path):
|
|
db = SessionDB(tmp_path / 'state.db')
|
|
try:
|
|
db.create_session('home', source='desktop')
|
|
db.create_session('other', source='desktop', cwd=str(tmp_path))
|
|
db.append_message('other', role='user', content='PRIVATE_PROFILE_CONTENT')
|
|
result = json.loads(session_search(db=db, current_session_id='home', scope='current_project'))
|
|
assert result['status'] == 'scope_unresolved'
|
|
assert 'PRIVATE_PROFILE_CONTENT' not in json.dumps(result)
|
|
finally:
|
|
db.close()
|