45 lines
2.0 KiB
Python
45 lines
2.0 KiB
Python
"""The Agent's inline recall path must select project scope itself."""
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
from agent.inline_tool_executors import InlineToolContext, _session_search
|
|
|
|
|
|
def test_inline_recall_binds_scope_and_keeps_profile_for_denial(monkeypatch):
|
|
from tools import session_search_tool
|
|
|
|
received = {}
|
|
def capture(**kw):
|
|
received.update(kw)
|
|
return json.dumps({'success': True})
|
|
monkeypatch.setattr(session_search_tool, 'session_search', capture)
|
|
db = object()
|
|
agent = SimpleNamespace(session_id='real_caller', _get_session_db_for_recall=lambda: db)
|
|
_session_search(agent, {'query':'history', 'profile':'other', 'scope':'legacy_profile'},
|
|
InlineToolContext(effective_task_id='task'))
|
|
assert received['scope'] == 'current_project'
|
|
assert received['current_session_id'] == 'real_caller'
|
|
assert received['profile'] == 'other'
|
|
assert received['db'] is db
|
|
|
|
|
|
def test_memory_provider_dispatch_receives_trusted_caller(monkeypatch):
|
|
from agent.inline_tool_executors import resolve_invoke_tool_executor
|
|
from agent import tool_executor
|
|
|
|
received = []
|
|
manager = SimpleNamespace(
|
|
has_tool=lambda name: name == 'fact_store',
|
|
handle_tool_call=lambda name, args, **kw: received.append(kw) or '{}')
|
|
agent = SimpleNamespace(session_id='current_identity', _memory_manager=manager,
|
|
_context_engine_tool_names=set())
|
|
args = {'action': 'list', 'session_id': 'forged_identity'}
|
|
execute = resolve_invoke_tool_executor(agent, 'fact_store')
|
|
execute(agent, args, InlineToolContext(effective_task_id='task'))
|
|
assert received[-1].get('session_id') == 'current_identity'
|
|
monkeypatch.setattr(tool_executor, '_start_quiet_tool_spinner', lambda *a, **kw: None)
|
|
ref = SimpleNamespace(name='fact_store', args=args, task_id='task', call_id='call', trace=[])
|
|
dispatch = tool_executor._resolve_sequential_dispatch(agent, ref, [])
|
|
dispatch.execute(args)
|
|
assert received[-1].get('session_id') == 'current_identity'
|