feat(skill): pin run revision bindings

This commit is contained in:
huanghuoguoguo
2026-09-28 10:22:01 +08:00
parent 1ab1a29b53
commit 9b8d1b4034
16 changed files with 657 additions and 274 deletions
@@ -5,6 +5,7 @@ import quart
from ...authz import Permission
from ...context import RequestContext
from .....operation_trace import service as settings_service
from .....skill.repository import SkillRevisionConflictError
from .. import group
@@ -76,6 +77,8 @@ class SkillsRouterGroup(group.RouterGroup):
previous = None
try:
skill = await self.ap.skill_service.update_skill(request_context, skill_name, data)
except SkillRevisionConflictError as exc:
return self.http_status(409, -1, str(exc))
except ValueError as exc:
return self.http_status(400, -1, str(exc))
@@ -115,6 +118,8 @@ class SkillsRouterGroup(group.RouterGroup):
include_hidden=include_hidden,
)
return self.success(data=result)
except SkillRevisionConflictError as exc:
return self.http_status(409, -1, str(exc))
except ValueError as exc:
return self.http_status(400, -1, str(exc))
@@ -140,11 +145,18 @@ class SkillsRouterGroup(group.RouterGroup):
async def write_skill_file(skill_name: str, path: str, request_context: RequestContext) -> quart.Response:
data = await quart.request.json
content = data.get('content', '')
base_revision = str(data.get('base_revision', '') or '').strip() or None
if content is None:
return self.http_status(400, -1, 'Missing required field: content')
try:
result = await self.ap.skill_service.write_skill_file(request_context, skill_name, path, content)
result = await self.ap.skill_service.write_skill_file(
request_context,
skill_name,
path,
content,
base_revision=base_revision,
)
return self.success(data=result)
except ValueError as exc:
return self.http_status(400, -1, str(exc))
+38 -5
View File
@@ -103,15 +103,34 @@ class SkillService:
await self._reload_skills(execution_context)
return self._serialize_skill(created)
async def import_skill_directory(self, context: TenantContext, path: str, data: dict) -> dict:
async def import_skill_directory(
self,
context: TenantContext,
path: str,
data: dict,
*,
base_revision: str | None = None,
) -> dict:
execution_context = await self._execution_context(context)
created = await self._repository().import_skill_directory(execution_context, path, data)
created = await self._repository().import_skill_directory(
execution_context,
path,
data,
base_revision=base_revision,
)
await self._reload_skills(execution_context)
return self._serialize_skill(created)
async def update_skill(self, context: TenantContext, skill_name: str, data: dict) -> dict:
execution_context = await self._execution_context(context)
updated = await self._repository().update_skill(execution_context, skill_name, data)
payload = dict(data)
base_revision = str(payload.pop('base_revision', '') or '').strip() or None
updated = await self._repository().update_skill(
execution_context,
skill_name,
payload,
base_revision=base_revision,
)
await self._reload_skills(execution_context)
return self._serialize_skill(updated)
@@ -142,9 +161,23 @@ class SkillService:
execution_context = await self._execution_context(context)
return await self._repository().read_skill_file(execution_context, skill_name, path)
async def write_skill_file(self, context: TenantContext, skill_name: str, path: str, content: str) -> dict:
async def write_skill_file(
self,
context: TenantContext,
skill_name: str,
path: str,
content: str,
*,
base_revision: str | None,
) -> dict:
execution_context = await self._execution_context(context)
result = await self._repository().write_skill_file(execution_context, skill_name, path, content)
result = await self._repository().write_skill_file(
execution_context,
skill_name,
path,
content,
base_revision=base_revision,
)
await self._reload_skills(execution_context)
return result
@@ -340,11 +340,14 @@ class NativeToolLoader(loader.ToolLoader):
if 'python_project' not in selected_skill:
python_project = skill_loader.should_prepare_skill_python_env(package_root)
if python_project:
revision_key = str(selected_skill.get('revision', '') or '').removeprefix('sha256:')
if not revision_key:
raise ValueError(f'Activated skill "{selected_skill_name}" has no pinned revision.')
parameters = dict(parameters)
parameters['command'] = skill_loader.wrap_skill_command_with_python_env(
command,
mount_path=skill_mount,
state_path=f'/workspace/.skill-envs/{selected_skill_name}',
state_path=f'/workspace/.skill-envs/{selected_skill_name}/{revision_key}',
)
# All exec calls (with or without skills) go through the same container
@@ -1119,25 +1122,27 @@ else:
skill_request = self._resolve_skill_relative_path(
query,
path,
include_visible=True,
include_visible=False,
include_activated=True,
)
skill_repository = getattr(self.ap, 'skill_repository', None)
if skill_request is not None and skill_repository is not None:
selected_skill, relative = skill_request
try:
result = await skill_repository.read_skill_file(
result = await skill_repository.read_skill_resource(
self._execution_context(query),
selected_skill['name'],
relative,
expected_revision=selected_skill.get('revision'),
)
return self._build_read_result_from_text(str(result.get('content', '')), parameters)
except Exception:
try:
result = await skill_repository.list_skill_files(
result = await skill_repository.list_skill_resources(
self._execution_context(query),
selected_skill['name'],
relative,
expected_revision=selected_skill.get('revision'),
)
entries = [entry['name'] for entry in result.get('entries', [])]
return self._build_directory_result(entries)
@@ -1147,7 +1152,7 @@ else:
host_location = self._resolve_host_location(
query,
path,
include_visible=True,
include_visible=False,
include_activated=True,
)
if host_location is None:
@@ -1161,22 +1166,22 @@ else:
path = parameters['path']
content = parameters['content']
self.ap.logger.info(f'write tool invoked: query_id={query.query_id} path={path} length={len(content)}')
encoding, _mode = self._write_options(parameters)
self._write_options(parameters)
skill_request = self._resolve_skill_relative_path(
query,
path,
include_visible=False,
include_activated=True,
)
skill_repository = getattr(self.ap, 'skill_repository', None)
if skill_request is not None and skill_repository is not None:
if encoding != 'text':
return {'ok': False, 'error': 'base64 writes to skill packages are not supported.'}
selected_skill, relative = skill_request
execution_context = self._execution_context(query)
await skill_repository.write_skill_file(execution_context, selected_skill['name'], relative, content)
await self.ap.skill_mgr.reload_skills(execution_context)
return {'ok': True, 'path': path}
if skill_request is not None:
return {
'ok': False,
'error': (
'Published Skill revisions are immutable. Copy the package to a writable '
'directory under /workspace/skill-drafts, edit it there, then call register_skill '
'with the activated revision as base_revision.'
),
}
host_location = self._resolve_host_location(
query,
@@ -1207,32 +1212,15 @@ else:
include_visible=False,
include_activated=True,
)
if skill_request is not None and getattr(self.ap, 'skill_repository', None) is not None:
selected_skill, relative = skill_request
try:
result = await self.ap.skill_repository.read_skill_file(
self._execution_context(query),
selected_skill['name'],
relative,
)
except Exception:
return {'ok': False, 'error': f'File not found: {path}'}
content = result.get('content', '')
count = content.count(old_string)
if count == 0:
return {'ok': False, 'error': 'old_string not found in file.'}
if count > 1:
return {'ok': False, 'error': f'old_string matches {count} locations; provide a more unique string.'}
new_content = content.replace(old_string, new_string, 1)
execution_context = self._execution_context(query)
await self.ap.skill_repository.write_skill_file(
execution_context,
selected_skill['name'],
relative,
new_content,
)
await self.ap.skill_mgr.reload_skills(execution_context)
return {'ok': True, 'path': path}
if skill_request is not None:
return {
'ok': False,
'error': (
'Published Skill revisions are immutable. Copy the package to a writable '
'directory under /workspace/skill-drafts, edit it there, then call register_skill '
'with the activated revision as base_revision.'
),
}
host_location = self._resolve_host_location(
query,
@@ -1532,7 +1520,7 @@ else:
host_location = self._resolve_host_location(
query,
path,
include_visible=True,
include_visible=False,
include_activated=True,
)
if host_location is None:
@@ -1554,7 +1542,7 @@ else:
host_location = self._resolve_host_location(
query,
path,
include_visible=True,
include_visible=False,
include_activated=True,
)
if host_location is None:
+78 -16
View File
@@ -74,24 +74,31 @@ def register_created_skill_visibility(query: pipeline_query.Query, skill_name: s
bound_skills.append(skill_name)
def build_execution_mounts(ap: app.Application, query: pipeline_query.Query) -> list[dict]:
"""Translate visible Core-owned packages into generic read-only mounts."""
"""Mount only immutable revisions pinned by this run's activations."""
mounts: list[dict] = []
for skill_name, skill_data in get_visible_skills(ap, query).items():
for skill_name, skill_data in get_activated_skills(query).items():
package_root = str(skill_data.get('package_root', '') or '').strip()
manifest_path = str(skill_data.get('manifest_path', '') or '').strip()
revision = str(skill_data.get('revision', '') or '').strip()
if not package_root:
continue
raise ValueError(f'Activated skill "{skill_name}" has no immutable package root.')
if not revision:
raise ValueError(f'Activated skill "{skill_name}" has no pinned revision.')
if not os.path.isdir(package_root):
ap.logger.warning(
f'Skill "{skill_name}" package_root missing on the Core filesystem '
f'({package_root}); skipping its execution mount. Reload the skill catalog.'
raise ValueError(
f'Activated skill "{skill_name}" pinned revision {revision} is unavailable; '
'the run cannot be recovered safely.'
)
continue
if not manifest_path or not os.path.isfile(manifest_path):
raise ValueError(f'Activated skill "{skill_name}" pinned revision {revision} has no publication manifest.')
mounts.append(
{
'host_path': package_root,
'mount_path': get_virtual_skill_mount_path(skill_name),
'mode': 'ro',
'content_digest': revision,
'manifest_path': manifest_path,
}
)
return mounts
@@ -111,14 +118,18 @@ def get_activated_skill(query: pipeline_query.Query, skill_name: str) -> dict |
return get_activated_skills(query).get(skill_name)
def register_activated_skill(query: pipeline_query.Query, skill_data: dict) -> None:
def register_activated_skill(query: pipeline_query.Query, skill_data: dict) -> dict:
if query.variables is None:
query.variables = {}
activated = query.variables.setdefault(ACTIVATED_SKILLS_KEY, {})
skill_name = str(skill_data.get('name', '') or '').strip()
if skill_name and skill_name not in activated:
activated[skill_name] = skill_data
revision = str(skill_data.get('revision', '') or '').strip()
if not skill_name or not revision:
raise ValueError('Activated Skills require a name and immutable revision.')
if skill_name not in activated:
activated[skill_name] = dict(skill_data)
return activated[skill_name]
def normalize_skill_names(value: typing.Any) -> list[str]:
@@ -139,17 +150,68 @@ def get_activated_skill_names(query: pipeline_query.Query) -> list[str]:
return normalize_skill_names(list(get_activated_skills(query).keys()))
def restore_activated_skills(
def get_activated_skill_bindings(query: pipeline_query.Query) -> list[dict[str, str]]:
"""Return exact bindings suitable for persistence and restart recovery."""
return [
{'name': name, 'revision': str(skill.get('revision', '') or '')}
for name, skill in get_activated_skills(query).items()
]
def normalize_skill_bindings(value: typing.Any) -> list[dict[str, str]]:
if not isinstance(value, list):
raise ValueError('Activated Skill recovery requires revision bindings.')
bindings: list[dict[str, str]] = []
seen: dict[str, str] = {}
for item in value:
if not isinstance(item, dict):
raise ValueError('Activated Skill recovery cannot use names without revisions.')
name = str(item.get('name', '') or '').strip()
revision = str(item.get('revision', '') or '').strip()
if not name or not revision:
raise ValueError('Each activated Skill binding requires name and revision.')
previous = seen.get(name)
if previous is not None and previous != revision:
raise ValueError(f'Conflicting pinned revisions supplied for Skill "{name}".')
if previous is None:
seen[name] = revision
bindings.append({'name': name, 'revision': revision})
return bindings
async def restore_activated_skills(
ap: app.Application,
query: pipeline_query.Query,
skill_names: typing.Any,
skill_bindings: typing.Any,
) -> list[str]:
"""Restore caller-provided names from the current visible skill set."""
"""Restore exact revisions or fail explicitly; never resolve latest by name."""
repository = getattr(ap, 'skill_repository', None)
if repository is None:
raise ValueError('Skill repository is unavailable during run recovery.')
context = ExecutionContext(
instance_uuid=str(getattr(query, 'instance_uuid', '') or ''),
workspace_uuid=str(getattr(query, 'workspace_uuid', '') or ''),
placement_generation=getattr(query, 'placement_generation', 0) or 0,
bot_uuid=getattr(query, 'bot_uuid', None),
pipeline_uuid=getattr(query, 'pipeline_uuid', None),
query_uuid=getattr(query, 'query_uuid', None),
)
bound_names = get_bound_skill_names(query)
restored: list[str] = []
for skill_name in normalize_skill_names(skill_names):
skill_data = get_visible_skill(ap, query, skill_name)
for binding in normalize_skill_bindings(skill_bindings):
skill_name = binding['name']
if bound_names is not None and skill_name not in bound_names:
raise ValueError(f'Skill "{skill_name}" is no longer authorized for this recoverable run.')
skill_data = await repository.get_skill(
context,
skill_name,
snapshot=True,
revision=binding['revision'],
)
if skill_data is None:
continue
raise ValueError(f'Skill "{skill_name}" pinned revision {binding["revision"]} is unavailable.')
register_activated_skill(query, skill_data)
restored.append(skill_name)
return restored
@@ -172,7 +172,8 @@ class SkillToolLoader(loader.ToolLoader):
if skill_data is None:
raise ValueError(f'Skill "{skill_name}" is no longer available; reload the skill catalog.')
skill_loader.register_activated_skill(query, skill_data)
# Keep the first revision pinned if the Skill is activated again in this run.
skill_data = skill_loader.register_activated_skill(query, skill_data)
await skill_loader.persist_activated_skill(self.ap, query, skill_name)
instructions = skill_data.get('instructions', '')
@@ -289,8 +290,8 @@ class SkillToolLoader(loader.ToolLoader):
if not skill_name:
raise ValueError('skill name is required')
# Create the skill
created = await skill_service.import_skill_directory(
base_revision = str(parameters.get('base_revision', '') or '').strip() or None
published = await skill_service.import_skill_directory(
execution_context,
host_path,
{
@@ -299,6 +300,7 @@ class SkillToolLoader(loader.ToolLoader):
'description': str(parameters.get('description') or scanned.get('description', '')).strip(),
'instructions': str(parameters.get('instructions') or scanned.get('instructions', '')),
},
base_revision=base_revision,
)
skill_loader.register_created_skill_visibility(query, skill_name)
@@ -306,7 +308,8 @@ class SkillToolLoader(loader.ToolLoader):
'registered': True,
'skill_name': skill_name,
'source_path': sandbox_path,
'skill': created,
'revision': published.get('revision'),
'skill': published,
}
def _resolve_workspace_directory(
@@ -405,9 +408,10 @@ class SkillToolLoader(loader.ToolLoader):
name=REGISTER_SKILL_TOOL_NAME,
human_desc='Register a skill from sandbox',
description=(
"Register a skill package from a directory under /workspace into LangBot's skill store. "
'Use this after creating or preparing a skill in the sandbox with exec/read/write/edit. '
"Publish a skill draft from a directory under /workspace into LangBot's skill store. "
'Use this after creating or preparing the draft with exec/read/write/edit. '
'The directory must contain a SKILL.md file. '
'Updating an existing Skill requires its current base_revision; conflicting updates are rejected. '
'After registration, the skill can be activated with the activate tool.'
),
parameters={
@@ -433,6 +437,10 @@ class SkillToolLoader(loader.ToolLoader):
'type': 'string',
'description': 'Optional instructions override.',
},
'base_revision': {
'type': 'string',
'description': 'Required current revision when publishing an update; omit for a new Skill.',
},
},
'required': ['path'],
'additionalProperties': False,
-44
View File
@@ -1,44 +0,0 @@
from __future__ import annotations
import typing
from ..provider.tools.loaders import skill as skill_loader
from ..api.http.context import ExecutionContext
if typing.TYPE_CHECKING:
from ..core import app
import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
# Skill activation is now handled through Tool Call mechanism (activate tool).
# This file is kept for potential future extensions but the text marker
# detection mechanism has been removed.
def register_activated_skill(
ap: app.Application,
query: pipeline_query.Query,
skill_name: str,
) -> bool:
"""Register an activated skill for sandbox mount path resolution.
This is called by the activate tool when a skill is activated via Tool Call.
"""
skill_mgr = getattr(ap, 'skill_mgr', None)
if skill_mgr is None:
return False
execution_context = ExecutionContext(
instance_uuid=str(getattr(query, 'instance_uuid', '') or ''),
workspace_uuid=str(getattr(query, 'workspace_uuid', '') or ''),
placement_generation=getattr(query, 'placement_generation', 0) or 0,
bot_uuid=getattr(query, 'bot_uuid', None),
pipeline_uuid=getattr(query, 'pipeline_uuid', None),
query_uuid=getattr(query, 'query_uuid', None),
)
skill_data = skill_mgr.get_skill_by_name(execution_context, skill_name)
if skill_data is None:
return False
skill_loader.register_activated_skill(query, skill_data)
return True
+59 -7
View File
@@ -5,7 +5,9 @@ import os
import weakref
from langbot_plugin.skill_store import (
SkillRevisionConflictError,
SkillRevisionMismatchError,
SkillRevisionNotFoundError,
SkillStore,
skill_namespace,
)
@@ -116,13 +118,29 @@ class SkillRepository:
async def list_skills(self, context: TenantContext) -> list[dict]:
return await self._call(context, 'list_skills')
async def get_skill(self, context: TenantContext, name: str, *, snapshot: bool = False) -> dict | None:
async def get_skill(
self,
context: TenantContext,
name: str,
*,
snapshot: bool = False,
revision: str | None = None,
) -> dict | None:
if revision is not None:
return await self._call(context, 'get_skill_snapshot', name, revision)
return await self._call(context, 'get_skill_snapshot' if snapshot else 'get_skill', name)
async def create_skill(self, context: TenantContext, skill: dict) -> dict:
return await self._call(context, 'create_skill', skill)
async def import_skill_directory(self, context: TenantContext, path: str, skill: dict) -> dict:
async def import_skill_directory(
self,
context: TenantContext,
path: str,
skill: dict,
*,
base_revision: str | None = None,
) -> dict:
namespace = self._namespace(context)
return await self._call(
context,
@@ -130,10 +148,24 @@ class SkillRepository:
path,
skill,
source_root=self._workspace_root(namespace),
base_revision=base_revision,
)
async def update_skill(self, context: TenantContext, name: str, skill: dict) -> dict:
return await self._call(context, 'update_skill', name, skill)
async def update_skill(
self,
context: TenantContext,
name: str,
skill: dict,
*,
base_revision: str | None,
) -> dict:
return await self._call(
context,
'update_skill',
name,
skill,
base_revision=base_revision,
)
async def delete_skill(self, context: TenantContext, name: str) -> None:
await self._call(context, 'delete_skill', name)
@@ -194,8 +226,23 @@ class SkillRepository:
expected_revision=expected_revision,
)
async def write_skill_file(self, context: TenantContext, name: str, path: str, content: str) -> dict:
return await self._call(context, 'write_skill_file', name, path, content)
async def write_skill_file(
self,
context: TenantContext,
name: str,
path: str,
content: str,
*,
base_revision: str | None,
) -> dict:
return await self._call(
context,
'write_skill_file',
name,
path,
content,
base_revision=base_revision,
)
async def preview_skill_zip(self, context: TenantContext, file_bytes: bytes, filename: str, **kwargs) -> list[dict]:
return await self._call(context, 'preview_zip_upload', file_bytes=file_bytes, filename=filename, **kwargs)
@@ -204,4 +251,9 @@ class SkillRepository:
return await self._call(context, 'install_zip_upload', file_bytes=file_bytes, filename=filename, **kwargs)
__all__ = ['SkillRepository', 'SkillRevisionMismatchError']
__all__ = [
'SkillRepository',
'SkillRevisionConflictError',
'SkillRevisionMismatchError',
'SkillRevisionNotFoundError',
]