fix(multi-instance): publish skill_manage cache resets and tolerate a concurrent namespace create (#7099)

* fix(multi-instance): publish skill_manage cache resets and tolerate a concurrent namespace create

Two leftovers from the multi-instance work (#7097).

The agent's skill_manage tool refreshed only its own process's skills
prompt cache after a create, edit, patch, delete, write_file or
remove_file, so a Gateway replica sharing the home volume kept rendering
that user's previous skill list. Every mutation now runs one drained tail,
as the skill routes do: the write and its history record, the local
refresh, then publish_skills_cache_reset(user_id=...) off the event loop.
A run cancelled while the write is in flight still refreshes and
publishes before the cancellation propagates. A marker that cannot be
written is logged inside the tail and raised as SkillCacheResetPublishError
with a message saying the change was applied, so the model does not retry
a write that succeeded.

Two provisioner replicas starting together against a missing
K8S_NAMESPACE both read 404 and both create; the loser's 409 escaped the
lifespan and crashed it. A 409 on create now counts as the namespace
existing, as the Pod and Service creates already do.

Also drops a stale duplicate of the multi-instance "declaration also
drives" note in CONFIGURATION.md.

* docs(changelog): reference #7099 for the multi-instance leftovers

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -1265,6 +1265,23 @@ This release closes that milestone with **621 merged pull requests**.
 
 ### Fixed
 
+- **skills:** A skill the agent creates, edits, patches or deletes with the
+  `skill_manage` tool now reaches every Gateway process sharing the home
+  volume, as a change made through the skills API already did. The tool
+  refreshed only its own process's skills prompt cache, so a peer replica kept
+  rendering the previous skill list for that user; it now publishes the same
+  user-scoped shared cache reset marker after the local refresh, off the event
+  loop. The write, its history record, the refresh and the marker settle as
+  one unit, so a run cancelled while the write is in flight no longer leaves
+  this or a peer process serving the stale list. When the marker cannot be
+  written, the tool reports that the change was applied but other processes
+  were not notified, instead of reading as a failed write. ([#7099])
+- **provisioner:** When two provisioner replicas start together and
+  `K8S_NAMESPACE` does not exist yet, the one that loses the create race no
+  longer crashes at startup: its `409 AlreadyExists` now counts as the
+  namespace existing, as the Pod and Service creates already did. Helm
+  deployments, where the provisioner runs inside the namespace it manages,
+  never took this path. ([#7099])
 - **projects:** A shelf upload or save-from-thread into a project the caller
   cannot write no longer leaves a directory behind or answers `500`. Both
   routes staged the body under `users/{user_id}/projects/{project_id}/` before
@@ -1274,7 +1291,6 @@ This release closes that milestone with **621 merged pull requests**.
   now check for an owned, active project first and answer the same `404`
   without writing; the shelf insert still re-checks under the project row lock.
   ([#7040])
-
 - **workflows:** A script workflow waiting on a long child Agent run no longer
   fails with `Worker turn limit` after about 209 seconds regardless of
   `timeout_seconds`. The supervisor re-sent an empty frame to the QuickJS worker
@@ -11860,3 +11876,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#6988]: https://github.com/bytedance/deer-flow/pull/6988
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
 [#7040]: https://github.com/bytedance/deer-flow/pull/7040
+[#7099]: https://github.com/bytedance/deer-flow/pull/7099

CHANGELOG_zh.md

@@ -1015,13 +1015,22 @@
 
 ### 修复
 
+- **技能：** agent 通过 `skill_manage` 工具创建、编辑、修补或删除的技能，现在会像通过
+  技能 API 所做的修改一样，传达到共享同一 home 卷的每个 Gateway 进程。此前该工具只刷新
+  本进程的技能提示缓存，对端副本会继续为该用户渲染旧的技能列表；现在它在本地刷新之后、
+  于事件循环之外发布同一个按用户限定的共享缓存重置标记。写入、历史记录、刷新与标记作为
+  一个整体完成，因此写入途中被取消的运行不再让本进程或对端继续使用过期列表。标记无法写入
+  时，工具会报告修改已生效但未能通知其他进程，而不会让人误以为写入失败。 ([#7099])
+- **provisioner：** 两个 provisioner 副本在 `K8S_NAMESPACE` 尚不存在时同时启动，输掉
+  创建竞争的一方不再在启动时崩溃：它收到的 `409 AlreadyExists` 现在视为命名空间已存在，
+  与 Pod 和 Service 的创建一致。Helm 部署中 provisioner 运行在其管理的命名空间内，从不
+  经过这条路径。 ([#7099])
 - **项目：** 向调用者无权写入的项目上传书架文档或从对话保存文件时，不再遗留目录，
   也不再返回 `500`。这两个路由此前先把请求体暂存到
   `users/{user_id}/projects/{project_id}/` 下才检查项目，因此不存在、属于他人或已归档
   的 id 会留下一个回收站对账永远不会清理的空 `{project_id}/documents/` 目录，而超过单个
   路径分量长度的 id 会因 `ENAMETOOLONG` 失败。现在会先确认项目归调用者所有且处于活跃
   状态，不写任何内容即返回同样的 `404`；书架插入仍会在项目行锁下再次检查。 ([#7040])
-
 - **浏览器：** `browser_screenshot` 不再替换线程 outputs 目录中已存在的文件。此前以
   `report.png` 保存的截图会覆盖用户自己的 `report.png`、更早的截图，或同一秒内生成的
   另一张默认命名截图；现在会像 `web_capture` 一样改用下一个空闲名称（`report-1.png`、
@@ -1030,7 +1039,6 @@
   上传相同，outputs 目录需要支持硬链接；不支持时工具会报错，而不会覆盖已有文件。发布前
   被替换的暂存截图（例如换成指向宿主机文件的符号链接）绝不会被链接到 outputs，工具会
   报错。 ([#6913])
-
 - **llm：** 流式响应已返回 `200` 之后才在流内报告的 provider 错误，现在会像以 HTTP
   状态码返回的同一错误一样重试。为了用 SSE keepalive 维持慢请求而提前返回 `200` 的
   OpenAI 兼容网关，会把之后发生的 504 作为 SSE `error` 事件发出；openai SDK 将其抛出为
@@ -9430,3 +9438,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#6988]: https://github.com/bytedance/deer-flow/pull/6988
 [#6991]: https://github.com/bytedance/deer-flow/pull/6991
 [#7040]: https://github.com/bytedance/deer-flow/pull/7040
+[#7099]: https://github.com/bytedance/deer-flow/pull/7099

backend/docs/CONFIGURATION.md

@@ -764,7 +764,6 @@ Notes:
 - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, the `auth.local.throttle_storage: memory` warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
 - Login lockouts are shared through the database: `auth.local.throttle_storage` (startup-only) defaults to `auto`, which keeps the per-IP failed-login counters for `POST /api/v1/auth/login/local` in the `login_throttle` table whenever `database.backend` is `sqlite` or `postgres`, so every instance enforces one `max_login_attempts` limit per client IP and a lockout on one instance holds on all of them. `memory` keeps the historical per-process counter (N instances give an attacker N × `max_login_attempts` guesses; a declared multi-instance deployment logs a warning), and `db` forces the table (it falls back to memory with a warning when the database backend is `memory`, and refuses to start when the configured database's engine is unavailable; `auto` falls back with a warning in that case). `max_login_attempts` and `lockout_seconds` stay live-read.
 - IM chat-to-thread bindings are shared through the database too: with `database.backend: sqlite` or `postgres` the `ChannelManager` keeps the `channel_name:chat_id[:topic_id]` → thread mapping of unbound IM conversations in the `channel_thread_bindings` table (migration `0038_channel_thread_bindings`), so a conversation created on one instance continues on the same thread when its next message lands on another. On the first start after upgrading, an existing `{base_dir}/channels/store.json` is imported once — only into an empty table, with `INSERT … ON CONFLICT DO NOTHING` so concurrently starting instances cannot duplicate a binding — and renamed `store.json.migrated`; a populated table leaves the file untouched. `memory` keeps the per-process JSON file.
-- The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count.
 - With the default DeerMem backend, the declaration also warns when the derived SQLite retrieval index sits inside the shared `storage_path` (the default `{storage_path}/.retrieval`). Set `memory.backend_config.retrieval_index_path` to an instance-local directory (relative values resolve against `storage_path`): SQLite WAL is unsupported on network filesystems, and the index is rebuilt from the Markdown facts at startup and re-synced per user scope after a peer writes.
 - A declared multi-instance deployment that enables a feature storing credentials (today `channel_connections.enabled: true`) also needs one shared `DEER_FLOW_CREDENTIALS_KEY`; see [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key).
 - IM channels: the gate refuses `channels.leader_lease.enabled: false` while a provider other than GitHub is enabled (under `channels.*`, or connectable from the web UI through `channel_connections.<provider>.enabled`), because every instance would otherwise open the platform connection with the same bot credentials. With `auto` (the default) or `true` the instances share each provider's connection through the `channel_leases` table; see [IM channel leader lease](#im-channel-leader-lease-channelsleader_lease).

backend/packages/harness/deerflow/skills/AGENTS.md

@@ -17,7 +17,7 @@
 - **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 invalidation hook for trusted MinIO/NFS/CSI writes. `SkillStorage` instances do not cache a catalog — `load_skills()` scans on every call — so the route clears all `(app_config, user_id)` entries and the rendered prompt-section LRU, waits up to the shared refresh timeout for the existing off-loop single-flight refresh, then publishes the shared reset marker (below). Each invalidation receives a generation-bound result handle; a successful scan atomically replaces the global enabled-skills cache, while a loader-level failure propagates to the HTTP waiter and preserves the last-known-good global cache. Per-user/config scans capture the refresh version and cannot repopulate shared caches if invalidation occurs while they are loading. A timed-out HTTP wait fails generically while the daemon refresh worker continues. Subsequent runs rescan after a successful reload; active runs keep their existing snapshot. Direct mount writes bypass install/edit validation, SkillScan, and history, so mounted roots are an operator-controlled trust boundary.
-- **Shared cache reset**: the prompt-layer caches in `agents/lead_agent/prompt.py` (global enabled-skills list, per-`(app_config, user_id)` entries, rendered `<skill_system>` LRU) are the only skill state that can diverge between Gateway processes sharing one home volume. Every mutation in `routers/skills.py` (install/upload, edit, delete, rollback, enable/disable, reload) refreshes its own process, then calls `publish_skills_cache_reset(user_id=...)`, which atomically replaces `.<extensions config name>.skills-cache-reset.json` beside the shared config under the extensions-config locks (`deerflow.config.shared_reset_marker`). Every cache lookup polls that marker at most once per second and, on a new signature, calls `_invalidate_enabled_skills_cache()` or, for a marker that names one `user_id` as the only publication since the last poll, `invalidate_user_skill_cache(user_id)`; the publishing process skips its own generation. `/skills/reload` runs refresh + publish through the same `await_drained(_drain_skill_mutation(...))` tail as the mutations, so a cancelled caller cannot skip the marker; it answers `scope: shared_config` when the marker was written and `scope: process` when no extensions config path resolves. Publication failures raise `SkillCacheResetPublishError` (a `RuntimeError`, never an `OSError`), so the handlers' `FileNotFoundError -> 404` / `ValueError -> 400` mappings cannot claim an applied change was not found; every handler reports it as a generic 500, and a 200 never means only one process was refreshed. The marker reaches processes mounting that directory, not replicas with independent filesystems. Tests: `tests/test_skills_cache_shared_reset.py`.
+- **Shared cache reset**: the prompt-layer caches in `agents/lead_agent/prompt.py` (global enabled-skills list, per-`(app_config, user_id)` entries, rendered `<skill_system>` LRU) are the only skill state that can diverge between Gateway processes sharing one home volume. Every mutation in `routers/skills.py` (install/upload, edit, delete, rollback, enable/disable, reload) and the agent `skill_manage` tool refreshes its own process, then calls `publish_skills_cache_reset(user_id=...)`, which atomically replaces `.<extensions config name>.skills-cache-reset.json` beside the shared config under the extensions-config locks (`deerflow.config.shared_reset_marker`). Every cache lookup polls that marker at most once per second and, on a new signature, calls `_invalidate_enabled_skills_cache()` or, for a marker that names one `user_id` as the only publication since the last poll, `invalidate_user_skill_cache(user_id)`; the publishing process skips its own generation. `/skills/reload` runs refresh + publish through the same `await_drained(_drain_skill_mutation(...))` tail as the mutations, so a cancelled caller cannot skip the marker; it answers `scope: shared_config` when the marker was written and `scope: process` when no extensions config path resolves. Publication failures raise `SkillCacheResetPublishError` (a `RuntimeError`, never an `OSError`), so the handlers' `FileNotFoundError -> 404` / `ValueError -> 400` mappings cannot claim an applied change was not found; every handler reports it as a generic 500, and a 200 never means only one process was refreshed. `skill_manage` drains write, history, refresh and publish as one unit too; its publish failure says the change was applied. 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.
 - **Projection manifest failures**: `_write_manifest` keeps UTF-8 JSON, private `mkstemp` permissions and same-directory replacement. On ordinary errors or interruptions, reuse `utils.atomic_write.cleanup_failed_atomic_write`; transfer descriptor ownership only after `fdopen` succeeds. Descriptor close, sibling unlink and warning logs must not mask the original error. Coverage: `tests/test_skill_projection_manifest_cleanup.py`.

backend/packages/harness/deerflow/tools/skill_manage_tool.py

@@ -9,6 +9,7 @@ import shutil
 import tempfile
 import threading
 from collections import deque
+from collections.abc import Awaitable, Callable
 from concurrent.futures import Future
 from pathlib import Path
 from typing import Any, NoReturn
@@ -16,7 +17,7 @@ from weakref import WeakValueDictionary
 
 from langchain.tools import tool
 
-from deerflow.agents.lead_agent.prompt import refresh_user_skills_system_prompt_cache_async
+from deerflow.agents.lead_agent.prompt import SkillCacheResetPublishError, publish_skills_cache_reset, refresh_user_skills_system_prompt_cache_async
 from deerflow.runtime.user_context import resolve_runtime_user_id
 from deerflow.skills.security_scanner import scan_skill_content
 from deerflow.skills.security_static_scanner import (
@@ -30,6 +31,7 @@ from deerflow.skills.storage.skill_storage import SkillStorage
 from deerflow.skills.types import SKILL_MD_FILE
 from deerflow.tools.sync import make_sync_tool_wrapper
 from deerflow.tools.types import Runtime
+from deerflow.utils.file_io import await_drained
 
 logger = logging.getLogger(__name__)
 
@@ -177,6 +179,37 @@ async def _to_thread(func, /, *args, **kwargs):
     return await asyncio.to_thread(func, *args, **kwargs)
 
 
+async def _write_and_record(skill_storage: SkillStorage, name: str, relative_path: str, content: str, record: dict[str, Any]) -> None:
+    await _to_thread(skill_storage.write_custom_skill, name, relative_path, content)
+    await _to_thread(skill_storage.append_history, name, record)
+
+
+async def _settle_skill_mutation(user_id: str, name: str, persist: Callable[[], Awaitable[object]]) -> None:
+    """Apply a skill change and propagate it to every Gateway process sharing the home volume.
+
+    The write, its history record, this process's prompt-cache refresh and the
+    shared skills cache reset marker settle as one drained unit, like the
+    mutation tails of the Gateway skill routes: cancelling ``asyncio.to_thread``
+    abandons the await, not the worker, so a cancelled run would otherwise
+    unwind with the write landed and this process and its peers still serving
+    the previous skill list. The marker write is filesystem IO and runs off the
+    event loop.
+    """
+
+    async def _tail() -> None:
+        await persist()
+        await refresh_user_skills_system_prompt_cache_async(user_id)
+        try:
+            await _to_thread(publish_skills_cache_reset, user_id=user_id)
+        except SkillCacheResetPublishError as exc:
+            # Logged here because a cancelled run never sees the raise below.
+            logger.error("Custom skill %r changed, but the shared skills cache reset could not be published", name, exc_info=True)
+            # The change is durable; say so, or the model reads a failed write and retries it.
+            raise SkillCacheResetPublishError(f"The change to custom skill '{name}' was applied, but other Gateway processes were not notified and may keep serving the previous skill list: {exc}") from exc
+
+    await await_drained(_tail())
+
+
 async def _skill_manage_impl(
     runtime: Runtime,
     action: str,
@@ -216,13 +249,8 @@ async def _skill_manage_impl(
             static_findings = await _scan_static_candidate_or_raise(name, {SKILL_MD_FILE: content})
             scan = await _scan_or_raise(content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
             scan["static_findings"] = static_findings
-            await _to_thread(skill_storage.write_custom_skill, name, SKILL_MD_FILE, content)
-            await _to_thread(
-                skill_storage.append_history,
-                name,
-                _history_record(action="create", file_path=SKILL_MD_FILE, prev_content=None, new_content=content, thread_id=thread_id, scanner=scan),
-            )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+            record = _history_record(action="create", file_path=SKILL_MD_FILE, prev_content=None, new_content=content, thread_id=thread_id, scanner=scan)
+            await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, content, record))
             return f"Created custom skill '{name}'."
         if action == "edit":
             await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
@@ -233,13 +261,8 @@ async def _skill_manage_impl(
             scan = await _scan_or_raise(content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
             scan["static_findings"] = static_findings
             prev_content = await _to_thread(skill_storage.read_custom_skill, name)
-            await _to_thread(skill_storage.write_custom_skill, name, SKILL_MD_FILE, content)
-            await _to_thread(
-                skill_storage.append_history,
-                name,
-                _history_record(action="edit", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan),
-            )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+            record = _history_record(action="edit", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan)
+            await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, content, record))
             return f"Updated custom skill '{name}'."
 
         if action == "patch":
@@ -258,29 +281,20 @@ async def _skill_manage_impl(
             static_findings = await _scan_static_candidate_or_raise(name, {SKILL_MD_FILE: new_content})
             scan = await _scan_or_raise(new_content, executable=False, location=f"{name}/{SKILL_MD_FILE}", static_findings=static_findings)
             scan["static_findings"] = static_findings
-            await _to_thread(skill_storage.write_custom_skill, name, SKILL_MD_FILE, new_content)
-            await _to_thread(
-                skill_storage.append_history,
-                name,
-                _history_record(action="patch", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=new_content, thread_id=thread_id, scanner=scan),
-            )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+            record = _history_record(action="patch", file_path=SKILL_MD_FILE, prev_content=prev_content, new_content=new_content, thread_id=thread_id, scanner=scan)
+            await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, SKILL_MD_FILE, new_content, record))
             return f"Patched custom skill '{name}' ({replacement_count} replacement(s) applied, {occurrences} match(es) found)."
 
         if action == "delete":
-            await _to_thread(
-                skill_storage.delete_custom_skill,
-                name,
-                history_meta=_history_record(
-                    action="delete",
-                    file_path=SKILL_MD_FILE,
-                    prev_content=None,
-                    new_content=None,
-                    thread_id=thread_id,
-                    scanner={"decision": "allow", "reason": "Deletion requested."},
-                ),
+            record = _history_record(
+                action="delete",
+                file_path=SKILL_MD_FILE,
+                prev_content=None,
+                new_content=None,
+                thread_id=thread_id,
+                scanner={"decision": "allow", "reason": "Deletion requested."},
             )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+            await _settle_skill_mutation(user_id, name, lambda: _to_thread(skill_storage.delete_custom_skill, name, history_meta=record))
             return f"Deleted custom skill '{name}'."
 
         if action == "write_file":
@@ -297,26 +311,24 @@ async def _skill_manage_impl(
             static_findings = await _scan_static_candidate_or_raise(name, {path: content}, skill_storage)
             scan = await _scan_or_raise(content, executable=executable, location=f"{name}/{path}", static_findings=static_findings)
             scan["static_findings"] = static_findings
-            await _to_thread(skill_storage.write_custom_skill, name, path, content)
-            await _to_thread(
-                skill_storage.append_history,
-                name,
-                _history_record(action="write_file", file_path=path, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan),
-            )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+            record = _history_record(action="write_file", file_path=path, prev_content=prev_content, new_content=content, thread_id=thread_id, scanner=scan)
+            await _settle_skill_mutation(user_id, name, lambda: _write_and_record(skill_storage, name, path, content, record))
             return f"Wrote '{path}' for custom skill '{name}'."
 
         if action == "remove_file":
             await _to_thread(skill_storage.ensure_custom_skill_is_editable, name)
             if path is None:
                 raise ValueError("path is required for remove_file.")
-            prev_content = await _to_thread(skill_storage.remove_custom_skill_file, name, path)
-            await _to_thread(
-                skill_storage.append_history,
-                name,
-                _history_record(action="remove_file", file_path=path, prev_content=prev_content, new_content=None, thread_id=thread_id, scanner={"decision": "allow", "reason": "Deletion requested."}),
-            )
-            await refresh_user_skills_system_prompt_cache_async(user_id)
+
+            async def _remove_and_record() -> None:
+                prev_content = await _to_thread(skill_storage.remove_custom_skill_file, name, path)
+                await _to_thread(
+                    skill_storage.append_history,
+                    name,
+                    _history_record(action="remove_file", file_path=path, prev_content=prev_content, new_content=None, thread_id=thread_id, scanner={"decision": "allow", "reason": "Deletion requested."}),
+                )
+
+            await _settle_skill_mutation(user_id, name, _remove_and_record)
             return f"Removed '{path}' from custom skill '{name}'."
 
         if await _to_thread(skill_storage.public_skill_exists, name):

backend/tests/blocking_io/test_skill_manage_tool.py

@@ -8,20 +8,26 @@ the Gateway's custom-skill edit route applies.
 
 Only the LLM security scan is stubbed; the app config is pinned so the lookup
 needs no ``config.yaml`` on disk, and the edit runs against the real per-user
-storage under ``DEER_FLOW_HOME``.
+storage under ``DEER_FLOW_HOME``. With a shared extensions config resolved, the
+edit also publishes the cross-replica skills cache reset marker (a lock file
+and an atomic replace), which must stay off the loop as well.
 """
 
 from __future__ import annotations
 
 import asyncio
+import json
 from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
 
 import deerflow.tools.skill_manage_tool as skill_manage_module
+from deerflow.agents.lead_agent import prompt as prompt_module
 from deerflow.config.app_config import AppConfig
+from deerflow.config.extensions_config import reset_extensions_config
 from deerflow.config.paths import get_paths
+from deerflow.config.shared_reset_marker import SharedResetMarkerTracker
 
 pytestmark = pytest.mark.asyncio
 
@@ -62,3 +68,29 @@ async def test_skill_manage_edit_does_not_block_event_loop(monkeypatch) -> None:
 
     assert result == f"Updated custom skill '{_SKILL_NAME}'."
     assert await asyncio.to_thread(skill_file.read_text, encoding="utf-8") == edited
+
+
+async def test_skill_manage_edit_publishes_the_shared_marker_off_the_loop(monkeypatch, tmp_path: Path) -> None:
+    await asyncio.to_thread(_install_skill)
+    config_path = tmp_path / "extensions_config.json"
+    await asyncio.to_thread(config_path.write_text, json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
+    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
+    reset_extensions_config()
+    monkeypatch.setattr(prompt_module, "_skills_cache_reset_tracker", SharedResetMarkerTracker(prompt_module.SKILLS_CACHE_RESET_MARKER))
+    config = AppConfig.model_validate({"sandbox": {"use": "test"}})
+    monkeypatch.setattr("deerflow.config.get_app_config", lambda: config)
+
+    async def _allow_scan(*args, **kwargs):
+        return SimpleNamespace(decision="allow", reason="anchor stub")
+
+    monkeypatch.setattr(skill_manage_module, "scan_skill_content", _allow_scan)
+
+    try:
+        result = await skill_manage_module.skill_manage_tool.coroutine(_RUNTIME, "edit", _SKILL_NAME, _SKILL_MD + "\nEdited.\n")
+    finally:
+        reset_extensions_config()
+
+    assert result == f"Updated custom skill '{_SKILL_NAME}'."
+    marker_path = await asyncio.to_thread(prompt_module.SKILLS_CACHE_RESET_MARKER.path_for, config_path)
+    payload = json.loads(await asyncio.to_thread(marker_path.read_text, encoding="utf-8"))
+    assert payload["user_id"] == "default"

backend/tests/test_provisioner_namespace.py

@@ -0,0 +1,74 @@
+"""Startup namespace creation in the provisioner must tolerate a concurrent replica.
+
+``_ensure_namespace()`` is the provisioner's one startup write. Two replicas
+that start together against a missing ``K8S_NAMESPACE`` both read 404 and both
+create; the loser's ``409 AlreadyExists`` used to escape the FastAPI lifespan
+and crash that process. A 409 on create means the namespace now exists, the
+same outcome the Pod and Service creates already accept.
+"""
+
+from __future__ import annotations
+
+import logging
+from unittest.mock import MagicMock
+
+import pytest
+
+
+def _core_v1(provisioner_module, *, read_status: int | None = 404, create_status: int | None = None) -> MagicMock:
+    core_v1 = MagicMock()
+    if read_status is not None:
+        core_v1.read_namespace.side_effect = provisioner_module.ApiException(status=read_status)
+    if create_status is not None:
+        core_v1.create_namespace.side_effect = provisioner_module.ApiException(status=create_status, reason="AlreadyExists" if create_status == 409 else "Forbidden")
+    return core_v1
+
+
+def test_ensure_namespace_treats_a_concurrent_create_as_already_existing(monkeypatch, provisioner_module, caplog):
+    core_v1 = _core_v1(provisioner_module, create_status=409)
+    monkeypatch.setattr(provisioner_module, "core_v1", core_v1)
+    monkeypatch.setattr(provisioner_module, "K8S_NAMESPACE", "deer-flow-race")
+
+    with caplog.at_level(logging.INFO, logger=provisioner_module.logger.name):
+        provisioner_module._ensure_namespace()
+
+    core_v1.create_namespace.assert_called_once()
+    assert core_v1.create_namespace.call_args.args[0].metadata.name == "deer-flow-race"
+    messages = [record.getMessage() for record in caplog.records if record.name == provisioner_module.logger.name]
+    assert "Namespace 'deer-flow-race' was created concurrently; using it" in messages
+    assert "Created namespace 'deer-flow-race'" not in messages
+
+
+def test_ensure_namespace_creates_a_missing_namespace(monkeypatch, provisioner_module):
+    core_v1 = _core_v1(provisioner_module)
+    monkeypatch.setattr(provisioner_module, "core_v1", core_v1)
+
+    provisioner_module._ensure_namespace()
+
+    core_v1.create_namespace.assert_called_once()
+
+
+def test_ensure_namespace_skips_create_when_the_namespace_exists(monkeypatch, provisioner_module):
+    core_v1 = _core_v1(provisioner_module, read_status=None)
+    monkeypatch.setattr(provisioner_module, "core_v1", core_v1)
+
+    provisioner_module._ensure_namespace()
+
+    core_v1.create_namespace.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "read_status, create_status",
+    [
+        pytest.param(404, 403, id="create-forbidden"),
+        pytest.param(403, None, id="read-forbidden"),
+    ],
+)
+def test_ensure_namespace_still_fails_on_other_api_errors(monkeypatch, provisioner_module, read_status, create_status):
+    core_v1 = _core_v1(provisioner_module, read_status=read_status, create_status=create_status)
+    monkeypatch.setattr(provisioner_module, "core_v1", core_v1)
+
+    with pytest.raises(provisioner_module.ApiException) as error:
+        provisioner_module._ensure_namespace()
+
+    assert error.value.status == 403

backend/tests/test_skill_manage_tool.py

@@ -14,6 +14,18 @@ from deerflow.skills.security_static_scanner import StaticScannerError
 skill_manage_module = importlib.import_module("deerflow.tools.skill_manage_tool")
 
 
+@pytest.fixture(autouse=True)
+def _record_skills_cache_reset_publications(monkeypatch):
+    """Keep the cross-replica marker out of whatever extensions config this checkout resolves.
+
+    Every mutation publishes it; tests/test_skills_cache_shared_reset.py covers
+    the real publication against a temporary shared config.
+    """
+    published: list[str | None] = []
+    monkeypatch.setattr(skill_manage_module, "publish_skills_cache_reset", lambda *, user_id=None: published.append(user_id))
+    return published
+
+
 def _skill_content(name: str, description: str = "Demo skill") -> str:
     return f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n"
 

backend/tests/test_skills_cache_shared_reset.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 
 import asyncio
 import json
+import threading
 import zipfile
 from pathlib import Path
 from types import SimpleNamespace
@@ -36,10 +37,11 @@ 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.config.shared_reset_marker import SharedResetChange, 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
+from deerflow.tools import skill_manage_tool as skill_manage_module
 
 _PROCESS_MESSAGE = "Skill caches invalidated; subsequent runs in this Gateway process will rescan the latest skills."
 
@@ -702,3 +704,178 @@ def test_publish_skills_cache_reset_wraps_filesystem_errors(shared_config, monke
     assert not isinstance(excinfo.value, OSError)
     assert isinstance(excinfo.value.__cause__, FileNotFoundError)
     assert missing not in str(excinfo.value)
+
+
+# --------------------------------------------------------------------------- #
+# Agent tool: skill_manage publishes the same marker as the routes
+# --------------------------------------------------------------------------- #
+
+_AGENT_USER = "alice"
+_AGENT_SKILL = "agent-skill"
+
+
+@pytest.fixture()
+def agent_skill_tool(shared_config, monkeypatch: pytest.MonkeyPatch, tmp_path: Path):
+    """The agent's ``skill_manage`` tool on per-user storage under ``tmp_path``.
+
+    Only the LLM scan is stubbed: the local refresh and the marker write are
+    the real ones, wrapped so ``events`` records their order and
+    ``publish_threads`` the thread each marker write ran on.
+    """
+    skills_root = tmp_path / "skills"
+    skill_dir = _user_custom_dir(tmp_path, _AGENT_USER) / _AGENT_SKILL
+    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.skills.security_scanner.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(skill_manage_module, "scan_skill_content", _allow_scan)
+    events: list[tuple[str, str | None]] = []
+    publish_threads: list[int] = []
+    real_refresh = skill_manage_module.refresh_user_skills_system_prompt_cache_async
+    real_publish = shared_config.marker.publish
+
+    async def _refresh(user_id: str) -> None:
+        events.append(("refresh", user_id))
+        await real_refresh(user_id)
+
+    def _publish(config_path, *, user_id=None):
+        events.append(("publish", user_id))
+        publish_threads.append(threading.get_ident())
+        return real_publish(config_path, user_id=user_id)
+
+    monkeypatch.setattr(skill_manage_module, "refresh_user_skills_system_prompt_cache_async", _refresh)
+    monkeypatch.setattr(shared_config.marker, "publish", _publish)
+
+    def _install() -> None:
+        """Put an editable skill with one support file on disk, bypassing the tool."""
+        (skill_dir / "references").mkdir(parents=True)
+        (skill_dir / "SKILL.md").write_text(_skill_content(_AGENT_SKILL), encoding="utf-8")
+        (skill_dir / "references" / "guide.md").write_text("guide", encoding="utf-8")
+
+    def _call(**kwargs):
+        runtime = SimpleNamespace(
+            context={"thread_id": "thread-1", "user_id": _AGENT_USER},
+            config={"configurable": {"thread_id": "thread-1", "user_id": _AGENT_USER}},
+        )
+        return skill_manage_module.skill_manage_tool.coroutine(runtime=runtime, name=_AGENT_SKILL, **kwargs)
+
+    return SimpleNamespace(call=_call, install=_install, events=events, publish_threads=publish_threads, skill_file=skill_dir / "SKILL.md")
+
+
+_AGENT_MUTATIONS = [
+    pytest.param({"action": "create", "content": _skill_content(_AGENT_SKILL)}, False, id="create"),
+    pytest.param({"action": "edit", "content": _skill_content(_AGENT_SKILL, "Edited skill")}, True, id="edit"),
+    pytest.param({"action": "patch", "find": "Demo skill", "replace": "Patched skill"}, True, id="patch"),
+    pytest.param({"action": "write_file", "path": "references/guide.md", "content": "new guide"}, True, id="write_file"),
+    pytest.param({"action": "remove_file", "path": "references/guide.md"}, True, id="remove_file"),
+    pytest.param({"action": "delete"}, True, id="delete"),
+]
+
+
+@pytest.mark.parametrize("arguments, existing", _AGENT_MUTATIONS)
+def test_skill_manage_tool_mutation_publishes_user_scoped_marker_after_local_refresh(shared_config, agent_skill_tool, arguments, existing) -> None:
+    """An agent-driven skill change reaches peers the way a route-driven one does.
+
+    The tool used to refresh only its own process, so a replica sharing the
+    home volume kept rendering the previous skill list for that user.
+    """
+    if existing:
+        agent_skill_tool.install()
+    peer_clock = _Clock()
+    peer = SharedResetMarkerTracker(shared_config.marker, clock=peer_clock)
+    assert peer.poll() is None
+
+    result = asyncio.run(agent_skill_tool.call(**arguments))
+
+    assert _AGENT_SKILL in result
+    assert agent_skill_tool.events == [("refresh", _AGENT_USER), ("publish", _AGENT_USER)]
+    assert agent_skill_tool.publish_threads and threading.get_ident() not in agent_skill_tool.publish_threads, "the marker write is filesystem IO and must run off the event loop"
+    peer_clock.advance(1.0)
+    assert peer.poll() == SharedResetChange(user_ids=frozenset({_AGENT_USER}), generation=_marker_payload(shared_config)["generation"])
+
+
+def test_skill_manage_tool_rejected_mutation_publishes_nothing(shared_config, agent_skill_tool) -> None:
+    agent_skill_tool.install()
+
+    with pytest.raises(ValueError, match="already exists"):
+        asyncio.run(agent_skill_tool.call(action="create", content=_skill_content(_AGENT_SKILL)))
+
+    assert agent_skill_tool.events == []
+    assert not _marker_path(shared_config).exists()
+
+
+def test_skill_manage_tool_marker_publication_failure_reports_the_applied_change(shared_config, agent_skill_tool, monkeypatch, caplog) -> None:
+    """The skill is on disk; the model must learn that only peer notification failed.
+
+    The error stays a ``SkillCacheResetPublishError``, never an ``OSError`` or
+    ``ValueError`` (which the tool raises for rejected input that changed
+    nothing), and names no server path. It is also logged inside the drained
+    tail, where a cancelled run cannot lose it.
+    """
+    missing = _marker_write_raises_file_not_found(monkeypatch, shared_config)
+
+    with caplog.at_level("ERROR", logger=skill_manage_module.logger.name), pytest.raises(prompt_module.SkillCacheResetPublishError) as excinfo:
+        asyncio.run(agent_skill_tool.call(action="create", content=_skill_content(_AGENT_SKILL)))
+
+    assert not isinstance(excinfo.value, (OSError, ValueError))
+    assert isinstance(excinfo.value.__cause__, prompt_module.SkillCacheResetPublishError)
+    message = str(excinfo.value)
+    assert f"'{_AGENT_SKILL}'" in message and "was applied" in message
+    assert missing not in message
+    assert agent_skill_tool.skill_file.read_text(encoding="utf-8") == _skill_content(_AGENT_SKILL)
+    assert agent_skill_tool.events == [("refresh", _AGENT_USER)]
+    assert [record.message for record in caplog.records if record.name == skill_manage_module.logger.name] == [f"Custom skill '{_AGENT_SKILL}' changed, but the shared skills cache reset could not be published"]
+
+
+@pytest.mark.asyncio
+async def test_skill_manage_tool_drains_write_refresh_and_publish_across_cancellation(shared_config, agent_skill_tool, monkeypatch) -> None:
+    """A run cancelled while the skill write is in flight must not skip the marker.
+
+    Cancelling ``asyncio.to_thread`` abandons the await, not the worker, so the
+    write lands anyway; unwinding before the refresh and the publication would
+    leave this process and every peer serving the previous skill list. Mirrors
+    the drained mutation tails of the skill routes.
+    """
+    await asyncio.to_thread(agent_skill_tool.install)
+    entered = threading.Event()
+    release = threading.Event()
+    to_thread = skill_manage_module._to_thread
+
+    async def _parking_to_thread(func, /, *args, **kwargs):
+        if getattr(func, "__name__", None) != "write_custom_skill":
+            return await to_thread(func, *args, **kwargs)
+
+        def _parked():
+            entered.set()
+            assert release.wait(5), "test never released the parked write"
+            return func(*args, **kwargs)
+
+        return await to_thread(_parked)
+
+    monkeypatch.setattr(skill_manage_module, "_to_thread", _parking_to_thread)
+    edited = _skill_content(_AGENT_SKILL, "Edited skill")
+    task = asyncio.create_task(agent_skill_tool.call(action="edit", content=edited))
+    try:
+        assert await asyncio.to_thread(entered.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 agent_skill_tool.events == [("refresh", _AGENT_USER), ("publish", _AGENT_USER)]
+    assert _marker_payload(shared_config)["user_id"] == _AGENT_USER
+    assert await asyncio.to_thread(agent_skill_tool.skill_file.read_text, encoding="utf-8") == edited

docker/provisioner/README.md

@@ -425,8 +425,8 @@ chart's `provisioner.replicas`; compose runs a single container):
 
 The one startup write is `_ensure_namespace()`: when `K8S_NAMESPACE` does not
 exist yet, two replicas starting together can both see `404` and race to create
-it; the loser's `409` is not caught, so that process exits and is restarted into
-the now-existing namespace. Under the Helm chart the provisioner runs inside the
+it; the loser's `409 AlreadyExists` is treated as "the namespace exists", so
+both replicas start. Under the Helm chart the provisioner runs inside the
 namespace it manages, so this path never runs there.
 
 ## Security Considerations

docker/provisioner/app.py

@@ -453,7 +453,15 @@ def _ensure_namespace() -> None:
                     },
                 )
             )
-            core_v1.create_namespace(ns)
+            try:
+                core_v1.create_namespace(ns)
+            except ApiException as create_exc:
+                # 409 = AlreadyExists: a replica starting alongside this one
+                # created it between our read and our create.
+                if create_exc.status != 409:
+                    raise
+                logger.info(f"Namespace '{K8S_NAMESPACE}' was created concurrently; using it")
+                return
             logger.info(f"Created namespace '{K8S_NAMESPACE}'")
         else:
             raise