github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(skills): share skill cache invalidation across Gateway replicas (#6495)

Browse Source 
* fix(skills): share skill cache invalidation across Gateway replicas

SkillStorage.load_skills() rescans disk and re-reads extensions_config.json
on every call, but the prompt layer in deerflow/agents/lead_agent/prompt.py
caches the enabled-skills list, the per-(app_config, user_id) entries and
the rendered <skill_system> section per process, and every skill mutation
in routers/skills.py only refreshed the process that handled it. With
several uvicorn workers or several Pods sharing one home volume, a skill
installed, edited, deleted, rolled back or toggled through replica A kept
being offered as the old skill set by replica B until B restarted, and
POST /api/skills/reload had to be sent to every Pod separately.

Reuse the MCP cache-reset mechanism. A new generic helper,
deerflow.config.shared_reset_marker, owns a hidden JSON marker beside the
resolved extensions config (.<config name>.<suffix>.json): publish()
replaces it atomically with a random generation under
extensions_config_write_lock + extensions_config_file_lock, chaining
previous_generation and an optional user_id; SharedResetMarkerTracker
polls it at most once per second (monotonic clock, never blocking behind a
concurrent poller), adopts the current state silently on its first poll,
and classifies a change as user-scoped only when exactly one publication
carrying that user happened since the last poll. deerflow/mcp/cache.py now
builds its .<config>.mcp-cache-reset.json marker through the same helper
with no behavior change.

prompt.py publishes .<config>.skills-cache-reset.json via
publish_skills_cache_reset(user_id=...) and checks the marker on every
cache lookup (global cache, per-config cache, prompt-section LRU, user
scoped entries), calling _invalidate_enabled_skills_cache() or
invalidate_user_skill_cache(). The publishing process remembers its own
generation so it does not retire the caches it just rebuilt. Every
mutation tail in routers/skills.py (install/upload, edit, delete,
rollback, enable/disable, reload) refreshes locally and then publishes the
marker off the event loop; /skills/reload answers scope=shared_config,
or scope=process when no extensions config path resolves, and a marker
publication failure is a 500 so a 200 never means only one process was
refreshed.

Docs: README, backend/docs/API.md, skills/config/mcp/gateway AGENTS.md,
and frontend skills.mdx (en/zh; also fixes the reference to a
non-existent skills/loader.py) describe the shared invalidation and drop
the per-Pod guidance. CHANGELOG entry under Unreleased / Fixed.

* docs(changelog): link the skills cache reset entry to #6495

* fix(skills): drain the reload tail and keep marker failures out of 404

Two review follow-ups on the shared skills cache reset (#6495).

POST /skills/reload awaited the local refresh and the marker publication
unshielded. CancelledError is a BaseException, so a caller that
disconnected between the two slipped past the `except Exception -> 500`
mapping, surfaced no error, and left every peer replica serving the old
skill set -- the one outcome the operator hook for external mount writes
exists to prevent. The refresh and the publication now run as one
`await_drained(_drain_skill_mutation("reload", ...))` tail, the same
discipline the install/edit/delete/rollback/toggle tails already use; the
exception -> 500 mapping and the scope response are unchanged.

Publication errors could surface as 404. resolve_shared_config_path()
only swallows the resolution-time FileNotFoundError; a config directory
that vanishes between resolution and the lock file / atomic replace still
raised FileNotFoundError from the marker write, and the install, edit,
delete and rollback handlers mapped that to HTTP 404 with the raw OS
error text for a skill that had in fact been applied. publish_skills_cache_reset()
now re-raises OSError (and a misconfigured project root's ValueError) as
SkillCacheResetPublishError, a RuntimeError that no handler maps to
404/400, so every handler reports it through its generic 500 branch; the
message names only the exception type, never a server path. The install
handler gains an explicit clause that pins that contract beside its 404
mapping.

Tests: cancelling reload after the refresh still publishes the marker;
a FileNotFoundError from the marker write yields 500 (not 404) for
install, edit, delete and rollback, 500 for the public toggle and the
generic reload error; the wrapper type is not an OSError.
This commit is contained in:
Nan Gao 2026-10-08 19:43:04 +08:00 committed by GitHub
parent 46eecf2950
commit 4bab280250
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 1609 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
CHANGELOG.md
View File

@ -804,6 +804,24 @@ This release closes that milestone with **439 merged pull requests**.
### Fixed
- **skills:** Skill changes made through one Gateway replica now reach the
skill list in every other replica's system prompt. `SkillStorage` rescans
disk on every call, but the prompt layer caches the enabled-skills list and
the rendered `<skill_system>` section per process, and installing, editing,
deleting, rolling back or toggling a skill (and `POST /api/skills/reload`)
only refreshed the process that handled the request, so with several
uvicorn workers or several Pods sharing one home volume the other replicas
kept offering the old skills until they restarted. Every one of those
mutations now also publishes `.extensions_config.json.skills-cache-reset.json`
beside the shared `extensions_config.json`, with the same atomic replace and
cross-process locks the config uses, and every cache lookup compares that
marker's signature at most once per second before serving a cached entry; a
marker scoped to one user's custom skills retires only that user's entries.
`/api/skills/reload` reports `scope: shared_config` when the marker was
written and `scope: process` when no extensions config path resolves, so
operators no longer need to call it on every Pod or worker separately. The
MCP cache reset's marker now shares the same `deerflow.config.shared_reset_marker`
helper. ([#6495])
- **persistence:** `scripts/migrate_user_isolation.py` now moves each legacy
thread to the user who owns it. It looked for thread owners in
`{base_dir}/deer-flow.db`, a file DeerFlow never creates (the database is
@ -9202,3 +9220,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
[#6448]: https://github.com/bytedance/deer-flow/pull/6448
[#6450]: https://github.com/bytedance/deer-flow/pull/6450
[#6481]: https://github.com/bytedance/deer-flow/pull/6481
[#6495]: https://github.com/bytedance/deer-flow/pull/6495

2
README.md
View File

@ -1357,7 +1357,7 @@ executing the command, and broker execution logs omit argument values. See the
[broker image guide](docker/lark-cli-broker/README.md) for timeout settings and
image rebuild requirements.
If a trusted operator manages the configured skills directory through an external mount such as MinIO, NFS, or CSI, an administrator can call `POST /api/skills/reload` after changing files. This invalidates skill prompt caches for the current Gateway process and waits up to the bounded refresh timeout so subsequent runs rescan the latest files; running tasks are unchanged. A loader-level filesystem failure returns a generic server error and preserves the last successfully loaded process cache rather than publishing an empty catalog. Uvicorn workers and Kubernetes Pods must each be targeted separately. Direct mount writes bypass the validation, SkillScan, and history applied by DeerFlow's install/edit APIs, so only operator-controlled systems should have write access.
If a trusted operator manages the configured skills directory through an external mount such as MinIO, NFS, or CSI, an administrator can call `POST /api/skills/reload` after changing files. This invalidates skill prompt caches in the handling Gateway process, waits up to the bounded refresh timeout, and then advances a durable reset marker (`.extensions_config.json.skills-cache-reset.json`) in the writable config directory. Every Gateway worker or Pod mounting that directory notices the marker within about a second of its next prompt build and rescans the latest files, so one call reaches every replica sharing the volume; the response reports `scope: shared_config`, or `scope: process` when no extensions config path is available. Running tasks are unchanged. Skill installs, edits, deletions, rollbacks and enable/disable toggles made through the API publish the same marker, so a change made on one replica takes effect on the others without a restart. A loader-level filesystem failure returns a generic server error and preserves the last successfully loaded process cache rather than publishing an empty catalog. Replicas with independent filesystems are not covered. Direct mount writes bypass the validation, SkillScan, and history applied by DeerFlow's install/edit APIs, so only operator-controlled systems should have write access.
Skill installs and agent-managed skill edits run through **SkillScan**, a native deterministic safety scanner before the LLM-based skill scanner. Phase 1 runs offline with no Semgrep/OpenGrep dependency, blocks high-confidence `CRITICAL` findings such as private keys or shell execution, and passes warning findings to the LLM scanner for contextual review. Code files (anything under `scripts/`, a script suffix such as `.py`, `.sh`, or `.js`, or an extensionless file starting with `#!`) that are not NUL-free UTF-8 text raise a warning and are still analyzed over a lossy decode, so a single stray byte cannot hide them from `CRITICAL` checks. The moderation adapter normalizes both plain-text model responses and LangChain Responses API text blocks before parsing the required JSON decision. Python instance-client exfiltration checks follow a minimal same-scope evidence chain: a simple name bound to a known client constructor, optional name-to-name aliases, and an actual outbound method or context-manager use supported by that constructor. Constructor roots must be proven imports; bare canonical-looking names are not inferred as modules. Nested scopes do not inherit client handles and inherit only constructor import aliases that are never rebound in the enclosing scope. Comprehensions, walrus-bearing statements, annotations, complex binding targets, unsupported operations, and ambiguous branch flows produce no finding from this signal; skipped constructs conservatively invalidate every name they may bind so stale client state cannot create a finding. A deterministic work budget or recursion limit reached by this best-effort analysis does not discard findings already collected for the file. Set `skill_scan.enabled: false` in `config.yaml` to disable only the deterministic analyzers; safe archive extraction and the LLM scanner still run.

2
backend/app/gateway/AGENTS.md
View File

@ -114,7 +114,7 @@ owner-scoped assistant version selection remains enabled.
| **Console** (`/api/console`) | Read-only cross-thread observability for the current user: `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly (no new `RunStore` methods); SQL only (503 on `database.backend: memory`). Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). Mixed currencies null cost/currency fields. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured |
| **MCP** (`/api/mcp`) | GET /config - raw/masked; PUT /config - bulk; PATCH /config - toggle; POST /config/servers - add; PUT /config/server - replace; DELETE /config/servers/{server_name:path} - bodyless. POST /cache/reset. Validate expanded, save raw; reload/reset; invalid -> 400. |
| **MCP Tasks** (`/api/threads/{id}/mcp-tasks`) | `GET /` - current user's durable tasks for one owned thread; `GET /{task_id}` - bounded result/input/status-error/cancellation-error detail, including cancellation attempt count, without remote task IDs or driver configuration |
| **Skills** (`/api/skills`) | `GET /` - list; `GET /{name}` - inspect; `PUT /{name}` - toggle; `POST /install` - install a thread-local .skill archive; `POST /install/upload` - admin-only multipart, authorized before parsing and capped at a 100 MiB file plus 1 MiB framing; `POST /reload` - invalidate process-local cache after trusted filesystem changes |
| **Skills** (`/api/skills`) | `GET /` - list; `GET /{name}` - inspect; `PUT /{name}` - toggle; `POST /install` - install a thread-local .skill archive; `POST /install/upload` - admin-only multipart, authorized before parsing and capped at a 100 MiB file plus 1 MiB framing; `POST /reload` - refresh, then publish the shared skills reset marker |
| **Subagents** (`/api/subagents`) | Admin managed-worker CRUD and listing. |
| **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_probed` / `sandbox_runtime_ready` (whether `lark-cli` will be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/config/credentials` - atomically switch the caller's per-user Lark app after validating the new `app_id`/`app_secret` through the official CLI's live tenant-token probe, revoke/remove the previous OAuth tokens, and restore the prior credential tree if the switch fails; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants. Config and auth flows carry a server-issued, per-user generation persisted under the credential lock; a rejected direct switch leaves the current generation unchanged, stale completions return 409, and browser re-registration uses the same token-clearing/revocation transaction as direct credential switches. |
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |

114
backend/app/gateway/routers/skills.py
View File

@ -18,7 +18,13 @@ from app.gateway.authz import (
from app.gateway.deps import get_config, get_optional_user_from_request, require_admin_user
from app.gateway.path_utils import resolve_thread_virtual_path
from app.gateway.skill_export import ExportClientDisconnected, SkillExportManifestResponse, SkillExportResponse, export_http_error, run_export_work
from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache, refresh_skills_system_prompt_cache_async, refresh_user_skills_system_prompt_cache_async
from deerflow.agents.lead_agent.prompt import (
SkillCacheResetPublishError,
clear_skills_system_prompt_cache,
publish_skills_cache_reset,
refresh_skills_system_prompt_cache_async,
refresh_user_skills_system_prompt_cache_async,
)
from deerflow.config.app_config import AppConfig
from deerflow.config.extensions_config import (
ExtensionsConfig,
@ -120,10 +126,13 @@ class SkillInstallResponse(BaseModel):
class SkillReloadResponse(BaseModel):
"""Response model for process-local skill cache invalidation."""
"""Response model for skill cache invalidation."""
success: bool = Field(..., description="Whether the skill caches were invalidated")
scope: Literal["process"] = Field(..., description="Reload scope; only the current Gateway process is affected")
scope: Literal["shared_config", "process"] = Field(
...,
description="Whether the reset was published through the shared extensions-config directory (every Gateway process mounting it rescans on its next prompt build) or only reached this process",
)
message: str = Field(..., description="Human-readable reload status")
@ -209,6 +218,28 @@ async def _drain_skill_mutation[T](op: str, persist: Callable[[], Coroutine[Any,
raise
async def _publish_skills_cache_reset(user_id: str | None) -> str | None:
"""Publish the shared skills cache reset marker off the event loop.
Every caller runs this at the end of a tail wrapped in
``await_drained(_drain_skill_mutation(...))`` -- the install, edit, delete,
rollback and toggle tails, and the ``/skills/reload`` refresh -- after the
change is durable on disk and this process has refreshed its own prompt
caches. The drain is what keeps a cancelled caller from skipping this step
and leaving peer Gateway processes serving the previous skill set; a call
outside such a tail would not have that guarantee. ``user_id`` scopes the
reset to one user's custom skills; ``None`` retires every user's caches
(public-skill state and whole-catalog reloads). Path resolution, the
sidecar lock and the atomic replace are filesystem work.
Failures surface as :class:`SkillCacheResetPublishError` (never an
``OSError``), so the handlers' ``FileNotFoundError -> 404`` and
``ValueError -> 400`` mappings cannot claim a successfully applied change
was not found; every handler reports it through its generic 500 branch.
"""
return await asyncio.to_thread(publish_skills_cache_reset, user_id=user_id)
def _load_user_skills(config: AppConfig) -> list[Skill]:
"""Load every skill visible to the caller's storage, enabled or not.
@ -284,14 +315,24 @@ async def _install_skill_archive(archive_path: Path, config: AppConfig) -> Skill
# absolute paths on construction: blocking filesystem IO.
storage = await asyncio.to_thread(_get_user_skill_storage, config)
result = await storage.ainstall_skill_from_archive(archive_path)
# The install and its prompt-cache refresh settle as one drained unit:
# a cancelled caller must not leave the freshly installed skill absent
# from (or a stale one still present in) the skills prompt cache.
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
# The install, its prompt-cache refresh and the shared reset marker
# settle as one drained unit: a cancelled caller must not leave the
# freshly installed skill absent from (or a stale one still present in)
# the skills prompt cache of this or any peer Gateway process.
user_id = get_effective_user_id()
await refresh_user_skills_system_prompt_cache_async(user_id)
await _publish_skills_cache_reset(user_id)
return SkillInstallResponse(**result)
try:
return await await_drained(_drain_skill_mutation("install", _persist_install))
except SkillCacheResetPublishError as e:
# The skill is installed; only the cross-replica publication failed.
# Not an OSError, so it would reach the generic 500 below anyway; the
# explicit clause pins that contract next to the 404/400 mappings and
# records that the install itself succeeded.
logger.error("Skill installed but the shared cache reset could not be published", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to install skill: {str(e)}") from e
except FileNotFoundError as e:
raise HTTPException(status_code=404, detail=str(e)) from e
except SkillAlreadyExistsError as e:
@ -453,21 +494,46 @@ async def upload_and_install_skill(
"/skills/reload",
response_model=SkillReloadResponse,
summary="Reload Skills",
description=("Invalidate skill prompt caches for all users in the current Gateway process. Subsequent runs rescan the configured skill directories; running tasks and other Gateway processes are unaffected."),
description=(
"Invalidate skill prompt caches for all users, then publish a reset marker beside the shared extensions config so every Gateway process "
"mounting that directory rescans the configured skill directories before its next prompt build. Running tasks are unaffected."
),
)
async def reload_skills(request: Request) -> SkillReloadResponse:
"""Invalidate process-local skill prompt caches after external file changes."""
"""Invalidate skill prompt caches after external file changes.
The local refresh runs first; the shared marker is published only once it
succeeded, and a publication failure is a server error. A ``200`` therefore
never means that only the handling process was refreshed. The refresh and
the publication run as one drained tail, like every other mutation in this
router: a caller that disconnects between the two cannot leave peer
processes serving the old skill set -- the one outcome this operator hook
for external mount writes exists to prevent. Without a resolvable
extensions config path there is no shared directory to publish into and
the response reports ``scope="process"``.
"""
await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL)
try:
async def _refresh_and_publish() -> str | None:
await refresh_skills_system_prompt_cache_async()
return await _publish_skills_cache_reset(None)
try:
generation = await await_drained(_drain_skill_mutation("reload", _refresh_and_publish))
except Exception as exc:
logger.exception("Failed to invalidate skills cache")
raise HTTPException(status_code=500, detail="Failed to invalidate skills cache.") from exc
if generation is None:
return SkillReloadResponse(
success=True,
scope="process",
message="Skill caches invalidated; subsequent runs in this Gateway process will rescan the latest skills.",
)
return SkillReloadResponse(
success=True,
scope="process",
message="Skill caches invalidated; subsequent runs in this Gateway process will rescan the latest skills.",
scope="shared_config",
message="Skill cache reset published through the shared config directory; subsequent runs in every Gateway process sharing it will rescan the latest skills.",
)
@ -612,7 +678,9 @@ async def update_custom_skill(skill_name: str, body: CustomSkillUpdateRequest, r
"scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
user_id = get_effective_user_id()
await refresh_user_skills_system_prompt_cache_async(user_id)
await _publish_skills_cache_reset(user_id)
await await_drained(_drain_skill_mutation("edit", _persist_edit))
return await _read_custom_skill_response(skill_name, config)
@ -653,7 +721,9 @@ async def delete_custom_skill(skill_name: str, request: Request, config: AppConf
"scanner": {"decision": "allow", "reason": "Deletion requested."},
},
)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
user_id = get_effective_user_id()
await refresh_user_skills_system_prompt_cache_async(user_id)
await _publish_skills_cache_reset(user_id)
await await_drained(_drain_skill_mutation("delete", _persist_delete))
return {"success": True}
@ -746,7 +816,9 @@ async def rollback_custom_skill(skill_name: str, body: SkillRollbackRequest, req
# cancelled caller unwinds.
await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, target_content)
await asyncio.to_thread(storage.append_history, skill_name, history_entry)
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
user_id = get_effective_user_id()
await refresh_user_skills_system_prompt_cache_async(user_id)
await _publish_skills_cache_reset(user_id)
await await_drained(_drain_skill_mutation("rollback", _persist_rollback))
return await _read_custom_skill_response(skill_name, config)
@ -906,17 +978,21 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
# and affects every user, so the prompt cache for ALL users must be
# invalidated. CUSTOM/LEGACY drops only the caller's cache, including the
# non-user-scoped fallback above, where other users' entries may stay stale.
# The state write and its cache
# invalidation settle as one drained unit: a cancelled caller must
# not leave the prompt cache serving the previous enablement.
# The state write, its cache invalidation and the shared reset marker
# (same scope: global for PUBLIC, the caller for CUSTOM/LEGACY) settle
# as one drained unit: a cancelled caller must not leave this or a
# peer process serving the previous enablement.
if skill.category == SkillCategory.PUBLIC:
# clear_skills_system_prompt_cache is sync; run it in a worker
# thread to avoid blocking the event loop. The lock inside it is
# cheap, but the async drop also keeps the test mock surface
# consistent (tests patch the async variant).
await asyncio.to_thread(clear_skills_system_prompt_cache)
await _publish_skills_cache_reset(None)
else:
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
user_id = get_effective_user_id()
await refresh_user_skills_system_prompt_cache_async(user_id)
await _publish_skills_cache_reset(user_id)
await await_drained(_drain_skill_mutation("state update", _persist_state))

45
backend/docs/API.md
View File

@ -952,9 +952,11 @@ Raw skill files, sidecars and empty directories are preserved. No hooks/scripts
#### Reload Skills
Invalidate the skill prompt caches for every user in the current Gateway
process. Subsequent runs rescan the configured public, custom, and legacy skill
directories; runs that have already started keep their existing skill snapshot.
Invalidate the skill prompt caches for every user, then publish a reset marker
beside the shared extensions config so every Gateway process mounting that
directory rescans the configured public, custom, and legacy skill directories
before its next prompt build. Runs that have already started keep their
existing skill snapshot.
```http
POST /api/skills/reload
@ -974,8 +976,8 @@ curl -X POST http://localhost:2026/api/skills/reload \
```json
{
"success": true,
"scope": "process",
"message": "Skill caches invalidated; subsequent runs in this Gateway process will rescan the latest skills."
"scope": "shared_config",
"message": "Skill cache reset published through the shared config directory; subsequent runs in every Gateway process sharing it will rescan the latest skills."
}
```
@ -983,17 +985,30 @@ curl -X POST http://localhost:2026/api/skills/reload \
malformed skills retain the existing parser behavior of being skipped and
logged. The endpoint returns `401` for unauthenticated callers, `403` for
non-admin users, and a generic `500` if the invalidation mechanism itself
fails or the process-local background scan does not finish within the cache
refresh timeout. A loader-level failure, such as an unavailable mounted root,
does not publish an empty catalog: the last successfully loaded process cache
remains available. A timed-out scan continues in its daemon worker and can
still populate the process cache when it finishes.
fails, the handling process's background scan does not finish within the cache
refresh timeout, or the reset marker cannot be written. A loader-level failure,
such as an unavailable mounted root, does not publish an empty catalog: the
last successfully loaded process cache remains available. A timed-out scan
continues in its daemon worker and can still populate the process cache when
it finishes.
The scope is deliberately process-local. Each Uvicorn worker or Kubernetes Pod
must be called directly; repeated requests through a load-balanced Service do
not guarantee that every instance is reached. External MinIO/NFS/CSI writes
bypass the validation, SkillScan, and history used by the install/edit APIs, so
the mounted directory must be writable only by trusted operators.
`shared_config` means the handling process refreshed itself and then atomically
replaced `.<extensions config name>.skills-cache-reset.json` next to the
resolved `extensions_config.json`. Every Gateway worker or Pod mounting that
directory compares the marker's signature (at most once per second) before it
serves a cached skill list, so one request reaches every replica sharing the
volume; it does not claim a deployment-wide broadcast when replicas use
independent filesystems. The skill install, edit, delete, rollback and
enable/disable endpoints publish the same marker after their own change is
durable, scoped to the calling user for custom skills; if that publication
fails after the change was applied, they return their generic `500` (never
`404`), and a client that disconnects mid-request does not skip it. When no extensions
config path can be resolved there is no shared directory to publish into; the
request still refreshes the current process and returns `"scope": "process"`
with the message `Skill caches invalidated; subsequent runs in this Gateway
process will rescan the latest skills.` External MinIO/NFS/CSI writes bypass
the validation, SkillScan, and history used by the install/edit APIs, so the
mounted directory must be writable only by trusted operators.
### File Uploads

100
backend/packages/harness/deerflow/agents/lead_agent/prompt.py
View File

@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any
from deerflow.agents.interaction_policy import RunInteractionPolicy
from deerflow.config.agents_config import load_agent_soul
from deerflow.config.shared_reset_marker import SharedResetChange, SharedResetMarker, SharedResetMarkerTracker, resolve_shared_config_path
from deerflow.config.subagents_config import (
DEFAULT_MAX_TOTAL_SUBAGENTS_PER_RUN,
clamp_subagent_concurrency,
@ -61,6 +62,98 @@ class _EnabledSkillsRefreshHandle:
_enabled_skills_refresh_waiters: list[_EnabledSkillsRefreshHandle] = []
# Cross-replica invalidation. ``SkillStorage.load_skills()`` rescans disk and
# re-reads ``extensions_config.json`` on every call, so the caches in this module
# are the only skill state that can go stale between Gateway processes sharing
# one home volume. Every skill mutation publishes
# ``.<extensions config name>.skills-cache-reset.json`` beside the shared config
# (``publish_skills_cache_reset``); every lookup below compares that marker's
# signature, throttled to one stat per second, before serving a cached entry.
# A marker that names a ``user_id`` retires only that user's entries, the same
# scope the local ``invalidate_user_skill_cache`` applies; anything else retires
# all three layers. The tracker is module state so tests can swap in a fake
# clock or call ``reset()``.
SKILLS_CACHE_RESET_MARKER = SharedResetMarker("skills-cache-reset")
_skills_cache_reset_tracker = SharedResetMarkerTracker(SKILLS_CACHE_RESET_MARKER)
def _check_shared_skills_cache_reset() -> None:
"""Retire caches that a reset published by another process has invalidated.
Runs on every cache lookup; the tracker bounds the filesystem cost to one
marker read per second and never blocks behind a concurrent poller. A
failing poll (an unreadable config directory, for example) keeps serving
the current cache: the next poll retries.
"""
try:
change = _skills_cache_reset_tracker.poll()
except Exception:
logger.debug("Could not poll the shared skills cache reset marker", exc_info=True)
return
if change is None:
return
_apply_shared_skills_cache_reset(change)
def _apply_shared_skills_cache_reset(change: SharedResetChange) -> None:
if change.user_ids is None:
logger.info("Shared skills cache reset observed (generation %s); invalidating enabled-skills caches", change.generation)
_invalidate_enabled_skills_cache()
return
for user_id in change.user_ids:
logger.info("Shared skills cache reset observed for user %s (generation %s); invalidating that user's enabled-skills cache", user_id, change.generation)
invalidate_user_skill_cache(user_id)
class SkillCacheResetPublishError(RuntimeError):
"""The shared skills cache reset marker could not be published.
Deliberately not an ``OSError`` subclass. The Gateway skill handlers map
``FileNotFoundError`` to HTTP 404 ("archive/skill not found") and
``ValueError`` to 400, and a marker that could not be written *after* a
skill was installed or edited successfully must not borrow either status;
this type falls through to their generic 500 branch instead. The original
filesystem error is chained as ``__cause__`` for logs; the message names
only its type, never a server path.
"""
def publish_skills_cache_reset(*, user_id: str | None = None) -> str | None:
"""Publish a skills cache reset to every Gateway process sharing the config directory.
Call after the skill change is durable on disk and this process has
refreshed its own caches: peers retire their prompt-layer caches on their
next lookup (within one poll interval), while this process remembers the
generation so it does not retire the caches it just rebuilt. ``user_id``
scopes the reset to one user's custom-skill entries, mirroring
:func:`invalidate_user_skill_cache`; omit it for public-skill and
whole-catalog changes.
Blocking filesystem IO (path resolution, a lock file, an atomic replace):
call via ``asyncio.to_thread`` from the event loop.
Returns:
The published generation, or ``None`` when no extensions config path
can be resolved and the reset therefore stayed process-local.
Raises:
SkillCacheResetPublishError: the config directory, lock file or marker
could not be written (an ``OSError`` such as the directory
vanishing between path resolution and the atomic replace), or the
project root the path resolution depends on is misconfigured
(``ValueError``). A missing explicit config *path* is not an
error: :func:`resolve_shared_config_path` maps it to ``None``.
"""
try:
config_path = resolve_shared_config_path()
if config_path is None:
return None
generation = SKILLS_CACHE_RESET_MARKER.publish(config_path, user_id=user_id)
except (OSError, ValueError) as exc:
raise SkillCacheResetPublishError(f"Could not publish the shared skills cache reset marker ({type(exc).__name__})") from exc
_skills_cache_reset_tracker.note_own_publication(generation)
return generation
def _load_enabled_skills_sync() -> list[Skill]:
return list(get_or_new_skill_storage().load_skills(enabled_only=True))
@ -110,6 +203,7 @@ def _refresh_enabled_skills_cache_worker() -> None:
def _ensure_enabled_skills_cache() -> threading.Event:
global _enabled_skills_refresh_active
_check_shared_skills_cache_reset()
with _enabled_skills_lock:
if _enabled_skills_refresh_active:
return _enabled_skills_refresh_event
@ -160,9 +254,12 @@ def _get_enabled_skills():
def get_cached_enabled_skills() -> list[Skill]:
"""Return the cached enabled-skills list, kicking off a background refresh on miss.
Safe to call from request paths: never blocks on disk I/O. Returns an empty
Safe to call from request paths: the skill directories are never scanned
inline, and the only filesystem work is the shared reset-marker check,
which is one small stat+read at most once per second. Returns an empty
list on cache miss; the next call will see the warmed result.
"""
_check_shared_skills_cache_reset()
with _enabled_skills_lock:
cached = _enabled_skills_cache
@ -188,6 +285,7 @@ def get_enabled_skills_for_config(app_config: AppConfig | None = None, user_id:
if app_config is None:
return _get_enabled_skills()
_check_shared_skills_cache_reset()
cache_key = (id(app_config), user_id or "default")
with _enabled_skills_lock:
cached = _enabled_skills_by_config_cache.get(cache_key)

2
backend/packages/harness/deerflow/config/AGENTS.md
View File

@ -122,6 +122,8 @@ that read's digest rather than an earlier probe's. Freshness reloads read the
probed path explicitly; a missing or invalid revision keeps the last-known-good
cache, including AppConfig's middleware snapshot, until a readable revision returns.
**Shared Reset Markers** (`shared_reset_marker.py`): caches derived from the extensions config can go stale without a byte of it changing (a remote MCP server's `tools/list`, a skill installed or edited on the shared volume by another process). `SharedResetMarker(suffix)` owns one hidden JSON file beside the resolved config, `.<config name>.<suffix>.json`; `publish()` replaces it atomically with a random generation under `extensions_config_write_lock` + `extensions_config_file_lock` (`publish_locked()` when the caller already holds both), chaining `previous_generation` and an optional `user_id`. `SharedResetMarkerTracker.poll()` re-reads the marker at most once per second (monotonic clock, never blocking behind a concurrent poller), adopts the current state silently on its first poll, and reports a `SharedResetChange`: `user_ids={user}` only when exactly one publication carrying that user happened since the last poll, otherwise `None` (retire everything), including deleted/unreadable markers and config-path switches. `note_own_publication()` lets the writer skip the redundant self-invalidation. `resolve_shared_config_path()` maps a missing explicit path to `None` (no shared directory: callers answer `scope=process`). Consumers: `mcp/cache.py` (`mcp-cache-reset`) and `agents/lead_agent/prompt.py` (`skills-cache-reset`). Tests: `tests/test_shared_reset_marker.py`.
### Config Schema
**`config.yaml`** key sections:

316
backend/packages/harness/deerflow/config/shared_reset_marker.py Normal file
View File

@ -0,0 +1,316 @@
"""Shared cache-reset markers for Gateway processes that share one config directory.
Several process-local caches are derived from ``extensions_config.json`` and the
files beside it: the MCP tools cache (``deerflow.mcp.cache``) and the skills
prompt caches (``deerflow.agents.lead_agent.prompt``). The file itself is
revalidated by content signature on every read, but a cache can also go stale
without a byte of that file changing: a remote MCP server changes its
``tools/list`` response, or a skill is installed, edited, deleted or toggled
on a shared volume by another Gateway worker or Pod.
A **shared reset marker** carries that invalidation. The writer replaces a
hidden JSON file beside the resolved extensions config
(``.<config name>.<suffix>.json``) with a fresh random generation, using the
same atomic write and the same cross-process locks the config itself uses.
Every process mounting that directory records the marker's ``(mtime, size,
sha256)`` signature (``deerflow.config.file_signature``) when it publishes a
cache and compares it on later lookups; a different signature retires the
cache. A random generation needs no read-modify-write counter and cannot lose
two concurrent resets: the final write still differs from every previously
observed signature.
The scope is deliberately the shared config *directory*: replicas with
independent filesystems are not covered, and a deployment without a resolvable
config path has no shared directory to publish into. Callers surface that
distinction as ``scope=shared_config`` versus ``scope=process``.
Only the marker payload is JSON. The ``previous_generation`` field chains
writes so a reader can tell "exactly one publication since my last look" (and
honor that publication's optional ``user_id`` scope) from "several, of which I
only see the last" (which must widen to a full reset).
"""
from __future__ import annotations
import json
import logging
import threading
import time
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from deerflow.config.file_signature import ConfigSignature, get_config_signature, read_config_with_signature
logger = logging.getLogger(__name__)
MARKER_VERSION = 1
#: Default minimum spacing between two marker reads by one tracker. The read is
#: one ``stat`` plus one small file read; the interval only bounds that cost on
#: hot cache lookups and is far below the cross-replica convergence budget.
DEFAULT_POLL_INTERVAL_SECONDS = 1.0
#: How many of its own recent generations a tracker remembers so it can skip the
#: redundant self-invalidation that would otherwise follow every local publish.
_OWN_GENERATIONS_LIMIT = 64
@dataclass(frozen=True)
class SharedResetMarkerPayload:
"""Parsed contents of a marker file. Every field is optional on read."""
generation: str | None = None
previous_generation: str | None = None
user_id: str | None = None
published_at: str | None = None
@classmethod
def from_json(cls, data: object) -> SharedResetMarkerPayload | None:
"""Return the payload a well-formed marker object carries, else ``None``."""
if not isinstance(data, dict):
return None
return cls(
generation=_optional_str(data.get("generation")),
previous_generation=_optional_str(data.get("previous_generation")),
user_id=_optional_str(data.get("user_id")),
published_at=_optional_str(data.get("published_at")),
)
def _optional_str(value: object) -> str | None:
return value if isinstance(value, str) and value else None
@dataclass(frozen=True)
class SharedResetChange:
"""One marker change observed by :class:`SharedResetMarkerTracker`.
``user_ids`` names the users whose caches the publication scoped itself to.
``None`` means the whole cache must be retired: a global publication, a
deleted or unreadable marker, a config-path switch, or several publications
between two polls of which only the last one is visible.
"""
user_ids: frozenset[str] | None
generation: str | None
class SharedResetMarker:
"""A named marker file colocated with the resolved extensions config."""
def __init__(self, suffix: str) -> None:
if not suffix or suffix in {".", ".."} or any(sep in suffix for sep in ("/", "\\")):
raise ValueError(f"shared reset marker suffix must be a plain file-name fragment, got {suffix!r}")
self._suffix = suffix
@property
def suffix(self) -> str:
return self._suffix
def path_for(self, config_path: Path) -> Path:
"""Return the marker path beside *config_path* (following a symlinked config)."""
path = Path(config_path)
target = path.resolve(strict=False) if path.is_symlink() else path
return target.parent / f".{target.name}.{self._suffix}.json"
def current_signature(self, config_path: Path | None) -> ConfigSignature | None:
"""Return the marker's current signature, or ``None`` without a config or marker."""
if config_path is None:
return None
return get_config_signature(self.path_for(config_path))
def read(self, config_path: Path) -> tuple[SharedResetMarkerPayload | None, ConfigSignature | None]:
"""Read the marker once, returning its payload and the signature of those bytes.
A missing marker is ``(None, None)``: no reset has been published yet.
A marker that exists but cannot be parsed keeps its signature so a
reader still notices that *something* was written, and reports no
payload so the reader widens to a full reset.
"""
marker_path = self.path_for(config_path)
try:
data, signature = read_config_with_signature(marker_path)
except OSError:
return None, None
try:
payload = SharedResetMarkerPayload.from_json(json.loads(data.decode("utf-8-sig")))
except (UnicodeDecodeError, ValueError):
logger.debug("Shared reset marker %s is not valid JSON; treating it as an unscoped reset", marker_path)
payload = None
return payload, signature
def publish(self, config_path: Path, *, user_id: str | None = None) -> str:
"""Publish a new generation under the extensions-config write locks.
Holds the process-local ``extensions_config_write_lock`` and the
cross-process ``extensions_config_file_lock`` for *config_path*, the
same discipline every ``extensions_config.json`` writer follows, so the
read of the previous generation and the replacement are one critical
section. Returns the published generation.
"""
from deerflow.config.extensions_config import extensions_config_file_lock, extensions_config_write_lock
with extensions_config_write_lock, extensions_config_file_lock(config_path):
return self.publish_locked(config_path, user_id=user_id)
def publish_locked(self, config_path: Path, *, user_id: str | None = None) -> str:
"""Publish a new generation; the caller already holds both config locks.
The marker is replaced atomically (temporary file + ``os.replace``,
with the same bind-mount fallback as the config itself), so a reader
never observes a truncated marker.
"""
from deerflow.config.extensions_config import atomic_write_extensions_config
previous, _signature = self.read(config_path)
generation = uuid.uuid4().hex
payload: dict[str, object] = {
"version": MARKER_VERSION,
"generation": generation,
"previous_generation": previous.generation if previous is not None else None,
"published_at": datetime.now(UTC).isoformat(),
}
if user_id:
payload["user_id"] = user_id
atomic_write_extensions_config(self.path_for(config_path), payload)
return generation
def resolve_shared_config_path() -> Path | None:
"""Resolve the extensions config path for marker purposes, or ``None``.
``ExtensionsConfig.resolve_config_path()`` raises ``FileNotFoundError``
when an explicit path or ``DEER_FLOW_EXTENSIONS_CONFIG_PATH`` names a file
that does not exist; that is right for callers that need the config, but a
marker reader runs on hot cache lookups and a marker writer only needs a
directory to publish into. Both treat that failure as "no shared config
directory right now", matching ``deerflow.mcp.cache._resolve_config_path``.
"""
from deerflow.config.extensions_config import ExtensionsConfig
try:
return ExtensionsConfig.resolve_config_path()
except FileNotFoundError:
logger.debug("Extensions config path could not be resolved for a shared reset marker; treating as unconfigured", exc_info=True)
return None
class SharedResetMarkerTracker:
"""Per-process observer of one :class:`SharedResetMarker`.
``poll()`` compares the marker against the last observed state at most once
per ``poll_interval_seconds`` (monotonic clock) and reports a
:class:`SharedResetChange` when the signature moved. The first poll adopts
the current state silently: a process that has not cached anything yet has
nothing to retire, and a restarted worker must not reset on the generation
that was already current when it started.
One instance per process is the production shape; tests build several so
two "processes" can watch the same marker file, and ``reset()`` returns an
instance to its never-polled state.
"""
def __init__(
self,
marker: SharedResetMarker,
*,
resolve_config_path: Callable[[], Path | None] = resolve_shared_config_path,
poll_interval_seconds: float = DEFAULT_POLL_INTERVAL_SECONDS,
clock: Callable[[], float] = time.monotonic,
) -> None:
if poll_interval_seconds < 0:
raise ValueError("poll_interval_seconds must not be negative")
self._marker = marker
self._resolve_config_path = resolve_config_path
self._poll_interval = poll_interval_seconds
self._clock = clock
self._lock = threading.Lock()
self.reset()
@property
def marker(self) -> SharedResetMarker:
return self._marker
def reset(self) -> None:
"""Forget every observation; the next ``poll()`` adopts silently."""
with self._lock:
self._observed = False
self._last_poll_at: float | None = None
self._last_config_path: Path | None = None
self._last_signature: ConfigSignature | None = None
self._last_generation: str | None = None
self._own_generations: list[str] = []
def note_own_publication(self, generation: str) -> None:
"""Remember a generation this process published itself.
The publishing process refreshes its own caches as part of the
mutation, so observing its own marker later must not retire them
again. An interleaved foreign publication still widens to a full reset
because the chain of ``previous_generation`` values no longer matches.
"""
with self._lock:
self._own_generations.append(generation)
del self._own_generations[:-_OWN_GENERATIONS_LIMIT]
def poll(self) -> SharedResetChange | None:
"""Report a marker change since the previous poll, honoring the throttle.
Never blocks on another thread's poll: a concurrent poller is already
doing the stat, so this call returns ``None`` immediately and the
caller serves its cache; the concurrent poller applies any change.
"""
if not self._lock.acquire(blocking=False):
return None
try:
now = self._clock()
if self._last_poll_at is not None and now - self._last_poll_at < self._poll_interval:
return None
self._last_poll_at = now
config_path = self._resolve_config_path()
if config_path is None:
payload, signature = None, None
else:
payload, signature = self._marker.read(config_path)
if not self._observed:
self._observed = True
self._record(config_path, payload, signature)
return None
if config_path == self._last_config_path and signature == self._last_signature:
return None
change = self._classify(config_path, payload)
self._record(config_path, payload, signature)
return change
finally:
self._lock.release()
def _record(self, config_path: Path | None, payload: SharedResetMarkerPayload | None, signature: ConfigSignature | None) -> None:
self._last_config_path = config_path
self._last_signature = signature
self._last_generation = payload.generation if payload is not None else None
def _classify(self, config_path: Path | None, payload: SharedResetMarkerPayload | None) -> SharedResetChange | None:
generation = payload.generation if payload is not None else None
# A config-path switch, a deleted marker or an unreadable one carry no
# scope information: retire everything.
if config_path != self._last_config_path or payload is None or generation is None:
return SharedResetChange(user_ids=None, generation=generation)
single_publication = payload.previous_generation == self._last_generation
if generation in self._own_generations:
self._own_generations.remove(generation)
if single_publication:
# Exactly one publication since the last poll, and it was ours.
return None
return SharedResetChange(user_ids=None, generation=generation)
if single_publication and payload.user_id:
return SharedResetChange(user_ids=frozenset({payload.user_id}), generation=generation)
return SharedResetChange(user_ids=None, generation=generation)

2
backend/packages/harness/deerflow/mcp/AGENTS.md
View File

@ -56,7 +56,7 @@
- **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values.
- **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(ownership domain, server_name, incarnation scope)` session when the registered entry is still the same `ClientSession` that failed. A late error from an old concurrent call cannot evict its replacement or a new in-flight creation. The failing call still surfaces its original error and is never replayed automatically; a later retry creates a fresh subprocess/session. Protocol timeouts, normal `isError=true` tool results, and interceptor failures do not evict a healthy stateful session.
- **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged. Bare-filename correlation must not replace a segment of a longer path: both `/` and `\` are separators at either boundary, even when the full-path pass leaves the reference unresolved. A trailing `\` immediately before a newline therefore counts as a separator too; that position is more often a Markdown hard break, but a backslash is also a legal filename character on POSIX, so forgoing that single rewrite is the conservative tradeoff. `tests/test_mcp_file_migration.py` covers this through the result converter with real workspace files.
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock, then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`.
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) additionally calls `publish_mcp_tools_cache_reset()`: it writes a random generation to a hidden JSON marker beside the resolved config under the same cross-process lock (`MCP_CACHE_RESET_MARKER`, a `deerflow.config.shared_reset_marker.SharedResetMarker` shared with the skills prompt-cache reset), then retires local state. Every worker mounting that directory records the marker signature when tools publish and invalidates on a later signature, so a remote `tools/list` change converges without a fake config edit; marker drift during discovery fences the stale result. The response reports this transport as `scope=shared_config` rather than claiming an implicit broadcast to replicas with independent filesystems. No resolvable config path keeps the legacy process-local behavior and is reported as `scope=process`.
- **Stdio launch policy at the HTTP boundary** (`routers/mcp.py::_validate_mcp_update_request`, shared by `PUT` and the enable branch of `PATCH`): a config file may express anything, but the API is untrusted input, so an API-registered stdio server must (a) name a bare executable from the allowlist — `_DEFAULT_MCP_STDIO_COMMAND_ALLOWLIST` = `{npx, uvx}`, extended by `DEER_FLOW_MCP_STDIO_COMMAND_ALLOWLIST`, with path separators, whitespace, and shell metacharacters rejected in `command`; (b) carry no `args` flag in `_ARBITRARY_EXEC_ARGS`; and (c) set no `env` name in `_CODE_INJECTING_ENV_VARS`. Checks (b) and (c) exist because the command check alone names a binary without constraining what that binary runs. The `env` denylist applies to **every** allowlisted command, and both denylists match `--flag=value` as well as `--flag value`. The `args` denylist's **scope depends on the command**, because where a launcher stops parsing its own flags is what decides whether a token is an exec flag at all:
- For a **package launcher** in `_PACKAGE_LAUNCHERS` (`{npx, uvx}`) only the launcher's own **option region** is screened. `npx`/`uvx` stop parsing their flags at the package name and hand every later token to the spawned server's argv, where `-c` is routinely "config" and `-e` "env" — screening those rejected ordinary third-party servers while covering nothing. A bare `--` ends the region too: only the *first* token after it is the package name. Finding that boundary needs each launcher's option **arity**, since a value is not a positional — `npx -p <pkg> -c '<command>'` **runs** the command (`-p` is `npm exec`'s `--package`, so `<pkg>` is its value and npm keeps parsing), so ending the region at the first non-flag token would walk straight past it. `_NPX_BOOLEAN_ARGS` is generated from `@npmcli/config`'s definitions (npm 10.9.4) minus the `-p` exec override; `_UVX_VALUE_ARGS` comes from `uvx --help` (uv 0.11.1). Regenerate these against a newer launcher rather than hand-editing. The unknown-option default is deliberately **opposite** per launcher, following the exec set rather than symmetry: npx owns real exec flags (`-c`/`--call`), so an unknown option consumes a value and keeps the region open (npm errors on options it does not define, so this cannot reject a working invocation); uvx owns no string-eval flag at all, so its screen is a tripwire, an unknown option consumes nothing, and uv's large boolean surface cannot over-block. uvx's exec set also drops the short spellings, because `-c` is uv's `--constraints` and `-p` its `--python`.

32
backend/packages/harness/deerflow/mcp/cache.py
View File

@ -4,13 +4,13 @@ import asyncio
import json
import logging
import threading
import uuid
from pathlib import Path
from langchain_core.tools import BaseTool
from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
from deerflow.config.file_signature import get_config_signature as _get_config_signature
from deerflow.config.shared_reset_marker import SharedResetMarker
from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config
logger = logging.getLogger(__name__)
@ -54,19 +54,20 @@ _initialized_without_config = False
# their next lookup.
_cache_reset_marker_signature: _ConfigSignature | None = None
#: ``.<extensions config name>.mcp-cache-reset.json`` beside the resolved config.
#: The generic marker/tracker contract lives in ``deerflow.config.shared_reset_marker``
#: and is shared with the skills prompt-cache reset.
MCP_CACHE_RESET_MARKER = SharedResetMarker("mcp-cache-reset")
def _cache_reset_marker_path(config_path: Path) -> Path:
"""Return the shared reset marker colocated with the extensions config."""
path = Path(config_path)
target = path.resolve(strict=False) if path.is_symlink() else path
return target.parent / f".{target.name}.mcp-cache-reset.json"
return MCP_CACHE_RESET_MARKER.path_for(config_path)
def _current_cache_reset_marker_signature(config_path: Path | None) -> _ConfigSignature | None:
"""Return the current shared-reset marker signature, if one exists."""
if config_path is None:
return None
return _get_config_signature(_cache_reset_marker_path(config_path))
return MCP_CACHE_RESET_MARKER.current_signature(config_path)
def _resolve_config_path() -> Path | None:
@ -602,19 +603,10 @@ def publish_mcp_tools_cache_reset() -> str | None:
reset_mcp_tools_cache()
return None
from deerflow.config.extensions_config import (
atomic_write_extensions_config,
extensions_config_file_lock,
extensions_config_write_lock,
)
generation = uuid.uuid4().hex
marker_path = _cache_reset_marker_path(config_path)
with extensions_config_write_lock, extensions_config_file_lock(config_path):
atomic_write_extensions_config(
marker_path,
{"version": 1, "generation": generation},
)
# The marker is replaced atomically under ``extensions_config_write_lock``
# and the cross-process ``extensions_config_file_lock``, the same discipline
# every ``extensions_config.json`` writer follows.
generation = MCP_CACHE_RESET_MARKER.publish(config_path)
# Publish-before-retire is intentional. A successful API response must
# never mean only the handling worker was refreshed; if publication fails,

3
backend/packages/harness/deerflow/skills/AGENTS.md
View File

@ -8,7 +8,8 @@
- **Location**: global public skills live under `deer-flow/skills/public/`; user-authored custom skills live under `{DEER_FLOW_HOME}/users/{user_id}/skills/custom/`; globally managed integration skills live under `{DEER_FLOW_HOME}/integrations/skills/{provider}/`; per-user integration credentials remain under `{DEER_FLOW_HOME}/users/{user_id}/integrations/{provider}/{config,data}`
- **Format**: Directory with `SKILL.md` (YAML frontmatter: name, description, license, allowed-tools as a spec-compatible string or YAML list, argument-hint, required-secrets). Exact portable spellings such as `Bash`, `WebFetch`, `WebSearch`, `Glob`, `Grep`, `Read`, `Write`, and `Edit` map to `bash`, `web_fetch`, `web_search`, `glob`, `grep`, `read_file`, `write_file`, and `str_replace`; lowercase or otherwise unknown scalar names and YAML-list entries preserve their exact runtime spelling. Argument-scoped entries remain literal and inactive because the tool policy does not inspect arguments; the scalar tokenizer keeps spaces, quotes, and escaped parentheses inside patterns intact.
- **Loading**: `load_skills()` recursively scans public, per-user custom, global integration, and legacy custom locations for `SKILL.md`, parses metadata, and reads enabled state from extensions_config.json plus per-user skill state for non-public categories; that directory is a package boundary, so no nested `SKILL.md` is registered as a runtime skill. A custom skill directory may be a one-level symlink to an external directory for compatibility with operator-managed skill trees; activation still rejects a symlinked `SKILL.md` or deeper path escape. Deleting such a skill unlinks the entry and leaves the external tree untouched. SkillScan has a deliberately narrower packaging rule: known eval fixtures are permitted as support data, while other nested `SKILL.md` files are reported as package defects. Ancestor-link cycles are pruned; independent aliases remain discoverable.
- **External reload**: `POST /api/skills/reload` is an admin-only, process-local invalidation hook for trusted MinIO/NFS/CSI writes. `SkillStorage` instances do not cache a catalog — `load_skills()` scans on every call — so the route clears all `(app_config, user_id)` entries and the rendered prompt-section LRU, then waits up to the shared refresh timeout for the existing off-loop single-flight refresh. Each invalidation receives a generation-bound result handle; a successful scan atomically replaces the global enabled-skills cache, while a loader-level failure propagates to the HTTP waiter and preserves the last-known-good global cache. Per-user/config scans capture the refresh version and cannot repopulate shared caches if invalidation occurs while they are loading. A timed-out HTTP wait fails generically while the daemon refresh worker continues. Subsequent runs rescan after a successful reload; active runs keep their existing snapshot. Each Uvicorn worker/Kubernetes Pod must be targeted separately. Direct mount writes bypass install/edit validation, SkillScan, and history, so mounted roots are an operator-controlled trust boundary.
- **External reload**: `POST /api/skills/reload` is an admin-only invalidation hook for trusted MinIO/NFS/CSI writes. `SkillStorage` instances do not cache a catalog — `load_skills()` scans on every call — so the route clears all `(app_config, user_id)` entries and the rendered prompt-section LRU, waits up to the shared refresh timeout for the existing off-loop single-flight refresh, then publishes the shared reset marker (below). Each invalidation receives a generation-bound result handle; a successful scan atomically replaces the global enabled-skills cache, while a loader-level failure propagates to the HTTP waiter and preserves the last-known-good global cache. Per-user/config scans capture the refresh version and cannot repopulate shared caches if invalidation occurs while they are loading. A timed-out HTTP wait fails generically while the daemon refresh worker continues. Subsequent runs rescan after a successful reload; active runs keep their existing snapshot. Direct mount writes bypass install/edit validation, SkillScan, and history, so mounted roots are an operator-controlled trust boundary.
- **Shared cache reset**: the prompt-layer caches in `agents/lead_agent/prompt.py` (global enabled-skills list, per-`(app_config, user_id)` entries, rendered `<skill_system>` LRU) are the only skill state that can diverge between Gateway processes sharing one home volume. Every mutation in `routers/skills.py` (install/upload, edit, delete, rollback, enable/disable, reload) refreshes its own process, then calls `publish_skills_cache_reset(user_id=...)`, which atomically replaces `.<extensions config name>.skills-cache-reset.json` beside the shared config under the extensions-config locks (`deerflow.config.shared_reset_marker`). Every cache lookup polls that marker at most once per second and, on a new signature, calls `_invalidate_enabled_skills_cache()` or, for a marker that names one `user_id` as the only publication since the last poll, `invalidate_user_skill_cache(user_id)`; the publishing process skips its own generation. `/skills/reload` runs refresh + publish through the same `await_drained(_drain_skill_mutation(...))` tail as the mutations, so a cancelled caller cannot skip the marker; it answers `scope: shared_config` when the marker was written and `scope: process` when no extensions config path resolves. Publication failures raise `SkillCacheResetPublishError` (a `RuntimeError`, never an `OSError`), so the handlers' `FileNotFoundError -> 404` / `ValueError -> 400` mappings cannot claim an applied change was not found; every handler reports it as a generic 500, and a 200 never means only one process was refreshed. The marker reaches processes mounting that directory, not replicas with independent filesystems. Tests: `tests/test_skills_cache_shared_reset.py`.
- **Tool policy**: Agent `allowed-tools` declarations apply dynamically only to slash-activated skills and skills captured in `ThreadState.skill_context` through configured `read_file` loads; passive enabled skills and skill allowlists do not clamp the baseline tool set. A lead custom Agent's explicit `skills` list is additionally enforced at the sandbox filesystem layer (see Sandbox projection); subagent skill lists still scope discovery and activation only because concurrently delegated subagents share the lead thread sandbox. Subagents render only skill discovery metadata at startup and reuse the same adjacent `SkillActivationMiddleware` + `SkillToolPolicyMiddleware` pair as the lead; their configured `skills` field limits discovery and activation instead of eagerly loading bodies or unioning policies. Slash policy is dominant for its run, preventing subsequently read skills from widening explicit authority; autonomous captured skills use the existing union only when no slash source exists. `tool_search` and `describe_skill` stay available as framework discovery infrastructure, while every discovered or promoted business tool still requires active-policy permission for schema visibility and execution; `task`, `list_background_tasks`, and `cancel_background_task` likewise require explicit declarations. Each active model call intentionally reloads the full live registry so enable/disable changes, frontmatter edits, and custom/public name-shadow winners take effect without a stale TTL or unsafe direct-path cache; all tool calls produced by that model step reuse the resulting source-and-path-signed decision. Registry failures and all-invalid active sets fail closed, while stale individual paths are skipped when another valid skill remains. The dynamic `allowed-tools` policy remains best-effort behavioral scoping: alternate loading paths are not captured and bounded autonomous context may evict entries.
- **Sandbox projection**: `skills/projection.py` materializes enabled-only shared trees at `{base_dir}/skills_view/public` and `{base_dir}/users/{user_id}/skills_view/{custom,legacy,integrations}`. A lead Agent with an explicit `skills` allowlist (including `[]`) gets the intersection of enabled public/user-visible skills and that allowlist at `{base_dir}/users/{user_id}/threads/{thread_id}/skills_view/{public,custom,legacy,integrations}`. `skills=None` keeps the shared zero-copy mount until a thread has used an explicit policy; later unrestricted runs repopulate the same stable thread root with all enabled skills (with host bash enabled, `LocalSandboxProvider` mounts the shared views instead). Rebuilds sign source state, view state, and normalized policy in a manifest, revoke every old category before adding the new policy, stage copies in temporary directories, and atomically replace files. Category root inodes stay stable for live bind mounts; concurrent readers can briefly see fewer skills during a policy change, never a skill revoked by the new policy. Policy-scoped copies reject absolute symlinks and relative symlinks that resolve outside their own skill package, preventing a permitted package from linking back to an omitted source. It copies files into the view (`_copy_into_view`) so a sandbox write cannot mutate the canonical skill inode; the operational trade-off is an O(total bytes) I/O and per-user/thread storage multiplier across rebuilds, prioritized for write isolation over zero-copy hardlinks. Steady-state freshness checks combine source and view metadata tree digests, so in-sandbox view tampering is detected and repaired on the next acquire. Storage writes, archive installs, deletes, and toggles rebuild shared scopes under a cross-process lock; Gateway boot ensures only the shared public view, user views are repaired lazily on acquire, and Agent thread views are recomputed before sandbox reuse. Managed integration packages are global, but their projected category is per-user because enabled state is isolated. User projection rebuilds re-read global enable state from disk instead of the process singleton, so a toggle handled by another Gateway worker is reflected on the next acquire. Gateway public-skill toggles take the public projection lock before the shared `extensions_config_write_lock`, re-read an existing config raw from disk, change only that skill's entry, and rebuild before responding; keep this as one worker-owned critical section so MCP writes cannot interleave and request cancellation cannot release either lock while the worker still runs. The shared public steady-state signature check runs without the global projection lock; stale/error paths take the lock and re-check before rebuilding or clearing. User and thread scope checks are serialized per scope. Projection failures clear the affected view before raising.
- **Injection (legacy / default)**: Enabled skills are listed in the agent system prompt with full metadata and container paths (`<available_skills>` block). Controlled by `skills.deferred_discovery: false` (default).

25
backend/tests/blocking_io/test_skills_reload.py
View File

@ -32,8 +32,33 @@ async def test_reload_skills_offloads_directory_scan(tmp_path: Path, monkeypatch
monkeypatch.setattr(skills_router, "require_admin_user", _noop_admin)
monkeypatch.setattr(prompt_module, "get_or_new_skill_storage", lambda **_kwargs: storage)
monkeypatch.setattr(prompt_module, "resolve_shared_config_path", lambda: None)
response = await skills_router.reload_skills(request=None)
assert response.success is True
assert response.scope == "process"
async def test_reload_skills_publishes_shared_marker_off_the_loop(tmp_path: Path, monkeypatch) -> None:
"""Resolving the shared config path and writing the reset marker are filesystem work."""
await asyncio.to_thread(_seed_skill, tmp_path)
storage = await asyncio.to_thread(LocalSkillStorage, host_path=str(tmp_path))
config_path = tmp_path / "extensions_config.json"
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
async def _noop_admin(_request, **_kwargs) -> None:
return None
monkeypatch.setattr(skills_router, "require_admin_user", _noop_admin)
monkeypatch.setattr(prompt_module, "get_or_new_skill_storage", lambda **_kwargs: storage)
try:
response = await skills_router.reload_skills(request=None)
finally:
prompt_module._skills_cache_reset_tracker.reset()
assert response.success is True
assert response.scope == "shared_config"
assert await asyncio.to_thread(lambda: prompt_module.SKILLS_CACHE_RESET_MARKER.path_for(config_path).exists())

288
backend/tests/test_shared_reset_marker.py Normal file
View File

@ -0,0 +1,288 @@
"""Shared cache-reset markers for Gateway processes that share one config directory.
``deerflow.config.shared_reset_marker`` is the generic helper behind the MCP
tools-cache reset (``.<extensions config>.mcp-cache-reset.json``) and the skills
prompt-cache reset (``.<extensions config>.skills-cache-reset.json``). A writer
publishes a fresh random generation with an atomic replace; every other process
compares the marker's ``(mtime, size, sha256)`` signature on its next lookup.
These tests drive two trackers side by side to simulate two Gateway processes
sharing one marker file.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
import deerflow.config.shared_reset_marker as shared_reset_marker_module
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.config.file_signature import get_config_signature
from deerflow.config.shared_reset_marker import (
SharedResetChange,
SharedResetMarker,
SharedResetMarkerTracker,
resolve_shared_config_path,
)
class _Clock:
def __init__(self, start: float = 1000.0) -> None:
self.now = start
def __call__(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
@pytest.fixture()
def config_path(tmp_path: Path) -> Path:
path = tmp_path / "extensions_config.json"
path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
return path
def _tracker(marker: SharedResetMarker, config_path: Path | None, clock: _Clock, **kwargs) -> SharedResetMarkerTracker:
return SharedResetMarkerTracker(marker, resolve_config_path=lambda: config_path, clock=clock, **kwargs)
def test_marker_path_is_hidden_sibling_of_the_config(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
assert marker.path_for(config_path) == config_path.parent / ".extensions_config.json.skills-cache-reset.json"
def test_marker_path_follows_a_symlinked_config(tmp_path: Path, config_path: Path) -> None:
link = tmp_path / "linked" / "extensions_config.json"
link.parent.mkdir()
link.symlink_to(config_path)
assert SharedResetMarker("skills-cache-reset").path_for(link) == config_path.parent / ".extensions_config.json.skills-cache-reset.json"
@pytest.mark.parametrize("suffix", ["", "has/slash", "has\\backslash", ".."])
def test_marker_rejects_unsafe_suffixes(suffix: str) -> None:
with pytest.raises(ValueError):
SharedResetMarker(suffix)
def test_publish_writes_an_atomic_json_marker(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
generation = marker.publish(config_path, user_id="alice")
marker_path = marker.path_for(config_path)
payload = json.loads(marker_path.read_text(encoding="utf-8"))
assert payload["version"] == 1
assert payload["generation"] == generation
assert payload["previous_generation"] is None
assert payload["user_id"] == "alice"
assert payload["published_at"]
assert not list(config_path.parent.glob(f".{marker_path.name}.*.tmp"))
def test_publish_chains_the_previous_generation_and_omits_user_id_for_global_resets(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
first = marker.publish(config_path, user_id="alice")
second = marker.publish(config_path)
payload = json.loads(marker.path_for(config_path).read_text(encoding="utf-8"))
assert first != second
assert payload["generation"] == second
assert payload["previous_generation"] == first
assert "user_id" not in payload
def test_read_tolerates_missing_and_malformed_markers(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
assert marker.read(config_path) == (None, None)
assert marker.current_signature(config_path) is None
assert marker.current_signature(None) is None
marker_path = marker.path_for(config_path)
marker_path.write_text("{not json", encoding="utf-8")
payload, signature = marker.read(config_path)
assert payload is None
assert signature == get_config_signature(marker_path)
def test_two_processes_share_one_marker(config_path: Path) -> None:
"""Process A publishes; process B sees the invalidation on its next poll."""
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
process_a = _tracker(marker, config_path, clock)
process_b = _tracker(marker, config_path, clock)
# Both processes adopt the current state (no marker yet) without a reset.
assert process_a.poll() is None
assert process_b.poll() is None
generation = marker.publish(config_path)
process_a.note_own_publication(generation)
clock.advance(1.0)
assert process_a.poll() is None, "the publishing process already refreshed itself"
change = process_b.poll()
assert change == SharedResetChange(user_ids=None, generation=generation)
clock.advance(1.0)
assert process_b.poll() is None, "an observed generation is not reported twice"
def test_user_scoped_publication_is_reported_with_its_user(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
process_b = _tracker(marker, config_path, clock)
assert process_b.poll() is None
generation = marker.publish(config_path, user_id="alice")
clock.advance(1.0)
assert process_b.poll() == SharedResetChange(user_ids=frozenset({"alice"}), generation=generation)
def test_multiple_publications_between_polls_fall_back_to_a_global_reset(config_path: Path) -> None:
"""Only the latest marker survives, so a skipped generation widens the reset."""
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
process_b = _tracker(marker, config_path, clock)
assert process_b.poll() is None
marker.publish(config_path, user_id="alice")
generation = marker.publish(config_path, user_id="bob")
clock.advance(1.0)
assert process_b.poll() == SharedResetChange(user_ids=None, generation=generation)
def test_own_publication_with_an_interleaved_foreign_write_still_resets(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
process_a = _tracker(marker, config_path, clock)
assert process_a.poll() is None
process_a.note_own_publication(marker.publish(config_path, user_id="alice"))
foreign = marker.publish(config_path, user_id="bob")
clock.advance(1.0)
assert process_a.poll() == SharedResetChange(user_ids=None, generation=foreign)
def test_poll_is_throttled_to_one_stat_per_interval(config_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
reads: list[Path] = []
original = shared_reset_marker_module.read_config_with_signature
def _counting_read(path: Path):
reads.append(path)
return original(path)
marker.publish(config_path)
monkeypatch.setattr(shared_reset_marker_module, "read_config_with_signature", _counting_read)
tracker = _tracker(marker, config_path, clock, poll_interval_seconds=1.0)
assert tracker.poll() is None
assert len(reads) == 1
clock.advance(0.5)
tracker.poll()
tracker.poll()
assert len(reads) == 1, "no repeated stat inside the throttle window"
clock.advance(0.5)
tracker.poll()
assert len(reads) == 2
def test_change_inside_the_throttle_window_is_seen_on_the_next_interval(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
tracker = _tracker(marker, config_path, clock)
assert tracker.poll() is None
generation = marker.publish(config_path)
clock.advance(0.2)
assert tracker.poll() is None, "throttled: the stat is skipped"
clock.advance(0.8)
assert tracker.poll() == SharedResetChange(user_ids=None, generation=generation)
def test_tracker_without_a_config_path_never_reports_a_change(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
tracker = _tracker(marker, None, clock)
assert tracker.poll() is None
marker.publish(config_path)
clock.advance(1.0)
assert tracker.poll() is None
def test_marker_deleted_after_adoption_is_a_global_reset(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
marker.publish(config_path)
tracker = _tracker(marker, config_path, clock)
assert tracker.poll() is None
marker.path_for(config_path).unlink()
clock.advance(1.0)
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
def test_malformed_marker_is_a_global_reset(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
tracker = _tracker(marker, config_path, clock)
assert tracker.poll() is None
marker.path_for(config_path).write_text("{not json", encoding="utf-8")
clock.advance(1.0)
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
def test_config_path_switch_after_adoption_is_a_global_reset(tmp_path: Path, config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
current = {"path": None}
tracker = SharedResetMarkerTracker(marker, resolve_config_path=lambda: current["path"], clock=clock)
assert tracker.poll() is None
current["path"] = config_path
clock.advance(1.0)
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
def test_reset_forgets_the_observed_state(config_path: Path) -> None:
marker = SharedResetMarker("skills-cache-reset")
clock = _Clock()
tracker = _tracker(marker, config_path, clock)
assert tracker.poll() is None
marker.publish(config_path)
tracker.reset()
assert tracker.poll() is None, "after a reset the first poll adopts silently"
clock.advance(1.0)
assert tracker.poll() is None
def test_resolve_shared_config_path_treats_a_missing_explicit_file_as_unconfigured(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
missing = tmp_path / "missing.json"
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(missing))
with pytest.raises(FileNotFoundError):
ExtensionsConfig.resolve_config_path()
assert resolve_shared_config_path() is None
def test_resolve_shared_config_path_returns_the_explicit_file(config_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
assert resolve_shared_config_path() == config_path

704
backend/tests/test_skills_cache_shared_reset.py Normal file
View File

@ -0,0 +1,704 @@
"""Cross-replica invalidation of the skills prompt caches.
``SkillStorage.load_skills()`` rescans disk and re-reads ``extensions_config.json``
on every call, so the only state that can go stale between Gateway replicas is
the prompt layer in ``deerflow.agents.lead_agent.prompt``: the global
enabled-skills cache, the per-``(app_config, user_id)`` cache and the rendered
``<skill_system>`` LRU. Every skill mutation handled by one replica therefore
publishes ``.<extensions config>.skills-cache-reset.json`` beside the shared
config; every other replica compares that marker's signature (throttled to one
stat per second) before serving a cached entry.
The prompt-layer tests play "process B" with the real module state and
"process A" by writing the marker directly, the way another replica would.
"""
from __future__ import annotations
import asyncio
import json
import zipfile
from pathlib import Path
from types import SimpleNamespace
from uuid import uuid4
import pytest
from _router_auth_helpers import make_authed_test_app
from fastapi import FastAPI
from fastapi.testclient import TestClient
import deerflow.config.shared_reset_marker as shared_reset_marker_module
from app.gateway.auth.models import User
from app.gateway.deps import get_config
from app.gateway.routers import skills as skills_router
from deerflow.agents.lead_agent import prompt as prompt_module
from deerflow.config.authorization_config import AuthorizationConfig
from deerflow.config.extensions_config import ExtensionsConfig, reset_extensions_config
from deerflow.config.file_signature import get_config_signature
from deerflow.config.paths import Paths
from deerflow.config.shared_reset_marker import SharedResetMarkerTracker
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
from deerflow.skills.types import Skill
_PROCESS_MESSAGE = "Skill caches invalidated; subsequent runs in this Gateway process will rescan the latest skills."
class _Clock:
def __init__(self, start: float = 1000.0) -> None:
self.now = start
def __call__(self) -> float:
return self.now
def advance(self, seconds: float) -> None:
self.now += seconds
def _reset_prompt_cache_state() -> None:
prompt_module._get_cached_skills_prompt_section.cache_clear()
with prompt_module._enabled_skills_lock:
prompt_module._enabled_skills_cache = None
prompt_module._enabled_skills_by_config_cache.clear()
prompt_module._enabled_skills_refresh_active = False
prompt_module._enabled_skills_refresh_version = 0
prompt_module._enabled_skills_refresh_event.clear()
prompt_module._enabled_skills_refresh_waiters.clear()
prompt_module._skills_cache_reset_tracker.reset()
@pytest.fixture()
def shared_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
"""One extensions config shared by every simulated replica, plus a fake clock."""
config_path = tmp_path / "shared" / "extensions_config.json"
config_path.parent.mkdir()
config_path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
reset_extensions_config()
clock = _Clock()
monkeypatch.setattr(prompt_module, "_skills_cache_reset_tracker", SharedResetMarkerTracker(prompt_module.SKILLS_CACHE_RESET_MARKER, clock=clock))
_reset_prompt_cache_state()
try:
yield SimpleNamespace(path=config_path, clock=clock, marker=prompt_module.SKILLS_CACHE_RESET_MARKER)
finally:
_reset_prompt_cache_state()
reset_extensions_config()
def _marker_path(shared) -> Path:
return shared.marker.path_for(shared.path)
def _marker_payload(shared) -> dict:
return json.loads(_marker_path(shared).read_text(encoding="utf-8"))
def _write_public_skill(root: Path, name: str, description: str) -> None:
skill_dir = root / "public" / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(f"---\nname: {name}\ndescription: {description}\n---\n# {name}\n", encoding="utf-8")
def _skills_config(skills_root: Path) -> SimpleNamespace:
return SimpleNamespace(
skills=SimpleNamespace(container_path="/mnt/skills", use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage", get_skills_path=lambda: skills_root),
skill_evolution=SimpleNamespace(enabled=False, moderation_model_name=None),
)
def _install_prompt_storage(monkeypatch: pytest.MonkeyPatch, skills_root: Path) -> LocalSkillStorage:
storage = LocalSkillStorage(host_path=str(skills_root))
monkeypatch.setattr(prompt_module, "get_or_new_skill_storage", lambda **_kwargs: storage)
monkeypatch.setattr(prompt_module, "get_or_new_user_skill_storage", lambda _user_id, **_kwargs: storage)
return storage
# --------------------------------------------------------------------------- #
# Prompt layer: replica B observes a marker written by replica A
# --------------------------------------------------------------------------- #
def test_peer_global_reset_invalidates_per_config_cache_and_prompt_section(shared_config, monkeypatch, tmp_path: Path) -> None:
skills_root = tmp_path / "skills"
_write_public_skill(skills_root, "changed-skill", "old description")
_install_prompt_storage(monkeypatch, skills_root)
config = _skills_config(skills_root)
before = prompt_module.get_enabled_skills_for_config(config, user_id="alice")
assert before[0].description == "old description"
assert "old description" in prompt_module.get_skills_prompt_section(app_config=config, user_id="alice")
assert prompt_module._get_cached_skills_prompt_section.cache_info().currsize == 1
# Replica A: the skill changed on the shared volume and A published a reset.
_write_public_skill(skills_root, "changed-skill", "new description")
shared_config.marker.publish(shared_config.path)
# Inside the throttle window B keeps serving its cache (bounded staleness).
assert prompt_module.get_enabled_skills_for_config(config, user_id="alice")[0].description == "old description"
shared_config.clock.advance(1.0)
after = prompt_module.get_enabled_skills_for_config(config, user_id="alice")
assert after[0].description == "new description"
assert prompt_module._get_cached_skills_prompt_section.cache_info().currsize == 0
assert "new description" in prompt_module.get_skills_prompt_section(app_config=config, user_id="alice")
def test_peer_user_scoped_reset_invalidates_only_that_user(shared_config, monkeypatch, tmp_path: Path) -> None:
skills_root = tmp_path / "skills"
_write_public_skill(skills_root, "changed-skill", "old description")
_install_prompt_storage(monkeypatch, skills_root)
config = _skills_config(skills_root)
assert prompt_module.get_enabled_skills_for_config(config, user_id="alice")[0].description == "old description"
assert prompt_module.get_enabled_skills_for_config(config, user_id="bob")[0].description == "old description"
_write_public_skill(skills_root, "changed-skill", "new description")
shared_config.marker.publish(shared_config.path, user_id="alice")
shared_config.clock.advance(1.0)
assert prompt_module.get_enabled_skills_for_config(config, user_id="alice")[0].description == "new description"
# Mirrors invalidate_user_skill_cache(): the other user's entry is untouched.
assert prompt_module.get_enabled_skills_for_config(config, user_id="bob")[0].description == "old description"
with prompt_module._enabled_skills_lock:
assert (id(config), "bob") in prompt_module._enabled_skills_by_config_cache
def test_peer_global_reset_refreshes_the_global_enabled_skills_cache(shared_config, monkeypatch, tmp_path: Path) -> None:
skills_root = tmp_path / "skills"
_write_public_skill(skills_root, "cached-skill", "old description")
storage = _install_prompt_storage(monkeypatch, skills_root)
old_skills = storage.load_skills(enabled_only=True)
_write_public_skill(skills_root, "cached-skill", "new description")
new_skills = storage.load_skills(enabled_only=True)
monkeypatch.setattr(prompt_module, "_load_enabled_skills_sync", lambda: list(new_skills))
with prompt_module._enabled_skills_lock:
prompt_module._enabled_skills_cache = list(old_skills)
assert prompt_module.get_cached_enabled_skills()[0].description == "old description"
shared_config.marker.publish(shared_config.path)
shared_config.clock.advance(1.0)
first_after = prompt_module.get_cached_enabled_skills()
assert prompt_module._enabled_skills_refresh_event.wait(timeout=5), "the reset must start the off-loop refresh"
assert prompt_module.get_cached_enabled_skills()[0].description == "new description"
# The last-known-good list is served while the refresh runs.
assert first_after[0].description in {"old description", "new description"}
def test_prompt_lookups_stat_the_marker_at_most_once_per_second(shared_config, monkeypatch, tmp_path: Path) -> None:
skills_root = tmp_path / "skills"
_write_public_skill(skills_root, "a-skill", "description")
_install_prompt_storage(monkeypatch, skills_root)
config = _skills_config(skills_root)
shared_config.marker.publish(shared_config.path)
reads: list[Path] = []
original = shared_reset_marker_module.read_config_with_signature
def _counting_read(path: Path):
reads.append(path)
return original(path)
monkeypatch.setattr(shared_reset_marker_module, "read_config_with_signature", _counting_read)
prompt_module.get_enabled_skills_for_config(config, user_id="alice")
prompt_module.get_enabled_skills_for_config(config, user_id="alice")
prompt_module.get_skills_prompt_section(app_config=config, user_id="alice")
assert reads == [_marker_path(shared_config)]
shared_config.clock.advance(1.0)
prompt_module.get_enabled_skills_for_config(config, user_id="alice")
assert len(reads) == 2
def test_publish_skills_cache_reset_without_config_path_is_process_local(shared_config, monkeypatch) -> None:
monkeypatch.setattr(prompt_module, "resolve_shared_config_path", lambda: None)
assert prompt_module.publish_skills_cache_reset() is None
assert not _marker_path(shared_config).exists()
def test_publish_skills_cache_reset_writes_marker_without_invalidating_the_publisher(shared_config, monkeypatch, tmp_path: Path) -> None:
skills_root = tmp_path / "skills"
_write_public_skill(skills_root, "a-skill", "description")
_install_prompt_storage(monkeypatch, skills_root)
config = _skills_config(skills_root)
prompt_module.get_enabled_skills_for_config(config, user_id="alice")
generation = prompt_module.publish_skills_cache_reset(user_id="alice")
assert generation is not None
assert _marker_payload(shared_config) == {
"version": 1,
"generation": generation,
"previous_generation": None,
"published_at": _marker_payload(shared_config)["published_at"],
"user_id": "alice",
}
shared_config.clock.advance(1.0)
prompt_module.get_enabled_skills_for_config(config, user_id="alice")
with prompt_module._enabled_skills_lock:
assert (id(config), "alice") in prompt_module._enabled_skills_by_config_cache, "the publishing process already refreshed itself"
# --------------------------------------------------------------------------- #
# Router: every mutation publishes the marker; reload reports its scope
# --------------------------------------------------------------------------- #
def _make_admin_user() -> User:
return User(email="admin-shared-reset@example.com", password_hash="x", system_role="admin", id=uuid4())
def _make_test_app(config) -> FastAPI:
if not hasattr(config, "authorization"):
config.authorization = AuthorizationConfig(enabled=False)
app = make_authed_test_app(user_factory=_make_admin_user)
app.state.config = config
app.dependency_overrides[get_config] = lambda: config
app.include_router(skills_router.router)
return app
def _skill_content(name: str, description: str = "Demo skill") -> str:
return f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n"
async def _allow_scan(*_args, **_kwargs):
from deerflow.skills.security_scanner import ScanResult
return ScanResult(decision="allow", reason="ok")
def _user_custom_dir(base_dir: Path, user_id: str = "default") -> Path:
return base_dir / "users" / user_id / "skills" / "custom"
@pytest.fixture()
def custom_skill_app(shared_config, monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
"""An admin app whose caller owns one editable custom skill on disk."""
skills_root = tmp_path / "skills"
custom_dir = _user_custom_dir(tmp_path) / "demo-skill"
custom_dir.mkdir(parents=True)
(custom_dir / "SKILL.md").write_text(_skill_content("demo-skill"), encoding="utf-8")
config = SimpleNamespace(
skills=SimpleNamespace(get_skills_path=lambda: skills_root, container_path="/mnt/skills", use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage"),
skill_evolution=SimpleNamespace(enabled=True, moderation_model_name=None),
)
monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
monkeypatch.setattr(skills_router, "scan_skill_content", _allow_scan)
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
events: list[tuple[str, str | None]] = []
async def _refresh(user_id: str) -> None:
events.append(("refresh", user_id))
real_publish = skills_router.publish_skills_cache_reset
def _publish(*, user_id=None):
events.append(("publish", user_id))
return real_publish(user_id=user_id)
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
monkeypatch.setattr(skills_router, "publish_skills_cache_reset", _publish)
return SimpleNamespace(app=_make_test_app(config), config=config, events=events, custom_dir=custom_dir, skills_root=skills_root, base_dir=tmp_path)
def test_custom_skill_edit_publishes_user_scoped_marker_after_local_refresh(shared_config, custom_skill_app) -> None:
assert not _marker_path(shared_config).exists()
with TestClient(custom_skill_app.app) as client:
response = client.put("/api/skills/custom/demo-skill", json={"content": _skill_content("demo-skill", "Edited skill")})
assert response.status_code == 200, response.text
assert custom_skill_app.events == [("refresh", "default"), ("publish", "default")]
assert _marker_payload(shared_config)["user_id"] == "default"
def test_custom_skill_rollback_publishes_marker(shared_config, custom_skill_app) -> None:
with TestClient(custom_skill_app.app) as client:
assert client.put("/api/skills/custom/demo-skill", json={"content": _skill_content("demo-skill", "Edited skill")}).status_code == 200
after_edit = get_config_signature(_marker_path(shared_config))
response = client.post("/api/skills/custom/demo-skill/rollback", json={"history_index": -1})
assert response.status_code == 200, response.text
assert response.json()["description"] == "Demo skill"
assert get_config_signature(_marker_path(shared_config)) != after_edit
assert custom_skill_app.events[-2:] == [("refresh", "default"), ("publish", "default")]
def test_custom_skill_delete_publishes_marker(shared_config, custom_skill_app) -> None:
with TestClient(custom_skill_app.app) as client:
response = client.delete("/api/skills/custom/demo-skill")
assert response.status_code == 200, response.text
assert not (custom_skill_app.custom_dir / "SKILL.md").exists()
assert custom_skill_app.events == [("refresh", "default"), ("publish", "default")]
assert _marker_payload(shared_config)["user_id"] == "default"
def test_archive_install_publishes_marker(shared_config, custom_skill_app, monkeypatch, tmp_path: Path) -> None:
archive = tmp_path / "install-skill.skill"
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr("install-skill/SKILL.md", _skill_content("install-skill"))
storage = UserScopedSkillStorage("default", host_path=str(custom_skill_app.skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr("deerflow.skills.installer.scan_skill_content", _allow_scan)
monkeypatch.setattr(skills_router, "resolve_thread_virtual_path", lambda thread_id, path: archive)
with TestClient(custom_skill_app.app) as client:
response = client.post("/api/skills/install", json={"thread_id": "thread-1", "path": "mnt/user-data/outputs/install-skill.skill"})
assert response.status_code == 200, response.text
assert (_user_custom_dir(custom_skill_app.base_dir) / "install-skill").exists()
assert custom_skill_app.events == [("refresh", "default"), ("publish", "default")]
assert _marker_payload(shared_config)["user_id"] == "default"
def test_archive_upload_publishes_marker(shared_config, custom_skill_app, monkeypatch) -> None:
class _Storage:
async def ainstall_skill_from_archive(self, archive_path: Path) -> dict:
return {"success": True, "skill_name": "uploaded-skill", "message": "Skill installed successfully"}
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: _Storage())
with TestClient(custom_skill_app.app) as client:
response = client.post("/api/skills/install/upload", files={"archive": ("uploaded-skill.skill", b"skill archive bytes", "application/octet-stream")})
assert response.status_code == 200, response.text
assert custom_skill_app.events == [("refresh", "default"), ("publish", "default")]
assert _marker_payload(shared_config)["user_id"] == "default"
def test_public_skill_toggle_publishes_global_marker(shared_config, monkeypatch) -> None:
def _load_skills(*, enabled_only: bool):
enabled = ExtensionsConfig.from_file(str(shared_config.path)).skills.get("public-skill", SimpleNamespace(enabled=True)).enabled
skill = Skill(
name="public-skill",
description="Description for public-skill",
license="MIT",
skill_dir=Path("/tmp/public-skill"),
skill_file=Path("/tmp/public-skill/SKILL.md"),
relative_path=Path("public-skill"),
category="public",
enabled=enabled,
)
return [] if enabled_only and not enabled else [skill]
events: list[tuple[str, str | None]] = []
real_publish = skills_router.publish_skills_cache_reset
def _publish(*, user_id=None):
events.append(("publish", user_id))
return real_publish(user_id=user_id)
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: SimpleNamespace(load_skills=_load_skills))
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "clear_skills_system_prompt_cache", lambda: events.append(("clear", None)))
monkeypatch.setattr(skills_router, "publish_skills_cache_reset", _publish)
with TestClient(_make_test_app(SimpleNamespace())) as client:
response = client.put("/api/skills/public-skill", json={"enabled": False})
assert response.status_code == 200, response.text
assert response.json()["enabled"] is False
assert json.loads(shared_config.path.read_text(encoding="utf-8"))["skills"]["public-skill"]["enabled"] is False
assert events == [("clear", None), ("publish", None)]
assert "user_id" not in _marker_payload(shared_config)
def test_custom_skill_toggle_publishes_user_scoped_marker(shared_config, monkeypatch) -> None:
from deerflow.skills.storage import user_scoped_skill_storage as uss_module
enabled_state = {"value": True}
class _FakeUserScopedStorage:
def load_skills(self, *, enabled_only: bool = False):
skill = Skill(
name="demo-skill",
description="Description for demo-skill",
license="MIT",
skill_dir=Path("/tmp/demo-skill"),
skill_file=Path("/tmp/demo-skill/SKILL.md"),
relative_path=Path("demo-skill"),
category="custom",
enabled=enabled_state["value"],
)
return [] if enabled_only and not skill.enabled else [skill]
def set_skill_enabled_state(self, name: str, enabled: bool) -> None:
enabled_state["value"] = enabled
events: list[tuple[str, str | None]] = []
async def _refresh(user_id: str) -> None:
events.append(("refresh", user_id))
real_publish = skills_router.publish_skills_cache_reset
def _publish(*, user_id=None):
events.append(("publish", user_id))
return real_publish(user_id=user_id)
monkeypatch.setattr(uss_module, "UserScopedSkillStorage", _FakeUserScopedStorage)
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: _FakeUserScopedStorage())
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "refresh_user_skills_system_prompt_cache_async", _refresh)
monkeypatch.setattr(skills_router, "publish_skills_cache_reset", _publish)
with TestClient(_make_test_app(SimpleNamespace())) as client:
response = client.put("/api/skills/demo-skill", json={"enabled": False})
assert response.status_code == 200, response.text
assert events == [("refresh", "default"), ("publish", "default")]
assert _marker_payload(shared_config)["user_id"] == "default"
def _make_reload_app() -> FastAPI:
config = SimpleNamespace(
skills=SimpleNamespace(get_skills_path=lambda: "/tmp/skills", container_path="/mnt/skills", use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage"),
skill_evolution=SimpleNamespace(enabled=True, moderation_model_name=None),
)
app = make_authed_test_app(user_factory=_make_admin_user)
app.state.config = config
app.dependency_overrides[get_config] = lambda: config
app.include_router(skills_router.router)
return app
def test_reload_publishes_marker_and_reports_shared_config_scope(shared_config, monkeypatch) -> None:
events: list[str] = []
async def _refresh() -> None:
events.append("refresh")
real_publish = skills_router.publish_skills_cache_reset
def _publish(*, user_id=None):
events.append("publish")
return real_publish(user_id=user_id)
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
monkeypatch.setattr(skills_router, "publish_skills_cache_reset", _publish)
with TestClient(_make_reload_app()) as client:
first = client.post("/api/skills/reload")
first_signature = get_config_signature(_marker_path(shared_config))
second = client.post("/api/skills/reload")
assert first.status_code == 200, first.text
assert first.json()["success"] is True
assert first.json()["scope"] == "shared_config"
assert "shared config directory" in first.json()["message"]
assert events == ["refresh", "publish", "refresh", "publish"]
assert second.status_code == 200
assert get_config_signature(_marker_path(shared_config)) != first_signature
assert "user_id" not in _marker_payload(shared_config)
def test_reload_without_config_path_falls_back_to_process_scope(shared_config, monkeypatch) -> None:
async def _refresh() -> None:
return None
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
monkeypatch.setattr(prompt_module, "resolve_shared_config_path", lambda: None)
with TestClient(_make_reload_app()) as client:
response = client.post("/api/skills/reload")
assert response.status_code == 200, response.text
assert response.json() == {"success": True, "scope": "process", "message": _PROCESS_MESSAGE}
assert not _marker_path(shared_config).exists()
def test_reload_marker_publication_failure_is_a_generic_server_error(shared_config, monkeypatch) -> None:
"""A 200 must never mean that only the handling process was refreshed."""
async def _refresh() -> None:
return None
def _failing_publish(*, user_id=None):
raise OSError("read-only volume at /srv/company/minio")
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
monkeypatch.setattr(skills_router, "publish_skills_cache_reset", _failing_publish)
with TestClient(_make_reload_app()) as client:
response = client.post("/api/skills/reload")
assert response.status_code == 500
assert response.json() == {"detail": "Failed to invalidate skills cache."}
assert "/srv/company/minio" not in response.text
# --------------------------------------------------------------------------- #
# Review follow-ups: cancellation drains the reload tail; publication errors
# are never reported as 404
# --------------------------------------------------------------------------- #
def _admin_request() -> SimpleNamespace:
return SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
@pytest.mark.asyncio
async def test_reload_drains_refresh_and_publish_across_cancellation(shared_config, monkeypatch) -> None:
"""A caller cancelled after the local refresh must not skip the shared marker.
``/skills/reload`` is the operator hook for external mount writes: if the
cancellation unwound between the refresh and the publish, every peer
replica would keep the old skill set, the one outcome the endpoint exists
to prevent. Mirrors the drained-tail tests of the other mutation routes.
"""
started = asyncio.Event()
release = asyncio.Event()
async def _blocked_refresh() -> None:
started.set()
await release.wait()
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _blocked_refresh)
assert not _marker_path(shared_config).exists()
task = asyncio.create_task(skills_router.reload_skills(_admin_request()))
try:
await asyncio.wait_for(started.wait(), 5)
task.cancel()
await asyncio.sleep(0.05)
task.cancel()
await asyncio.sleep(0.05)
assert not task.done(), "the drained tail must keep running until it settles"
release.set()
with pytest.raises(asyncio.CancelledError):
await task
finally:
release.set()
if not task.done():
task.cancel()
await asyncio.gather(task, return_exceptions=True)
assert _marker_path(shared_config).exists(), "the shared marker must be published even though the caller was cancelled"
assert "user_id" not in _marker_payload(shared_config)
def _marker_write_raises_file_not_found(monkeypatch: pytest.MonkeyPatch, shared_config) -> str:
"""Simulate the config directory vanishing between path resolution and the atomic write."""
missing = str(_marker_path(shared_config))
def _publish(config_path, *, user_id=None):
raise FileNotFoundError(2, "No such file or directory", missing)
monkeypatch.setattr(prompt_module.SKILLS_CACHE_RESET_MARKER, "publish", _publish)
return missing
def test_install_marker_publication_failure_is_500_not_404(shared_config, custom_skill_app, monkeypatch, tmp_path: Path) -> None:
"""A FileNotFoundError from the marker write must not borrow the 'archive not found' status."""
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
archive = tmp_path / "install-skill.skill"
with zipfile.ZipFile(archive, "w") as zf:
zf.writestr("install-skill/SKILL.md", _skill_content("install-skill"))
storage = UserScopedSkillStorage("default", host_path=str(custom_skill_app.skills_root))
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: storage)
monkeypatch.setattr("deerflow.skills.installer.scan_skill_content", _allow_scan)
monkeypatch.setattr(skills_router, "resolve_thread_virtual_path", lambda thread_id, path: archive)
with TestClient(custom_skill_app.app) as client:
response = client.post("/api/skills/install", json={"thread_id": "thread-1", "path": "mnt/user-data/outputs/install-skill.skill"})
assert response.status_code == 500, response.text
assert response.json()["detail"].startswith("Failed to install skill:")
assert missing not in response.text
# The install itself succeeded; only the cross-replica publication failed.
assert (_user_custom_dir(custom_skill_app.base_dir) / "install-skill").exists()
assert custom_skill_app.events == [("refresh", "default"), ("publish", "default")]
def test_edit_marker_publication_failure_is_500_not_404(shared_config, custom_skill_app, monkeypatch) -> None:
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
with TestClient(custom_skill_app.app) as client:
response = client.put("/api/skills/custom/demo-skill", json={"content": _skill_content("demo-skill", "Edited skill")})
assert response.status_code == 500, response.text
assert response.json()["detail"].startswith("Failed to update custom skill:")
assert missing not in response.text
assert "Edited skill" in (custom_skill_app.custom_dir / "SKILL.md").read_text(encoding="utf-8")
def test_delete_marker_publication_failure_is_500_not_404(shared_config, custom_skill_app, monkeypatch) -> None:
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
with TestClient(custom_skill_app.app) as client:
response = client.delete("/api/skills/custom/demo-skill")
assert response.status_code == 500, response.text
assert response.json()["detail"].startswith("Failed to delete custom skill:")
assert missing not in response.text
def test_rollback_marker_publication_failure_is_500_not_404(shared_config, custom_skill_app, monkeypatch) -> None:
with TestClient(custom_skill_app.app) as client:
assert client.put("/api/skills/custom/demo-skill", json={"content": _skill_content("demo-skill", "Edited skill")}).status_code == 200
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
response = client.post("/api/skills/custom/demo-skill/rollback", json={"history_index": -1})
assert response.status_code == 500, response.text
assert response.json()["detail"].startswith("Failed to roll back custom skill:")
assert missing not in response.text
def test_public_toggle_marker_publication_failure_is_500(shared_config, monkeypatch) -> None:
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
skill = Skill(
name="public-skill",
description="Description for public-skill",
license="MIT",
skill_dir=Path("/tmp/public-skill"),
skill_file=Path("/tmp/public-skill/SKILL.md"),
relative_path=Path("public-skill"),
category="public",
enabled=True,
)
monkeypatch.setattr(skills_router, "_get_user_skill_storage", lambda cfg: SimpleNamespace(load_skills=lambda *, enabled_only: [skill]))
monkeypatch.setattr(skills_router, "get_effective_user_id", lambda: "default")
monkeypatch.setattr(skills_router, "clear_skills_system_prompt_cache", lambda: None)
with TestClient(_make_test_app(SimpleNamespace())) as client:
response = client.put("/api/skills/public-skill", json={"enabled": False})
assert response.status_code == 500, response.text
assert response.json()["detail"].startswith("Failed to update skill:")
assert missing not in response.text
assert json.loads(shared_config.path.read_text(encoding="utf-8"))["skills"]["public-skill"]["enabled"] is False
def test_reload_marker_write_file_not_found_is_a_generic_server_error(shared_config, monkeypatch) -> None:
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
async def _refresh() -> None:
return None
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
with TestClient(_make_reload_app()) as client:
response = client.post("/api/skills/reload")
assert response.status_code == 500
assert response.json() == {"detail": "Failed to invalidate skills cache."}
assert missing not in response.text
def test_publish_skills_cache_reset_wraps_filesystem_errors(shared_config, monkeypatch) -> None:
"""The dedicated error type is not an OSError, so no handler can map it to 404/400."""
missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
with pytest.raises(prompt_module.SkillCacheResetPublishError) as excinfo:
prompt_module.publish_skills_cache_reset(user_id="alice")
assert not isinstance(excinfo.value, OSError)
assert isinstance(excinfo.value.__cause__, FileNotFoundError)
assert missing not in str(excinfo.value)

12
backend/tests/test_skills_reload.py
View File

@ -61,6 +61,16 @@ def _reset_prompt_cache_state() -> None:
prompt_module._enabled_skills_refresh_version = 0
prompt_module._enabled_skills_refresh_event.clear()
prompt_module._enabled_skills_refresh_waiters.clear()
prompt_module._skills_cache_reset_tracker.reset()
def _without_shared_config_path(monkeypatch) -> None:
"""Pin the process-local fallback: no extensions config path, no shared marker.
The shared-config path (marker published, ``scope: shared_config``) is
covered by ``tests/test_skills_cache_shared_reset.py``.
"""
monkeypatch.setattr(prompt_module, "resolve_shared_config_path", lambda: None)
def test_admin_can_reload_skills(monkeypatch) -> None:
@ -71,6 +81,7 @@ def test_admin_can_reload_skills(monkeypatch) -> None:
calls += 1
monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
_without_shared_config_path(monkeypatch)
app = _make_app(system_role="admin")
with TestClient(app) as client:
@ -113,6 +124,7 @@ def test_reload_worker_failure_returns_500_preserves_last_good_cache_and_can_ret
return recovered_skills
monkeypatch.setattr(prompt_module, "_load_enabled_skills_sync", _load_enabled_skills)
_without_shared_config_path(monkeypatch)
_reset_prompt_cache_state()
with prompt_module._enabled_skills_lock:
prompt_module._enabled_skills_cache = last_good_skills

2
frontend/src/content/en/harness/skills.mdx
View File

@ -68,7 +68,7 @@ DeerFlow ships with the following public skills:
### Discovery and loading
`load_skills()` in `skills/loader.py` scans both `public/` and `custom/` directories under the configured skills path. It re-reads `ExtensionsConfig.from_file()` on every call, which means enabling or disabling a skill through the Gateway API takes effect immediately in the running agent runtime without a restart.
`SkillStorage.load_skills()` in `skills/storage/` (`LocalSkillStorage` and the per-user `UserScopedSkillStorage`) scans the `public/`, per-user `custom/`, integration and legacy directories under the configured skills path. It rescans disk and re-reads the extensions config on every call, so enabling or disabling a skill through the Gateway API takes effect immediately in the running agent runtime without a restart. The only cached skill state is the prompt layer in `agents/lead_agent/prompt.py` (the enabled-skills list and the rendered `<skill_system>` section); every skill mutation and `POST /api/skills/reload` refresh it in the handling process and publish a `.extensions_config.json.skills-cache-reset.json` marker beside the shared `extensions_config.json`, which every other Gateway process mounting that directory checks (at most once per second) before its next prompt build. One API call therefore reaches every replica sharing the volume; the reload response reports `scope: shared_config`, or `scope: process` when no extensions config path is available.
### Parsing

2
frontend/src/content/zh/harness/skills.mdx
View File

@ -64,7 +64,7 @@ DeerFlow 内置以下公共技能：
### 发现和加载
`skills/loader.py` 中的 `load_skills()` 扫描配置技能路径下的 `public/` 和 `custom/` 目录。它每次调用都重新读取 `ExtensionsConfig.from_file()`，这意味着通过 Gateway API 启用或禁用技能会立即在运行中的 LangGraph 服务器中生效，无需重启。
`skills/storage/` 中的 `SkillStorage.load_skills()`（`LocalSkillStorage` 以及按用户隔离的 `UserScopedSkillStorage`）扫描配置技能路径下的 `public/`、按用户的 `custom/`、集成和 legacy 目录。它每次调用都重新扫描磁盘并重新读取 extensions 配置，这意味着通过 Gateway API 启用或禁用技能会立即在运行中的 Agent 运行时中生效，无需重启。唯一被缓存的技能状态是 `agents/lead_agent/prompt.py` 中的提示词层（已启用技能列表和渲染后的 `<skill_system>` 段）；每一次技能变更以及 `POST /api/skills/reload` 都会先刷新处理请求的进程，再在共享的 `extensions_config.json` 旁写入 `.extensions_config.json.skills-cache-reset.json` 标记文件，其他挂载同一目录的 Gateway 进程会在下一次构建提示词前检查该标记（每秒最多一次）。因此一次 API 调用即可覆盖所有共享该卷的副本；reload 响应返回 `scope: shared_config`，在没有可用 extensions 配置路径时返回 `scope: process`。
### 解析