diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md index 2e3f9dbaf0..9ec301437e 100644 --- a/backend/app/gateway/routers/AGENTS.md +++ b/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. diff --git a/backend/app/gateway/routers/skills.py b/backend/app/gateway/routers/skills.py index 29a9293d00..5a2952ab76 100644 --- a/backend/app/gateway/routers/skills.py +++ b/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) diff --git a/backend/tests/AGENTS.md b/backend/tests/AGENTS.md index 6bc54aa519..7c4ac099e0 100644 --- a/backend/tests/AGENTS.md +++ b/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. diff --git a/backend/tests/test_skills_custom_multiprocess.py b/backend/tests/test_skills_custom_multiprocess.py new file mode 100644 index 0000000000..6b34558091 --- /dev/null +++ b/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 diff --git a/backend/tests/test_skills_custom_router.py b/backend/tests/test_skills_custom_router.py index 7a9524b7d2..d98a56d1e2 100644 --- a/backend/tests/test_skills_custom_router.py +++ b/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]