diff --git a/docs/review/box-architecture.md b/docs/review/box-architecture.md index 5cbae1566..64adf4627 100644 --- a/docs/review/box-architecture.md +++ b/docs/review/box-architecture.md @@ -582,6 +582,8 @@ volumes: > Core 的 SkillRepository 是 `langbot_plugin.skill_store.SkillStore` 的异步 Workspace 适配层。默认 `skills.root` 保持原 `data/box/skills/tenants/...` 布局，升级时无需移动已安装 Skill；旧 `box.local.skills_root` 仅作为在线升级 fallback，并将在下一大版本删除。Box 只消费 Core 下发的通用只读 mount。 +自定义 `skills.root` 若位于 Box 的 `allowed_mount_roots` 之外，Skill 仍可管理和读取，但脚本执行的只读挂载会被拒绝。执行这些 Skill 时，需将该路径加入允许的挂载根；分离部署还需让 Core 和 Box Runtime 在相同路径看见这份目录。 + ### Session scope Pipeline 与 Runner 配置不再暴露 sandbox session 模板。Host 将当前平台会话/事件 scope 规范化后哈希成固定长度的 `lb-box-`；相同 scope 稳定复用，不同 scope 隔离，缺少 identity 时拒绝执行。SDK/Python 与 scoped MCP bridge 的工具调用遵守同一规则。详见 [box-session-scope.md](./box-session-scope.md)。 diff --git a/src/langbot/pkg/agent/runner/orchestrator.py b/src/langbot/pkg/agent/runner/orchestrator.py index c2cf5f9cb..de6327a40 100644 --- a/src/langbot/pkg/agent/runner/orchestrator.py +++ b/src/langbot/pkg/agent/runner/orchestrator.py @@ -150,10 +150,15 @@ class AgentRunOrchestrator: resources=resources, ) + authorized_skill_names = [ + str(skill['skill_name']) for skill in resources.get('skills', []) if skill.get('skill_name') + ] + prepare_execution_query(execution_query, event, authorized_skill_names) + session_query_id = None if adapter_context: if execution_query is not None: - skill_loader.restore_activated_skills_from_state( + await skill_loader.restore_activated_skills_from_state( self.ap, execution_query, context.get('state', {}), @@ -164,10 +169,6 @@ class AgentRunOrchestrator: if 'params' in adapter_context: context['adapter']['extra']['params'] = adapter_context['params'] - authorized_skill_names = [ - str(skill['skill_name']) for skill in resources.get('skills', []) if skill.get('skill_name') - ] - prepare_execution_query(execution_query, event, authorized_skill_names) context['variables'] = { key: value for key, value in (execution_query.variables or {}).items() diff --git a/src/langbot/pkg/api/http/service/maintenance.py b/src/langbot/pkg/api/http/service/maintenance.py index cdaf32091..4fe3cc642 100644 --- a/src/langbot/pkg/api/http/service/maintenance.py +++ b/src/langbot/pkg/api/http/service/maintenance.py @@ -148,6 +148,10 @@ class MaintenanceService: scoped_storage_path = Path('data/storage') / self.ap.storage_mgr.scoped_prefix(context) roots = [('storage', scoped_storage_path)] + skill_repository = getattr(self.ap, 'skill_repository', None) + if skill_repository is not None: + roots.append(('skills', Path(skill_repository.storage_root(context)))) + sections = await asyncio.to_thread(self._collect_sections, roots) runtime_processes = await self._runtime_storage_processes( context, diff --git a/src/langbot/pkg/box/runner.py b/src/langbot/pkg/box/runner.py index fa1b8aab6..6be620f88 100644 --- a/src/langbot/pkg/box/runner.py +++ b/src/langbot/pkg/box/runner.py @@ -152,8 +152,6 @@ class RunnerBoxService: if set(request.options) - {'image'}: raise BoxValidationError('Only the Box image can be specified; resource limits are Host-owned') spec = {'session_id': session_id, **request.options} - if query is not None: - spec['extra_mounts'] = self.box.build_skill_extra_mounts(query) result = await self.box.create_session(context, spec) return self.public_session(result) @@ -174,7 +172,6 @@ class RunnerBoxService: return self.binding_info(current) fields = ('image', 'network', 'cpus', 'memory_mb', 'pids_limit', 'read_only_rootfs', 'persistent') spec = {key: item[key] for key in fields if key in item} - spec['extra_mounts'] = self.box.build_skill_extra_mounts(query) binding = RunBoxBinding(run_id, box_id, spec, run_id) object.__setattr__(query, '_box_binding', binding) return self.binding_info(binding) diff --git a/src/langbot/pkg/provider/tools/loaders/native.py b/src/langbot/pkg/provider/tools/loaders/native.py index 1c6d6e345..74b067320 100644 --- a/src/langbot/pkg/provider/tools/loaders/native.py +++ b/src/langbot/pkg/provider/tools/loaders/native.py @@ -352,7 +352,7 @@ class NativeToolLoader(loader.ToolLoader): # All exec calls (with or without skills) go through the same container # via execute_tool. Skills are mounted at /workspace/.skills/{name}/ - # via extra_mounts built by BoxService. + # via read-only mounts composed by the Core skill loader. result = await self.ap.box_service.execute_tool( parameters, query, diff --git a/src/langbot/pkg/provider/tools/loaders/skill.py b/src/langbot/pkg/provider/tools/loaders/skill.py index 9669b7982..7ddaab86d 100644 --- a/src/langbot/pkg/provider/tools/loaders/skill.py +++ b/src/langbot/pkg/provider/tools/loaders/skill.py @@ -73,6 +73,7 @@ def register_created_skill_visibility(query: pipeline_query.Query, skill_name: s if isinstance(bound_skills, list) and skill_name not in bound_skills: bound_skills.append(skill_name) + def build_execution_mounts(ap: app.Application, query: pipeline_query.Query) -> list[dict]: """Mount only immutable revisions pinned by this run's activations.""" @@ -217,20 +218,30 @@ async def restore_activated_skills( return restored -def restore_activated_skills_from_state( +async def restore_activated_skills_from_state( ap: app.Application, query: pipeline_query.Query, state: dict[str, dict[str, typing.Any]], ) -> list[str]: - """Restore persisted activated skill names into Query variables. - - The state value stores names only. Full skill metadata is rebuilt from the - current pipeline-visible skill cache so removed or unbound skills remain - unavailable to native exec/write/edit. - """ + """Restore the exact activated Skill revisions stored by the Host.""" conversation_state = state.get('conversation', {}) if isinstance(state, dict) else {} - skill_names = normalize_skill_names(conversation_state.get(ACTIVATED_SKILL_NAMES_STATE_KEY)) - return restore_activated_skills(ap, query, skill_names) + if not isinstance(conversation_state, dict): + raise ValueError('Activated Skill conversation state must be an object.') + if ACTIVATED_SKILL_NAMES_STATE_KEY not in conversation_state: + return [] + bindings = conversation_state[ACTIVATED_SKILL_NAMES_STATE_KEY] + if isinstance(bindings, list) and bindings and all(isinstance(item, str) for item in bindings): + logger = getattr(ap, 'logger', None) + if logger is not None: + workspace_uuid = getattr(query, 'workspace_uuid', None) or getattr( + getattr(query, '_execution_context', None), 'workspace_uuid', 'unknown' + ) + logger.warning( + f'Legacy name-only activated Skill state in Workspace {workspace_uuid}; ' + 'reactivate Skills to pin their revisions.' + ) + return [] + return await restore_activated_skills(ap, query, bindings) async def persist_activated_skill( @@ -238,10 +249,10 @@ async def persist_activated_skill( query: pipeline_query.Query, skill_name: str, ) -> None: - """Persist activated skill names into host-owned conversation state. + """Persist activated Skill revision bindings into Host-owned conversation state. - ``activate`` runs host-side. This writes the run's current activated skill - names to the conversation-scope ``host.activated_skills`` snapshot so a later + ``activate`` runs host-side. This writes the run's pinned Skill revisions + to the conversation-scope ``host.activated_skills`` snapshot so a later run can restore them via ``restore_activated_skills_from_state``. Host writes here and a runner ``state.updated`` to the same key follow last-write-wins. @@ -273,7 +284,7 @@ async def persist_activated_skill( await store.state_set( scope_key=conversation_scope_key, state_key=ACTIVATED_SKILL_NAMES_STATE_KEY, - value=get_activated_skill_names(query), + value=get_activated_skill_bindings(query), runner_id=str(session.get('runner_id', '') or ''), binding_identity=str(state_context.get('binding_identity', 'unknown') or 'unknown'), scope='conversation', diff --git a/src/langbot/pkg/provider/tools/toolmgr.py b/src/langbot/pkg/provider/tools/toolmgr.py index fe6f8dfd4..e3fb26519 100644 --- a/src/langbot/pkg/provider/tools/toolmgr.py +++ b/src/langbot/pkg/provider/tools/toolmgr.py @@ -376,7 +376,8 @@ class ToolManager: if source in {'builtin', 'native'}: return await self.native_tool_loader.get_tool(name) if source == 'skill': - return await self.skill_tool_loader.get_tool(name) + sandbox_available = await self._workspace_sandbox_available(context) + return await self.skill_tool_loader.get_tool(name, sandbox_available=sandbox_available) if source == 'plugin': if not source_id: return None @@ -512,11 +513,11 @@ class ToolManager: telemetry_source = 'native' exists = await loader.has_tool(name) elif source == 'skill': - if not sandbox_available: + if not sandbox_available and self.skill_tool_loader.is_sandbox_tool(name): raise ToolNotFoundError(name) loader = self.skill_tool_loader telemetry_source = 'skill' - exists = await loader.has_tool(name) + exists = await loader.has_tool(name, sandbox_available=sandbox_available) elif source == 'plugin' and source_id: loader = self.plugin_tool_loader telemetry_source = 'plugin' @@ -537,6 +538,9 @@ class ToolManager: if not exists: raise ToolNotFoundError(name) + if source == 'skill' and isinstance(getattr(query, 'variables', None), dict): + query.variables['_skill_execution_available'] = sandbox_available + async def invoke_selected_tool() -> typing.Any: if source == 'mcp': return await loader.invoke_tool( diff --git a/src/langbot/pkg/skill/repository.py b/src/langbot/pkg/skill/repository.py index 6aaaa0c95..0d1de38dc 100644 --- a/src/langbot/pkg/skill/repository.py +++ b/src/langbot/pkg/skill/repository.py @@ -103,6 +103,9 @@ class SkillRepository: def _workspace_root(self, namespace: str) -> str: return os.path.join(self._default_workspace(), 'tenants', namespace) + def storage_root(self, context: TenantContext) -> str: + return self._store.scoped(self._namespace(context)).root + async def _call(self, context: TenantContext, method_name: str, *args, **kwargs): execution_context = await self._validated_execution_context(context) namespace = self._namespace(execution_context) diff --git a/tests/integration_tests/box/test_cloud_box_admission_integration.py b/tests/integration_tests/box/test_cloud_box_admission_integration.py index 0d01bc1d9..4be9ccfc6 100644 --- a/tests/integration_tests/box/test_cloud_box_admission_integration.py +++ b/tests/integration_tests/box/test_cloud_box_admission_integration.py @@ -411,7 +411,39 @@ async def test_cloud_core_skills_mount_generically_and_do_not_require_box_entitl ) assert [skill['name'] for skill in await repository.list_skills(ineligible)] == ['docs-only'] with pytest.raises(EntitlementUnavailableError): - await service.execute_tool({'command': 'true'}, _query(ineligible, 92)) + await RunnerBoxService(service).acquire(ineligible, {'reuse_key': 'global'}, _query(ineligible, 92)) + finally: + server_task.cancel() + client_task.cancel() + await runtime.shutdown() + + +@pytest.mark.asyncio +async def test_activated_skill_mount_changes_recreate_shared_box_session(tmp_path): + service, runtime, backend, entitlements, server_task, client_task = await _stack(tmp_path) + context = _context('workspace-a') + entitlements.snapshots[context.workspace_uuid] = _snapshot(context.workspace_uuid) + try: + first_query = await _bound_query(service, context, 101) + await service.execute_tool({'command': 'echo before activation'}, first_query) + assert len(backend.started_specs) == 1 + + skill = await service.ap.skill_repository.create_skill( + context, {'name': 'demo', 'instructions': 'Run a script'} + ) + second_query = await _bound_query(service, context, 102) + skill_loader.register_activated_skill(second_query, skill) + await service.execute_tool( + {'command': 'echo with skill'}, + second_query, + read_only_mounts=skill_loader.build_execution_mounts(service.ap, second_query), + ) + assert len(backend.started_specs) == 2 + assert backend.stopped_sessions == [backend.started_specs[0].session_id] + + await service.execute_tool({'command': 'echo first query again'}, first_query) + assert len(backend.started_specs) == 3 + assert backend.stopped_sessions == [backend.started_specs[0].session_id] * 2 finally: server_task.cancel() client_task.cancel() diff --git a/tests/unit_tests/agent/test_orchestrator_integration.py b/tests/unit_tests/agent/test_orchestrator_integration.py index 1c5f2ea66..955e41ccd 100644 --- a/tests/unit_tests/agent/test_orchestrator_integration.py +++ b/tests/unit_tests/agent/test_orchestrator_integration.py @@ -1007,7 +1007,7 @@ class TestQueryEntrySessionQueryId: @pytest.mark.asyncio async def test_no_query_id_for_pure_event_first_flow(self, clean_agent_state): - """Pure event-first flow has query_id=None in session.""" + """Pure event-first flow restores authorized Skill revisions without a Query.""" from langbot.pkg.agent.runner.host_models import ( AgentEventEnvelope, AgentBinding, @@ -1018,6 +1018,8 @@ class TestQueryEntrySessionQueryId: ) from langbot_plugin.api.entities.builtin.runner.input import AgentInput from langbot_plugin.api.entities.builtin.runner.delivery import DeliveryContext + from langbot.pkg.agent.runner.persistent_state_store import get_persistent_state_store + from langbot.pkg.provider.tools.loaders.skill import ACTIVATED_SKILL_NAMES_STATE_KEY, ACTIVATED_SKILLS_KEY db_engine = clean_agent_state descriptor = make_descriptor() @@ -1030,6 +1032,10 @@ class TestQueryEntrySessionQueryId: ] ) ap = FakeApplication(plugin_connector, db_engine) + revision = 'sha256:' + '1' * 64 + ap.skill_repository = types.SimpleNamespace( + get_skill=AsyncMock(return_value={'name': 'demo', 'revision': revision}) + ) async def build_resource_context(execution_query): from langbot.pkg.provider.tools.loaders.mcp import ( @@ -1085,10 +1091,21 @@ class TestQueryEntrySessionQueryId: 'mcp-resource-agent-read-enabled': True, }, resource_policy=ResourcePolicy(), - state_policy=StatePolicy(enable_state=False, state_scopes=[]), + state_policy=StatePolicy(enable_state=True, state_scopes=['conversation']), delivery_policy=DeliveryPolicy(enable_streaming=True, enable_reply=True), ) + success, error = await get_persistent_state_store(db_engine).apply_update_from_event( + event.model_copy(update={'workspace_id': TEST_CONTEXT.workspace_uuid}), + binding, + descriptor, + 'conversation', + ACTIVATED_SKILL_NAMES_STATE_KEY, + [{'name': 'demo', 'revision': revision}], + None, + ) + assert success is True and error is None + messages = [ message async for message in orchestrator.run( @@ -1114,6 +1131,7 @@ class TestQueryEntrySessionQueryId: assert execution_query.message_event.type == event.event_type assert '_host_box_scope' not in execution_query.variables assert execution_query.variables['_pipeline_bound_skills'] == ['demo', 'hidden'] + assert execution_query.variables[ACTIVATED_SKILLS_KEY]['demo']['revision'] == revision assert execution_query.variables['_pipeline_mcp_resource_attachments'][0]['server_uuid'] == 'srv-1' assert execution_query.variables['_pipeline_mcp_resource_agent_read_enabled'] is True assert 'MCP resource context selected by LangBot host:' in plugin_connector.contexts[0]['input']['text'] @@ -1344,8 +1362,9 @@ class TestQueryEntryAdapterHostCapabilities: assert snapshot['conversation']['external.test_key'] == 'test_value' @pytest.mark.asyncio - async def test_run_from_query_restores_activated_skills_from_state(self, clean_agent_state): - """Persisted activated skill names are restored into the next Query run.""" + @pytest.mark.parametrize('legacy_state', [False, True]) + async def test_run_from_query_restores_activated_skills_from_state(self, clean_agent_state, legacy_state): + """Pinned revisions restore; old name-only state requires reactivation.""" from langbot.pkg.agent.runner.persistent_state_store import get_persistent_state_store from langbot.pkg.provider.tools.loaders.skill import ( ACTIVATED_SKILL_NAMES_STATE_KEY, @@ -1363,6 +1382,10 @@ class TestQueryEntryAdapterHostCapabilities: ] ) ap = FakeApplication(plugin_connector, db_engine) + revision = 'sha256:' + '1' * 64 + ap.skill_repository = types.SimpleNamespace( + get_skill=AsyncMock(return_value={'name': 'demo', 'revision': revision}) + ) orchestrator = AgentRunOrchestrator(ap, FakeRegistry(descriptor)) query = make_query() @@ -1376,7 +1399,7 @@ class TestQueryEntryAdapterHostCapabilities: descriptor, 'conversation', ACTIVATED_SKILL_NAMES_STATE_KEY, - ['demo'], + ['demo'] if legacy_state else [{'name': 'demo', 'revision': revision}], None, ) assert success is True @@ -1385,7 +1408,16 @@ class TestQueryEntryAdapterHostCapabilities: messages = [message async for message in orchestrator.run_from_query(query)] assert len(messages) == 1 - assert query.variables[ACTIVATED_SKILLS_KEY]['demo']['name'] == 'demo' + if legacy_state: + assert ACTIVATED_SKILLS_KEY not in query.variables + assert ap.skill_repository.get_skill.await_count == 0 + assert ( + len([warning for warning in ap.logger.warnings if 'Legacy name-only activated Skill state' in warning]) + == 1 + ) + else: + assert query.variables[ACTIVATED_SKILLS_KEY]['demo']['name'] == 'demo' + assert query.variables[ACTIVATED_SKILLS_KEY]['demo']['revision'] == revision @pytest.mark.asyncio async def test_event_log_and_transcript_written(self, clean_agent_state): diff --git a/tests/unit_tests/api/service/test_maintenance_service.py b/tests/unit_tests/api/service/test_maintenance_service.py index b4b00e6fb..16f27aa38 100644 --- a/tests/unit_tests/api/service/test_maintenance_service.py +++ b/tests/unit_tests/api/service/test_maintenance_service.py @@ -383,6 +383,7 @@ class TestMaintenanceServiceGetStorageAnalysis: handler=plugin_handler, ), box_service=box_service, + skill_repository=SimpleNamespace(storage_root=Mock(return_value='/core/skills/tenants/test')), ) service = MaintenanceService(ap) service._path_size = Mock(return_value=10) @@ -394,12 +395,17 @@ class TestMaintenanceServiceGetStorageAnalysis: result = await service.get_storage_analysis(TEST_CONTEXT) - assert result['total_size_bytes'] == 115 + assert result['total_size_bytes'] == 125 + assert ( + next(section for section in result['sections'] if section['key'] == 'skills')['path'] + == '/core/skills/tenants/test' + ) assert result['processes'][1]['directories'][0]['key'] == 'artifacts' assert result['processes'][2]['directories'][1]['key'] == 'mcp' assert result['processes'][2]['managed_processes'] == 2 plugin_handler.get_storage_analysis.assert_awaited_once_with() box_service.get_storage_analysis.assert_awaited_once_with(TEST_CONTEXT) + ap.skill_repository.storage_root.assert_called_once_with(TEST_CONTEXT) async def test_get_storage_analysis_postgresql(self): """Handles PostgreSQL database type.""" diff --git a/tests/unit_tests/box/test_box_service.py b/tests/unit_tests/box/test_box_service.py index 610163c68..4ae7e3517 100644 --- a/tests/unit_tests/box/test_box_service.py +++ b/tests/unit_tests/box/test_box_service.py @@ -585,7 +585,6 @@ async def test_box_service_reconnect_restores_workspace_and_runs_cleanup( connector.reconnect.assert_awaited_once() service._ensure_default_workspace.assert_called_once() service._purge_attachment_dirs.assert_awaited_once() - app.skill_mgr.reload_skills.assert_not_awaited() assert service.available is True diff --git a/tests/unit_tests/box/test_runner_box.py b/tests/unit_tests/box/test_runner_box.py index ad49482f5..1f462d2cc 100644 --- a/tests/unit_tests/box/test_runner_box.py +++ b/tests/unit_tests/box/test_runner_box.py @@ -24,7 +24,6 @@ def service(): require_workspace_sandbox=AsyncMock(side_effect=lambda context: context), _action_context=lambda context: context, client=SimpleNamespace(get_sessions=AsyncMock(side_effect=lambda **kw: list(sessions.values()))), - build_skill_extra_mounts=lambda query: [], ) return RunnerBoxService(box) diff --git a/tests/unit_tests/provider/test_skill_tools.py b/tests/unit_tests/provider/test_skill_tools.py index 7c25371a6..7396c00ed 100644 --- a/tests/unit_tests/provider/test_skill_tools.py +++ b/tests/unit_tests/provider/test_skill_tools.py @@ -261,6 +261,33 @@ class TestSkillToolLoader: handled by SkillService via the HTTP API / web UI instead. """ + @pytest.mark.asyncio + async def test_frozen_skill_source_runs_activation_without_box(self): + from langbot.pkg.provider.tools.loaders.skill_authoring import SkillToolLoader + from langbot.pkg.provider.tools.toolmgr import ToolManager + + skill = _make_skill_data(name='demo', instructions='Read this first') + ap = _make_ap() + ap.skill_mgr = _make_skill_manager({'demo': skill}) + ap.skill_repository = SimpleNamespace(get_skill=AsyncMock(return_value=skill)) + ap.box_service = SimpleNamespace(is_workspace_sandbox_available=AsyncMock(return_value=False)) + manager = ToolManager(ap) + manager.skill_tool_loader = SkillToolLoader(ap) + query = _make_query(_execution_context=_CONTEXT) + source_ref = {'source': 'skill', 'source_id': None} + + result = await manager.execute_func_call('activate', {'skill_name': 'demo'}, query, source_ref) + + assert result['activated'] is True + assert result['revision'] == skill['revision'] + assert result['capabilities']['execution_available'] is False + assert query.variables['_skill_execution_available'] is False + assert await manager.get_tool_by_source(_CONTEXT, 'activate', source_ref) is not None + assert await manager.get_tool_by_source(_CONTEXT, 'register_skill', source_ref) is None + + ap.box_service.is_workspace_sandbox_available.return_value = True + assert await manager.get_tool_by_source(_CONTEXT, 'register_skill', source_ref) is not None + @pytest.mark.asyncio async def test_activate_returns_instructions_and_registers_skill(self): from langbot.pkg.provider.tools.loaders.skill_authoring import (