diff --git a/CHANGELOG.md b/CHANGELOG.md index c27e0bd91f..1abd166e4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 `` 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 diff --git a/README.md b/README.md index 7b6b9d8f46..d776dc6a29 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index fd96971e55..24b3f21eae 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -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 | diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index dce8b835d6..29a9293d00 100644 --- a/backend/app/gateway/routers/skills.py +++ b/backend/app/gateway/routers/skills.py @@ -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)) diff --git a/backend/docs/API.md b/backend/docs/API.md index cfa865621e..2fcf4f927d 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -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 `..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 diff --git a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index 0047bc1dd2..0ad06297c9 100644 --- a/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/backend/packages/harness/deerflow/agents/lead_agent/prompt.py @@ -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 +# ``..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) diff --git a/backend/packages/harness/deerflow/config/AGENTS.md b/backend/packages/harness/deerflow/config/AGENTS.md index 2ed1ebda12..d25b160747 100644 --- a/backend/packages/harness/deerflow/config/AGENTS.md +++ b/backend/packages/harness/deerflow/config/AGENTS.md @@ -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, `...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: diff --git a/backend/packages/harness/deerflow/config/shared_reset_marker.py b/backend/packages/harness/deerflow/config/shared_reset_marker.py new file mode 100644 index 0000000000..3a183ce5fc --- /dev/null +++ b/backend/packages/harness/deerflow/config/shared_reset_marker.py @@ -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 +(``...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) diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index bc600d3ccc..25f6b0b379 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -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 -c ''` **runs** the command (`-p` is `npm exec`'s `--package`, so `` 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`. diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index 921d6d0982..5ab5014441 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -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 +#: ``..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, diff --git a/backend/packages/harness/deerflow/skills/AGENTS.md b/backend/packages/harness/deerflow/skills/AGENTS.md index 666c58904e..65671347e8 100644 --- a/backend/packages/harness/deerflow/skills/AGENTS.md +++ b/backend/packages/harness/deerflow/skills/AGENTS.md @@ -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 `` 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 `..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 (`` block). Controlled by `skills.deferred_discovery: false` (default). diff --git a/backend/tests/blocking_io/test_skills_reload.py b/backend/tests/blocking_io/test_skills_reload.py index b54a00d172..b96839094f 100644 --- a/backend/tests/blocking_io/test_skills_reload.py +++ b/backend/tests/blocking_io/test_skills_reload.py @@ -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()) diff --git a/backend/tests/test_shared_reset_marker.py b/backend/tests/test_shared_reset_marker.py new file mode 100644 index 0000000000..ad6de7bfaa --- /dev/null +++ b/backend/tests/test_shared_reset_marker.py @@ -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 (``..mcp-cache-reset.json``) and the skills +prompt-cache reset (``..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 diff --git a/backend/tests/test_skills_cache_shared_reset.py b/backend/tests/test_skills_cache_shared_reset.py new file mode 100644 index 0000000000..20cac1b199 --- /dev/null +++ b/backend/tests/test_skills_cache_shared_reset.py @@ -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 +```` LRU. Every skill mutation handled by one replica therefore +publishes ``..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) diff --git a/backend/tests/test_skills_reload.py b/backend/tests/test_skills_reload.py index bd7ddbf61c..9a05008e5b 100644 --- a/backend/tests/test_skills_reload.py +++ b/backend/tests/test_skills_reload.py @@ -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 diff --git a/frontend/src/content/en/harness/skills.mdx b/frontend/src/content/en/harness/skills.mdx index e92ee13951..6ffd8f4662 100644 --- a/frontend/src/content/en/harness/skills.mdx +++ b/frontend/src/content/en/harness/skills.mdx @@ -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 `` 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 diff --git a/frontend/src/content/zh/harness/skills.mdx b/frontend/src/content/zh/harness/skills.mdx index 1940e0ac5d..99fb90fc35 100644 --- a/frontend/src/content/zh/harness/skills.mdx +++ b/frontend/src/content/zh/harness/skills.mdx @@ -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` 中的提示词层（已启用技能列表和渲染后的 `` 段）；每一次技能变更以及 `POST /api/skills/reload` 都会先刷新处理请求的进程，再在共享的 `extensions_config.json` 旁写入 `.extensions_config.json.skills-cache-reset.json` 标记文件，其他挂载同一目录的 Gateway 进程会在下一次构建提示词前检查该标记（每秒最多一次）。因此一次 API 调用即可覆盖所有共享该卷的副本；reload 响应返回 `scope: shared_config`，在没有可用 extensions 配置路径时返回 `scope: process`。 ### 解析