From 2ede08eda5851d4d463742e6fedb4910e5c5e3c4 Mon Sep 17 00:00:00 2001
From: raychen <815315825@qq.com>
Date: Thu, 18 Jun 2026 16:17:20 +0800
Subject: [PATCH] =?UTF-8?q?feature:=20=E4=BC=98=E5=8C=96skill?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- 本地沙箱环境注入文件支持软连接
- skill文件读取支持缓存
---
 docs/mkdocs/en/skill.md                       |   6 +-
 docs/mkdocs/zh/skill.md                       |   6 +-
 examples/skills/agent/agent.py                |   2 +-
 examples/skills/agent/prompts.py              |   3 +
 examples/skills/agent/tools.py                |  20 +-
 .../local/test_local_ws_runtime.py            |  53 ++++-
 tests/code_executors/test_types.py            |   4 +-
 tests/code_executors/utils/test_meta.py       |   1 +
 tests/skills/test_repository.py               |  61 ++++++
 tests/skills/test_utils.py                    |   1 +
 ...est_copy_stager.py => test_file_stager.py} |   4 +-
 trpc_agent_sdk/agents/core/README.md          | 167 +++++++++++++++-
 trpc_agent_sdk/code_executors/_types.py       |   3 +
 .../code_executors/local/_local_ws_runtime.py |  76 +++++---
 trpc_agent_sdk/code_executors/utils/_meta.py  |   3 +
 trpc_agent_sdk/skills/__init__.py             |   2 +
 trpc_agent_sdk/skills/_repository.py          | 183 +++++++++++++++++-
 trpc_agent_sdk/skills/_toolset.py             |   4 +-
 trpc_agent_sdk/skills/_utils.py               |   5 +
 trpc_agent_sdk/skills/stager/_base_stager.py  |   3 +-
 trpc_agent_sdk/skills/tools/__init__.py       |   4 +-
 .../{_copy_stager.py => _file_stager.py}      |  27 ++-
 trpc_agent_sdk/skills/tools/_skill_exec.py    |   2 +-
 trpc_agent_sdk/skills/tools/_skill_load.py    |   2 +-
 trpc_agent_sdk/skills/tools/_skill_run.py     |   2 +-
 25 files changed, 565 insertions(+), 79 deletions(-)
 rename tests/skills/tools/{test_copy_stager.py => test_file_stager.py} (97%)
 rename trpc_agent_sdk/skills/tools/{_copy_stager.py => _file_stager.py} (73%)

diff --git a/docs/mkdocs/en/skill.md b/docs/mkdocs/en/skill.md
index f76a235d..4be0c437 100644
--- a/docs/mkdocs/en/skill.md
+++ b/docs/mkdocs/en/skill.md
@@ -100,6 +100,7 @@ from trpc_agent_sdk.agents import LlmAgent
 from trpc_agent_sdk.models import OpenAIModel
 from trpc_agent_sdk.skills import SkillToolSet
 from trpc_agent_sdk.skills import create_default_skill_repository
+from trpc_agent_sdk.skills.tools import LinkSkillStager
 from trpc_agent_sdk.code_executors import create_local_workspace_runtime
 from trpc_agent_sdk.code_executors import create_container_workspace_runtime
 # Cube is an optional extra (`pip install 'trpc-agent-py[cube]'`); import lazily.
@@ -114,11 +115,12 @@ workspace_runtime = create_local_workspace_runtime()
 #   workspace_runtime = create_cube_workspace_runtime(executor)
 
 # Create skill repository
-repository = create_default_skill_repository("./skills", workspace_runtime=workspace_runtime)
+repository = create_default_skill_repository("./skills", workspace_runtime=workspace_runtime, use_cached_repository=True)
 
 # Create skill tool set with optional artifact save options
 skill_tool_set = SkillToolSet(
     repository=repository,
+    skill_stager=LinkSkillStager(),
     # run_tool_kwargs is an optional tool parameter
     run_tool_kwargs={
         "save_as_artifacts": True,  # Whether to save as artifact files
@@ -137,6 +139,8 @@ agent = LlmAgent(
     )
 ```
 
+*Note: Starting after version 1.1.10, skill loading and injection were optimized to support caching skill content and symlink-based staging in the local sandbox, avoiding full directory copies.*
+
 **Prompt example**:
 
 The `INSTRUCTION` should include complete skill usage workflow guidance:
diff --git a/docs/mkdocs/zh/skill.md b/docs/mkdocs/zh/skill.md
index 760e5eb1..1c5789f0 100644
--- a/docs/mkdocs/zh/skill.md
+++ b/docs/mkdocs/zh/skill.md
@@ -100,6 +100,7 @@ from trpc_agent_sdk.agents import LlmAgent
 from trpc_agent_sdk.models import OpenAIModel
 from trpc_agent_sdk.skills import SkillToolSet
 from trpc_agent_sdk.skills import create_default_skill_repository
+from trpc_agent_sdk.skills.tools import LinkSkillStager
 from trpc_agent_sdk.code_executors import create_local_workspace_runtime
 from trpc_agent_sdk.code_executors import create_container_workspace_runtime
 # Cube 是可选 extra（`pip install 'trpc-agent-py[cube]'`），按需引入。
@@ -114,11 +115,12 @@ workspace_runtime = create_local_workspace_runtime()
 #   workspace_runtime = create_cube_workspace_runtime(executor)
 
 # 创建技能仓库
-repository = create_default_skill_repository("./skills", workspace_runtime=workspace_runtime)
+repository = create_default_skill_repository("./skills", workspace_runtime=workspace_runtime, use_cached_repository=True)
 
 # 创建技能工具集，可配置工件保存选项
 skill_tool_set = SkillToolSet(
     repository=repository,
+    skill_stager=LinkSkillStager(),
     # run_tool_kwargs 属于工具可选参数
     run_tool_kwargs={
         "save_as_artifacts": True,  # 是否存储为制品文件
@@ -137,6 +139,8 @@ agent = LlmAgent(
     )
 ```
 
+*注意：在版本 1.1.10（不包含）之后，优化了 skill 的加载和注入机制，支持缓存 skill 内容和本地沙箱环境软连的方式来避免拷贝*
+
 **提示词示例**：
 
 在 `INSTRUCTION` 中应包含完整的技能使用工作流指导：
diff --git a/examples/skills/agent/agent.py b/examples/skills/agent/agent.py
index 0d70ba87..810ab0ba 100644
--- a/examples/skills/agent/agent.py
+++ b/examples/skills/agent/agent.py
@@ -25,7 +25,7 @@ def create_agent():
     """Create a skill run agent to demonstrate the various capabilities of an LLM agent."""
 
     # Create tools
-    skill_tool_set, skill_repository = create_skill_tool_set()
+    skill_tool_set, skill_repository = create_skill_tool_set(is_link_stager=True, use_cached_repository=True)
 
     return LlmAgent(
         name="skill_run_agent",
diff --git a/examples/skills/agent/prompts.py b/examples/skills/agent/prompts.py
index bc5f9b39..278235ae 100644
--- a/examples/skills/agent/prompts.py
+++ b/examples/skills/agent/prompts.py
@@ -23,6 +23,9 @@
 skill_run inputs/outputs fields to map files instead of shell commands
 like cp or mv where possible.
 
+When a task needs host files inside the workspace, call skill_load first so
+shared work/inputs links are ready, then copy or map files into work/inputs/.
+
 When using a skill, follow this workflow:
 1. First call skill_load to load the skill documentation
 2. Always call skill_list_docs immediately after skill_load to verify what documents have been loaded,
diff --git a/examples/skills/agent/tools.py b/examples/skills/agent/tools.py
index 42bc0a99..a10e8724 100644
--- a/examples/skills/agent/tools.py
+++ b/examples/skills/agent/tools.py
@@ -12,6 +12,8 @@
 from trpc_agent_sdk.code_executors import create_local_workspace_runtime
 from trpc_agent_sdk.skills import ENV_SKILLS_ROOT
 from trpc_agent_sdk.skills import SkillToolSet
+from trpc_agent_sdk.skills.tools import LinkSkillStager
+from trpc_agent_sdk.skills.tools import CopySkillStager
 from trpc_agent_sdk.skills import create_default_skill_repository
 
 
@@ -37,8 +39,13 @@ def _create_workspace_runtime(**kwargs: Any) -> BaseWorkspaceRuntime:
     return create_local_workspace_runtime(**kwargs)
 
 
-def create_skill_tool_set() -> SkillToolSet:
-    """Create a new skill tool set."""
+def create_skill_tool_set(is_link_stager: bool = True, use_cached_repository: bool = True) -> SkillToolSet:
+    """Create a new skill tool set.
+
+    Args:
+        is_link_stager: Whether to use link stager.
+        use_cached_repository: Whether to use cached repository.
+    """
     tool_kwargs = {
         "save_as_artifacts": True,
         "omit_inline_content": False,
@@ -46,5 +53,10 @@ def create_skill_tool_set() -> SkillToolSet:
     workspace_runtime_args = {}
     workspace_runtime = _create_workspace_runtime(**workspace_runtime_args)
     skill_paths = _get_skill_paths()
-    repository = create_default_skill_repository(skill_paths, workspace_runtime=workspace_runtime)
-    return SkillToolSet(repository=repository, run_tool_kwargs=tool_kwargs), repository
+    # use_cached_repository: Whether to use cached repository.
+    repository = create_default_skill_repository(skill_paths, workspace_runtime=workspace_runtime, 
+                                                 use_cached_repository=use_cached_repository)
+    skill_stager = LinkSkillStager() if is_link_stager else CopySkillStager()
+    # skill_stager: The stager to use for staging skills.
+    skill_toolset = SkillToolSet(repository=repository, run_tool_kwargs=tool_kwargs, skill_stager=skill_stager)
+    return skill_toolset, repository
diff --git a/tests/code_executors/local/test_local_ws_runtime.py b/tests/code_executors/local/test_local_ws_runtime.py
index b054507f..657194a4 100644
--- a/tests/code_executors/local/test_local_ws_runtime.py
+++ b/tests/code_executors/local/test_local_ws_runtime.py
@@ -241,13 +241,43 @@ async def test_stage_directory(self):
         assert (dest / "file.txt").read_text() == "content"
         assert (dest / "sub" / "nested.txt").read_text() == "nested"
 
+    @pytest.mark.asyncio
+    async def test_stage_directory_link_mode(self):
+        src_dir = Path(self.tmpdir) / "source_link"
+        src_dir.mkdir()
+        (src_dir / "file.txt").write_text("content")
+        (src_dir / "sub").mkdir()
+        (src_dir / "sub" / "nested.txt").write_text("nested")
+
+        await self.fs.stage_directory(self.ws, str(src_dir), "dest_link", WorkspaceStageOptions(mode="link"))
+
+        dest = Path(self.tmpdir) / "dest_link"
+        assert dest.is_dir()
+        assert (dest / "file.txt").is_symlink()
+        assert (dest / "sub").is_symlink()
+        assert (dest / "file.txt").read_text() == "content"
+        assert (dest / "sub" / "nested.txt").read_text() == "nested"
+
+    @pytest.mark.asyncio
+    async def test_stage_directory_link_mode_keeps_root_mutable_for_stager_links(self):
+        src_dir = Path(self.tmpdir) / "source_link_root"
+        src_dir.mkdir()
+        (src_dir / "file.txt").write_text("content")
+
+        await self.fs.stage_directory(self.ws, str(src_dir), "dest_link_root", WorkspaceStageOptions(mode="link"))
+
+        dest = Path(self.tmpdir) / "dest_link_root"
+        (dest / "out").symlink_to("../out")
+        assert (dest / "out").is_symlink()
+        assert not (src_dir / "out").exists()
+
     @pytest.mark.asyncio
     async def test_stage_directory_read_only(self):
         src_dir = Path(self.tmpdir) / "src_ro"
         src_dir.mkdir()
         (src_dir / "file.txt").write_text("readonly")
 
-        await self.fs.stage_directory(self.ws, str(src_dir), "dest_ro", WorkspaceStageOptions(read_only=True))
+        await self.fs.stage_directory(self.ws, str(src_dir), "dest_ro", WorkspaceStageOptions(read_only=True, mode="copy"))
         dest_file = Path(self.tmpdir) / "dest_ro" / "file.txt"
         mode = dest_file.stat().st_mode
         assert not (mode & 0o222)  # no write bits
@@ -259,7 +289,7 @@ async def test_stage_directory_read_only_via_fs_flag(self):
         (src_dir / "file.txt").write_text("fs_readonly")
 
         fs_ro = LocalWorkspaceFS(read_only_staged_skill=True)
-        await fs_ro.stage_directory(self.ws, str(src_dir), "dest_fs_ro", WorkspaceStageOptions())
+        await fs_ro.stage_directory(self.ws, str(src_dir), "dest_fs_ro", WorkspaceStageOptions(mode="copy"))
         dest_file = Path(self.tmpdir) / "dest_fs_ro" / "file.txt"
         mode = dest_file.stat().st_mode
         assert not (mode & 0o222)
@@ -462,8 +492,8 @@ async def test_fetch_bytes_budget_above_size(self):
         assert data == b"hello world"
         assert raw == 11
 
-    # --- _copy_directory ---
-    def test_copy_directory(self):
+    # --- _put_directory ---
+    def test_put_directory_copy_mode(self):
         src = Path(self.tmpdir) / "copy_src"
         src.mkdir()
         (src / "a.txt").write_text("a")
@@ -471,16 +501,16 @@ def test_copy_directory(self):
         (src / "sub" / "b.txt").write_text("b")
 
         dst = Path(self.tmpdir) / "copy_dst"
-        self.fs._copy_directory(str(src), str(dst))
+        self.fs._put_directory(self.ws, str(src), "copy_dst", mode="copy")
 
         assert (dst / "a.txt").read_text() == "a"
         assert (dst / "sub" / "b.txt").read_text() == "b"
 
-    def test_copy_directory_empty(self):
+    def test_put_directory_copy_mode_empty(self):
         src = Path(self.tmpdir) / "empty_src"
         src.mkdir()
         dst = Path(self.tmpdir) / "empty_dst"
-        self.fs._copy_directory(str(src), str(dst))
+        self.fs._put_directory(self.ws, str(src), "empty_dst", mode="copy")
         assert dst.exists()
 
     # --- _make_tree_read_only ---
@@ -551,8 +581,9 @@ async def test_stage_inputs_host_copy(self):
         specs = [WorkspaceInputSpec(src=f"host://{host_dir}", dst="work/inputs/data", mode="copy")]
         await self.fs.stage_inputs(self.ws, specs)
 
-        copied = Path(self.tmpdir) / "work" / "inputs"
-        assert copied.exists()
+        copied = Path(self.tmpdir) / "work" / "inputs" / "data"
+        assert copied.is_dir()
+        assert (copied / "data.txt").read_text() == "host data"
 
     @pytest.mark.asyncio
     async def test_stage_inputs_host_link(self):
@@ -564,7 +595,9 @@ async def test_stage_inputs_host_link(self):
         await self.fs.stage_inputs(self.ws, specs)
 
         linked = Path(self.tmpdir) / "work" / "inputs" / "linked"
-        assert linked.is_symlink() or linked.exists()
+        assert linked.is_dir()
+        assert (linked / "link.txt").is_symlink()
+        assert (linked / "link.txt").read_text() == "link data"
 
     @pytest.mark.asyncio
     async def test_stage_inputs_workspace(self):
diff --git a/tests/code_executors/test_types.py b/tests/code_executors/test_types.py
index 4a7fe97d..38228cfe 100644
--- a/tests/code_executors/test_types.py
+++ b/tests/code_executors/test_types.py
@@ -294,10 +294,11 @@ class TestWorkspaceStageOptions:
 
     def test_create_workspace_stage_options(self):
         """Test creating workspace stage options."""
-        options = WorkspaceStageOptions(read_only=True, allow_mount=True)
+        options = WorkspaceStageOptions(read_only=True, allow_mount=True, mode="link")
 
         assert options.read_only is True
         assert options.allow_mount is True
+        assert options.mode == "link"
 
     def test_create_workspace_stage_options_defaults(self):
         """Test creating workspace stage options with defaults."""
@@ -305,6 +306,7 @@ def test_create_workspace_stage_options_defaults(self):
 
         assert options.read_only is False
         assert options.allow_mount is False
+        assert options.mode == "link"
 
 
 class TestWorkspaceCapabilities:
diff --git a/tests/code_executors/utils/test_meta.py b/tests/code_executors/utils/test_meta.py
index 9dba4a82..45d9d6cb 100644
--- a/tests/code_executors/utils/test_meta.py
+++ b/tests/code_executors/utils/test_meta.py
@@ -184,6 +184,7 @@ def test_creates_all_subdirs(self):
                 assert (Path(tmpdir) / name).is_dir()
                 assert name in paths
                 assert paths[name] == Path(tmpdir) / name
+            assert (Path(tmpdir) / DIR_WORK / "inputs").is_dir()
 
     def test_creates_metadata_file(self):
         with tempfile.TemporaryDirectory() as tmpdir:
diff --git a/tests/skills/test_repository.py b/tests/skills/test_repository.py
index 6f24581e..6c205096 100644
--- a/tests/skills/test_repository.py
+++ b/tests/skills/test_repository.py
@@ -26,6 +26,7 @@
 from trpc_agent_sdk.skills._repository import (
     BASE_DIR_PLACEHOLDER,
     BaseSkillRepository,
+    CachedFsSkillRepository,
     FsSkillRepository,
     create_default_skill_repository,
 )
@@ -203,6 +204,65 @@ def test_summaries(self, tmp_path):
         assert "a-skill" in names
         assert "b-skill" in names
 
+    def test_base_repository_reads_body_for_summaries(self, tmp_path):
+        _create_skill_dir(tmp_path, "summary-baseline", "Summary", body="# Large Body\n")
+        repo = FsSkillRepository(str(tmp_path))
+        original_read_text = Path.read_text
+
+        with patch.object(Path, "read_text", autospec=True, side_effect=original_read_text) as mock_read_text:
+            summaries = repo.summaries()
+
+        assert [summary.name for summary in summaries] == ["summary-baseline"]
+        assert mock_read_text.call_count == 1
+
+    def test_cached_repository_summaries_do_not_read_skill_body(self, tmp_path):
+        _create_skill_dir(tmp_path, "summary-only", "Summary", body="# Large Body\n")
+
+        with patch.object(Path, "read_text", side_effect=AssertionError("body should not be read")):
+            repo = CachedFsSkillRepository(str(tmp_path))
+            summaries = repo.summaries()
+
+        assert [summary.name for summary in summaries] == ["summary-only"]
+        assert summaries[0].description == "Summary"
+
+    def test_base_repository_does_not_reuse_skill_body(self, tmp_path):
+        _create_skill_dir(tmp_path, "uncached-skill", "Uncached", body="# Uncached Body\n")
+        repo = FsSkillRepository(str(tmp_path))
+        original_read_text = Path.read_text
+
+        with patch.object(Path, "read_text", autospec=True, side_effect=original_read_text) as mock_read_text:
+            repo.get("uncached-skill")
+            repo.get("uncached-skill")
+
+        assert mock_read_text.call_count == 2
+
+    def test_cached_repository_get_reuses_cached_skill_body(self, tmp_path):
+        _create_skill_dir(tmp_path, "cached-skill", "Cached", body="# Cached Body\n")
+        repo = CachedFsSkillRepository(str(tmp_path))
+        original_read_text = Path.read_text
+
+        with patch.object(Path, "read_text", autospec=True, side_effect=original_read_text) as mock_read_text:
+            first = repo.get("cached-skill")
+            second = repo.get("cached-skill")
+
+        assert first.body == second.body
+        assert mock_read_text.call_count == 1
+
+    def test_cached_repository_get_refreshes_cached_body_when_skill_file_changes(self, tmp_path):
+        skill_dir = _create_skill_dir(tmp_path, "mutable-skill", "Before", body="# Before\n")
+        skill_file = skill_dir / "SKILL.md"
+        repo = CachedFsSkillRepository(str(tmp_path))
+        first = repo.get("mutable-skill")
+
+        skill_file.write_text("---\nname: mutable-skill\ndescription: After\n---\n# After changed body\n",
+                              encoding="utf-8")
+
+        second = repo.get("mutable-skill")
+
+        assert "Before" in first.body
+        assert "After changed body" in second.body
+        assert second.summary.description == "After"
+
     def test_refresh(self, tmp_path):
         repo = FsSkillRepository(str(tmp_path))
         assert repo.skill_list() == []
@@ -311,6 +371,7 @@ class TestCreateDefaultSkillRepository:
     def test_creates_repository(self, tmp_path):
         _create_skill_dir(tmp_path, "test")
         repo = create_default_skill_repository(str(tmp_path))
+        assert isinstance(repo, CachedFsSkillRepository)
         assert isinstance(repo, FsSkillRepository)
         assert "test" in repo.skill_list()
 
diff --git a/tests/skills/test_utils.py b/tests/skills/test_utils.py
index 609be43e..52d189ac 100644
--- a/tests/skills/test_utils.py
+++ b/tests/skills/test_utils.py
@@ -130,6 +130,7 @@ def test_creates_subdirectories(self, tmp_path):
         assert len(paths) == 4
         for p in paths.values():
             assert p.exists()
+        assert (tmp_path / "work" / "inputs").is_dir()
 
     def test_creates_metadata_file(self, tmp_path):
         from trpc_agent_sdk.code_executors import META_FILE_NAME
diff --git a/tests/skills/tools/test_copy_stager.py b/tests/skills/tools/test_file_stager.py
similarity index 97%
rename from tests/skills/tools/test_copy_stager.py
rename to tests/skills/tools/test_file_stager.py
index 57876804..7564ad9c 100644
--- a/tests/skills/tools/test_copy_stager.py
+++ b/tests/skills/tools/test_file_stager.py
@@ -3,7 +3,7 @@
 # Copyright (C) 2026 Tencent. All rights reserved.
 #
 # tRPC-Agent-Python is licensed under Apache-2.0.
-"""Unit tests for trpc_agent_sdk.skills.tools._copy_stager.
+"""Unit tests for trpc_agent_sdk.skills.tools._file_stager.
 
 Covers:
 - normalize_workspace_skill_dir: valid paths, edge cases, rejected paths
@@ -18,7 +18,7 @@
 
 import pytest
 
-from trpc_agent_sdk.skills.tools._copy_stager import (
+from trpc_agent_sdk.skills.tools._file_stager import (
     CopySkillStager,
     normalize_workspace_skill_dir,
     _normalize_skill_stage_result,
diff --git a/trpc_agent_sdk/agents/core/README.md b/trpc_agent_sdk/agents/core/README.md
index 4dbbaa2d..98a9b0b4 100644
--- a/trpc_agent_sdk/agents/core/README.md
+++ b/trpc_agent_sdk/agents/core/README.md
@@ -53,7 +53,7 @@
 2. 执行旧状态迁移（兼容历史 key）与 turn 模式清理
 3. 注入 skill 概览（始终执行）
 4. 读取 loaded skills 并按 `max_loaded_skills` 裁剪
-5. `load_mode=session` 在 temp-only 设计下与 `turn` 读取语义一致
+5. 按 `load_mode` 解析状态键并读取（见下文 **temp-only**）
 6. 根据 `tool_result_mode` 分流：
    - `False`：直接向 system instruction 注入 `[Loaded]`、`Docs loaded`、`[Doc]`
    - `True`：跳过注入，交给 `SkillsToolResultRequestProcessor` 处理
@@ -61,6 +61,17 @@
 
 ### 2.3 状态语义
 
+**temp-only 状态模型**：技能 loaded/docs/tools 状态不再维护 `temp:skill:*` 与 `user:skill:*` 两套键，也不再做 temp → user 的 promote（`_maybe_promote_skill_state_for_session` 为 no-op）。统一由 `loaded_state_key()` 等函数（[`_state_keys.py`](../../skills/_state_keys.py)）按 `load_mode` 决定键名：
+
+| `load_mode` | 键名 | 生命周期 |
+|-------------|------|----------|
+| `turn` / `once` | 带 `temp:` 前缀，如 `temp:skill:loaded_by_agent:<agent>/<skill>` | `turn` 每轮 invocation 清空；`once` 注入后 offload |
+| `session` | 去掉 `temp:` 前缀，如 `skill:loaded_by_agent:<agent>/<skill>` | 整个 session 内保留 |
+
+读取一律走 `session_state + state_delta` 合并视图（`_snapshot_state`），`turn` 与 `session` 共用同一套解析函数，仅键前缀与清理策略不同。
+
+各模式清理策略：
+
 - `turn`
   - 每次 invocation 开始清理一次技能状态
   - 对应：`_maybe_clear_skill_state_for_turn`
@@ -68,10 +79,7 @@
   - 本轮用完后清理，避免持续占用上下文
   - 对应：`_maybe_offload_loaded_skills`
 - `session`
-  - 在 temp-only 状态模型下，不再维护 `user:skill:*` 双键
-  - 对应：`_maybe_promote_skill_state_for_session`（当前为 no-op）
-
-读取策略为单一 temp key 读取（`session_state + state_delta` 视图）。
+  - 不做 turn 级清空，也不做 once 级 offload；状态写入无 `temp:` 前缀的键
 
 ### 2.4 关键参数
 
@@ -247,3 +255,152 @@
    - 是否注入了 guidance
    - 是否注入了 loaded context
    - 是否发生了 offload/clear
+
+## 附录
+
+### A. `turn` 与 `once` 的区别
+
+二者都属于**临时状态**（键名带 `temp:` 前缀），差异在于**何时清空 loaded 状态**，以及**同一 invocation 内是否会反复注入 skill 正文**。
+
+> **术语（避免与后文「轮」混淆）**
+>
+> | 术语 | 含义 |
+> |------|------|
+> | **invocation** | 用户发**一条消息**后，Runner 从开始处理到结束的整段流程（其间可有多次 LLM 调用、多次 tool 调用） |
+> | **LLM 调用** | 每次调用模型前执行一次 `process_llm_request`（下文记为 LLM #1、#2、#3 …） |
+> | **用户消息 #N** | 第 N 条用户输入，通常对应第 N 个 invocation |
+>
+> `turn` 的清空发生在**新 invocation 开始时**（仅一次），不是每次 LLM 调用前。因此：
+> - **同一 invocation 内**（同一条用户消息、多次 LLM）：`skill_load` 写入的 state **会保留**，故 LLM #2、#3 都能读到并重复注入；
+> - **跨 invocation**（用户消息 #1 → 用户消息 #2）：消息 #2 开始时 **会清空** 消息 #1 留下的 state。
+>
+> 下文「同一轮内」= 同一 invocation；「下一轮」= 下一条用户消息（新 invocation）。二者不矛盾。
+
+#### 对比一览
+
+| | `turn` | `once` |
+|---|--------|--------|
+| **清空时机** | 每次新 invocation（用户新发一条消息）**开始时**清空 | 每次将 skill 内容注入请求**之后**清空（offload） |
+| **同一 invocation 内多步 Agent 循环** | `skill_load` 后的状态会保留，**每一步 LLM 调用都会重新注入** skill 正文 | 注入 loaded 后 offload；**后续 LLM 不再从 state 注入**（除非再次 `skill_load`） |
+| **跨用户消息** | 下一条用户消息（新 invocation）开始时清空 | 不在 invocation 开始时统一清空；靠「注入后 offload」释放 state |
+| **典型用途** | 一轮内多步工具调用都需要完整 skill 上下文 | 控制 token：skill 正文只进一次 prompt，之后靠历史 / tool result |
+| **对应函数** | `_maybe_clear_skill_state_for_turn` | `_maybe_offload_loaded_skills` |
+
+#### 清空边界（常见误解）
+
+下文「清」均指清除 **skill loaded/docs/tools 的 session state**，不是清除聊天历史。
+
+**`turn`：跨 invocation（用户消息）清，同一 invocation 内不清**
+
+- **一次请求** = 用户发一条消息 → 一次 `run_async(..., new_message=...)` → 一个 invocation。
+- **一次请求里的多轮** = 同一条消息内 Agent 循环（LLM #1 → 工具 → LLM #2 → …，多次 `process_llm_request`）。
+- `turn` 只在**新 invocation 的第一次** `process_llm_request` 时清空（`processor:skills:turn_init` 保证同一条消息内后续 LLM 不再清）。
+
+```text
+用户消息 #1（invocation A）
+  开始 → [清] 只清「更早 invocation」留下的 state
+  LLM #1 → skill_load → LLM #2 → skill_run → LLM #3   ← 中间不再清
+
+用户消息 #2（invocation B）
+  开始 → [清] 清掉消息 #1 结束时残留的 loaded 标记
+```
+
+归纳：**同一条用户消息内的多次 LLM 不清；下一条用户消息（新 invocation）开始时清。**
+
+**`once`：注入 loaded 之后清，不是每个 LLM 轮次都清**
+
+- offload 仅在 `_maybe_offload_loaded_skills` 中触发，且要求本次 `process_llm_request` 读到的 `loaded` **非空**（见 `_skill_processor.py`）。
+- 尚未 `skill_load` 的 LLM 调用（`loaded` 为空）**不会**触发 offload。
+
+```text
+用户消息 #1（一个 invocation）
+  LLM #1：尚无 loaded → 不 offload
+  skill_load → 写入 state
+  LLM #2：注入 [Loaded] ... → [offload 清 state]
+  skill_run
+  LLM #3：state 已空 → 不再从 state 注入（除非再次 skill_load）
+```
+
+归纳：**不是「一次请求里每一轮 LLM 都清」，而是「有 loaded 且本次请求完成注入后清」。**
+
+**一句话对比**
+
+| 模式 | 清空边界 | 同一条用户消息内多次 LLM |
+|------|----------|--------------------------|
+| `turn` | **新用户消息**开始时清 | state **保留**，每步 LLM 都可能重复注入 skill 正文 |
+| `once` | **每次注入 loaded 内容之后**清 | 通常只在 `skill_load` 后的那一次 LLM 从 state 注入正文 |
+| `session` | 不做 turn 级开头清、不做 once 级注入后清 | 键名去掉 `temp:`，可跨多条用户消息保留 |
+
+#### 举例：一轮内先 `skill_load` 再 `skill_run`
+
+用户问：「用 data-analysis 分析 CSV」。Agent 在同一 invocation 内通常会经历多次 LLM 调用：
+
+1. 第 1 次 LLM → 决定调用 `skill_load`
+2. 执行 `skill_load` → 写入 state
+3. 第 2 次 LLM → 构建请求、注入 skill 内容
+4. 决定调用 `skill_run`
+5. 第 3 次 LLM → 继续推理
+
+**`turn` 模式**
+
+```text
+用户消息 #1 开始
+  → [清空] 上一轮 skill 状态
+  → LLM #1：只有 skill 概览，尚无 loaded 正文
+  → skill_load("data-analysis")          // 写入 state
+  → LLM #2：system 里注入 [Loaded] ...   // 同一 invocation 内 state 仍在
+  → skill_run(...)
+  → LLM #3：再次注入 [Loaded] ...        // 同一 invocation 内重复注入（尚未跨用户消息）
+```
+
+特点：同一轮内每一步 LLM 请求都能看到完整 skill 正文，适合多步推理时上下文需持续「在线」；代价是 token 重复消耗。
+
+**`once` 模式**
+
+```text
+用户消息 #1
+  → LLM #1：概览
+  → skill_load("data-analysis")
+  → LLM #2：注入 [Loaded] ... → [offload 清空 state]
+  → skill_run(...)
+  → LLM #3：state 已空，不再从 state 注入 skill 正文
+        （若开启 tool_result_mode，正文可能在 history 的 function_response 里）
+```
+
+特点：skill 正文只在 `skill_load` 后的**那一次**请求里注入，随后从 session state 删除，避免后续每步 LLM 重复塞入 SKILL.md；适合省 token，后续步骤依赖对话历史或 `SkillsToolResultRequestProcessor` 物化结果。
+
+#### 举例：连续两条用户消息
+
+**`turn`**
+
+```text
+用户消息 #1：skill_load + 完成任务
+  → 结束时 state 里可能仍有 loaded 标记
+
+用户消息 #2 开始（新 invocation）
+  → [清空] 消息 #1 的 skill 状态
+  → 若本条消息要再用 skill，需重新 skill_load
+```
+
+**跨 invocation**（用户消息 #1 → 用户消息 #2）时，消息 #1 的 skill state **不会**带到消息 #2；这与消息 #1 **内部** LLM #2、#3 之间 state 仍保留并不冲突。
+
+**`once`**
+
+```text
+用户消息 #1
+  → 每次注入后 offload，通常不会长期保留 loaded state
+
+用户消息 #2
+  → 不会在 invocation 开始时主动统一 wipe
+  → 一般仍需重新 skill_load；重点在于「单次注入后立即释放 state」
+```
+
+#### 与 `session` 对比（选型参考）
+
+| 场景 | 建议 |
+|------|------|
+| 一轮内多次 LLM + 工具，每步都要完整 skill 文档 | `turn`（默认） |
+| skill 正文很大，只想注入一次，后续靠历史 | `once` |
+| 同一会话多轮对话都要复用已加载 skill | `session` |
+
+`examples/skills` 未显式配置 `load_mode` 时一般为 **`turn`**。若 skill 文档较大且一轮内会多次调 LLM，可尝试 `once` 配合 `tool_result_mode=True`，将正文物化进 tool result，而不是每步重复写入 system instruction。
diff --git a/trpc_agent_sdk/code_executors/_types.py b/trpc_agent_sdk/code_executors/_types.py
index 66698cbb..b1e2648a 100644
--- a/trpc_agent_sdk/code_executors/_types.py
+++ b/trpc_agent_sdk/code_executors/_types.py
@@ -184,6 +184,9 @@ class WorkspaceStageOptions(BaseModel):
     allow_mount: bool = False
     """ whether to allow mount"""
 
+    mode: str = "link"
+    """ staging mode hint, e.g. 'copy' or 'link'. Default is 'link'."""
+
 
 class WorkspaceCapabilities(BaseModel):
     """
diff --git a/trpc_agent_sdk/code_executors/local/_local_ws_runtime.py b/trpc_agent_sdk/code_executors/local/_local_ws_runtime.py
index 2f6bfa53..466e6a01 100644
--- a/trpc_agent_sdk/code_executors/local/_local_ws_runtime.py
+++ b/trpc_agent_sdk/code_executors/local/_local_ws_runtime.py
@@ -152,6 +152,13 @@ class LocalWorkspaceFS(BaseWorkspaceFS):
     def __init__(self, read_only_staged_skill: bool = False):
         self.read_only_staged_skill = read_only_staged_skill
 
+    @staticmethod
+    def _normalize_stage_mode(mode: str) -> str:
+        mode = (mode or "copy").strip().lower()
+        if mode not in {"copy", "link"}:
+            raise ValueError(f"unsupported local workspace stage mode: {mode!r}")
+        return mode
+
     @override
     async def put_files(
         self,
@@ -189,11 +196,15 @@ async def stage_directory(
             to: Destination path relative to workspace
             opt: Staging options
         """
-        self._put_directory(ws, src, dst)
+        mode = self._normalize_stage_mode(opt.mode)
+        self._put_directory(ws, src, dst, mode=mode)
 
         # Make tree read-only if requested
         ro = opt.read_only or self.read_only_staged_skill
-        if ro:
+        # Link mode uses symlinks to host files. chmod would follow symlinks on
+        # many platforms and mutate the source skill tree, so leave protection
+        # to the skill stager's symlink-aware chmod pass.
+        if ro and mode != "link":
             if dst:
                 dst = Path(ws.path) / Path(dst)
             else:
@@ -205,9 +216,10 @@ def _put_directory(
         ws: WorkspaceInfo,
         src: str,
         dst: str,
+        mode: str = "copy",
     ) -> None:
         """
-        Copy an entire directory from host into workspace.
+        Put a host path into workspace using copy or link mode.
 
         Args:
             ctx: Context for the operation
@@ -217,31 +229,44 @@ def _put_directory(
         """
         src = os.path.abspath(src)
         dst = path_join(ws.path, dst)
-        self._copy_directory(src, dst)
+        src_path = Path(src)
+        if mode == "link" and src_path.is_dir():
+            self._link_directory(src, dst)
+        elif mode == "link":
+            make_symlink(ws.path, os.path.relpath(dst, ws.path), src)
+        else:
+            copy_path(src, dst)
 
-    def _copy_directory(
+    def _link_directory(
         self,
         src: str,
         dst: str,
     ) -> None:
-        """Copy directory recursively.
+        """Stage a directory by symlinking its direct children.
 
-        Args:
-            src: Source directory path
-            dst: Destination directory path
+        The destination root is a real directory so later staging logic can add
+        workspace-local links such as ``out`` / ``work`` without mutating the
+        original skill directory. Direct child directories such as ``scripts``
+        or ``references`` are linked as directories to avoid walking large
+        skill trees.
         """
         src_path = Path(src)
         dst_path = Path(dst)
+        if dst_path.exists() or dst_path.is_symlink():
+            if dst_path.is_symlink() or dst_path.is_file():
+                dst_path.unlink()
+            elif not dst_path.is_dir():
+                raise ValueError(f"cannot stage into non-directory path: {dst}")
         dst_path.mkdir(parents=True, exist_ok=True)
 
-        for item in src_path.rglob("*"):
-            rel_path = item.relative_to(src)
-            target = dst_path / rel_path
-            if item.is_dir():
-                target.mkdir(parents=True, exist_ok=True)
-            else:
-                target.parent.mkdir(parents=True, exist_ok=True)
-                shutil.copy2(item, target)
+        for item in src_path.iterdir():
+            target = dst_path / item.name
+            if target.exists() or target.is_symlink():
+                if target.is_dir() and not target.is_symlink():
+                    shutil.rmtree(target)
+                else:
+                    target.unlink()
+            target.symlink_to(item.resolve(strict=False), target_is_directory=item.is_dir())
 
     def _make_tree_read_only(
         self,
@@ -401,31 +426,20 @@ async def stage_inputs(
                 # Handle host inputs
                 host_path = spec.src[len("host://"):]
                 resolved = host_path
-                if mode == "link":
-                    make_symlink(ws.path, dst.as_posix(), host_path)
-                else:
-                    self._put_directory(ws, host_path, dst.parent.as_posix())
+                self._put_directory(ws, host_path, dst.as_posix(), mode=mode)
             elif spec.src.startswith("workspace://"):
                 # Handle workspace inputs
                 rel = spec.src[len("workspace://"):]
                 src = path_join(ws.path, rel)
                 resolved = rel
-
-                if mode == "link":
-                    make_symlink(ws.path, dst.as_posix(), src)
-                else:
-                    copy_path(src, path_join(ws.path, dst.as_posix()))
+                self._put_directory(ws, src, dst.as_posix(), mode=mode)
             elif spec.src.startswith("skill://"):
                 # Handle skill inputs
                 rest = spec.src[len("skill://"):]
                 src_base = Path(ws.path) / DIR_SKILLS
                 src = path_join(src_base.as_posix(), rest)
                 resolved = src
-
-                if mode == "link":
-                    make_symlink(ws.path, dst.as_posix(), src)
-                else:
-                    copy_path(src, path_join(ws.path, dst.as_posix()))
+                self._put_directory(ws, src, dst.as_posix(), mode=mode)
             else:
                 raise ValueError(f"unsupported input: {spec.src}")
 
diff --git a/trpc_agent_sdk/code_executors/utils/_meta.py b/trpc_agent_sdk/code_executors/utils/_meta.py
index 34e35648..933a4ba3 100644
--- a/trpc_agent_sdk/code_executors/utils/_meta.py
+++ b/trpc_agent_sdk/code_executors/utils/_meta.py
@@ -105,6 +105,9 @@ def ensure_layout(root: Path | str) -> dict[str, Path]:
     for p in paths.values():
         Path(p).mkdir(parents=True, exist_ok=True)
 
+    # Host/user files are staged under work/inputs before skill_load links them.
+    (root / DIR_WORK / "inputs").mkdir(parents=True, exist_ok=True)
+
     # Initialize metadata if missing
     meta_file = root / META_FILE_NAME
     if not meta_file.exists():
diff --git a/trpc_agent_sdk/skills/__init__.py b/trpc_agent_sdk/skills/__init__.py
index 2f429d41..36cfd3f1 100644
--- a/trpc_agent_sdk/skills/__init__.py
+++ b/trpc_agent_sdk/skills/__init__.py
@@ -56,6 +56,7 @@
 from ._dynamic_toolset import DynamicSkillToolSet
 from ._registry import SkillRegistry
 from ._repository import BaseSkillRepository
+from ._repository import CachedFsSkillRepository
 from ._repository import FsSkillRepository
 from ._repository import VisibilityFilter
 from ._repository import create_default_skill_repository
@@ -135,6 +136,7 @@
     "DynamicSkillToolSet",
     "SkillRegistry",
     "BaseSkillRepository",
+    "CachedFsSkillRepository",
     "FsSkillRepository",
     "VisibilityFilter",
     "create_default_skill_repository",
diff --git a/trpc_agent_sdk/skills/_repository.py b/trpc_agent_sdk/skills/_repository.py
index 38fd71fc..6c1fe053 100644
--- a/trpc_agent_sdk/skills/_repository.py
+++ b/trpc_agent_sdk/skills/_repository.py
@@ -17,6 +17,7 @@
 
 import abc
 import os
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Callable
 from typing import List
@@ -45,6 +46,16 @@
 VisibilityFilter = Callable[[SkillSummary], bool]
 
 
+@dataclass
+class _SkillFileCacheEntry:
+    """Cached parse result for a SKILL.md file."""
+
+    mtime_ns: int
+    size: int
+    front_matter: dict[str, str]
+    body: str | None = None
+
+
 class BaseSkillRepository(abc.ABC):
     """
     Base class for a source of skills.
@@ -278,7 +289,7 @@ def _index_one(self, dirpath: str, skill_file_path: Path) -> bool:
             return False
         self._discovered_skill_files.add(skill_file_key)
 
-        front_matter, _ = self._read_skill_file(skill_file_path)
+        front_matter = self._read_skill_front_matter(skill_file_path)
         name = front_matter.get("name", "").strip()
         if not name:
             name = Path(dirpath).name.strip()
@@ -345,11 +356,39 @@ def _prune_deleted_skills(self) -> None:
         if stale_files:
             self._discovered_skill_files.difference_update(stale_files)
 
+    def _read_skill_front_matter(self, path: Path) -> dict[str, str]:
+        """Read front matter for indexing/summary paths.
+
+        The base repository intentionally keeps the original no-cache behavior
+        so it can be used as a performance baseline against
+        CachedFsSkillRepository.
+        """
+        front_matter, _ = self._read_skill_file(path)
+        return front_matter
+
     @classmethod
-    def _read_skill_file(cls, path: Path) -> tuple[dict[str, str], str]:
+    def _parse_front_matter_yaml(cls, raw_yaml: str) -> dict[str, str]:
+        try:
+            parsed = yaml.safe_load(raw_yaml) or {}
+            if not isinstance(parsed, dict):
+                return {}
+        except Exception:
+            return {}
+        out: dict[str, str] = {}
+        for k, v in parsed.items():
+            key = str(k).strip()
+            if not key:
+                continue
+            if v is None:
+                out[key] = ""
+            else:
+                out[key] = str(v)
+        return out
+
+    def _read_skill_file(self, path: Path) -> tuple[dict[str, str], str]:
         """Read the skill file and return the front matter and body."""
         content = path.read_text(encoding="utf-8")
-        return cls.from_markdown(content)
+        return self.from_markdown(content)
 
     # ------------------------------------------------------------------
     # Public API
@@ -381,7 +420,7 @@ def summaries(self) -> List[SkillSummary]:
         for name in sorted(self._skill_paths):
             skill_file_path = Path(self._skill_paths[name]) / SKILL_FILE
             try:
-                front_matter, _ = self._read_skill_file(skill_file_path)
+                front_matter = self._read_skill_front_matter(skill_file_path)
                 summary = SkillSummary(
                     name=front_matter.get("name", "").strip(),
                     description=front_matter.get("description", "").strip(),
@@ -526,27 +565,151 @@ def _parse_tools_from_body(body: str) -> list[str]:
         return tool_names
 
 
+class CachedFsSkillRepository(FsSkillRepository):
+    """Filesystem skill repository with SKILL.md frontmatter/body caching.
+
+    Cache entries are keyed by the resolved SKILL.md path and invalidated when
+    the file's ``mtime_ns`` or size changes, or when the file is deleted.
+    Summary/index paths read only front matter; full body is read lazily by
+    :meth:`get`.
+    """
+
+    def __init__(self, *roots: str, **kwargs):
+        self._skill_file_cache: dict[str, _SkillFileCacheEntry] = {}
+        super().__init__(*roots, **kwargs)
+
+    @override
+    def _index(self) -> None:
+        self._prune_skill_file_cache()
+        super()._index()
+
+    @override
+    def _prune_deleted_skills(self) -> None:
+        super()._prune_deleted_skills()
+        self._prune_skill_file_cache()
+
+    @staticmethod
+    def _skill_file_cache_key(path: Path) -> str:
+        return str(path.resolve(strict=False))
+
+    @staticmethod
+    def _safe_file_signature(path: Path) -> tuple[int, int] | None:
+        try:
+            stat = path.stat()
+            return stat.st_mtime_ns, stat.st_size
+        except OSError:
+            return None
+
+    def _get_cached_skill_file(self, path: Path) -> _SkillFileCacheEntry | None:
+        """Return a valid cache entry, or clear it when the file changed/deleted."""
+        cache_key = self._skill_file_cache_key(path)
+        signature = self._safe_file_signature(path)
+        if signature is None:
+            self._skill_file_cache.pop(cache_key, None)
+            return None
+        cached = self._skill_file_cache.get(cache_key)
+        if cached and (cached.mtime_ns, cached.size) == signature:
+            return cached
+        self._skill_file_cache.pop(cache_key, None)
+        return None
+
+    def _set_cached_skill_file(
+        self,
+        path: Path,
+        *,
+        front_matter: dict[str, str],
+        body: str | None,
+    ) -> _SkillFileCacheEntry:
+        signature = self._safe_file_signature(path)
+        if signature is None:
+            raise FileNotFoundError(str(path))
+        cache_key = self._skill_file_cache_key(path)
+        entry = _SkillFileCacheEntry(
+            mtime_ns=signature[0],
+            size=signature[1],
+            front_matter=front_matter,
+            body=body,
+        )
+        self._skill_file_cache[cache_key] = entry
+        return entry
+
+    def _prune_skill_file_cache(self) -> None:
+        """Drop cache entries for deleted SKILL.md files."""
+        stale_keys = [cache_key for cache_key in self._skill_file_cache if not Path(cache_key).is_file()]
+        for cache_key in stale_keys:
+            self._skill_file_cache.pop(cache_key, None)
+
+    @override
+    def _read_skill_front_matter(self, path: Path) -> dict[str, str]:
+        """Read only the SKILL.md front matter, avoiding body I/O for summaries."""
+        cached = self._get_cached_skill_file(path)
+        if cached is not None:
+            return dict(cached.front_matter)
+
+        front_matter = self._read_front_matter_only(path)
+        self._set_cached_skill_file(path, front_matter=front_matter, body=None)
+        return dict(front_matter)
+
+    @classmethod
+    def _read_front_matter_only(cls, path: Path) -> dict[str, str]:
+        """Parse YAML front matter without reading the Markdown body."""
+        with path.open("r", encoding="utf-8", newline=None) as file:
+            first_line = file.readline()
+            if first_line != "---\n":
+                return {}
+
+            yaml_lines: list[str] = []
+            for line in file:
+                if line == "---\n":
+                    return cls._parse_front_matter_yaml("".join(yaml_lines))
+                yaml_lines.append(line)
+
+        # Match from_markdown's behavior for an unclosed front matter block.
+        return {}
+
+    @override
+    def _read_skill_file(self, path: Path) -> tuple[dict[str, str], str]:
+        """Read full SKILL.md content, using cached body when valid."""
+        cached = self._get_cached_skill_file(path)
+        if cached is not None and cached.body is not None:
+            return dict(cached.front_matter), cached.body
+
+        content = path.read_text(encoding="utf-8")
+        front_matter, body = self.from_markdown(content)
+        self._set_cached_skill_file(path, front_matter=front_matter, body=body)
+        return dict(front_matter), body
+
+
 def create_default_skill_repository(
     *roots: str,
     workspace_runtime: Optional[BaseWorkspaceRuntime] = None,
     enable_hot_reload: bool = True,
-) -> FsSkillRepository:
+    use_cached_repository: bool = True,
+) -> BaseSkillRepository:
     """Create a new filesystem skill repository.
 
     Args:
         roots: Root directories (or URLs) to scan for skills.
         workspace_runtime: Optional workspace runtime.
         enable_hot_reload: Whether to enable skill hot reload checks.
+        use_cached_repository: Whether to use cached repository.
     Returns:
         A configured :class:`FsSkillRepository`.
     """
     if workspace_runtime is None:
         workspace_runtime = create_local_workspace_runtime()
-    return FsSkillRepository(
-        *roots,
-        workspace_runtime=workspace_runtime,
-        enable_hot_reload=enable_hot_reload,
-    )
+    if use_cached_repository:
+        return CachedFsSkillRepository(
+            *roots,
+            workspace_runtime=workspace_runtime,
+            enable_hot_reload=enable_hot_reload,
+        )
+    else:
+        return FsSkillRepository(
+            *roots,
+            workspace_runtime=workspace_runtime,
+            enable_hot_reload=enable_hot_reload,
+        )
 
 
 SkillRepositoryResolver: TypeAlias = Callable[[InvocationContext], BaseSkillRepository]
diff --git a/trpc_agent_sdk/skills/_toolset.py b/trpc_agent_sdk/skills/_toolset.py
index d1d186ed..4499dc9f 100644
--- a/trpc_agent_sdk/skills/_toolset.py
+++ b/trpc_agent_sdk/skills/_toolset.py
@@ -50,7 +50,7 @@
 from .tools import WorkspaceKillSessionTool
 from .tools import CreateWorkspaceNameCallback
 from .tools import default_create_ws_name_callback
-from .tools import CopySkillStager
+from .tools import LinkSkillStager
 from .stager import Stager
 
 
@@ -104,7 +104,7 @@ def __init__(self,
         )
         self._skill_config = skill_config or DEFAULT_SKILL_CONFIG
         self._create_ws_name_cb = create_ws_name_cb or default_create_ws_name_callback
-        self._skill_stager = skill_stager or CopySkillStager()
+        self._skill_stager = skill_stager or LinkSkillStager()
         self._load_tool = SkillLoadTool(repository=self._repository,
                                         repo_resolver=repo_resolver,
                                         skill_stager=self._skill_stager,
diff --git a/trpc_agent_sdk/skills/_utils.py b/trpc_agent_sdk/skills/_utils.py
index 61bccc00..163667a6 100644
--- a/trpc_agent_sdk/skills/_utils.py
+++ b/trpc_agent_sdk/skills/_utils.py
@@ -107,6 +107,11 @@ def ensure_layout(root: Union[Path, str]) -> dict[str, Path]:
         if not path.exists():
             os.makedirs(path, mode=0o755, exist_ok=True)
 
+    # Host/user files are staged under work/inputs before skill_load links them.
+    inputs_dir = root / DIR_WORK / "inputs"
+    if not inputs_dir.exists():
+        os.makedirs(inputs_dir, mode=0o755, exist_ok=True)
+
     # Initialize metadata if missing
     metadata_file = root / META_FILE_NAME
     if not metadata_file.exists():
diff --git a/trpc_agent_sdk/skills/stager/_base_stager.py b/trpc_agent_sdk/skills/stager/_base_stager.py
index 2775912b..ef12a0ba 100644
--- a/trpc_agent_sdk/skills/stager/_base_stager.py
+++ b/trpc_agent_sdk/skills/stager/_base_stager.py
@@ -54,6 +54,7 @@ def __init__(self) -> None:
         # Deduplicate noisy link warnings within one process lifetime.
         # Key format: "<invocation_id>|<skill_name>|<stderr>".
         self._link_error_warned_keys: set[str] = set()
+        self._stage_mode: str = "link"
 
     # ------------------------------------------------------------------
     # Public API
@@ -93,7 +94,7 @@ async def stage_skill(self, request: SkillStageRequest) -> SkillStageResult:
         await self.remove_workspace_path(ctx, runtime, ws, dest)
 
         fs = runtime.fs(ctx)
-        await fs.stage_directory(ws, root, dest, WorkspaceStageOptions(), ctx)
+        await fs.stage_directory(ws, root, dest, WorkspaceStageOptions(mode=self._stage_mode), ctx)
 
         await self._link_workspace_dirs(ctx, runtime, ws, name)
         await self._read_only_except_symlinks(ctx, runtime, ws, dest)
diff --git a/trpc_agent_sdk/skills/tools/__init__.py b/trpc_agent_sdk/skills/tools/__init__.py
index a6fc68d5..b4b99629 100644
--- a/trpc_agent_sdk/skills/tools/__init__.py
+++ b/trpc_agent_sdk/skills/tools/__init__.py
@@ -5,7 +5,8 @@
 
 from ._common import CreateWorkspaceNameCallback
 from ._common import default_create_ws_name_callback
-from ._copy_stager import CopySkillStager
+from ._file_stager import CopySkillStager
+from ._file_stager import LinkSkillStager
 from ._save_artifact import SaveArtifactTool
 from ._skill_exec import SkillExecTool
 from ._skill_list import skill_list
@@ -24,6 +25,7 @@
     "CreateWorkspaceNameCallback",
     "default_create_ws_name_callback",
     "CopySkillStager",
+    "LinkSkillStager",
     "SaveArtifactTool",
     "SkillExecTool",
     "skill_list",
diff --git a/trpc_agent_sdk/skills/tools/_copy_stager.py b/trpc_agent_sdk/skills/tools/_file_stager.py
similarity index 73%
rename from trpc_agent_sdk/skills/tools/_copy_stager.py
rename to trpc_agent_sdk/skills/tools/_file_stager.py
index 0ca52884..30b704b1 100644
--- a/trpc_agent_sdk/skills/tools/_copy_stager.py
+++ b/trpc_agent_sdk/skills/tools/_file_stager.py
@@ -3,10 +3,13 @@
 # Copyright (C) 2026 Tencent. All rights reserved.
 #
 # tRPC-Agent-Python is licensed under Apache-2.0.
-"""Skill staging interface and default copy-based implementation.
+"""Skill staging interface implementation using file system.
 
 This module provides the CopySkillStager class which is responsible for staging
-skills to the workspace.
+skills to the workspace using file system.
+
+The actual filesystem work — digest check, ``stage_directory``, symlink
+creation, read-only chmod — is delegated to :class:`~trpc_agent.skills.stager.Stager`.
 """
 
 from __future__ import annotations
@@ -36,7 +39,7 @@ def normalize_workspace_skill_dir(raw: str) -> str:
     """Normalize and validate a workspace-relative skill directory.
 
     Mirrors Go's ``normalizeWorkspaceSkillDir``.  Strips leading slashes,
-    normalises path separators, and ensures the result stays within a
+    normalizes path separators, and ensures the result stays within a
     known workspace root.
 
     Raises:
@@ -60,7 +63,7 @@ def normalize_workspace_skill_dir(raw: str) -> str:
 
 
 def _normalize_skill_stage_result(result: SkillStageResult) -> SkillStageResult:
-    """Return *result* with :attr:`workspace_skill_dir` normalised.
+    """Return *result* with :attr:`workspace_skill_dir` normalized.
 
     Mirrors Go's ``normalizeSkillStageResult``.
 
@@ -71,17 +74,29 @@ def _normalize_skill_stage_result(result: SkillStageResult) -> SkillStageResult:
 
 
 class CopySkillStager(Stager):
-    """Default stager: copies the skill directory into ``skills/<name>``.
+    """Stager: copies the skill directory into ``skills/<name>``.
 
     The actual filesystem work — digest check, ``stage_directory``, symlink
     creation, read-only chmod — is delegated to :class:`~trpc_agent.skills.stager.Stager`.
     """
 
+    def __init__(self) -> None:
+        super().__init__()
+        self._stage_mode: str = "copy"
+
     async def stage_skill(self, request: SkillStageRequest) -> SkillStageResult:
-        """Stage the skill and return the normalised workspace skill dir."""
+        """Stage the skill and return the normalized workspace skill dir."""
         if request.repository is None:
             raise ValueError(_ERR_REPO_NOT_CONFIGURED)
 
         result = await super().stage_skill(request)
 
         return _normalize_skill_stage_result(result)
+
+
+class LinkSkillStager(CopySkillStager):
+    """Stager: links the skill directory into ``skills/<name>`` and links the shared workspace dirs."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._stage_mode: str = "link"
diff --git a/trpc_agent_sdk/skills/tools/_skill_exec.py b/trpc_agent_sdk/skills/tools/_skill_exec.py
index 0b0a428b..b093088b 100644
--- a/trpc_agent_sdk/skills/tools/_skill_exec.py
+++ b/trpc_agent_sdk/skills/tools/_skill_exec.py
@@ -86,7 +86,7 @@
 from ._common import default_create_ws_name_callback
 from ._common import inline_json_schema_refs
 from ._common import require_non_empty
-from ._copy_stager import SkillStageRequest
+from ._file_stager import SkillStageRequest
 from ._skill_run import SkillRunInput
 from ._skill_run import SkillRunOutput
 from ._skill_run import SkillRunTool
diff --git a/trpc_agent_sdk/skills/tools/_skill_load.py b/trpc_agent_sdk/skills/tools/_skill_load.py
index eddf57a9..caeacb72 100644
--- a/trpc_agent_sdk/skills/tools/_skill_load.py
+++ b/trpc_agent_sdk/skills/tools/_skill_load.py
@@ -35,7 +35,7 @@
 from ._common import CreateWorkspaceNameCallback
 from ._common import default_create_ws_name_callback
 from ._common import set_staged_workspace_dir
-from ._copy_stager import CopySkillStager
+from ._file_stager import CopySkillStager
 from .._repository import SkillRepositoryResolver
 
 
diff --git a/trpc_agent_sdk/skills/tools/_skill_run.py b/trpc_agent_sdk/skills/tools/_skill_run.py
index 0102d9ae..ca106c67 100644
--- a/trpc_agent_sdk/skills/tools/_skill_run.py
+++ b/trpc_agent_sdk/skills/tools/_skill_run.py
@@ -52,7 +52,7 @@
 from ._common import default_create_ws_name_callback
 from ._common import get_staged_workspace_dir
 from ._common import inline_json_schema_refs
-from ._copy_stager import CopySkillStager
+from ._file_stager import CopySkillStager
 
 # ---------------------------------------------------------------------------
 # Constants