"""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