mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-31 07:30:43 +00:00
d53ec767af
* fix(persistence): keep LF bytes in skill and agent text writes Text mode translates a newline into os.linesep on write, so on Windows every caller-supplied LF reached disk as CRLF while readers use universal newlines or decode bytes explicitly. The difference stayed invisible: the API returned LF, the file held CRLF, and a skill history record's prev_content described a revision no API view could return (#7174). Pass newline="" at the ten writers that persist or judge caller text: the local and user-scoped skill writers, the skill history JSONL, the user skill-states JSON, the SOUL.md/config.yaml staging in FileAgentStore, the managed-subagent store, setup_agent's SOUL.md publish, the USER.md route, and the two temp copies the frontmatter validator and the static scanner judge. The tests emulate the Windows translation on io.open, so the regression runs on the Linux CI shards instead of only on a Windows host: 10 failed before this commit, 10 pass after. * fix(skills): preserve candidate newlines during static scan --------- Co-authored-by: Felix Shi <204130665+shi-felix@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
388 lines
18 KiB
Python
388 lines
18 KiB
Python
"""Tool for creating and evolving custom skills."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import shutil
|
|
import tempfile
|
|
import threading
|
|
from collections import deque
|
|
from collections.abc import Awaitable, Callable
|
|
from concurrent.futures import Future
|
|
from pathlib import Path
|
|
from typing import Any, NoReturn
|
|
from weakref import WeakValueDictionary
|
|
|
|
from langchain.tools import tool
|
|
|
|
from deerflow.agents.lead_agent.prompt import SkillCacheResetPublishError, publish_skills_cache_reset, refresh_user_skills_system_prompt_cache_async
|
|
from deerflow.runtime.user_context import resolve_runtime_user_id
|
|
from deerflow.skills.package_files import is_code_file
|
|
from deerflow.skills.security_scanner import scan_skill_content
|
|
from deerflow.skills.security_static_scanner import (
|
|
StaticFinding,
|
|
StaticScanBlockedError,
|
|
StaticScannerError,
|
|
enforce_static_scan,
|
|
)
|
|
from deerflow.skills.storage import get_or_new_user_skill_storage
|
|
from deerflow.skills.storage.skill_storage import SkillStorage
|
|
from deerflow.skills.types import SKILL_MD_FILE
|
|
from deerflow.tools.sync import make_sync_tool_wrapper
|
|
from deerflow.tools.types import Runtime
|
|
from deerflow.utils.file_io import await_drained
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class _SkillLock:
|
|
"""Per-(user, skill) lock shared by every event loop that runs this tool.
|
|
|
|
The sync wrapper gives each call its own ``asyncio.run`` loop; an
|
|
``asyncio.Lock`` binds to one of them and strands waiters on the others
|
|
(#5846). Each waiter here parks on its own future, woken on its own loop,
|
|
so waiting holds no executor thread. Like ``asyncio.Lock``, release hands
|
|
the lock straight to the oldest waiter, so a later call cannot overtake it.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._state = threading.Lock()
|
|
self._locked = False
|
|
self._waiters: deque[tuple[Future[None], asyncio.AbstractEventLoop]] = deque()
|
|
|
|
def locked(self) -> bool:
|
|
with self._state:
|
|
return self._locked
|
|
|
|
async def __aenter__(self) -> None:
|
|
with self._state:
|
|
if not self._locked:
|
|
self._locked = True
|
|
return
|
|
waiter: Future[None] = Future()
|
|
self._waiters.append((waiter, asyncio.get_running_loop()))
|
|
try:
|
|
await asyncio.wrap_future(waiter)
|
|
except BaseException:
|
|
# cancel() fails only once release has handed us the lock: pass it
|
|
# on, or the callers queued behind us would wait forever.
|
|
if not waiter.cancel():
|
|
self._release()
|
|
raise
|
|
|
|
async def __aexit__(self, *_exc: object) -> None:
|
|
self._release()
|
|
|
|
def _release(self) -> None:
|
|
with self._state:
|
|
if not self._locked:
|
|
raise RuntimeError("release unlocked skill lock")
|
|
while self._waiters:
|
|
waiter, loop = self._waiters.popleft()
|
|
# Claiming the future makes a racing cancellation a no-op, so
|
|
# the lock is either handed over here or skipped to the next.
|
|
if not loop.is_closed() and waiter.set_running_or_notify_cancel():
|
|
break
|
|
else:
|
|
self._locked = False
|
|
return
|
|
# Outside the state lock: the result callback schedules the waiter's
|
|
# wakeup on its own loop.
|
|
waiter.set_result(None)
|
|
|
|
|
|
# Lock granularity: (user_id, skill_name) to avoid cross-user blocking.
|
|
_skill_locks: WeakValueDictionary[tuple[str, str], _SkillLock] = WeakValueDictionary()
|
|
_skill_locks_guard = threading.Lock()
|
|
|
|
|
|
def _get_lock(user_id: str, name: str) -> _SkillLock:
|
|
key = (user_id, name)
|
|
with _skill_locks_guard:
|
|
lock = _skill_locks.get(key)
|
|
if lock is None:
|
|
lock = _SkillLock()
|
|
_skill_locks[key] = lock
|
|
return lock
|
|
|
|
|
|
def _get_thread_id(runtime: Runtime | None) -> str | None:
|
|
if runtime is None:
|
|
return None
|
|
if runtime.context and runtime.context.get("thread_id"):
|
|
return runtime.context.get("thread_id")
|
|
return runtime.config.get("configurable", {}).get("thread_id")
|
|
|
|
|
|
def _history_record(*, action: str, file_path: str, prev_content: str | None, new_content: str | None, thread_id: str | None, scanner: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"action": action,
|
|
"author": "agent",
|
|
"thread_id": thread_id,
|
|
"file_path": file_path,
|
|
"prev_content": prev_content,
|
|
"new_content": new_content,
|
|
"scanner": scanner,
|
|
}
|
|
|
|
|
|
async def _scan_or_raise(content: str, *, executable: bool, location: str, static_findings: list[StaticFinding] | None = None) -> dict[str, Any]:
|
|
# In-graph: the graph root already attached tracing (see the INVARIANT in
|
|
# agents/lead_agent/agent.py), so the scan model must not attach it again.
|
|
result = await scan_skill_content(content, executable=executable, location=location, static_findings=static_findings or [], attach_tracing=False)
|
|
if result.decision == "block":
|
|
raise ValueError(f"Security scan blocked the write: {result.reason}")
|
|
if executable and result.decision != "allow":
|
|
raise ValueError(f"Security scan rejected executable content: {result.reason}")
|
|
return {"decision": result.decision, "reason": result.reason}
|
|
|
|
|
|
def _raise_static_block(error: StaticScanBlockedError) -> NoReturn:
|
|
payload = {
|
|
"skill_name": error.skill_name,
|
|
"findings": error.findings,
|
|
}
|
|
raise ValueError(f"{error} Findings: {json.dumps(payload, ensure_ascii=False)}") from error
|
|
|
|
|
|
def _raise_static_scan_failure(name: str, error: StaticScannerError) -> NoReturn:
|
|
raise ValueError(f"Static security scan failed for skill '{name}': {error}") from error
|
|
|
|
|
|
async def _scan_static_candidate_or_raise(name: str, updates: dict[str, str], skill_storage: SkillStorage | None = None) -> list[StaticFinding]:
|
|
def _scan_candidate() -> list[StaticFinding]:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
skill_dir = Path(tmp) / name
|
|
if skill_storage is None:
|
|
skill_dir.mkdir(parents=True)
|
|
else:
|
|
from deerflow.skills.mutations.guard import managed_read
|
|
|
|
with managed_read(skill_storage):
|
|
shutil.copytree(skill_storage.get_custom_skill_dir(name), skill_dir)
|
|
for relative_path, content in updates.items():
|
|
target = skill_dir / relative_path
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
# Match the bytes skill storage publishes, including on Windows.
|
|
# Default text mode would rewrite LF before the scanner reads it.
|
|
target.write_text(content, encoding="utf-8", newline="")
|
|
return enforce_static_scan(skill_dir, skill_name=name)
|
|
|
|
try:
|
|
return await _to_thread(_scan_candidate)
|
|
except StaticScanBlockedError as e:
|
|
_raise_static_block(e)
|
|
except StaticScannerError as e:
|
|
_raise_static_scan_failure(name, e)
|
|
|
|
|
|
async def _to_thread(func, /, *args, **kwargs):
|
|
return await asyncio.to_thread(func, *args, **kwargs)
|
|
|
|
|
|
async def _write_and_record(skill_storage: SkillStorage, name: str, relative_path: str, content: str, record: dict[str, Any]) -> None:
|
|
await _to_thread(skill_storage.write_custom_skill, name, relative_path, content)
|
|
await _to_thread(skill_storage.append_history, name, record)
|
|
|
|
|
|
async def _settle_skill_mutation(user_id: str, name: str, persist: Callable[[], Awaitable[object]]) -> None:
|
|
"""Apply a skill change and propagate it to every Gateway process sharing the home volume.
|
|
|
|
The write, its history record, this process's prompt-cache refresh and the
|
|
shared skills cache reset marker settle as one drained unit, like the
|
|
mutation tails of the Gateway skill routes: cancelling ``asyncio.to_thread``
|
|
abandons the await, not the worker, so a cancelled run would otherwise
|
|
unwind with the write landed and this process and its peers still serving
|
|
the previous skill list. The marker write is filesystem IO and runs off the
|
|
event loop.
|
|
"""
|
|
|
|
async def _tail() -> None:
|
|
await persist()
|
|
await refresh_user_skills_system_prompt_cache_async(user_id)
|
|
try:
|
|
await _to_thread(publish_skills_cache_reset, user_id=user_id)
|
|
except SkillCacheResetPublishError as exc:
|
|
# Logged here because a cancelled run never sees the raise below.
|
|
logger.error("Custom skill %r changed, but the shared skills cache reset could not be published", name, exc_info=True)
|
|
# The change is durable; say so, or the model reads a failed write and retries it.
|
|
raise SkillCacheResetPublishError(f"The change to custom skill '{name}' was applied, but other Gateway processes were not notified and may keep serving the previous skill list: {exc}") from exc
|
|
|
|
await await_drained(_tail())
|
|
|
|
|
|
async def _skill_manage_impl(
|
|
runtime: Runtime,
|
|
action: str,
|
|
name: str,
|
|
content: str | None = None,
|
|
path: str | None = None,
|
|
find: str | None = None,
|
|
replace: str | None = None,
|
|
expected_count: int | None = None,
|
|
) -> str:
|
|
"""Manage custom skills under skills/custom/.
|
|
|
|
Args:
|
|
action: One of create, patch, edit, delete, write_file, remove_file.
|
|
name: Skill name in hyphen-case.
|
|
content: New file content for create, edit, or write_file.
|
|
path: Supporting file path for write_file or remove_file.
|
|
find: Existing text to replace for patch.
|
|
replace: Replacement text for patch.
|
|
expected_count: Optional expected number of replacements for patch.
|
|
"""
|
|
name = SkillStorage.validate_skill_name(name)
|
|
user_id = resolve_runtime_user_id(runtime)
|
|
lock = _get_lock(user_id, name)
|
|
thread_id = _get_thread_id(runtime)
|
|
# The lookup resolves the app config (a config.yaml stat on every call)
|
|
# and, on a cold cache, the project root: blocking filesystem IO.
|
|
skill_storage = await _to_thread(get_or_new_user_skill_storage, user_id)
|
|
|
|
async with lock:
|
|
if action == "create":
|
|
if await _to_thread(skill_storage.custom_skill_exists, name):
|
|
raise ValueError(f"Custom skill '{name}' already exists.")
|
|
if content is None:
|
|
raise ValueError("content is required for create.")
|
|
await _to_thread(skill_storage.validate_skill_markdown_content, name, content)
|
|
static_findings = await _scan_static_candidate_or_raise(name, {SKILL_MD_FILE: content})
|
|
scan = await _scan_or_raise(content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
|
|
scan["static_findings"] = static_findings
|
|
record = _history_record(action="create", file_path=SKILL_MD_FILE, prev_content=None, new_content=content, thread_id=thread_id, scanner=scan)
|
|
await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, content, record))
|
|
return f"Created custom skill '{name}'."
|
|
if action == "edit":
|
|
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
|
|
if content is None:
|
|
raise ValueError("content is required for edit.")
|
|
await _to_thread(skill_storage.validate_skill_markdown_content, name, content)
|
|
static_findings = await _scan_static_candidate_or_raise(name, {SKILL_MD_FILE: content})
|
|
scan = await _scan_or_raise(content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
|
|
scan["static_findings"] = static_findings
|
|
prev_content = await _to_thread(skill_storage.read_custom_skill, name)
|
|
record = _history_record(action="edit", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan)
|
|
await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, content, record))
|
|
return f"Updated custom skill '{name}'."
|
|
|
|
if action == "patch":
|
|
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
|
|
if find is None or replace is None:
|
|
raise ValueError("find and replace are required for patch.")
|
|
prev_content = await _to_thread(skill_storage.read_custom_skill, name)
|
|
occurrences = prev_content.count(find)
|
|
if occurrences == 0:
|
|
raise ValueError("Patch target not found in SKILL.md.")
|
|
if expected_count is not None and occurrences != expected_count:
|
|
raise ValueError(f"Expected {expected_count} replacements but found {occurrences}.")
|
|
replacement_count = expected_count if expected_count is not None else 1
|
|
new_content = prev_content.replace(find, replace, replacement_count)
|
|
await _to_thread(skill_storage.validate_skill_markdown_content, name, new_content)
|
|
static_findings = await _scan_static_candidate_or_raise(name, {SKILL_MD_FILE: new_content})
|
|
scan = await _scan_or_raise(new_content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
|
|
scan["static_findings"] = static_findings
|
|
record = _history_record(action="patch", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=new_content, thread_id=thread_id, scanner=scan)
|
|
await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, new_content, record))
|
|
return f"Patched custom skill '{name}' ({replacement_count} replacement(s) applied, {occurrences} match(es) found)."
|
|
|
|
if action == "delete":
|
|
record = _history_record(
|
|
action="delete",
|
|
file_path=SKILL_MD_FILE,
|
|
prev_content=None,
|
|
new_content=None,
|
|
thread_id=thread_id,
|
|
scanner={"decision": "allow", "reason": "Deletion requested."},
|
|
)
|
|
await _settle_skill_mutation(user_id, name, lambda: _to_thread(skill_storage.delete_custom_skill, name, history_meta=record))
|
|
return f"Deleted custom skill '{name}'."
|
|
|
|
if action == "write_file":
|
|
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
|
|
if path is None or content is None:
|
|
raise ValueError("path and content are required for write_file.")
|
|
target = await _to_thread(skill_storage.ensure_safe_support_path, name, path)
|
|
if await _to_thread(target.is_dir):
|
|
raise ValueError(f"Supporting file path '{path}' is a directory, not a file.")
|
|
from deerflow.skills.mutations.guard import read_optional_text
|
|
|
|
prev_content = await _to_thread(read_optional_text, skill_storage, target)
|
|
# Executable content is whatever the shared package-file rule classifies
|
|
# as code: a file under the top-level ``scripts/`` directory, a code-suffix
|
|
# file anywhere in the package, or an extensionless file that starts with
|
|
# a shebang. Classify with that shared rule instead of re-deriving it; a
|
|
# substring test here also caught static support files such as
|
|
# references/scripts/overview.md and templates/my-scripts/setup.md, while
|
|
# the ``scripts/``-member rule alone would downgrade runnable files such as
|
|
# references/scripts/helper.py or assets/my-scripts/tool.js to static.
|
|
executable = is_code_file(path, content[:8].encode("utf-8", "surrogatepass"))
|
|
static_findings = await _scan_static_candidate_or_raise(name, {path: content}, skill_storage)
|
|
scan = await _scan_or_raise(content, executable=executable, location=f"{name}/{path}", static_findings=static_findings)
|
|
scan["static_findings"] = static_findings
|
|
record = _history_record(action="write_file", file_path=path, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan)
|
|
await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, path, content, record))
|
|
return f"Wrote '{path}' for custom skill '{name}'."
|
|
|
|
if action == "remove_file":
|
|
await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
|
|
if path is None:
|
|
raise ValueError("path is required for remove_file.")
|
|
|
|
async def _remove_and_record() -> None:
|
|
prev_content = await _to_thread(skill_storage.remove_custom_skill_file, name, path)
|
|
await _to_thread(
|
|
skill_storage.append_history,
|
|
name,
|
|
_history_record(action="remove_file", file_path=path, prev_content=prev_content, new_content=None, thread_id=thread_id, scanner={"decision": "allow", "reason": "Deletion requested."}),
|
|
)
|
|
|
|
await _settle_skill_mutation(user_id, name, _remove_and_record)
|
|
return f"Removed '{path}' from custom skill '{name}'."
|
|
|
|
if await _to_thread(skill_storage.public_skill_exists, name):
|
|
# public_skill_exists covers both built-in (PUBLIC) and legacy (LEGACY)
|
|
# skills; the UserScopedSkillStorage override distinguishes them in
|
|
# ensure_custom_skill_is_editable with category-specific messages.
|
|
raise ValueError(f"'{name}' is a read-only skill (built-in or legacy shared). To customise it, create your own version with the same name.")
|
|
raise ValueError(f"Unsupported action '{action}'.")
|
|
|
|
|
|
@tool("skill_manage", parse_docstring=True)
|
|
async def skill_manage_tool(
|
|
runtime: Runtime,
|
|
action: str,
|
|
name: str,
|
|
content: str | None = None,
|
|
path: str | None = None,
|
|
find: str | None = None,
|
|
replace: str | None = None,
|
|
expected_count: int | None = None,
|
|
) -> str:
|
|
"""Manage custom skills under skills/custom/.
|
|
|
|
Args:
|
|
action: One of create, patch, edit, delete, write_file, remove_file.
|
|
name: Skill name in hyphen-case.
|
|
content: New file content for create, edit, or write_file.
|
|
path: Supporting file path for write_file or remove_file.
|
|
find: Existing text to replace for patch.
|
|
replace: Replacement text for patch.
|
|
expected_count: Optional expected number of replacements for patch.
|
|
"""
|
|
return await _skill_manage_impl(
|
|
runtime=runtime,
|
|
action=action,
|
|
name=name,
|
|
content=content,
|
|
path=path,
|
|
find=find,
|
|
replace=replace,
|
|
expected_count=expected_count,
|
|
)
|
|
|
|
|
|
skill_manage_tool.func = make_sync_tool_wrapper(_skill_manage_impl, "skill_manage")
|