|
|
@@ -5,6 +5,7 @@ import tempfile
|
|
|
import threading
|
|
|
import time
|
|
|
import unittest
|
|
|
+from contextvars import ContextVar
|
|
|
from pathlib import Path
|
|
|
from types import SimpleNamespace
|
|
|
from unittest.mock import patch
|
|
|
@@ -24,6 +25,7 @@ class ComputerPolicyTests(unittest.TestCase):
|
|
|
def setUp(self):
|
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
|
self.home = Path(self.temp.name)
|
|
|
+ self.profile_scope = ContextVar('fixture_home', default=self.home)
|
|
|
self.agent = ReviewAgent(session_id='stored', enabled_toolsets=['all'], review_calls=[],
|
|
|
skip_context_files=False, load_soul_identity=False,
|
|
|
_background_review_agent=None, _background_review_run=None)
|
|
|
@@ -36,7 +38,11 @@ class ComputerPolicyTests(unittest.TestCase):
|
|
|
computer_policy=dict(self.session['_yaoyao_computer_policy']))
|
|
|
bridge._bindings.clear(); bridge._generations.clear(); bridge._children.clear()
|
|
|
bridge._bindings['runtime'] = self.binding
|
|
|
- self.patches = [patch.dict(sys.modules, {'agent.redact': SimpleNamespace(redact_sensitive_text=lambda text, **kw: text)}),
|
|
|
+ self.patches = [patch.dict(sys.modules, {
|
|
|
+ 'agent.redact': SimpleNamespace(redact_sensitive_text=lambda text, **kw: text),
|
|
|
+ 'hermes_constants': SimpleNamespace(get_hermes_home=self.profile_scope.get,
|
|
|
+ set_hermes_home_override=self.profile_scope.set, reset_hermes_home_override=self.profile_scope.reset),
|
|
|
+ }),
|
|
|
patch.object(bridge, '_server', return_value=self.server),
|
|
|
patch.object(bridge, '_check_session_mode'),
|
|
|
patch.object(bridge, '_transport_identity', return_value=self.binding.owner),
|
|
|
@@ -47,6 +53,46 @@ class ComputerPolicyTests(unittest.TestCase):
|
|
|
for p in reversed(self.patches): p.stop()
|
|
|
self.temp.cleanup()
|
|
|
|
|
|
+ def test_native_catalog_uses_binding_profile_and_restores_request_context(self):
|
|
|
+ current = ContextVar('fixture_profile', default=self.home)
|
|
|
+ constants = SimpleNamespace(set_hermes_home_override=current.set, reset_hermes_home_override=current.reset)
|
|
|
+ named = self.home / 'profiles' / 'yaoer'
|
|
|
+ self.binding.profile_home = named
|
|
|
+ self.binding.computer_policy = None
|
|
|
+ self.binding.toolset = 'yaoyao_turn_current'
|
|
|
+ other_turn = 'yaoyao_turn_other'
|
|
|
+ scopes = []
|
|
|
+
|
|
|
+ def definitions(enabled_toolsets, **kwargs):
|
|
|
+ scopes.append(current.get())
|
|
|
+ names = ['named_skill', self.binding.toolset, other_turn] if current.get() == named else ['default_skill']
|
|
|
+ return [{'function': {'name': name}} for name in names
|
|
|
+ if enabled_toolsets == ['all'] or name in enabled_toolsets]
|
|
|
+
|
|
|
+ model_tools = SimpleNamespace(get_tool_definitions=definitions, get_toolset_for_tool=lambda name: name)
|
|
|
+ with patch.dict(sys.modules, {'hermes_constants': constants, 'model_tools': model_tools}):
|
|
|
+ bridge._refresh_agent_tools(self.agent, self.binding)
|
|
|
+ self.assertEqual(scopes, [named, named])
|
|
|
+ self.assertEqual(self.agent.valid_tool_names, {'named_skill', self.binding.toolset})
|
|
|
+ self.assertNotIn(other_turn, self.agent.enabled_toolsets)
|
|
|
+ self.assertEqual(current.get(), self.home)
|
|
|
+
|
|
|
+ def test_catalog_failure_restores_profile_context(self):
|
|
|
+ current = ContextVar('fixture_profile_failure', default=self.home)
|
|
|
+ constants = SimpleNamespace(set_hermes_home_override=current.set, reset_hermes_home_override=current.reset)
|
|
|
+ self.binding.profile_home = self.home / 'profiles' / 'yaoer'
|
|
|
+ self.binding.computer_policy = None
|
|
|
+
|
|
|
+ def fail(**kwargs):
|
|
|
+ self.assertEqual(current.get(), self.binding.profile_home)
|
|
|
+ raise RuntimeError('fixture catalog failed')
|
|
|
+
|
|
|
+ model_tools = SimpleNamespace(get_tool_definitions=fail, get_toolset_for_tool=lambda name: name)
|
|
|
+ with patch.dict(sys.modules, {'hermes_constants': constants, 'model_tools': model_tools}):
|
|
|
+ with self.assertRaisesRegex(RuntimeError, 'fixture catalog failed'):
|
|
|
+ bridge._refresh_agent_tools(self.agent, self.binding)
|
|
|
+ self.assertEqual(current.get(), self.home)
|
|
|
+
|
|
|
def test_isolated_skills_and_vm_tools_use_native_catalog_but_host_operations_require_approval(self):
|
|
|
for name in ['skills_list', 'skill_view', 'vision_analyze', 'yaoyao_computer_shell_aabbcc']:
|
|
|
self.assertIsNone(bridge.computer_directive(session_id='stored', tool_name=name, args={}))
|