42 lines
1.3 KiB
Python
42 lines
1.3 KiB
Python
"""A skill snapshot is reusable only inside one explicitly enabled tool call."""
|
|
|
|
from collections.abc import Callable
|
|
from contextvars import ContextVar
|
|
from dataclasses import dataclass
|
|
from functools import wraps
|
|
from typing import TYPE_CHECKING
|
|
|
|
from deerflow.config.runtime_performance_config import get_runtime_performance_config
|
|
|
|
if TYPE_CHECKING:
|
|
from deerflow.sandbox.skill_view import SkillSandboxView
|
|
|
|
|
|
@dataclass
|
|
class SkillViewScope:
|
|
view: "SkillSandboxView | None" = None
|
|
error: Exception | None = None
|
|
|
|
|
|
_scope: ContextVar[SkillViewScope | None] = ContextVar("sandbox_skill_view_scope", default=None)
|
|
|
|
|
|
def get_active_skill_view_scope() -> SkillViewScope | None:
|
|
return _scope.get()
|
|
|
|
|
|
def with_skill_view_scope[**P, R](func: Callable[P, R]) -> Callable[P, R]:
|
|
@wraps(func)
|
|
def wrapped(*args: P.args, **kwargs: P.kwargs) -> R:
|
|
if not get_runtime_performance_config().skill_view_reuse_enabled:
|
|
return func(*args, **kwargs)
|
|
token = _scope.set(SkillViewScope())
|
|
try:
|
|
return func(*args, **kwargs)
|
|
finally:
|
|
# Do not leak a previously enabled skill into the next tool call,
|
|
# including when a tool raises or shares an executor thread.
|
|
_scope.reset(token)
|
|
|
|
return wrapped
|