fix(skills): stop rebuilding the thread skill view every turn on network (#7333)

* fix(skills): stop rebuilding the thread skill view every turn on network filesystems

* fix(skills): repair thread views with stable content fingerprints

---------

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

README.md

@@ -1692,6 +1692,12 @@ When you install `.skill` archives through the Gateway, DeerFlow accepts standar
 
 Disabling a skill also removes it from the sandbox filesystem view, so shell commands and structured file tools follow the same enabled state. Local, Docker/AIO, hostPath provisioner, and newly created E2B sandboxes source `/mnt/skills` from enabled-only projections that update when public, custom, legacy, or managed integration skills are toggled, edited, created, deleted, or installed. Structured `read_file` calls (including line ranges and read-before-write checks) use the sandbox provider's mount mapping, so the user identity captured when the sandbox was acquired remains authoritative. Managed integration packages remain shared, while their projected filesystem visibility follows each user's enabled state. Multi-worker Gateways re-read on-disk enable state while rebuilding user projections, so a toggle handled by one worker is honored by another worker's next sandbox acquire. Existing E2B sandboxes retain their creation-time snapshot until they are recreated. PVC-backed provisioner skills keep their configured PVC snapshot/layout for now; dynamic PVC materialization is tracked separately.
 
+Chat-specific skill views used by an Agent's skill allowlist verify copied file
+contents before sandbox reuse. Edited instructions, missing or injected files,
+and changed links are repaired from the canonical skills on the next acquire;
+inode and timestamp drift on shared storage does not force a rebuild. Integrity
+checks read the projected files, so larger packages increase acquire-time read I/O.
+
 Skill projection manifest saves preserve the original error or interruption even if temporary-file cleanup or its logging also fails. Manifest files retain their private filesystem permissions.
 
 ```

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

@@ -19,7 +19,7 @@
 - **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) 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 (compatibility metadata)**: `allowed-tools` declarations are parsed, validated, exported, and described for interop with other agent frameworks, but they never filter model-visible schemas, refuse calls, narrow a subagent, or rewrite `tool_search` promotions. Real tool authority lives in Agent configuration and authorization: assembly-time provider filtering, runtime Guardrail enforcement, guardrail allowlists, and the deferred-promotion catalog. A lead custom Agent's explicit `skills` list is still enforced at the sandbox filesystem layer (see Sandbox projection); subagent skill lists scope discovery and activation only because concurrently delegated subagents share the lead thread sandbox. Subagents render skill discovery metadata at startup and reuse the same `SkillActivationMiddleware` as the lead; their configured `skills` field limits discovery and activation, not tools.
-- **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.
+- **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. Source and shared-view freshness use metadata digests. Thread views instead sign file bytes, executable flags, entry types/paths and link targets; inode/size/timestamp drift must not recopy an unchanged view. These checks read O(total projected bytes) with bounded memory on each acquire, preserving next-acquire repair of sandbox edits, missing/injected entries and changed links. Never read special files or follow links while fingerprinting; read failures retain fail-closed cleanup. Thread policy version 2 invalidates old metadata-only manifests. The manifest view signature is written and checked with the same algorithm; do not restore an unused metadata walk. Regressions: `tests/test_skill_projection.py`. 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`.
 - **Custom skill writes**: `LocalSkillStorage.write_custom_skill` and `UserScopedSkillStorage.write_custom_skill` publish through a same-directory temporary and `Path.replace`, then make the written path sandbox-readable. Cleanup uses `cleanup_failed_atomic_write`, including after a successful replace has already consumed the temporary name, so an unlink or warning failure cannot replace the write error. `_write_skill_states` keeps its indented JSON and transfers the descriptor only after `fdopen` succeeds; cleanup runs for `BaseException`, including `KeyboardInterrupt`, and warnings replace the previous silent `OSError` swallow. Coverage: `tests/test_skill_storage_atomic_write_cleanup.py`.
 - **Injection (legacy / default)**: Enabled skills are listed in the agent system prompt with full metadata and container paths (`<available_skills>` block). Controlled by `skills.deferred_discovery: false` (default).

backend/packages/harness/deerflow/skills/projection.py

@@ -8,6 +8,7 @@ import json
 import logging
 import os
 import shutil
+import stat
 import tempfile
 import threading
 from collections.abc import Iterator
@@ -37,7 +38,7 @@ _process_locks: dict[Path, threading.RLock] = {}
 _held_projection_locks = threading.local()
 _MANIFEST_VERSION = 1
 _MAX_REBUILD_ATTEMPTS = 2
-_THREAD_PROJECTION_POLICY_VERSION = 1
+_THREAD_PROJECTION_POLICY_VERSION = 2
 
 
 @dataclass(frozen=True)
@@ -448,13 +449,56 @@ def _write_manifest(scope_root: Path, source_signature: str, view_signature: str
         raise
 
 
+def _thread_view_signature(paths: SkillProjectionPaths) -> str:
+    """Sign projected bytes/layout without NFS-sensitive inode/size/timestamps.
+
+    Unlike source/shared-view metadata checks, this reads every projected file
+    in bounded chunks. That cost preserves next-acquire repair of sandbox edits
+    without recopying the whole view when only filesystem metadata settles.
+    Symlinks are signed as links, never opened; special files are not read.
+    """
+    digest = hashlib.sha256()
+    for category in ("public", "custom", "legacy", "integrations"):
+        digest.update(f"root:{category}\0".encode())
+        stack = [(getattr(paths, category), Path("."))]
+        while stack:
+            current, relative_root = stack.pop()
+            with os.scandir(current) as entries:
+                ordered = sorted(entries, key=lambda entry: entry.name)
+            child_dirs: list[tuple[Path, Path]] = []
+            for entry in ordered:
+                relative = relative_root / entry.name
+                if entry.is_symlink():
+                    value = ["link", os.readlink(entry.path)]
+                elif entry.is_dir(follow_symlinks=False):
+                    value = ["dir"]
+                    child_dirs.append((Path(entry.path), relative))
+                elif entry.is_file(follow_symlinks=False):
+                    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0)
+                    fd = os.open(entry.path, flags)
+                    try:
+                        metadata = os.fstat(fd)
+                        if not stat.S_ISREG(metadata.st_mode):
+                            raise ValueError(f"Projected skill file changed type: {entry.path}")
+                        with os.fdopen(fd, "rb", closefd=False) as stream:
+                            content = hashlib.file_digest(stream, "sha256").hexdigest()
+                        value = ["file", content, metadata.st_mode & 0o111]
+                    finally:
+                        os.close(fd)
+                else:
+                    value = ["special", stat.S_IFMT(entry.stat(follow_symlinks=False).st_mode)]
+                digest.update(json.dumps([category, relative.as_posix(), value], separators=(",", ":")).encode())
+            stack.extend(reversed(child_dirs))
+    return digest.hexdigest()
+
+
 def _view_signature(paths: SkillProjectionPaths, scope: str) -> str:
+    if scope == "thread":
+        return _thread_view_signature(paths)
     digest = hashlib.sha256()
     if scope == "public":
         _update_tree_digest(digest, paths.public, "public_view")
-    elif scope in {"user", "thread"}:
-        if scope == "thread":
-            _update_tree_digest(digest, paths.public, "public_view")
+    elif scope == "user":
         _update_tree_digest(digest, paths.custom, "custom_view")
         _update_tree_digest(digest, paths.legacy, "legacy_view")
         _update_tree_digest(digest, paths.integrations, "integrations_view")
@@ -489,9 +533,13 @@ def _thread_projection_is_fresh(
     if manifest_before is None or manifest_before.get("version") != _MANIFEST_VERSION:
         return False
     source_sig = _thread_source_signature(storage, allowed_skills)
+    if manifest_before.get("source_signature") != source_sig:
+        return False
+    # Thread view metadata can settle after a rebuild on shared storage. Check
+    # bytes/layout instead, retaining repair of out-of-band edits on any FS.
     view_sig = _view_signature(paths, "thread")
     manifest_after = _read_manifest(scope_root)
-    return manifest_before == manifest_after and manifest_before.get("source_signature") == source_sig and manifest_before.get("view_signature") == view_sig
+    return manifest_before == manifest_after and manifest_before.get("view_signature") == view_sig
 
 
 def _load_public_skills(storage: SkillStorage, *, enabled_only: bool) -> list[Skill]:

backend/tests/test_skill_projection.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import json
+import os
 import shutil
 import zipfile
 from concurrent.futures import ThreadPoolExecutor
@@ -952,3 +954,129 @@ def test_projection_read_windows_lock_retries_only_contention(tmp_path, monkeypa
                 pytest.fail("Must not enter after a permanent lock error")
         assert caught.value.errno == errno.EBADF
         assert calls == [FakeMsvcrt.LK_NBLCK]
+
+
+def test_thread_projection_stays_fresh_when_view_metadata_drifts(projection_env) -> None:
+    """Identical bytes with new inodes/timestamps never trigger another copy."""
+    from deerflow.skills import projection as projection_module
+
+    env = projection_env
+    _write_skill(env.skills_root / "public", "alpha")
+    projected = ensure_thread_skill_projection(env.storage, "thread-view-drift", {"alpha"})
+    target = projected.public / "alpha" / "SKILL.md"
+    original = target.read_bytes()
+    with patch.object(projection_module, "_rebuild_thread_locked", wraps=projection_module._rebuild_thread_locked) as rebuild:
+        for offset in range(3):
+            replacement = target.with_suffix(".tmp")
+            replacement.write_bytes(original)
+            replacement.chmod(target.stat().st_mode)
+            replacement.replace(target)
+            os.utime(target, ns=(1_000_000_000 + offset, 2_000_000_000 + offset))
+            ensure_thread_skill_projection(env.storage, "thread-view-drift", {"alpha"})
+    assert not rebuild.called
+
+
+def test_thread_projection_rebuilds_when_a_source_skill_changes(projection_env) -> None:
+    from deerflow.skills import projection as projection_module
+
+    env = projection_env
+    _write_skill(env.skills_root / "public", "alpha", "v1")
+    allowed = {"alpha"}
+    ensure_thread_skill_projection(env.storage, "thread-source-change", allowed)
+
+    _write_skill(env.skills_root / "public", "alpha", "a longer replacement body")
+
+    with patch.object(
+        projection_module,
+        "_rebuild_thread_locked",
+        wraps=projection_module._rebuild_thread_locked,
+    ) as rebuild:
+        ensure_thread_skill_projection(env.storage, "thread-source-change", allowed)
+
+    assert rebuild.called
+
+
+@pytest.mark.parametrize("category", ["public", "custom", "legacy", "integrations"])
+def test_thread_view_instruction_tampering_is_repaired(projection_env, category) -> None:
+    env = projection_env
+    roots = {
+        "public": env.skills_root / "public",
+        "custom": env.storage.get_user_custom_root(),
+        "legacy": env.skills_root / "custom",
+        "integrations": env.paths.integration_skills_dir() / "provider",
+    }
+    source = _write_skill(roots[category], "alpha")
+    projected = ensure_thread_skill_projection(env.storage, "thread-tamper", {"alpha"})
+    relative = Path("provider/alpha/SKILL.md") if category == "integrations" else Path("alpha/SKILL.md")
+    target = getattr(projected, category) / relative
+    target.write_text("MUTATED\n", encoding="utf-8")
+
+    ensure_thread_skill_projection(env.storage, "thread-tamper", {"alpha"})
+
+    assert target.read_bytes() == source.read_bytes()
+
+
+@pytest.mark.parametrize("damage", ["overwrite", "delete", "inject_file", "inject_directory", "symlink", "fifo", "executable"])
+def test_thread_view_support_tampering_is_repaired(projection_env, damage) -> None:
+    env = projection_env
+    source = _write_skill(env.skills_root / "public", "alpha").parent
+    (source / "run.py").write_text("print('original')\n", encoding="utf-8")
+    (source / "run.py").chmod(0o755)
+    projected = ensure_thread_skill_projection(env.storage, "thread-support-tamper", {"alpha"})
+    package = projected.public / "alpha"
+    target = package / "run.py"
+    external = env.root / "external.txt"
+    external.write_text("outside the projection", encoding="utf-8")
+    if damage == "overwrite":
+        metadata = target.stat()
+        target.write_text("print('modified')\n", encoding="utf-8")
+        os.utime(target, ns=(metadata.st_atime_ns, metadata.st_mtime_ns))
+    elif damage == "delete":
+        target.unlink()
+    elif damage == "inject_file":
+        (package / "injected.txt").write_text("injected", encoding="utf-8")
+    elif damage == "inject_directory":
+        (package / "injected").mkdir()
+    elif damage == "symlink":
+        target.unlink()
+        try:
+            target.symlink_to(external)
+        except OSError as exc:
+            if getattr(exc, "winerror", None) == 1314:
+                pytest.skip("Windows symlink creation requires SeCreateSymbolicLinkPrivilege")
+            raise
+    elif damage == "fifo":
+        if not hasattr(os, "mkfifo"):
+            pytest.skip("FIFOs require POSIX")
+        target.unlink()
+        os.mkfifo(target)
+    elif damage == "executable":
+        if os.name == "nt":
+            pytest.skip("Windows chmod does not expose POSIX executable bits")
+        target.chmod(0o644)
+
+    ensure_thread_skill_projection(env.storage, "thread-support-tamper", {"alpha"})
+
+    assert not target.is_symlink()
+    assert target.is_file()
+    assert target.read_bytes() == (source / "run.py").read_bytes()
+    assert sorted(path.name for path in package.iterdir()) == ["SKILL.md", "run.py"]
+    if os.name != "nt":
+        assert target.stat().st_mode & 0o111 == 0o111
+    assert external.read_text(encoding="utf-8") == "outside the projection"
+
+
+def test_thread_projection_upgrades_metadata_only_manifest(projection_env) -> None:
+    env = projection_env
+    source = _write_skill(env.skills_root / "public", "alpha")
+    projected = ensure_thread_skill_projection(env.storage, "thread-old-manifest", {"alpha"})
+    target = projected.public / "alpha" / "SKILL.md"
+    manifest_path = projected.public.parent / ".projection-manifest.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    manifest["view_signature"] = "old-metadata-only-signature"
+    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+    target.write_text("MUTATED\n", encoding="utf-8")
+
+    ensure_thread_skill_projection(env.storage, "thread-old-manifest", {"alpha"})
+
+    assert target.read_bytes() == source.read_bytes()