feat(skill): pin run revision bindings

2026-09-16 23:07:14 +00:00 · 2026-09-12 13:04:11 +08:00
parent 2fe4b117a4
commit 4d438f8821
16 changed files with 657 additions and 284 deletions
@@ -351,13 +351,26 @@ async def test_cloud_core_skills_mount_generically_and_do_not_require_box_entitl
                'instructions': 'Run scripts/main.py',
            },
        )
-        await repository.write_skill_file(first, 'runner', 'scripts/main.py', "print('ok')")
-        await repository.write_skill_file(first, 'runner', 'requirements.txt', 'requests==2.32.0\n')
+        script_revision = await repository.write_skill_file(
+            first,
+            'runner',
+            'scripts/main.py',
+            "print('ok')",
+            base_revision=own_skill['revision'],
+        )
+        await repository.write_skill_file(
+            first,
+            'runner',
+            'requirements.txt',
+            'requests==2.32.0\n',
+            base_revision=script_revision['revision'],
+        )
        refreshed_skill = await repository.get_skill(first, 'runner')
        assert refreshed_skill is not None
        assert refreshed_skill['python_project'] is True
        await service.ap.skill_mgr.reload_skills(first)
        query = _query(first, 91)
+        skill_loader.register_activated_skill(query, refreshed_skill)
        await service.execute_tool(
            {
                'command': 'python /workspace/.skills/runner/scripts/main.py',
@@ -369,9 +382,10 @@ async def test_cloud_core_skills_mount_generically_and_do_not_require_box_entitl

        mounted_spec = backend.started_specs[-1]
        assert len(mounted_spec.extra_mounts) == 1
-        assert mounted_spec.extra_mounts[0].host_path == own_skill['package_root']
+        assert mounted_spec.extra_mounts[0].host_path == refreshed_skill['package_root']
        assert mounted_spec.extra_mounts[0].mount_path == '/workspace/.skills/runner'
        assert mounted_spec.extra_mounts[0].mode.value == 'ro'
+        assert mounted_spec.extra_mounts[0].content_digest == refreshed_skill['revision']

        assert await repository.get_skill(first, 'private') is None
        await repository.create_skill(
@@ -1939,29 +1939,36 @@ def test_disconnect_callback_does_not_schedule_without_running_event_loop():


 class TestBuildSkillExecutionMounts:
-    """Robustness of skill mount construction against a stale skill cache.
-
-    The three sandbox backends behave inconsistently when a skill's
-    package_root no longer exists on disk (nsjail aborts the whole sandbox
-    start, Docker silently auto-creates a root-owned empty directory, E2B
-    silently skips). Mount construction must filter these out up front so
-    the backend never sees a bad mount.
-    """
+    """Execution materializes only revisions pinned by this run."""

    def _make_app(self, logger, skills):
        app = make_app(logger)
        app.skill_mgr = SimpleNamespace(skills=skills, get_skills=Mock(return_value=skills))
        return app

-    def test_skips_skill_with_missing_package_root(self):
+    def test_mounts_only_activated_revision(self):
        logger = Mock()
        with tempfile.TemporaryDirectory() as live_dir:
+            manifest_path = os.path.join(live_dir, 'manifest.json')
+            with open(manifest_path, 'w', encoding='utf-8') as file:
+                file.write('{}')
            skills = {
-                'alive': {'name': 'alive', 'package_root': live_dir},
-                'ghost': {'name': 'ghost', 'package_root': '/nonexistent/path/should/never/exist'},
+                'alive': {
+                    'name': 'alive',
+                    'package_root': live_dir,
+                    'manifest_path': manifest_path,
+                    'revision': 'sha256:' + '1' * 64,
+                },
+                'visible-not-activated': {
+                    'name': 'visible-not-activated',
+                    'package_root': live_dir,
+                    'manifest_path': manifest_path,
+                    'revision': 'sha256:' + '2' * 64,
+                },
            }
            app = self._make_app(logger, skills)
            query = make_query()
+            skill_loader.register_activated_skill(query, skills['alive'])

            mounts = skill_loader.build_execution_mounts(app, query)

@@ -1970,42 +1977,56 @@ class TestBuildSkillExecutionMounts:
                    'host_path': live_dir,
                    'mount_path': '/workspace/.skills/alive',
                    'mode': 'ro',
+                    'content_digest': 'sha256:' + '1' * 64,
+                    'manifest_path': manifest_path,
                }
            ]
-            # Warning logged so operators can see what was dropped
-            assert any(
-                'ghost' in str(call.args[0]) and 'package_root missing' in str(call.args[0])
-                for call in logger.warning.call_args_list
-            )

-    def test_rejects_missing_core_paths_when_filesystem_not_shared(self):
-        """Core owns package paths even when Box is a separate process."""
+    def test_missing_pinned_revision_fails_instead_of_being_skipped(self):
        logger = Mock()
        skills = {
-            'a': {'name': 'a', 'package_root': '/box/skills/a'},
-            'b': {'name': 'b', 'package_root': '/box/skills/b'},
+            'a': {
+                'name': 'a',
+                'package_root': '/box/skills/a',
+                'manifest_path': '/box/skills/manifest.json',
+                'revision': 'sha256:' + '1' * 64,
+            }
        }
        app = self._make_app(logger, skills)
+        query = make_query()
+        skill_loader.register_activated_skill(query, skills['a'])

-        mounts = skill_loader.build_execution_mounts(app, make_query())
+        with pytest.raises(ValueError, match='cannot be recovered safely'):
+            skill_loader.build_execution_mounts(app, query)

-        assert mounts == []
-        assert len(logger.warning.call_args_list) == 2
-
-    def test_skips_skill_with_empty_package_root(self):
+    def test_rejects_activated_skill_with_empty_package_root(self):
        logger = Mock()
        skills = {
-            'no_root': {'name': 'no_root', 'package_root': ''},
-            'whitespace': {'name': 'whitespace', 'package_root': '   '},
+            'no_root': {
+                'name': 'no_root',
+                'package_root': '',
+                'manifest_path': '',
+                'revision': 'sha256:' + '1' * 64,
+            }
        }
        app = self._make_app(logger, skills)
+        query = make_query()
+        skill_loader.register_activated_skill(query, skills['no_root'])

-        assert skill_loader.build_execution_mounts(app, make_query()) == []
+        with pytest.raises(ValueError, match='no immutable package root'):
+            skill_loader.build_execution_mounts(app, query)

    def test_empty_package_root_skipped_even_when_not_shared(self):
        """An empty package_root is always invalid regardless of topology."""
        logger = Mock()
-        skills = {'no_root': {'name': 'no_root', 'package_root': ''}}
+        skills = {
+            'no_root': {
+                'name': 'no_root',
+                'package_root': '',
+                'manifest_path': '',
+                'revision': 'sha256:' + '1' * 64,
+            }
+        }
        app = self._make_app(logger, skills)

        assert skill_loader.build_execution_mounts(app, make_query()) == []
@@ -61,7 +61,9 @@ def _make_skill_data(
        'description': kwargs.pop('description', f'Description of {name}'),
        'instructions': instructions,
        'package_root': package_root,
+        'manifest_path': kwargs.pop('manifest_path', ''),
        'entry_file': entry_file,
+        'revision': kwargs.pop('revision', 'sha256:' + '1' * 64),
        **kwargs,
    }

@@ -109,58 +111,6 @@ class TestSkillManagerCache:
        repository.list_skills.assert_awaited_once_with(_CONTEXT)


-class TestSkillActivationHelper:
-    """Skill activation is now Tool-Call based.
-
-    The legacy text-marker mechanism (``[ACTIVATE_SKILL: x]`` detection,
-    ``build_activation_prompt_for_skills``, ``remove_activation_marker``,
-    ``prepare_skill_activation``) has been removed. Activation now goes
-    through ``skill.activation.register_activated_skill``, invoked by the
-    ``activate`` Tool Call.
-    """
-
-    def test_register_activated_skill_records_known_skill(self):
-        from langbot.pkg.skill.activation import register_activated_skill
-        from langbot.pkg.provider.tools.loaders.skill import ACTIVATED_SKILLS_KEY
-        from langbot.pkg.skill.manager import SkillManager
-
-        ap = _make_ap()
-        mgr = SkillManager(ap)
-        mgr._skills_by_scope[mgr._scope_key(_CONTEXT)] = {
-            'primary': _make_skill_data(name='primary', instructions='Primary instructions'),
-        }
-        ap.skill_mgr = mgr
-
-        query = _make_query()
-
-        assert register_activated_skill(ap, query, 'primary') is True
-        assert set(query.variables[ACTIVATED_SKILLS_KEY].keys()) == {'primary'}
-        assert query.variables[ACTIVATED_SKILLS_KEY]['primary']['name'] == 'primary'
-
-    def test_register_activated_skill_rejects_unknown_skill(self):
-        from langbot.pkg.skill.activation import register_activated_skill
-        from langbot.pkg.provider.tools.loaders.skill import ACTIVATED_SKILLS_KEY
-        from langbot.pkg.skill.manager import SkillManager
-
-        ap = _make_ap()
-        mgr = SkillManager(ap)
-        mgr._skills_by_scope[mgr._scope_key(_CONTEXT)] = {'primary': _make_skill_data(name='primary')}
-        ap.skill_mgr = mgr
-
-        query = _make_query()
-
-        assert register_activated_skill(ap, query, 'missing') is False
-        assert ACTIVATED_SKILLS_KEY not in query.variables
-
-    def test_register_activated_skill_without_skill_manager_returns_false(self):
-        from langbot.pkg.skill.activation import register_activated_skill
-
-        ap = _make_ap()  # no skill_mgr attribute
-        query = _make_query()
-
-        assert register_activated_skill(ap, query, 'primary') is False
-
-
 class TestSkillPathHelpers:
    def test_get_visible_skills_filters_by_bound_names(self):
        from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY, get_visible_skills
@@ -178,28 +128,49 @@ class TestSkillPathHelpers:

        assert list(result.keys()) == ['visible']

-    def test_restore_activated_skills_uses_caller_provided_names_and_visibility(self):
+    @pytest.mark.asyncio
+    async def test_restore_activated_skills_uses_exact_revision_bindings(self):
        from langbot.pkg.provider.tools.loaders.skill import (
            ACTIVATED_SKILLS_KEY,
            PIPELINE_BOUND_SKILLS_KEY,
+            get_activated_skill_bindings,
            get_activated_skill_names,
            restore_activated_skills,
        )

        ap = _make_ap()
-        ap.skill_mgr = _make_skill_manager(
-            {
-                'visible': _make_skill_data(name='visible'),
-                'hidden': _make_skill_data(name='hidden'),
-            }
+        pinned = _make_skill_data(name='visible', revision='sha256:' + '2' * 64)
+        ap.skill_repository = SimpleNamespace(
+            get_skill=AsyncMock(return_value=pinned),
        )
        query = _make_query(variables={PIPELINE_BOUND_SKILLS_KEY: ['visible']})

-        restored = restore_activated_skills(ap, query, ['visible', 'hidden', 'visible', ''])
+        restored = await restore_activated_skills(
+            ap,
+            query,
+            [{'name': 'visible', 'revision': pinned['revision']}],
+        )

        assert restored == ['visible']
        assert list(query.variables[ACTIVATED_SKILLS_KEY].keys()) == ['visible']
        assert get_activated_skill_names(query) == ['visible']
+        assert get_activated_skill_bindings(query) == [{'name': 'visible', 'revision': pinned['revision']}]
+        ap.skill_repository.get_skill.assert_awaited_once_with(
+            _CONTEXT,
+            'visible',
+            snapshot=True,
+            revision=pinned['revision'],
+        )
+
+    @pytest.mark.asyncio
+    async def test_restore_rejects_name_only_state(self):
+        from langbot.pkg.provider.tools.loaders.skill import restore_activated_skills
+
+        ap = _make_ap()
+        ap.skill_repository = SimpleNamespace(get_skill=AsyncMock())
+
+        with pytest.raises(ValueError, match='cannot use names without revisions'):
+            await restore_activated_skills(ap, _make_query(), ['visible'])

    def test_resolve_virtual_skill_path_allows_visible_skill_reads(self):
        from langbot.pkg.provider.tools.loaders.skill import (
@@ -314,6 +285,37 @@ class TestSkillToolLoader:
        assert '<package-root>' not in result['content']
        assert set(query.variables[ACTIVATED_SKILLS_KEY].keys()) == {'demo'}

+    @pytest.mark.asyncio
+    async def test_reactivation_keeps_first_revision_and_instructions(self):
+        from langbot.pkg.provider.tools.loaders.skill_authoring import SkillToolLoader
+
+        v1 = _make_skill_data(
+            name='demo',
+            instructions='version one',
+            revision='sha256:' + '1' * 64,
+        )
+        v2 = _make_skill_data(
+            name='demo',
+            instructions='version two',
+            revision='sha256:' + '2' * 64,
+        )
+        ap = _make_ap()
+        ap.skill_mgr = _make_skill_manager({'demo': v1})
+        ap.skill_repository = SimpleNamespace(
+            get_skill=AsyncMock(side_effect=[v1, v2]),
+        )
+        ap.box_service = SimpleNamespace(is_workspace_sandbox_available=AsyncMock(return_value=False))
+        loader = SkillToolLoader(ap)
+        query = _make_query()
+
+        first = await loader.invoke_tool('activate', {'skill_name': 'demo'}, query)
+        second = await loader.invoke_tool('activate', {'skill_name': 'demo'}, query)
+
+        assert first['revision'] == v1['revision']
+        assert second['revision'] == v1['revision']
+        assert 'version one' in second['content']
+        assert 'version two' not in second['content']
+
    @pytest.mark.asyncio
    async def test_activate_unknown_skill_raises(self):
        from langbot.pkg.provider.tools.loaders.skill_authoring import (
@@ -381,11 +383,55 @@ class TestSkillToolLoader:
                'description': 'Imported from clone',
                'instructions': 'Do work',
            },
+            base_revision=None,
        )
        assert result['registered'] is True
        assert result['skill_name'] == 'cloned-skill'
        assert result['source_path'] == '/workspace/repo'

+    @pytest.mark.asyncio
+    async def test_register_skill_forwards_base_revision_for_update(self):
+        from langbot.pkg.provider.tools.loaders.skill_authoring import (
+            REGISTER_SKILL_TOOL_NAME,
+            SkillToolLoader,
+        )
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            draft = os.path.join(tmpdir, 'skill-drafts', 'demo')
+            os.makedirs(draft)
+            revision = 'sha256:' + '3' * 64
+            ap = _make_ap()
+            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={
+                        'name': 'demo',
+                        'display_name': 'Demo',
+                        'description': 'Updated',
+                        'instructions': 'v2',
+                    }
+                ),
+                import_skill_directory=AsyncMock(
+                    return_value=_make_skill_data(name='demo', revision='sha256:' + '4' * 64)
+                ),
+            )
+
+            result = await SkillToolLoader(ap).invoke_tool(
+                REGISTER_SKILL_TOOL_NAME,
+                {
+                    'path': '/workspace/skill-drafts/demo',
+                    'base_revision': revision,
+                },
+                _make_query(),
+            )
+
+        assert result['revision'] == 'sha256:' + '4' * 64
+        assert ap.skill_service.import_skill_directory.await_args.kwargs['base_revision'] == revision
+
    @pytest.mark.asyncio
    async def test_register_skill_rejects_workspace_escape(self):
        from langbot.pkg.provider.tools.loaders.skill_authoring import (
@@ -552,7 +598,7 @@ class TestNativeToolLoaderSkillPaths:
    @pytest.mark.asyncio
    async def test_read_visible_skill_file(self):
        from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
-        from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY
+        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill

        with tempfile.TemporaryDirectory() as tmpdir:
            skill_md = os.path.join(tmpdir, 'SKILL.md')
@@ -568,10 +614,12 @@ class TestNativeToolLoaderSkillPaths:
            ap.skill_mgr = _make_skill_manager({'demo': _make_skill_data(name='demo', package_root=tmpdir)})
            loader = NativeToolLoader(ap)

+            query = _make_query(query_id='q1')
+            register_activated_skill(query, ap.skill_mgr.skills['demo'])
            result = await loader.invoke_tool(
                'read',
                {'path': '/workspace/.skills/demo/SKILL.md'},
-                _make_query(query_id='q1', variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']}),
+                query,
            )

            assert result['ok'] is True
@@ -581,7 +629,7 @@ class TestNativeToolLoaderSkillPaths:
    @pytest.mark.asyncio
    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
+        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill

        with tempfile.TemporaryDirectory() as tmpdir:
            with open(os.path.join(tmpdir, 'SKILL.md'), 'w', encoding='utf-8') as file_obj:
@@ -593,14 +641,12 @@ class TestNativeToolLoaderSkillPaths:
                shares_filesystem_with_box=False,
            )
            ap.skill_repository = SimpleNamespace(
-                read_skill_file=AsyncMock(return_value={'content': 'repository-content'})
+                read_skill_resource=AsyncMock(return_value={'content': 'repository-content'})
            )
            ap.skill_mgr = _make_skill_manager({'demo': _make_skill_data(name='demo', package_root=tmpdir)})
            loader = NativeToolLoader(ap)
-            query = _make_query(
-                query_id='q-external-read',
-                variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']},
-            )
+            query = _make_query(query_id='q-external-read')
+            register_activated_skill(query, ap.skill_mgr.skills['demo'])

            result = await loader.invoke_tool(
                'read',
@@ -611,12 +657,17 @@ class TestNativeToolLoaderSkillPaths:
            assert result['ok'] is True
            assert result['content'] == 'repository-content'
            assert 'core-host-secret' not in repr(result)
-            ap.skill_repository.read_skill_file.assert_awaited_once_with(_CONTEXT, 'demo', 'SKILL.md')
+            ap.skill_repository.read_skill_resource.assert_awaited_once_with(
+                _CONTEXT,
+                'demo',
+                'SKILL.md',
+                expected_revision='sha256:' + '1' * 64,
+            )

    @pytest.mark.asyncio
    async def test_core_owned_skill_path_does_not_depend_on_runtime_topology(self):
        from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
-        from langbot.pkg.provider.tools.loaders.skill import PIPELINE_BOUND_SKILLS_KEY
+        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill

        with tempfile.TemporaryDirectory() as tmpdir:
            with open(os.path.join(tmpdir, 'secret.txt'), 'w', encoding='utf-8') as file_obj:
@@ -629,10 +680,8 @@ class TestNativeToolLoaderSkillPaths:
            )
            ap.skill_mgr = _make_skill_manager({'demo': _make_skill_data(name='demo', package_root=tmpdir)})
            loader = NativeToolLoader(ap)
-            query = _make_query(
-                query_id='q-external-no-protocol',
-                variables={PIPELINE_BOUND_SKILLS_KEY: ['demo']},
-            )
+            query = _make_query(query_id='q-external-no-protocol')
+            register_activated_skill(query, ap.skill_mgr.skills['demo'])

            result = await loader.invoke_tool(
                'grep',
@@ -652,13 +701,20 @@ class TestNativeToolLoaderSkillPaths:
        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill

        with tempfile.TemporaryDirectory() as tmpdir:
+            manifest_path = os.path.join(tmpdir, 'manifest.json')
+            with open(manifest_path, 'w', encoding='utf-8') as file:
+                file.write('{}')
            ap = _make_ap()
            ap.box_service = SimpleNamespace(
                available=True,
                default_workspace=tmpdir,
                execute_tool=AsyncMock(return_value={'ok': True}),
            )
-            skill_data = _make_skill_data(name='demo', package_root=tmpdir)
+            skill_data = _make_skill_data(
+                name='demo',
+                package_root=tmpdir,
+                manifest_path=manifest_path,
+            )
            ap.skill_mgr = _make_skill_manager(
                {'demo': skill_data},
                refresh_skill_from_disk=Mock(),
@@ -689,40 +745,45 @@ class TestNativeToolLoaderSkillPaths:
        from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill

-        ap = _make_ap()
-        ap.box_service = SimpleNamespace(
-            available=True,
-            shares_filesystem_with_box=False,
-            execute_tool=AsyncMock(return_value={'ok': True}),
-        )
-        skill_data = _make_skill_data(
-            name='demo',
-            package_root='/box-runtime/skills/tenants/workspace/demo',
-            python_project=True,
-        )
-        ap.skill_mgr = _make_skill_manager(
-            {'demo': skill_data},
-            refresh_skill_from_disk=Mock(),
-        )
-        loader = NativeToolLoader(ap)
-        query = _make_query(query_id='q-external', launcher_type='person', launcher_id='123')
-        register_activated_skill(query, skill_data)
+        with tempfile.TemporaryDirectory() as tmpdir:
+            manifest_path = os.path.join(tmpdir, 'manifest.json')
+            with open(manifest_path, 'w', encoding='utf-8') as file:
+                file.write('{}')
+            ap = _make_ap()
+            ap.box_service = SimpleNamespace(
+                available=True,
+                shares_filesystem_with_box=False,
+                execute_tool=AsyncMock(return_value={'ok': True}),
+            )
+            skill_data = _make_skill_data(
+                name='demo',
+                package_root=tmpdir,
+                manifest_path=manifest_path,
+                python_project=True,
+            )
+            ap.skill_mgr = _make_skill_manager(
+                {'demo': skill_data},
+                refresh_skill_from_disk=Mock(),
+            )
+            loader = NativeToolLoader(ap)
+            query = _make_query(query_id='q-external', launcher_type='person', launcher_id='123')
+            register_activated_skill(query, skill_data)

-        result = await loader.invoke_tool(
-            'exec',
-            {
-                'command': 'python /workspace/.skills/demo/scripts/run.py',
-                'workdir': '/workspace/.skills/demo',
-            },
-            query,
-        )
+            result = await loader.invoke_tool(
+                'exec',
+                {
+                    'command': 'python /workspace/.skills/demo/scripts/run.py',
+                    'workdir': '/workspace/.skills/demo',
+                },
+                query,
+            )

        assert result['ok'] is True
        tool_parameters = ap.box_service.execute_tool.await_args.args[0]
        wrapped = tool_parameters['command']
-        assert '_LB_VENV_DIR="/workspace/.skill-envs/demo/.venv"' in wrapped
+        assert f'_LB_VENV_DIR="/workspace/.skill-envs/demo/{"1" * 64}/.venv"' in wrapped
        assert 'root = "/workspace/.skills/demo"' in wrapped
-        assert '/box-runtime/skills/tenants/workspace/demo' not in wrapped
+        assert tmpdir not in wrapped
        assert 'skill_name' not in ap.box_service.execute_tool.await_args.kwargs

    @pytest.mark.asyncio
@@ -744,3 +805,33 @@ class TestNativeToolLoaderSkillPaths:
                    {'path': '/workspace/.skills/demo/notes.txt', 'content': 'hi'},
                    query,
                )
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('tool_name', ['write', 'edit'])
+    async def test_published_revision_rejects_direct_mutation(self, tool_name):
+        from langbot.pkg.provider.tools.loaders.native import NativeToolLoader
+        from langbot.pkg.provider.tools.loaders.skill import register_activated_skill
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            ap = _make_ap()
+            ap.box_service = SimpleNamespace(available=True, default_workspace=tmpdir)
+            skill = _make_skill_data(name='demo', package_root=tmpdir)
+            ap.skill_mgr = _make_skill_manager({'demo': skill})
+            loader = NativeToolLoader(ap)
+            query = _make_query(query_id='immutable')
+            register_activated_skill(query, skill)
+            parameters = (
+                {'path': '/workspace/.skills/demo/notes.txt', 'content': 'hi'}
+                if tool_name == 'write'
+                else {
+                    'path': '/workspace/.skills/demo/notes.txt',
+                    'old_string': 'old',
+                    'new_string': 'new',
+                }
+            )
+
+            result = await loader.invoke_tool(tool_name, parameters, query)
+
+            assert result['ok'] is False
+            assert 'immutable' in result['error']
+            assert '/workspace/skill-drafts' in result['error']
@@ -1,9 +1,10 @@
 from types import SimpleNamespace
+from unittest.mock import Mock

 import pytest

 from langbot.pkg.api.http.context import ExecutionContext
-from langbot.pkg.skill.repository import SkillRepository, SkillRevisionMismatchError
+from langbot.pkg.skill.repository import SkillRepository, SkillRevisionConflictError


 _CONTEXT = ExecutionContext(
@@ -79,7 +80,7 @@ def test_repository_keeps_old_box_root_only_for_online_upgrade(tmp_path):
 async def test_repository_crud_and_reads_do_not_require_box(tmp_path):
    repository = _repository(tmp_path)

-    await repository.create_skill(
+    created = await repository.create_skill(
        _CONTEXT,
        {
            'name': 'docs-only',
@@ -93,11 +94,12 @@ async def test_repository_crud_and_reads_do_not_require_box(tmp_path):
        'docs-only',
        'references/guide.md',
        '# Guide\n\nNo execution needed.',
+        base_revision=created['revision'],
    )

    skill = await repository.get_skill(_CONTEXT, 'docs-only', snapshot=True)
    assert skill is not None
-    assert skill['revision'].startswith('stat-v1:')
+    assert skill['revision'].startswith('sha256:')
    assert [item['name'] for item in await repository.list_skills(_CONTEXT)] == ['docs-only']

    listed = await repository.list_skill_resources(
@@ -120,27 +122,47 @@ async def test_repository_crud_and_reads_do_not_require_box(tmp_path):


@pytest.mark.asyncio
-async def test_repository_rejects_traversal_and_stale_revision(tmp_path):
+async def test_repository_rejects_traversal_and_conflicting_update(tmp_path):
    repository = _repository(tmp_path)
-    await repository.create_skill(
+    created = await repository.create_skill(
        _CONTEXT,
        {'name': 'safe', 'description': 'Safe', 'instructions': 'Use the reference.'},
    )
-    await repository.write_skill_file(_CONTEXT, 'safe', 'reference.md', 'first')
+    first = await repository.write_skill_file(
+        _CONTEXT,
+        'safe',
+        'reference.md',
+        'first',
+        base_revision=created['revision'],
+    )
    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(
+    second = await repository.write_skill_file(
+        _CONTEXT,
+        'safe',
+        'reference.md',
+        'second',
+        base_revision=first['revision'],
+    )
+    pinned = await repository.read_skill_resource(
+        _CONTEXT,
+        'safe',
+        'reference.md',
+        expected_revision=skill['revision'],
+    )
+    assert pinned['content'] == 'first'
+    with pytest.raises(SkillRevisionConflictError, match='changed since'):
+        await repository.update_skill(
            _CONTEXT,
            'safe',
-            'reference.md',
-            expected_revision=skill['revision'],
+            {'instructions': 'stale'},
+            base_revision=first['revision'],
        )
+    assert second['revision'] != first['revision']


@pytest.mark.asyncio
@@ -207,3 +229,89 @@ async def test_repository_imports_only_from_the_fenced_workspace(tmp_path):
    (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))
+
+
+@pytest.mark.asyncio
+async def test_existing_run_keeps_v1_while_new_run_resolves_v2(tmp_path):
+    from langbot.pkg.provider.tools.loaders import skill as skill_loader
+
+    repository = _repository(tmp_path)
+    created = await repository.create_skill(
+        _CONTEXT,
+        {'name': 'runner', 'instructions': 'Run scripts/main.py'},
+    )
+    v1_write = await repository.write_skill_file(
+        _CONTEXT,
+        'runner',
+        'scripts/main.py',
+        "print('v1')",
+        base_revision=created['revision'],
+    )
+    v1 = await repository.get_skill(_CONTEXT, 'runner', snapshot=True)
+    old_run = SimpleNamespace(variables={})
+    skill_loader.register_activated_skill(old_run, v1)
+
+    await repository.write_skill_file(
+        _CONTEXT,
+        'runner',
+        'scripts/main.py',
+        "print('v2')",
+        base_revision=v1_write['revision'],
+    )
+    v2 = await repository.get_skill(_CONTEXT, 'runner', snapshot=True)
+    new_run = SimpleNamespace(variables={})
+    skill_loader.register_activated_skill(new_run, v2)
+
+    old_resource = await repository.read_skill_resource(
+        _CONTEXT,
+        'runner',
+        'scripts/main.py',
+        expected_revision=v1['revision'],
+    )
+    new_resource = await repository.read_skill_resource(
+        _CONTEXT,
+        'runner',
+        'scripts/main.py',
+        expected_revision=v2['revision'],
+    )
+    app = SimpleNamespace(logger=Mock())
+    old_mount = skill_loader.build_execution_mounts(app, old_run)[0]
+    new_mount = skill_loader.build_execution_mounts(app, new_run)[0]
+
+    assert old_resource['content'] == "print('v1')"
+    assert new_resource['content'] == "print('v2')"
+    assert old_mount['host_path'] == v1['package_root']
+    assert new_mount['host_path'] == v2['package_root']
+    assert old_mount['content_digest'] == v1['revision']
+    assert new_mount['content_digest'] == v2['revision']
+
+
+@pytest.mark.asyncio
+async def test_deleted_skill_can_restore_exact_recoverable_run_revision(tmp_path):
+    from langbot.pkg.provider.tools.loaders import skill as skill_loader
+
+    repository = _repository(tmp_path)
+    published = await repository.create_skill(
+        _CONTEXT,
+        {'name': 'recoverable', 'instructions': 'Pinned instructions'},
+    )
+    await repository.delete_skill(_CONTEXT, 'recoverable')
+    app = SimpleNamespace(skill_repository=repository)
+    query = SimpleNamespace(
+        variables={skill_loader.PIPELINE_BOUND_SKILLS_KEY: ['recoverable']},
+        instance_uuid=_CONTEXT.instance_uuid,
+        workspace_uuid=_CONTEXT.workspace_uuid,
+        placement_generation=_CONTEXT.placement_generation,
+        bot_uuid=None,
+        pipeline_uuid=None,
+        query_uuid='recovered-query',
+    )
+
+    restored = await skill_loader.restore_activated_skills(
+        app,
+        query,
+        [{'name': 'recoverable', 'revision': published['revision']}],
+    )
+
+    assert restored == ['recoverable']
+    assert skill_loader.get_activated_skill(query, 'recoverable')['revision'] == published['revision']
@@ -31,11 +31,11 @@ class TestSkillRepositoryBoundary:
        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'}),
+            create_skill=AsyncMock(return_value={'name': 'x', 'instructions': 'Do work', 'revision': 'sha256:v1'}),
+            update_skill=AsyncMock(return_value={'name': 'x', 'instructions': 'Updated', 'revision': 'sha256:v2'}),
            delete_skill=AsyncMock(),
            read_skill_file=AsyncMock(return_value={'path': 'a.txt', 'content': 'hello'}),
-            write_skill_file=AsyncMock(return_value={'path': 'a.txt'}),
+            write_skill_file=AsyncMock(return_value={'path': 'a.txt', 'revision': 'sha256:v3'}),
        )
        return SimpleNamespace(
            skill_mgr=SimpleNamespace(reload_skills=AsyncMock()),
@@ -62,12 +62,33 @@ class TestSkillRepositoryBoundary:
        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')
+        await service.update_skill(
+            _CONTEXT,
+            'x',
+            {'instructions': 'Updated', 'base_revision': 'sha256:v1'},
+        )
+        await service.write_skill_file(
+            _CONTEXT,
+            'x',
+            'a.txt',
+            'hello',
+            base_revision='sha256:v2',
+        )

        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')
+        ap.skill_repository.update_skill.assert_awaited_once_with(
+            _CONTEXT,
+            'x',
+            {'instructions': 'Updated'},
+            base_revision='sha256:v1',
+        )
+        ap.skill_repository.write_skill_file.assert_awaited_once_with(
+            _CONTEXT,
+            'x',
+            'a.txt',
+            'hello',
+            base_revision='sha256:v2',
+        )

    @pytest.mark.asyncio
    async def test_get_skill_returns_repository_revision(self):