feat(skill): use SDK store without box execution

tests/unit_tests/box/test_box_service.py

@@ -1978,7 +1978,7 @@ class TestBuildSkillExtraMounts:
                 {
                     'host_path': live_dir,
                     'mount_path': '/workspace/.skills/alive',
-                    'mode': 'rw',
+                    'mode': 'ro',
                 }
             ]
             # Warning logged so operators can see what was dropped
@@ -2003,8 +2003,8 @@ class TestBuildSkillExtraMounts:
         mounts = service.build_skill_extra_mounts(make_query())
 
         assert mounts == [
-            {'host_path': '/box/skills/a', 'mount_path': '/workspace/.skills/a', 'mode': 'rw'},
-            {'host_path': '/box/skills/b', 'mount_path': '/workspace/.skills/b', 'mode': 'rw'},
+            {'host_path': '/box/skills/a', 'mount_path': '/workspace/.skills/a', 'mode': 'ro'},
+            {'host_path': '/box/skills/b', 'mount_path': '/workspace/.skills/b', 'mode': 'ro'},
         ]
         # No skill is dropped, so no "missing" warning should be logged.
         assert not any('package_root missing' in str(call.args[0]) for call in logger.warning.call_args_list)

tests/unit_tests/provider/test_skill_tools.py

@@ -67,15 +67,10 @@ def _make_skill_data(
 
 
 class TestSkillManagerCache:
-    """The Box runtime is the only source of truth — SkillManager just holds
-    an in-memory cache populated by ``reload_skills``. There is no local
-    filesystem reader anymore."""
+    """SkillManager caches the Core-owned SkillRepository catalog."""
 
     def test_refresh_skill_from_disk_reports_cache_presence(self):
-        """Box is the only source of truth for skill content. refresh_skill_from_disk
-        now just reports whether the skill is still in the in-memory cache —
-        the actual content refresh is driven by SkillService awaiting
-        ``reload_skills`` after every Box mutation."""
+        """Disk mutations are reflected by an explicit repository reload."""
         from langbot.pkg.skill.manager import SkillManager
 
         ap = _make_ap()
@@ -92,67 +87,26 @@ class TestSkillManagerCache:
         assert mgr.refresh_skill_from_disk(_CONTEXT, '') is False
 
     @pytest.mark.asyncio
-    async def test_reload_skills_drops_box_skills_with_missing_package_root(self):
-        """When LangBot shares a filesystem with Box (local stdio mode) and Box
-        reports a skill whose package_root is gone from that shared filesystem,
-        the cache must drop it instead of keeping a stale entry that would later
-        produce a bad mount."""
+    async def test_reload_skills_uses_repository_when_box_is_disabled(self):
         from langbot.pkg.skill.manager import SkillManager
 
-        with tempfile.TemporaryDirectory() as live_dir:
-            ghost_dir = os.path.join(live_dir, '_does_not_exist')
-            box_service = SimpleNamespace(
-                available=True,
-                shares_filesystem_with_box=True,
-                list_skills=AsyncMock(
-                    return_value=[
-                        _make_skill_data(name='alive', package_root=live_dir),
-                        _make_skill_data(name='ghost', package_root=ghost_dir),
-                    ]
-                ),
-            )
-
-            ap = _make_ap()
-            ap.box_service = box_service
-            mgr = SkillManager(ap)
-
-            await mgr.reload_skills(_CONTEXT)
-
-        assert list(mgr.get_skills(_CONTEXT)) == ['alive']
-        # Warning fired with the dropped skill name so operators can see it.
-        warning_messages = [str(call.args[0]) for call in ap.logger.warning.call_args_list]
-        assert any('ghost' in msg and 'package_root missing' in msg for msg in warning_messages)
-
-    @pytest.mark.asyncio
-    async def test_reload_skills_trusts_box_paths_when_filesystem_not_shared(self):
-        """In separated deployments (Docker Compose, k8s sidecar,
-        --standalone-box, remote endpoint) the package_root reported by Box
-        lives on the Box runtime's filesystem and is not resolvable on the
-        LangBot side. The cache must keep every Box-reported skill rather than
-        dropping them all via a local isdir() check."""
-        from langbot.pkg.skill.manager import SkillManager
-
-        box_service = SimpleNamespace(
-            available=True,
-            shares_filesystem_with_box=False,
+        repository = SimpleNamespace(
             list_skills=AsyncMock(
                 return_value=[
-                    _make_skill_data(name='alpha', package_root='/box/skills/alpha'),
-                    _make_skill_data(name='beta', package_root='/box/skills/beta'),
+                    _make_skill_data(name='alpha', package_root='/skills/alpha'),
+                    _make_skill_data(name='beta', package_root='/skills/beta'),
                 ]
             ),
         )
-
         ap = _make_ap()
-        ap.box_service = box_service
+        ap.box_service = SimpleNamespace(available=False, enabled=False)
+        ap.skill_repository = repository
         mgr = SkillManager(ap)
 
         await mgr.reload_skills(_CONTEXT)
 
         assert sorted(mgr.get_skills(_CONTEXT)) == ['alpha', 'beta']
-        # No skill dropped → no "package_root missing" warning.
-        warning_messages = [str(call.args[0]) for call in ap.logger.warning.call_args_list]
-        assert not any('package_root missing' in msg for msg in warning_messages)
+        repository.list_skills.assert_awaited_once_with(_CONTEXT)
 
 
 class TestSkillActivationHelper:
@@ -322,7 +276,7 @@ class TestSkillPathHelpers:
 
 
 class TestSkillToolLoader:
-    """The skill tool surface is now just ``activate`` + ``register_skill``.
+    """Skill activation and resources are independent from sandbox execution.
 
     The legacy CRUD authoring tools (create/list/get/update/delete/
     import_skill_from_directory/reload_skills) were removed; skill CRUD is
@@ -338,8 +292,11 @@ class TestSkillToolLoader:
         from langbot.pkg.provider.tools.loaders.skill import ACTIVATED_SKILLS_KEY
 
         skill = _make_skill_data(name='demo', package_root='/data/skills/demo', instructions='Step 1')
+        skill['revision'] = 'sha256:demo'
         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))
 
         loader = SkillToolLoader(ap)
         query = _make_query()
@@ -348,9 +305,13 @@ class TestSkillToolLoader:
 
         assert result['activated'] is True
         assert result['skill_name'] == 'demo'
-        assert result['mount_path'] == '/workspace/.skills/demo'
+        assert result['mount_path'] is None
+        assert result['revision'] == 'sha256:demo'
+        assert result['capabilities']['resources_readable'] is True
+        assert result['capabilities']['execution_available'] is False
         assert result['activated_skill_names'] == ['demo']
         assert 'Step 1' in result['content']
+        assert '<package-root>' not in result['content']
         assert set(query.variables[ACTIVATED_SKILLS_KEY].keys()) == {'demo'}
 
     @pytest.mark.asyncio
@@ -384,7 +345,11 @@ class TestSkillToolLoader:
             os.makedirs(repo_dir)
 
             ap = _make_ap()
-            ap.box_service = SimpleNamespace(default_workspace=tmpdir, available=True)
+            ap.box_service = SimpleNamespace(
+                default_workspace=tmpdir,
+                available=True,
+                require_workspace_sandbox=AsyncMock(return_value=_CONTEXT),
+            )
             ap.skill_service = SimpleNamespace(
                 scan_directory_async=AsyncMock(
                     return_value={
@@ -394,7 +359,7 @@ class TestSkillToolLoader:
                         'instructions': 'Do work',
                     }
                 ),
-                create_skill=AsyncMock(
+                import_skill_directory=AsyncMock(
                     return_value=_make_skill_data(name='cloned-skill', package_root=os.path.realpath(repo_dir))
                 ),
             )
@@ -407,14 +372,14 @@ class TestSkillToolLoader:
             )
 
         ap.skill_service.scan_directory_async.assert_awaited_once_with(_CONTEXT, os.path.realpath(repo_dir))
-        ap.skill_service.create_skill.assert_awaited_once_with(
+        ap.skill_service.import_skill_directory.assert_awaited_once_with(
             _CONTEXT,
+            os.path.realpath(repo_dir),
             {
                 'name': 'cloned-skill',
                 'display_name': 'Cloned Skill',
                 'description': 'Imported from clone',
                 'instructions': 'Do work',
-                'package_root': os.path.realpath(repo_dir),
             },
         )
         assert result['registered'] is True
@@ -430,8 +395,15 @@ class TestSkillToolLoader:
 
         with tempfile.TemporaryDirectory() as tmpdir:
             ap = _make_ap()
-            ap.box_service = SimpleNamespace(default_workspace=tmpdir, available=True)
-            ap.skill_service = SimpleNamespace(scan_directory_async=AsyncMock(), create_skill=AsyncMock())
+            ap.box_service = SimpleNamespace(
+                default_workspace=tmpdir,
+                available=True,
+                require_workspace_sandbox=AsyncMock(return_value=_CONTEXT),
+            )
+            ap.skill_service = SimpleNamespace(
+                scan_directory_async=AsyncMock(),
+                import_skill_directory=AsyncMock(),
+            )
 
             loader = SkillToolLoader(ap)
 
@@ -451,7 +423,11 @@ class TestSkillToolLoader:
 
         with tempfile.TemporaryDirectory() as tmpdir:
             ap = _make_ap()  # no skill_service attribute
-            ap.box_service = SimpleNamespace(default_workspace=tmpdir, available=True)
+            ap.box_service = SimpleNamespace(
+                default_workspace=tmpdir,
+                available=True,
+                require_workspace_sandbox=AsyncMock(return_value=_CONTEXT),
+            )
 
             loader = SkillToolLoader(ap)
 
@@ -463,21 +439,22 @@ class TestSkillToolLoader:
                 )
 
     @pytest.mark.asyncio
-    async def test_tools_hidden_when_sandbox_backend_unavailable(self):
+    async def test_read_only_tools_remain_when_sandbox_unavailable(self):
         from langbot.pkg.provider.tools.loaders.skill_authoring import SkillToolLoader
 
         ap = _make_ap()
         ap.skill_mgr = SimpleNamespace(skills={})
-        ap.box_service = SimpleNamespace(
-            available=True,
-            get_backend_status=AsyncMock(return_value={'backend': {'available': False}}),
-        )
+        ap.skill_repository = SimpleNamespace()
 
         loader = SkillToolLoader(ap)
         await loader.initialize()
 
-        assert await loader.get_tools() == []
-        assert await loader.has_tool('activate') is False
+        assert sorted(tool.name for tool in await loader.get_tools(sandbox_available=False)) == [
+            'activate',
+            'list_skill_resources',
+            'read_skill_resource',
+        ]
+        assert await loader.has_tool('activate') is True
         assert await loader.has_tool('register_skill') is False
 
     @pytest.mark.asyncio
@@ -486,38 +463,89 @@ class TestSkillToolLoader:
 
         ap = _make_ap()
         ap.skill_mgr = _make_skill_manager({'demo': _make_skill_data(name='demo')})
-        ap.box_service = SimpleNamespace(
-            available=True,
-            get_backend_status=AsyncMock(return_value={'backend': {'available': True}}),
-        )
+        ap.skill_repository = SimpleNamespace()
 
         loader = SkillToolLoader(ap)
         await loader.initialize()
 
-        tools = await loader.get_tools()
+        tools = await loader.get_tools(sandbox_available=True)
 
-        assert sorted(tool.name for tool in tools) == ['activate', 'register_skill']
+        assert sorted(tool.name for tool in tools) == [
+            'activate',
+            'list_skill_resources',
+            'read_skill_resource',
+            'register_skill',
+        ]
         assert await loader.has_tool('activate') is True
-        assert await loader.has_tool('register_skill') is True
+        assert await loader.has_tool('register_skill', sandbox_available=True) is True
 
     @pytest.mark.asyncio
-    async def test_tools_reappear_after_box_backend_recovers(self):
+    async def test_register_skill_appears_after_sandbox_recovers(self):
         from langbot.pkg.provider.tools.loaders.skill_authoring import SkillToolLoader
 
         ap = _make_ap()
         ap.skill_mgr = SimpleNamespace(skills={'demo': _make_skill_data(name='demo')})
-        ap.box_service = SimpleNamespace(
-            available=False,
-            get_backend_status=AsyncMock(return_value={'backend': {'available': True}}),
-        )
+        ap.skill_repository = SimpleNamespace()
 
         loader = SkillToolLoader(ap)
         await loader.initialize()
-        assert await loader.get_tools() == []
+        assert 'register_skill' not in {tool.name for tool in await loader.get_tools(sandbox_available=False)}
+        assert 'register_skill' in {tool.name for tool in await loader.get_tools(sandbox_available=True)}
 
-        ap.box_service.available = True
+    @pytest.mark.asyncio
+    async def test_resources_require_activation_and_use_pinned_revision(self):
+        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill
+        from langbot.pkg.provider.tools.loaders.skill_authoring import SkillToolLoader
 
-        assert sorted(tool.name for tool in await loader.get_tools()) == ['activate', 'register_skill']
+        skill = _make_skill_data(name='demo', instructions='Read references')
+        skill['revision'] = 'sha256:demo'
+        ap = _make_ap()
+        ap.skill_mgr = _make_skill_manager({'demo': skill})
+        ap.skill_repository = SimpleNamespace(
+            list_skill_resources=AsyncMock(
+                return_value={'entries': [{'path': 'references/a.md'}], 'revision': 'sha256:demo'}
+            ),
+            read_skill_resource=AsyncMock(
+                return_value={
+                    'path': 'references/a.md',
+                    'content': 'reference text',
+                    'revision': 'sha256:demo',
+                    'mime_type': 'text/markdown',
+                }
+            ),
+        )
+        loader = SkillToolLoader(ap)
+        query = _make_query()
+
+        with pytest.raises(ValueError, match='must be activated'):
+            await loader.invoke_tool(
+                'read_skill_resource',
+                {'skill_name': 'demo', 'path': 'references/a.md'},
+                query,
+            )
+
+        register_activated_skill(query, skill)
+        listed = await loader.invoke_tool('list_skill_resources', {'skill_name': 'demo'}, query)
+        read = await loader.invoke_tool(
+            'read_skill_resource',
+            {'skill_name': 'demo', 'path': 'references/a.md', 'revision': 'sha256:demo'},
+            query,
+        )
+
+        assert listed['entries'][0]['path'] == 'references/a.md'
+        assert read['content'] == 'reference text'
+        ap.skill_repository.list_skill_resources.assert_awaited_once_with(
+            _CONTEXT,
+            'demo',
+            '.',
+            expected_revision='sha256:demo',
+        )
+        ap.skill_repository.read_skill_resource.assert_awaited_once_with(
+            _CONTEXT,
+            'demo',
+            'references/a.md',
+            expected_revision='sha256:demo',
+        )
 
 
 class TestNativeToolLoaderSkillPaths:
@@ -551,7 +579,7 @@ class TestNativeToolLoaderSkillPaths:
             assert result['truncated'] is False
 
     @pytest.mark.asyncio
-    async def test_external_runtime_read_never_interprets_package_root_on_core_host(self):
+    async def test_external_runtime_read_uses_core_skill_repository(self):
         from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
         from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY
 
@@ -563,7 +591,9 @@ class TestNativeToolLoaderSkillPaths:
             ap.box_service = SimpleNamespace(
                 available=True,
                 shares_filesystem_with_box=False,
-                read_skill_file=AsyncMock(return_value={'content': 'runtime-owned-content'}),
+            )
+            ap.skill_repository = SimpleNamespace(
+                read_skill_file=AsyncMock(return_value={'content': 'repository-content'})
             )
             ap.skill_mgr = _make_skill_manager({'demo': _make_skill_data(name='demo', package_root=tmpdir)})
             loader = NativeToolLoader(ap)
@@ -579,9 +609,9 @@ class TestNativeToolLoaderSkillPaths:
             )
 
             assert result['ok'] is True
-            assert result['content'] == 'runtime-owned-content'
+            assert result['content'] == 'repository-content'
             assert 'core-host-secret' not in repr(result)
-            ap.box_service.read_skill_file.assert_awaited_once_with(_CONTEXT, 'demo', 'SKILL.md')
+            ap.skill_repository.read_skill_file.assert_awaited_once_with(_CONTEXT, 'demo', 'SKILL.md')
 
     @pytest.mark.asyncio
     async def test_external_runtime_rejects_skill_host_fallback_without_protocol_capability(self):

tests/unit_tests/provider/test_tool_manager_native.py

@@ -55,10 +55,20 @@ class StubLoader:
             for tool in self._tools
         ]
 
-    async def has_tool(self, *args) -> bool:
+    async def get_tool(self, name: str, **_kwargs):
+        return next((tool for tool in self._tools if tool.name == name), None)
+
+    async def has_tool(self, *args, **_kwargs) -> bool:
         name = args[-1]
         return any(tool.name == name for tool in self._tools)
 
+    def recognizes_tool(self, name: str) -> bool:
+        return any(tool.name == name for tool in self._tools)
+
+    @staticmethod
+    def is_sandbox_tool(name: str) -> bool:
+        return name == 'register_skill'
+
     async def invoke_tool(self, name: str, parameters: dict, query):
         return self._invoke_result(name, parameters, query) if callable(self._invoke_result) else self._invoke_result
 
@@ -145,7 +155,7 @@ async def test_tool_manager_routes_native_tool_calls():
 
 
 @pytest.mark.asyncio
-async def test_tool_manager_hides_sandbox_and_skill_tools_without_workspace_entitlement():
+async def test_tool_manager_keeps_read_only_skill_tools_without_workspace_entitlement():
     box_service = SimpleNamespace(is_workspace_sandbox_available=AsyncMock(return_value=False))
     manager = ToolManager(SimpleNamespace(box_service=box_service))
     manager.native_tool_loader = StubLoader([make_tool('exec')])
@@ -156,8 +166,8 @@ async def test_tool_manager_hides_sandbox_and_skill_tools_without_workspace_enti
     tools = await manager.get_all_tools(_CONTEXT, include_skill_authoring=True)
     catalog = await manager.get_tool_catalog(_CONTEXT, include_skill_authoring=True)
 
-    assert [tool.name for tool in tools] == ['plugin_tool', 'mcp_tool']
-    assert [item['name'] for item in catalog] == ['plugin_tool', 'mcp_tool']
+    assert [tool.name for tool in tools] == ['activate', 'plugin_tool', 'mcp_tool']
+    assert [item['name'] for item in catalog] == ['activate', 'plugin_tool', 'mcp_tool']
     assert box_service.is_workspace_sandbox_available.await_count == 2
 
 
@@ -176,9 +186,10 @@ async def test_tool_manager_rechecks_workspace_entitlement_before_native_invocat
         query_uuid=None,
     )
 
-    with pytest.raises(Exception, match='exec'):
-        await manager.execute_func_call('exec', {'command': 'pwd'}, query=query)
+    result = await manager.execute_func_call('exec', {'command': 'pwd'}, query=query)
 
+    assert result['code'] == 'sandbox_unavailable'
+    assert result['tool'] == 'exec'
     box_service.is_workspace_sandbox_available.assert_awaited_once_with(_CONTEXT)
 
 

tests/unit_tests/test_preproc.py

@@ -190,6 +190,7 @@ async def test_preproc_injects_skill_index_into_system_prompt():
     preproc_module, entities_module = _import_preproc_modules()
 
     app = _make_app(skill_service=SimpleNamespace())
+    app.tool_mgr.get_all_tools.return_value = [SimpleNamespace(name='activate')]
     addendum = '\n\nAvailable Skills:\n- demo (demo): Demo skill.\n\nCall activate ...'
     app.skill_mgr.build_skill_aware_prompt_addition = Mock(return_value=addendum)
 
@@ -204,6 +205,22 @@ async def test_preproc_injects_skill_index_into_system_prompt():
     head = query.prompt.messages[0]
     assert head.role == 'system'
     assert head.content.endswith(addendum)
+    assert query.variables['_skill_execution_available'] is False
+
+
+@pytest.mark.asyncio
+async def test_preproc_does_not_advertise_activation_when_tool_is_filtered_out():
+    preproc_module, entities_module = _import_preproc_modules()
+
+    app = _make_app(skill_service=SimpleNamespace())
+    addendum = '\n\nAvailable Skills:\n- demo (demo): Demo skill.\n\nCall activate ...'
+    app.skill_mgr.build_skill_aware_prompt_addition = Mock(return_value=addendum)
+
+    query = _make_query()
+    result = await stage_process_capture(preproc_module, app, query)
+
+    assert result.result_type == entities_module.ResultType.CONTINUE
+    assert addendum not in query.prompt.messages[0].content
 
 
 @pytest.mark.asyncio

tests/unit_tests/test_skill_repository.py

@@ -0,0 +1,175 @@
+from types import SimpleNamespace
+
+import pytest
+
+from langbot.pkg.api.http.context import ExecutionContext
+from langbot.pkg.skill.repository import SkillRepository, SkillRevisionMismatchError
+
+
+_CONTEXT = ExecutionContext(
+    instance_uuid='instance-a',
+    workspace_uuid='workspace-a',
+    placement_generation=1,
+)
+
+
+async def _binding(workspace_uuid, *, expected_generation):
+    return SimpleNamespace(
+        instance_uuid=_CONTEXT.instance_uuid,
+        workspace_uuid=workspace_uuid,
+        placement_generation=expected_generation,
+    )
+
+
+def _repository(tmp_path) -> SkillRepository:
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(
+            get_execution_binding=_binding,
+        ),
+        instance_config=SimpleNamespace(
+            data={
+                'box': {
+                    'enabled': False,
+                    'local': {
+                        'host_root': str(tmp_path / 'box'),
+                        'skills_root': 'skills',
+                    },
+                }
+            }
+        ),
+    )
+    return SkillRepository(app)
+
+
+@pytest.mark.asyncio
+async def test_repository_crud_and_reads_do_not_require_box(tmp_path):
+    repository = _repository(tmp_path)
+
+    await repository.create_skill(
+        _CONTEXT,
+        {
+            'name': 'docs-only',
+            'display_name': 'Docs only',
+            'description': 'Read-only guidance',
+            'instructions': 'Read references/guide.md.',
+        },
+    )
+    await repository.write_skill_file(
+        _CONTEXT,
+        'docs-only',
+        'references/guide.md',
+        '# Guide\n\nNo execution needed.',
+    )
+
+    skill = await repository.get_skill(_CONTEXT, 'docs-only', snapshot=True)
+    assert skill is not None
+    assert skill['revision'].startswith('sha256:')
+    assert [item['name'] for item in await repository.list_skills(_CONTEXT)] == ['docs-only']
+
+    listed = await repository.list_skill_resources(
+        _CONTEXT,
+        'docs-only',
+        'references',
+        expected_revision=skill['revision'],
+    )
+    assert listed['entries'][0]['path'] == 'references/guide.md'
+    assert listed['entries'][0]['mime_type'] == 'text/markdown'
+
+    resource = await repository.read_skill_resource(
+        _CONTEXT,
+        'docs-only',
+        'references/guide.md',
+        expected_revision=skill['revision'],
+    )
+    assert resource['content'].startswith('# Guide')
+    assert resource['revision'] == skill['revision']
+
+
+@pytest.mark.asyncio
+async def test_repository_rejects_traversal_and_stale_revision(tmp_path):
+    repository = _repository(tmp_path)
+    await repository.create_skill(
+        _CONTEXT,
+        {'name': 'safe', 'description': 'Safe', 'instructions': 'Use the reference.'},
+    )
+    await repository.write_skill_file(_CONTEXT, 'safe', 'reference.md', 'first')
+    skill = await repository.get_skill(_CONTEXT, 'safe', snapshot=True)
+    assert skill is not None
+
+    with pytest.raises(ValueError, match='stay within'):
+        await repository.read_skill_resource(_CONTEXT, 'safe', '../secret.txt')
+
+    await repository.write_skill_file(_CONTEXT, 'safe', 'reference.md', 'second')
+    with pytest.raises(SkillRevisionMismatchError, match='reactivate'):
+        await repository.read_skill_resource(
+            _CONTEXT,
+            'safe',
+            'reference.md',
+            expected_revision=skill['revision'],
+        )
+
+
+@pytest.mark.asyncio
+async def test_repository_scopes_skills_by_workspace(tmp_path):
+    repository = _repository(tmp_path)
+    other_context = ExecutionContext(
+        instance_uuid='instance-a',
+        workspace_uuid='workspace-b',
+        placement_generation=1,
+    )
+
+    await repository.create_skill(_CONTEXT, {'name': 'private', 'instructions': 'A'})
+
+    assert [skill['name'] for skill in await repository.list_skills(_CONTEXT)] == ['private']
+    assert await repository.list_skills(other_context) == []
+
+
+@pytest.mark.asyncio
+async def test_repository_rejects_stale_workspace_placement(tmp_path):
+    repository = _repository(tmp_path)
+
+    async def stale_binding(workspace_uuid, *, expected_generation):
+        return SimpleNamespace(
+            instance_uuid=_CONTEXT.instance_uuid,
+            workspace_uuid=workspace_uuid,
+            placement_generation=expected_generation + 1,
+        )
+
+    repository.ap.workspace_service.get_execution_binding = stale_binding
+    with pytest.raises(ValueError, match='stale Workspace placement'):
+        await repository.list_skills(_CONTEXT)
+
+
+@pytest.mark.asyncio
+async def test_repository_imports_only_from_the_fenced_workspace(tmp_path):
+    repository = _repository(tmp_path)
+    namespace = repository._namespace(_CONTEXT)
+    source = tmp_path / 'box' / 'default' / 'tenants' / namespace / 'draft'
+    source.mkdir(parents=True)
+    (source / 'SKILL.md').write_text(
+        '---\nname: draft\ndescription: Draft skill\n---\n\nFollow the guide.',
+        encoding='utf-8',
+    )
+    (source / 'guide.md').write_text('Imported resource', encoding='utf-8')
+
+    scanned = await repository.scan_skill_directory(_CONTEXT, str(source))
+    imported = await repository.import_skill_directory(
+        _CONTEXT,
+        str(source),
+        {
+            'name': scanned['name'],
+            'display_name': scanned['display_name'],
+            'description': scanned['description'],
+            'instructions': scanned['instructions'],
+        },
+    )
+
+    assert imported['name'] == 'draft'
+    resource = await repository.read_skill_file(_CONTEXT, 'draft', 'guide.md')
+    assert resource['content'] == 'Imported resource'
+
+    outside = tmp_path / 'outside'
+    outside.mkdir()
+    (outside / 'SKILL.md').write_text('Outside', encoding='utf-8')
+    with pytest.raises(ValueError, match='trusted source root'):
+        await repository.scan_skill_directory(_CONTEXT, str(outside))

tests/unit_tests/test_skill_service.py

@@ -23,102 +23,73 @@ def _workspace_service():
     )
 
 
-class TestRequireBoxForWrite:
-    """Box is the only source of truth for skills — there is no local
-    filesystem fallback. Every write and (most) read methods refuse cleanly
-    when the Box runtime is disabled, unreachable, or simply not installed."""
+class TestSkillRepositoryBoundary:
+    """Skill management and reads remain available without Box execution."""
 
-    def _ap_with_disabled_box(self):
+    @staticmethod
+    def _ap_with_repository():
+        repository = SimpleNamespace(
+            list_skills=AsyncMock(return_value=[{'name': 'x', 'instructions': 'Do work'}]),
+            get_skill=AsyncMock(return_value={'name': 'x', 'instructions': 'Do work', 'revision': 'sha256:x'}),
+            create_skill=AsyncMock(return_value={'name': 'x', 'instructions': 'Do work'}),
+            update_skill=AsyncMock(return_value={'name': 'x', 'instructions': 'Updated'}),
+            delete_skill=AsyncMock(),
+            read_skill_file=AsyncMock(return_value={'path': 'a.txt', 'content': 'hello'}),
+            write_skill_file=AsyncMock(return_value={'path': 'a.txt'}),
+        )
         return SimpleNamespace(
             skill_mgr=SimpleNamespace(reload_skills=AsyncMock()),
             workspace_service=_workspace_service(),
-            box_service=SimpleNamespace(
-                available=False,
-                enabled=False,
-                _connector_error='Box runtime is disabled in config (box.enabled = false)',
-            ),
-        )
-
-    def _ap_with_failed_box(self):
-        return SimpleNamespace(
-            skill_mgr=SimpleNamespace(reload_skills=AsyncMock()),
-            workspace_service=_workspace_service(),
-            box_service=SimpleNamespace(
-                available=False,
-                enabled=True,
-                _connector_error='docker daemon not running',
-            ),
+            box_service=SimpleNamespace(available=False, enabled=False),
+            skill_repository=repository,
         )
 
     @pytest.mark.asyncio
-    async def test_create_skill_refused_when_box_disabled(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='disabled in config'):
-            await service.create_skill(_CONTEXT, {'name': 'x'})
+    async def test_list_and_read_work_when_box_disabled(self):
+        ap = self._ap_with_repository()
+        service = SkillService(ap)
+
+        assert await service.list_skills(_CONTEXT) == [{'name': 'x', 'instructions': 'Do work'}]
+        assert await service.read_skill_file(_CONTEXT, 'x', 'a.txt') == {
+            'path': 'a.txt',
+            'content': 'hello',
+        }
+        ap.skill_repository.read_skill_file.assert_awaited_once_with(_CONTEXT, 'x', 'a.txt')
 
     @pytest.mark.asyncio
-    async def test_create_skill_refused_when_box_failed(self):
-        service = SkillService(self._ap_with_failed_box())
-        with pytest.raises(ValueError, match='docker daemon not running'):
-            await service.create_skill(_CONTEXT, {'name': 'x'})
+    async def test_create_update_and_write_work_when_box_disabled(self):
+        ap = self._ap_with_repository()
+        service = SkillService(ap)
+
+        await service.create_skill(_CONTEXT, {'name': 'x'})
+        await service.update_skill(_CONTEXT, 'x', {'instructions': 'Updated'})
+        await service.write_skill_file(_CONTEXT, 'x', 'a.txt', 'hello')
+
+        ap.skill_repository.create_skill.assert_awaited_once_with(_CONTEXT, {'name': 'x'})
+        ap.skill_repository.update_skill.assert_awaited_once_with(_CONTEXT, 'x', {'instructions': 'Updated'})
+        ap.skill_repository.write_skill_file.assert_awaited_once_with(_CONTEXT, 'x', 'a.txt', 'hello')
 
     @pytest.mark.asyncio
-    async def test_update_skill_refused_when_box_disabled(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='Editing a skill requires the Box runtime'):
-            await service.update_skill(_CONTEXT, 'x', {})
+    async def test_get_skill_returns_repository_revision(self):
+        ap = self._ap_with_repository()
+        service = SkillService(ap)
+
+        skill = await service.get_skill(_CONTEXT, 'x')
+
+        assert skill['revision'] == 'sha256:x'
+        ap.skill_repository.get_skill.assert_awaited_once_with(_CONTEXT, 'x', snapshot=True)
 
     @pytest.mark.asyncio
-    async def test_write_skill_file_refused_when_box_disabled(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='Editing skill files requires the Box runtime'):
-            await service.write_skill_file(_CONTEXT, 'x', 'a.txt', 'hi')
-
-    @pytest.mark.asyncio
-    async def test_install_from_github_refused_when_box_disabled(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='Installing a skill from GitHub'):
-            await service.install_from_github(
-                _CONTEXT,
-                {'owner': 'o', 'repo': 'r', 'asset_url': 'https://example/x.zip'},
-            )
-
-    @pytest.mark.asyncio
-    async def test_install_from_zip_upload_refused_when_box_disabled(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='Installing a skill from upload'):
-            await service.install_from_zip_upload(
-                _CONTEXT,
-                file_bytes=b'',
-                filename='x.zip',
-            )
-
-    @pytest.mark.asyncio
-    async def test_create_skill_refused_when_box_service_missing_entirely(self):
-        """No ap.box_service attribute at all (truly minimal setup):
-        Box is the only source of truth, so creation must still refuse."""
+    async def test_missing_repository_is_explicit(self):
         service = SkillService(
             SimpleNamespace(
                 skill_mgr=SimpleNamespace(reload_skills=AsyncMock()),
                 workspace_service=_workspace_service(),
             )
         )
-        with pytest.raises(ValueError, match='not initialised'):
+        with pytest.raises(ValueError, match='repository is not initialised'):
             await service.create_skill(_CONTEXT, {'name': 'x'})
 
-    @pytest.mark.asyncio
-    async def test_list_skills_returns_empty_when_box_unavailable(self):
-        """list_skills should render an empty surface (not crash) so the
-        skills page can show a banner instead of a broken state."""
-        service = SkillService(self._ap_with_disabled_box())
-        assert await service.list_skills(_CONTEXT) == []
-
-    @pytest.mark.asyncio
-    async def test_read_skill_file_refused_when_box_unavailable(self):
-        service = SkillService(self._ap_with_disabled_box())
-        with pytest.raises(ValueError, match='Reading a skill file'):
-            await service.read_skill_file(_CONTEXT, 'x', 'a.txt')
-
 
 class TestGithubSkillArchiveLimits:
     @staticmethod