fix(skills): record the content actually overwritten in custom-skill history (#6153)

* fix(skills): record the content actually overwritten in custom-skill history

Scope:
- backend/app/gateway/routers/skills.py: move the prev_content read of
  update_custom_skill and the current-content read of rollback_custom_skill
  into their drained persist tails (read -> write -> append_history in one
  worker thread, no awaits in between). The blocked-rollback branch now
  builds its informational entry separately. No behavior contract changes:
  status codes, cancellation draining, and prompt-cache refresh order are
  untouched.
- backend/tests/test_skills_custom_router.py: two regression tests that
  inject a concurrent write during the scan phase and assert the history
  entry's prev_content equals the revision actually overwritten on disk.

Why: prev_content was read before the security scans, so a concurrent
revision landing between that read and the persisted write was overwritten
on disk but absent from the history chain, making it unreachable by a later
rollback and desynchronizing the audit record.

Tests: uv run pytest tests/test_skills_custom_router.py --basetemp=... ->
36/36 passed (34 existing + 2 new); ruff check and ruff format --check clean
on both files. Python 3.12.13, Windows 11.

* fix(skills): serialize custom-skill mutations and pin the concurrency regression to the real window

- Add a module-level mutation lock shared by the edit, delete, and rollback
  paths, covering the full read -> write -> history-append sequence so a
  concurrent PUT cannot land between a request's predecessor read and its
  write (previously the projection lock started inside write_custom_skill,
  leaving that window open).
- Replace the two regression tests: they now pause the first mutation inside
  its storage write after the predecessor read (deterministic gate), complete
  a second real HTTP PUT inside that window, and assert the history chain
  stays replayable. Verified both tests fail against the base implementation
  (2 failed) and pass with the lock (2 passed); full file 36 passed,
  ruff check + format clean.

* fix(skills): scope the custom-skill mutation lock to the owning user

The lock serializing the read -> write -> append sequence was one process-wide
lock, and that sequence wraps write_custom_skill's entire user projection
rebuild. A user waiting on their own projection lock, or copying a large or
slow linked package, therefore stalled every other user's edit, rollback, and
delete of unrelated custom skills, even though their skill trees, histories,
and projection roots are disjoint. The base storage only serialized those
mutations per user projection.

Key the lock on the canonical owning-user storage root instead, so one user's
mutations remain a single critical section — including across separately
constructed storage objects, which a hot-reloaded config or a later request
creates — while two users never share it.

The read -> write -> append sequence still cannot interleave within a user, and
the lock is still acquired and released inside the worker thread's own with
block, so caller cancellation cannot strand it.

* fix(skills): serialize custom history across gateway processes

---------

Co-authored-by: sjr666666 <sjr666666@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/gateway/routers/AGENTS.md

@@ -11,3 +11,11 @@ credentials. Import may rename locally but must remain owner-scoped and
 create-only: collisions return 409, never overwrite or upsert. Ordinary creation
 and import share `_persist_new_agent`, which uses `run_drained_write` so a
 cancelled request still drains the owned persistence worker.
+
+Custom-skill edit, delete and rollback hold the owning root's process mutex and
+`.custom.mutation.lock` sidecar across predecessor read, storage mutation and
+history append. Peer Gateway processes sharing that root must use this same
+critical section to preserve history ordering. Keep the sidecar distinct from
+the nested projection lock, retain it across requests, and acquire/release both
+off-loop inside the drained worker. Unrelated users retain independent locks;
+external filesystem writers do not participate in this history guarantee.

backend/app/gateway/routers/skills.py

@@ -1,7 +1,9 @@
 import asyncio
 import logging
 import tempfile
-from collections.abc import AsyncGenerator, Callable, Coroutine
+import threading
+from collections.abc import AsyncGenerator, Callable, Coroutine, Iterator
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any, BinaryIO, Literal
 
@@ -55,6 +57,12 @@ from deerflow.utils.thread_id import ThreadId
 
 logger = logging.getLogger(__name__)
 
+try:
+    import fcntl
+except ImportError:  # pragma: no cover - Windows
+    fcntl = None  # type: ignore[assignment]
+    import msvcrt
+
 router = APIRouter(prefix="/api", tags=["skills"])
 
 _ADMIN_REQUIRED_DETAIL = "Admin privileges required to manage skills."
@@ -152,6 +160,86 @@ class SkillRollbackRequest(BaseModel):
     history_index: int = Field(default=-1, description="History entry index to restore from, defaulting to the latest change.")
 
 
+# Custom-skill mutations (edit, delete, rollback) each span three steps —
+# read the content being replaced, write the new content, append the history
+# entry. Those steps must not interleave between two requests: a request that
+# reads its predecessor, loses the CPU, and writes later silently overwrites
+# the winner's revision while recording the stale predecessor in history,
+# making the overwritten revision unreachable by rollback. Running each
+# mutation's steps inside one worker-thread function does not make them a
+# critical section — two requests can both be inside their own workers — so
+# the cooperating mutation paths share a lock around the whole sequence.
+#
+# The lock is scoped to the owning user's storage: the
+# critical section includes write_custom_skill's projection rebuild, which
+# copies that user's whole skill tree (including a large linked package) and
+# waits on the user's projection lock. A process-wide lock would therefore
+# queue every other user's edit, rollback, and delete behind one user's slow
+# filesystem work, even though their skill trees, histories, and projection
+# roots are disjoint. Keying on the resolved root — not the user_id string,
+# and not one lock per storage instance — keeps a single lock per user when a
+# hot-reloaded config or a later request builds a fresh storage object.
+_custom_skill_mutation_locks: dict[str, threading.Lock] = {}
+_custom_skill_mutation_locks_guard = threading.Lock()
+
+
+def _custom_skill_mutation_lock(storage: SkillStorage) -> threading.Lock:
+    """Return the process-local part of the owning user's mutation lock."""
+    scope = _custom_skill_mutation_scope(storage)
+    with _custom_skill_mutation_locks_guard:
+        return _custom_skill_mutation_locks.setdefault(scope, threading.Lock())
+
+
+def _custom_skill_mutation_scope(storage: SkillStorage) -> str:
+    """The storage scope whose custom-skill mutations must serialize together.
+
+    User-scoped storage reports the canonical root its mutations touch, so two
+    users never share a lock while two storage objects for one user always do.
+    Legacy local storage uses its global custom root. Non-filesystem test
+    doubles have one process-local scope.
+    """
+    get_user_custom_root = getattr(storage, "get_user_custom_root", None)
+    if callable(get_user_custom_root):
+        return str(Path(get_user_custom_root()).resolve())
+    get_skills_root_path = getattr(storage, "get_skills_root_path", None)
+    if callable(get_skills_root_path):
+        return str((Path(get_skills_root_path()) / SkillCategory.CUSTOM.value).resolve())
+    return "global"
+
+
+@contextmanager
+def _custom_skill_mutation(storage: SkillStorage) -> Iterator[None]:
+    """Hold read → write → history across threads and peer Gateway processes.
+
+    Use a stable sidecar outside the skill tree, distinct from projection's
+    lock: write/delete acquire the projection lock while this one is held.
+    Acquiring that same file lock twice would deadlock on POSIX. Both locks
+    are acquired and released in the drained mutation worker, off the loop.
+    """
+    with _custom_skill_mutation_lock(storage):
+        scope = _custom_skill_mutation_scope(storage)
+        if scope == "global":
+            yield
+            return
+        root = Path(scope)
+        lock_path = root.parent / f".{root.name}.mutation.lock"
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        with lock_path.open("a", encoding="utf-8") as lock_file:
+            if fcntl is not None:
+                fcntl.flock(lock_file, fcntl.LOCK_EX)
+            else:  # pragma: no cover - Windows
+                lock_file.seek(0)
+                msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1)
+            try:
+                yield
+            finally:
+                if fcntl is not None:
+                    fcntl.flock(lock_file, fcntl.LOCK_UN)
+                else:  # pragma: no cover - Windows
+                    lock_file.seek(0)
+                    msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
+
+
 def _skill_to_response(skill: Skill) -> SkillResponse:
     """Convert a Skill object to a SkillResponse."""
     return SkillResponse(
@@ -658,26 +746,35 @@ async def update_custom_skill(skill_name: str, body: CustomSkillUpdateRequest, r
         scan = await scan_skill_content(body.content, executable=False, location=f"{skill_name}/{SKILL_MD_FILE}", app_config=config, static_findings=static_findings)
         if scan.decision == "block":
             raise HTTPException(status_code=400, detail=f"Security scan blocked the edit: {scan.reason}")
-        prev_content = await asyncio.to_thread(storage.read_custom_skill, skill_name)
 
         async def _persist_edit() -> None:
-            # The write and its history entry must settle together, and the
-            # prompt cache must not stay stale behind them: a cancelled caller
-            # drains the whole mutation tail instead of cutting it mid-sequence.
-            await asyncio.to_thread(storage.write_custom_skill, skill_name, SKILL_MD_FILE, body.content)
-            await asyncio.to_thread(
-                storage.append_history,
-                skill_name,
-                {
-                    "action": "human_edit",
-                    "author": "human",
-                    "thread_id": None,
-                    "file_path": SKILL_MD_FILE,
-                    "prev_content": prev_content,
-                    "new_content": body.content,
-                    "scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
-                },
-            )
+            def _write_and_record() -> None:
+                # The write and its history entry must settle together, and the
+                # prompt cache must not stay stale behind them: a cancelled caller
+                # drains the whole mutation tail instead of cutting it mid-sequence.
+                # The read → write → append sequence runs under the owning
+                # user's mutation lock so a concurrent edit cannot land between
+                # the predecessor read and the write: prev_content must record
+                # what this mutation actually overwrites, or the overwritten
+                # revision silently drops out of the history chain and becomes
+                # unreachable by rollback.
+                with _custom_skill_mutation(storage):
+                    prev_content = storage.read_custom_skill(skill_name)
+                    storage.write_custom_skill(skill_name, SKILL_MD_FILE, body.content)
+                    storage.append_history(
+                        skill_name,
+                        {
+                            "action": "human_edit",
+                            "author": "human",
+                            "thread_id": None,
+                            "file_path": SKILL_MD_FILE,
+                            "prev_content": prev_content,
+                            "new_content": body.content,
+                            "scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
+                        },
+                    )
+
+            await asyncio.to_thread(_write_and_record)
             user_id = get_effective_user_id()
             await refresh_user_skills_system_prompt_cache_async(user_id)
             await _publish_skills_cache_reset(user_id)
@@ -707,20 +804,25 @@ async def delete_custom_skill(skill_name: str, request: Request, config: AppConf
         async def _persist_delete() -> None:
             # Same cancellation contract as the edit and rollback tails: the
             # deletion and its history record settle before a cancelled caller
-            # unwinds, and the prompt cache reflects the removal.
-            await asyncio.to_thread(
-                storage.delete_custom_skill,
-                skill_name,
-                history_meta={
-                    "action": "human_delete",
-                    "author": "human",
-                    "thread_id": None,
-                    "file_path": SKILL_MD_FILE,
-                    "prev_content": None,
-                    "new_content": None,
-                    "scanner": {"decision": "allow", "reason": "Deletion requested."},
-                },
-            )
+            # unwinds, and the prompt cache reflects the removal. The deletion
+            # joins the owning user's mutation lock so it cannot interleave with
+            # a concurrent edit or rollback's read → write → append sequence.
+            def _delete_and_record() -> None:
+                with _custom_skill_mutation(storage):
+                    storage.delete_custom_skill(
+                        skill_name,
+                        history_meta={
+                            "action": "human_delete",
+                            "author": "human",
+                            "thread_id": None,
+                            "file_path": SKILL_MD_FILE,
+                            "prev_content": None,
+                            "new_content": None,
+                            "scanner": {"decision": "allow", "reason": "Deletion requested."},
+                        },
+                    )
+
+            await asyncio.to_thread(_delete_and_record)
             user_id = get_effective_user_id()
             await refresh_user_skills_system_prompt_cache_async(user_id)
             await _publish_skills_cache_reset(user_id)
@@ -795,27 +897,49 @@ async def rollback_custom_skill(skill_name: str, body: SkillRollbackRequest, req
             skill_file = storage.get_custom_skill_file(skill_name)
             return skill_file.read_text(encoding="utf-8") if skill_file.exists() else None
 
-        current_content = await asyncio.to_thread(_read_current_content)
-        history_entry = {
-            "action": "rollback",
-            "author": "human",
-            "thread_id": None,
-            "file_path": SKILL_MD_FILE,
-            "prev_content": current_content,
-            "new_content": target_content,
-            "rollback_from_ts": record.get("ts"),
-            "scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
-        }
         if scan.decision == "block":
-            await asyncio.to_thread(storage.append_history, skill_name, history_entry)
+            # The blocked path overwrites nothing, so the entry is purely
+            # informational and an early read is safe here.
+            blocked_entry = {
+                "action": "rollback",
+                "author": "human",
+                "thread_id": None,
+                "file_path": SKILL_MD_FILE,
+                "prev_content": await asyncio.to_thread(_read_current_content),
+                "new_content": target_content,
+                "rollback_from_ts": record.get("ts"),
+                "scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
+            }
+            await asyncio.to_thread(storage.append_history, skill_name, blocked_entry)
             raise HTTPException(status_code=400, detail=f"Rollback blocked by security scanner: {scan.reason}")
 
         async def _persist_rollback() -> None:
-            # The restore write and its history entry must settle together, and
-            # the prompt cache must reflect the restored content, before a
-            # 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)
+            def _restore_and_record() -> None:
+                # The restore write and its history entry must settle together, and
+                # the prompt cache must reflect the restored content, before a
+                # cancelled caller unwinds. The read → write → append sequence runs
+                # under the owning user's mutation lock (same contract as the edit
+                # path): the recorded prev_content must be what this rollback
+                # actually overwrites, which requires that no concurrent mutation
+                # can land between the predecessor read and the write.
+                with _custom_skill_mutation(storage):
+                    current_content = _read_current_content()
+                    storage.write_custom_skill(skill_name, SKILL_MD_FILE, target_content)
+                    storage.append_history(
+                        skill_name,
+                        {
+                            "action": "rollback",
+                            "author": "human",
+                            "thread_id": None,
+                            "file_path": SKILL_MD_FILE,
+                            "prev_content": current_content,
+                            "new_content": target_content,
+                            "rollback_from_ts": record.get("ts"),
+                            "scanner": {"decision": scan.decision, "reason": scan.reason, "static_findings": static_findings},
+                        },
+                    )
+
+            await asyncio.to_thread(_restore_and_record)
             user_id = get_effective_user_id()
             await refresh_user_skills_system_prompt_cache_async(user_id)
             await _publish_skills_cache_reset(user_id)

backend/tests/AGENTS.md

@@ -2,6 +2,12 @@
 
 Backend tests must preserve the runtime invariants they exercise without changing production execution topology.
 
+`test_skills_custom_multiprocess.py` starts independent spawn workers sharing a
+temporary custom-skill root. Park edit/rollback before the write and before the
+history append, observe the peer reaching the mutation lock, then check both
+exclusion and the persisted predecessor chain. Always release and reap workers
+on assertion failures; do not replace this with two threads sharing a mutex.
+
 Search coercion ownership tests audit executable helper bodies/call sites and
 inline integer assignments. Include Exa and SearXNG in the deferred census;
 pin new-provider copies and docstring false positives with source fixtures.

backend/tests/test_skills_custom_multiprocess.py

@@ -0,0 +1,136 @@
+"""History ordering for independent Gateway processes sharing a skills root."""
+
+import asyncio
+import json
+import multiprocessing
+from types import SimpleNamespace
+
+import pytest
+
+
+def _content(description: str) -> str:
+    return f"---\nname: demo-skill\ndescription: {description}\n---\n\n# Demo\n"
+
+
+def _mutate_in_process(base_dir, action, content, gate_phase, ready, start, attempted, parked, release, done):
+    # Use spawn, not fork: each worker has its own lock registry and storage.
+    from app.gateway.routers import skills as router
+    from deerflow.config.paths import Paths
+    from deerflow.skills.security_scanner import ScanResult
+    from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
+
+    config = SimpleNamespace(
+        skills=SimpleNamespace(get_skills_path=lambda: base_dir / "skills", container_path="/mnt/skills"),
+        skill_evolution=SimpleNamespace(enabled=True, moderation_model_name=None),
+    )
+    with pytest.MonkeyPatch.context() as patch:
+        patch.setattr("deerflow.config.get_app_config", lambda: config)
+        patch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=base_dir))
+        patch.setattr("deerflow.config.paths._paths", None)
+        storage = UserScopedSkillStorage("default", host_path=str(base_dir / "skills"))
+        patch.setattr(router, "_get_user_skill_storage", lambda cfg: storage)
+        patch.setattr(router, "get_effective_user_id", lambda: "default")
+
+        async def scan(*args, **kwargs):
+            return ScanResult(decision="allow", reason="offline test")
+
+        async def static_scan(*args, **kwargs):
+            return []
+
+        async def refresh(*args, **kwargs):
+            pass
+
+        patch.setattr(router, "scan_skill_content", scan)
+        patch.setattr(router, "_scan_static_skill_markdown_or_raise", static_scan)
+        patch.setattr(router, "refresh_user_skills_system_prompt_cache_async", refresh)
+        patch.setattr(router, "_publish_skills_cache_reset", refresh)
+        original_lock = router._custom_skill_mutation_lock
+
+        def observed_lock(storage):
+            attempted.set()
+            return original_lock(storage)
+
+        patch.setattr(router, "_custom_skill_mutation_lock", observed_lock)
+        if gate_phase:
+            method = "write_custom_skill" if gate_phase == "write" else "append_history"
+            original = getattr(UserScopedSkillStorage, method)
+
+            def gated(self, *args, **kwargs):
+                parked.set()
+                assert release.wait(30), "parent did not release the parked mutation"
+                return original(self, *args, **kwargs)
+
+            patch.setattr(UserScopedSkillStorage, method, gated)
+
+        request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(system_role="admin")))
+        ready.set()
+        assert start.wait(30), "parent did not start the mutation"
+        if action == "edit":
+            asyncio.run(router.update_custom_skill("demo-skill", router.CustomSkillUpdateRequest(content=content), request, config))
+        else:
+            asyncio.run(router.rollback_custom_skill("demo-skill", router.SkillRollbackRequest(history_index=0), request, config))
+        done.set()
+
+
+@pytest.mark.parametrize("action", ["edit", "rollback"])
+@pytest.mark.parametrize("gate_phase", ["write", "history"])
+def test_custom_skill_history_serializes_independent_processes(tmp_path, action, gate_phase):
+    """A peer must wait through both the predecessor read and history append."""
+    custom_root = tmp_path / "users" / "default" / "skills" / "custom"
+    skill_file = custom_root / "demo-skill" / "SKILL.md"
+    skill_file.parent.mkdir(parents=True)
+    initial, restored, edited, concurrent = (_content(label) for label in ("Initial", "Restored", "Edited", "Concurrent"))
+    skill_file.write_text(initial, encoding="utf-8")
+    history_file = custom_root / ".history" / "demo-skill.jsonl"
+    history_file.parent.mkdir()
+    prior = [{"action": "human_edit", "prev_content": restored, "new_content": initial}] if action == "rollback" else []
+    history_file.write_text("".join(json.dumps(entry) + "\n" for entry in prior), encoding="utf-8")
+
+    context = multiprocessing.get_context("spawn")
+    parked, release = context.Event(), context.Event()
+    workers = []
+    starts, ready_events, attempted_events, done_events = [], [], [], []
+    for worker_action, worker_content, phase in [(action, edited, gate_phase), ("edit", concurrent, None)]:
+        ready, start, attempted, done = (context.Event() for _ in range(4))
+        workers.append(context.Process(target=_mutate_in_process, args=(tmp_path, worker_action, worker_content, phase, ready, start, attempted, parked, release, done)))
+        ready_events.append(ready)
+        starts.append(start)
+        attempted_events.append(attempted)
+        done_events.append(done)
+
+    try:
+        for worker in workers:
+            worker.start()
+        for ready in ready_events:
+            assert ready.wait(30), "worker did not initialize"
+        starts[0].set()
+        assert parked.wait(10), "first mutation did not reach the gate"
+        starts[1].set()
+        assert attempted_events[1].wait(10), "peer did not reach the mutation lock"
+        assert not done_events[1].wait(2), "peer mutated the skill inside the first process's critical section"
+        release.set()
+        for worker in workers:
+            worker.join(30)
+            assert worker.exitcode == 0
+        assert all(done.is_set() for done in done_events)
+    finally:
+        release.set()
+        for start in starts:
+            start.set()
+        for worker in workers:
+            if worker.pid is not None:
+                worker.join(10)
+                if worker.is_alive():
+                    worker.terminate()
+                    worker.join(10)
+                worker.close()
+
+    entries = [json.loads(line) for line in history_file.read_text(encoding="utf-8").splitlines()]
+    first_content = edited if action == "edit" else restored
+    assert len(entries) == len(prior) + 2
+    assert entries[len(prior)]["action"] == ("human_edit" if action == "edit" else "rollback")
+    assert entries[len(prior)]["prev_content"] == initial
+    assert entries[len(prior)]["new_content"] == first_content
+    assert entries[-1]["prev_content"] == first_content
+    assert entries[-1]["new_content"] == concurrent
+    assert skill_file.read_text(encoding="utf-8") == concurrent

backend/tests/test_skills_custom_router.py

@@ -1,12 +1,16 @@
 import errno
 import json
 import stat
+import threading
+import time
 import zipfile
 from io import BytesIO
 from pathlib import Path
 from types import SimpleNamespace
 
+import httpx
 import pytest
+import uvicorn
 from _router_auth_helpers import make_authed_test_app
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
@@ -51,7 +55,7 @@ def _make_skill(name: str, *, enabled: bool) -> Skill:
     )
 
 
-def _make_test_app(config) -> FastAPI:
+def _make_test_app(config, *, user_factory=None, bind_current_user: bool = False) -> FastAPI:
     # The listing/detail routes read config.authorization.fail_closed even
     # when authorization is disabled (mirroring list_models). Many tests here
     # build minimal SimpleNamespace configs; backfill the real disabled
@@ -59,7 +63,10 @@ def _make_test_app(config) -> FastAPI:
     # crashing on a missing attribute.
     if not hasattr(config, "authorization"):
         config.authorization = AuthorizationConfig(enabled=False)
-    app = make_authed_test_app(user_factory=_make_admin_user)
+    # bind_current_user=True is what makes the route's real
+    # get_effective_user_id() resolve the stub user, so a test can drive two
+    # requests as two different users.
+    app = make_authed_test_app(user_factory=user_factory or _make_admin_user, bind_current_user=bind_current_user)
     app.state.config = config  # kept for any startup-style reads
     app.dependency_overrides[get_config] = lambda: config
     app.include_router(skills_router.router)
@@ -1764,3 +1771,341 @@ async def test_install_skill_archive_drains_install_and_cache_refresh_across_can
 
     assert _user_custom_dir(tmp_path, "default").joinpath("install-skill").exists()
     assert refresh_calls == [("refresh", "default")]
+
+
+def _run_live_server(app) -> tuple[uvicorn.Server, threading.Thread, str]:
+    """Serve the app on a real socket so two requests can run concurrently."""
+    server = uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=0, log_level="error"))
+    thread = threading.Thread(target=server.run, daemon=True)
+    thread.start()
+    deadline = time.time() + 10
+    while not server.started and time.time() < deadline:
+        time.sleep(0.05)
+    assert server.started, "uvicorn test server did not start"
+    port = server.servers[0].sockets[0].getsockname()[1]
+    return server, thread, f"http://127.0.0.1:{port}"
+
+
+def _gate_first_write(monkeypatch) -> tuple[threading.Event, threading.Event]:
+    """Pause the first mutation inside its storage write, after its
+    predecessor read and before the write lands, so the test can complete a
+    second mutation inside the read -> write window."""
+    first_write_entered = threading.Event()
+    second_put_done = threading.Event()
+    original_write = UserScopedSkillStorage.write_custom_skill
+
+    def _gated_write(self, name, relative_path, content):
+        if not first_write_entered.is_set():
+            first_write_entered.set()
+            # When the mutation paths are serialized this wait times out —
+            # the second mutation cannot land inside the first one's critical
+            # section — and the write proceeds with a still-true predecessor.
+            # Without serialization the gate releases as soon as the test
+            # signals that the second PUT has fully completed, leaving the
+            # first entry's prev_content stale.
+            second_put_done.wait(timeout=2)
+        return original_write(self, name, relative_path, content)
+
+    monkeypatch.setattr(UserScopedSkillStorage, "write_custom_skill", _gated_write)
+    return first_write_entered, second_put_done
+
+
+def test_custom_skill_update_history_survives_concurrent_put(monkeypatch, tmp_path):
+    """A PUT completing between request A's predecessor read and A's write
+    must still be recorded in history: A's entry may not claim it overwrote
+    a revision that was already gone, and B's entry must record what it
+    actually overwrote, so every overwritten revision stays reachable by
+    rollback."""
+    skills_root = tmp_path / "skills"
+    from deerflow.config.paths import Paths
+
+    user_custom = _user_custom_dir(tmp_path, "default")
+    custom_dir = user_custom / "demo-skill"
+    custom_dir.mkdir(parents=True, exist_ok=True)
+    initial_content = _skill_content("demo-skill")
+    edited_content = _skill_content("demo-skill", "Edited skill")
+    concurrent_content = _skill_content("demo-skill", "Concurrent edit")
+    (custom_dir / "SKILL.md").write_text(initial_content, 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)
+
+    async def _refresh(user_id: str):
+        return None
+
+    monkeypatch.setattr("app.gateway.routers.skills.refresh_user_skills_system_prompt_cache_async", _refresh)
+    monkeypatch.setattr("app.gateway.routers.skills.get_effective_user_id", lambda: "default")
+    monkeypatch.setattr("app.gateway.routers.skills.scan_skill_content", lambda *args, **kwargs: _async_scan("allow", "ok"))
+
+    first_write_entered, second_put_done = _gate_first_write(monkeypatch)
+
+    app = _make_test_app(config)
+    server, server_thread, base_url = _run_live_server(app)
+    try:
+        with httpx.Client(base_url=base_url, timeout=30) as client_a, httpx.Client(base_url=base_url, timeout=30) as client_b:
+            a_result: dict = {}
+
+            def _request_a():
+                a_result["response"] = client_a.put("/api/skills/custom/demo-skill", json={"content": edited_content})
+
+            thread_a = threading.Thread(target=_request_a)
+            thread_a.start()
+            assert first_write_entered.wait(timeout=10), "request A never reached its write"
+            b_response = client_b.put("/api/skills/custom/demo-skill", json={"content": concurrent_content})
+            second_put_done.set()
+            thread_a.join(timeout=30)
+            assert a_result["response"].status_code == 200
+            assert b_response.status_code == 200
+
+            history_response = client_a.get("/api/skills/custom/demo-skill/history")
+    finally:
+        server.should_exit = True
+        server_thread.join(timeout=10)
+
+    assert history_response.status_code == 200
+    entries = history_response.json()["history"]
+    assert [entry["action"] for entry in entries] == ["human_edit", "human_edit"]
+    # First entry: A edited from the initial revision it actually overwrote.
+    assert entries[0]["prev_content"] == initial_content
+    assert entries[0]["new_content"] == edited_content
+    # Second entry: B landed after A and recorded A's revision as its
+    # predecessor — the chain stays replayable, so B remains reachable by
+    # a rollback from the latest entry.
+    assert entries[1]["prev_content"] == edited_content
+    assert entries[1]["new_content"] == concurrent_content
+    assert (custom_dir / "SKILL.md").read_text(encoding="utf-8") == concurrent_content
+
+
+def test_custom_skill_rollback_history_survives_concurrent_put(monkeypatch, tmp_path):
+    """Same contract as the edit path, for rollback: a concurrent PUT
+    completing between the rollback's predecessor read and its restore write
+    must end up recorded as the rollback's actually-overwritten content (or
+    serialized after it), never silently dropped from the history chain."""
+    skills_root = tmp_path / "skills"
+    from deerflow.config.paths import Paths
+
+    user_custom = _user_custom_dir(tmp_path, "default")
+    custom_dir = user_custom / "demo-skill"
+    custom_dir.mkdir(parents=True, exist_ok=True)
+    original_content = _skill_content("demo-skill")
+    current_content = _skill_content("demo-skill", "Current skill")
+    concurrent_content = _skill_content("demo-skill", "Concurrent edit")
+    (custom_dir / "SKILL.md").write_text(current_content, 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)
+
+    async def _refresh(user_id: str):
+        return None
+
+    monkeypatch.setattr("app.gateway.routers.skills.refresh_user_skills_system_prompt_cache_async", _refresh)
+    monkeypatch.setattr("app.gateway.routers.skills.get_effective_user_id", lambda: "default")
+    monkeypatch.setattr("app.gateway.routers.skills.scan_skill_content", lambda *args, **kwargs: _async_scan("allow", "ok"))
+
+    # One prior human_edit entry so the rollback has something to restore.
+    storage = UserScopedSkillStorage("default", host_path=str(skills_root))
+    history_file = storage.get_skill_history_file("demo-skill")
+    history_file.parent.mkdir(parents=True, exist_ok=True)
+    history_file.write_text(
+        '{"action":"human_edit","prev_content":' + json.dumps(original_content) + ',"new_content":' + json.dumps(current_content) + "}\n",
+        encoding="utf-8",
+    )
+
+    first_write_entered, second_put_done = _gate_first_write(monkeypatch)
+
+    app = _make_test_app(config)
+    server, server_thread, base_url = _run_live_server(app)
+    try:
+        with httpx.Client(base_url=base_url, timeout=30) as client_a, httpx.Client(base_url=base_url, timeout=30) as client_b:
+            a_result: dict = {}
+
+            def _request_a():
+                a_result["response"] = client_a.post("/api/skills/custom/demo-skill/rollback", json={"history_index": -1})
+
+            thread_a = threading.Thread(target=_request_a)
+            thread_a.start()
+            assert first_write_entered.wait(timeout=10), "rollback never reached its write"
+            b_response = client_b.put("/api/skills/custom/demo-skill", json={"content": concurrent_content})
+            second_put_done.set()
+            thread_a.join(timeout=30)
+            assert a_result["response"].status_code == 200
+            assert b_response.status_code == 200
+
+            history_response = client_a.get("/api/skills/custom/demo-skill/history")
+    finally:
+        server.should_exit = True
+        server_thread.join(timeout=10)
+
+    assert history_response.status_code == 200
+    entries = history_response.json()["history"]
+    assert [entry["action"] for entry in entries] == ["human_edit", "rollback", "human_edit"]
+    # The rollback restored the prior revision it actually overwrote.
+    assert entries[1]["prev_content"] == current_content
+    assert entries[1]["new_content"] == original_content
+    # The concurrent edit serialized after the rollback and recorded the
+    # restored revision as its predecessor — the chain stays replayable.
+    assert entries[2]["prev_content"] == original_content
+    assert entries[2]["new_content"] == concurrent_content
+    assert (custom_dir / "SKILL.md").read_text(encoding="utf-8") == concurrent_content
+
+
+def _user_with_id(user_id: str) -> User:
+    """An admin user with a fixed id, so its per-user skill root is predictable."""
+    from uuid import UUID
+
+    return User(email=f"{user_id}@example.com", password_hash="x", system_role="admin", id=UUID(user_id))
+
+
+def _users_in_request_order(*users: User):
+    """Hand one stub user to each request, in arrival order, then repeat the last."""
+    remaining = list(users)
+
+    def _factory() -> User:
+        return remaining.pop(0) if remaining else users[-1]
+
+    return _factory
+
+
+def _gate_write_by_content(monkeypatch, gated_content: str, concurrent_content: str) -> tuple[threading.Event, threading.Event, threading.Event]:
+    """Pause one mutation inside its storage write, after its predecessor read.
+
+    The gated request stays there until the test releases it, so a second
+    mutation either lands inside that window (its own user's lock) or cannot
+    start at all (one lock shared by both users).
+    """
+    gated_write_entered = threading.Event()
+    concurrent_write_entered = threading.Event()
+    release_gated_write = threading.Event()
+    original_write = UserScopedSkillStorage.write_custom_skill
+
+    def _gated_write(self, name, relative_path, content):
+        if content == gated_content:
+            gated_write_entered.set()
+            release_gated_write.wait(timeout=30)
+        elif content == concurrent_content:
+            concurrent_write_entered.set()
+        return original_write(self, name, relative_path, content)
+
+    monkeypatch.setattr(UserScopedSkillStorage, "write_custom_skill", _gated_write)
+    return gated_write_entered, concurrent_write_entered, release_gated_write
+
+
+def test_custom_skill_mutation_lock_is_scoped_to_the_owning_user(tmp_path, monkeypatch):
+    """One mutation lock per owning-user storage root, not one per process.
+
+    Every request (and every hot-reloaded config) builds its own storage object,
+    so identity across instances is what keeps one user's mutations serialized;
+    two users that own disjoint custom-skill trees must not share the lock that
+    also spans write_custom_skill's whole projection rebuild.
+    """
+    from app.gateway.routers.skills import _custom_skill_mutation_lock
+    from deerflow.config.paths import Paths
+
+    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: Paths(base_dir=tmp_path))
+    monkeypatch.setattr("deerflow.config.paths._paths", None)
+
+    alice = UserScopedSkillStorage("alice", host_path=str(tmp_path / "skills"))
+    alice_second_instance = UserScopedSkillStorage("alice", host_path=str(tmp_path / "skills"))
+    bob = UserScopedSkillStorage("bob", host_path=str(tmp_path / "skills"))
+
+    assert _custom_skill_mutation_lock(alice) is _custom_skill_mutation_lock(alice_second_instance)
+    assert _custom_skill_mutation_lock(alice) is not _custom_skill_mutation_lock(bob)
+
+
+def test_custom_skill_update_does_not_serialize_an_unrelated_user(monkeypatch, tmp_path):
+    """One user's in-flight mutation must not block a different user's.
+
+    The locked sequence wraps write_custom_skill, which rebuilds the user's
+    skill projection and can therefore be slow (a large or linked package, a
+    projection lock held elsewhere). With one process-wide lock, Alice's
+    rebuild stalls Bob's edit of an unrelated skill; scoped to the owning
+    user's storage, Bob's write reaches storage while Alice is still inside
+    hers, and each user's own history chain stays intact.
+    """
+    skills_root = tmp_path / "skills"
+    alice = _user_with_id("00000000-0000-4000-8000-0000000000a1")
+    bob = _user_with_id("00000000-0000-4000-8000-0000000000b2")
+    alice_custom = _user_custom_dir(tmp_path, str(alice.id)) / "demo-skill"
+    bob_custom = _user_custom_dir(tmp_path, str(bob.id)) / "demo-skill"
+    alice_custom.mkdir(parents=True, exist_ok=True)
+    bob_custom.mkdir(parents=True, exist_ok=True)
+    alice_initial = _skill_content("demo-skill")
+    bob_initial = _skill_content("demo-skill", "Bob's own skill")
+    alice_edited = _skill_content("demo-skill", "Edited by Alice")
+    bob_edited = _skill_content("demo-skill", "Edited by Bob")
+    (alice_custom / "SKILL.md").write_text(alice_initial, encoding="utf-8")
+    (bob_custom / "SKILL.md").write_text(bob_initial, encoding="utf-8")
+
+    from deerflow.config.paths import Paths
+
+    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)
+
+    async def _refresh(user_id: str):
+        return None
+
+    monkeypatch.setattr("app.gateway.routers.skills.refresh_user_skills_system_prompt_cache_async", _refresh)
+    monkeypatch.setattr("app.gateway.routers.skills.scan_skill_content", lambda *args, **kwargs: _async_scan("allow", "ok"))
+
+    gated_write_entered, concurrent_write_entered, release_gated_write = _gate_write_by_content(monkeypatch, alice_edited, bob_edited)
+
+    app = _make_test_app(config, user_factory=_users_in_request_order(alice, bob), bind_current_user=True)
+    server, server_thread, base_url = _run_live_server(app)
+    bob_progressed = False
+    try:
+        with httpx.Client(base_url=base_url, timeout=30) as client_a, httpx.Client(base_url=base_url, timeout=30) as client_b:
+            a_result: dict = {}
+            b_result: dict = {}
+
+            def _request_a():
+                a_result["response"] = client_a.put("/api/skills/custom/demo-skill", json={"content": alice_edited})
+
+            def _request_b():
+                b_result["response"] = client_b.put("/api/skills/custom/demo-skill", json={"content": bob_edited})
+
+            thread_a = threading.Thread(target=_request_a)
+            thread_a.start()
+            assert gated_write_entered.wait(timeout=10), "Alice's edit never reached its write"
+
+            thread_b = threading.Thread(target=_request_b)
+            thread_b.start()
+            # Alice is parked inside her mutation; Bob must not wait for her.
+            bob_progressed = concurrent_write_entered.wait(timeout=5)
+
+            release_gated_write.set()
+            thread_a.join(timeout=30)
+            thread_b.join(timeout=30)
+    finally:
+        release_gated_write.set()
+        server.should_exit = True
+        server_thread.join(timeout=10)
+
+    assert bob_progressed, "Bob's edit waited on Alice's in-flight mutation instead of its own user's lock"
+    assert a_result["response"].status_code == 200
+    assert b_result["response"].status_code == 200
+    assert (alice_custom / "SKILL.md").read_text(encoding="utf-8") == alice_edited
+    assert (bob_custom / "SKILL.md").read_text(encoding="utf-8") == bob_edited
+
+    # Each user kept its own chain: the concurrent peer never entered the
+    # other user's history.
+    alice_history = UserScopedSkillStorage(str(alice.id), host_path=str(skills_root)).read_history("demo-skill")
+    bob_history = UserScopedSkillStorage(str(bob.id), host_path=str(skills_root)).read_history("demo-skill")
+    assert [entry["prev_content"] for entry in alice_history] == [alice_initial]
+    assert [entry["new_content"] for entry in alice_history] == [alice_edited]
+    assert [entry["prev_content"] for entry in bob_history] == [bob_initial]
+    assert [entry["new_content"] for entry in bob_history] == [bob_edited]