diff --git a/reme2/component/file_graph/base_file_graph.py b/reme2/component/file_graph/base_file_graph.py
index 07504545..f979c4f5 100644
--- a/reme2/component/file_graph/base_file_graph.py
+++ b/reme2/component/file_graph/base_file_graph.py
@@ -26,8 +26,15 @@ class BaseFileGraph(BaseComponent):
         """Delete nodes from the graph."""
 
     @abstractmethod
-    async def get_nodes(self, paths: list[str]) -> list[FileNode]:
-        """Get nodes from the graph."""
+    async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
+        """Get nodes from the graph.
+
+        ``paths=None`` (the default) returns every real node — the
+        graph's "scan everything" entry point. Pass an explicit list
+        for path-known lookups; ``[]`` returns ``[]`` (empty input,
+        empty output). Virtual placeholders (nodes that exist only
+        because something links to them) are filtered out either way.
+        """
 
     @abstractmethod
     async def rebuild_links(self) -> None:
diff --git a/reme2/component/file_graph/local_file_graph.py b/reme2/component/file_graph/local_file_graph.py
index 9b13832d..3af6692a 100644
--- a/reme2/component/file_graph/local_file_graph.py
+++ b/reme2/component/file_graph/local_file_graph.py
@@ -98,7 +98,9 @@ class LocalFileGraph(BaseFileGraph):
             if demoted:
                 self._pending.setdefault(path, set()).update(demoted)
 
-    async def get_nodes(self, paths: list[str]) -> list[FileNode]:
+    async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
+        if paths is None:
+            return list(self._nodes.values())
         return [self._nodes[p] for p in paths if p in self._nodes]
 
     async def rebuild_links(self) -> None:
diff --git a/reme2/component/file_graph/neo4j_file_graph.py b/reme2/component/file_graph/neo4j_file_graph.py
index da5865d6..2add7b81 100644
--- a/reme2/component/file_graph/neo4j_file_graph.py
+++ b/reme2/component/file_graph/neo4j_file_graph.py
@@ -231,20 +231,34 @@ class Neo4jFileGraph(BaseFileGraph):
             paths=paths,
         )
 
-    async def get_nodes(self, paths: list[str]) -> list[FileNode]:
-        """Return only real nodes (virtual placeholders are filtered)."""
-        if not paths:
+    async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
+        """Return real nodes (virtual placeholders filtered).
+
+        ``paths=None`` streams every real node ordered by path. An
+        explicit ``[]`` returns ``[]`` without hitting the database.
+        """
+        if paths is not None and not paths:
             return []
         async with self._session() as session:
-            rec = await session.run(
-                """
-                UNWIND $paths AS p
-                MATCH (f:File {path: p})
-                WHERE f.links_json IS NOT NULL
-                RETURN f
-                """,
-                paths=list(paths),
-            )
+            if paths is None:
+                rec = await session.run(
+                    """
+                    MATCH (f:File)
+                    WHERE f.links_json IS NOT NULL
+                    RETURN f
+                    ORDER BY f.path ASC
+                    """,
+                )
+            else:
+                rec = await session.run(
+                    """
+                    UNWIND $paths AS p
+                    MATCH (f:File {path: p})
+                    WHERE f.links_json IS NOT NULL
+                    RETURN f
+                    """,
+                    paths=list(paths),
+                )
             rows = [row["f"] async for row in rec]
         return [self._row_to_node(row) for row in rows]
 
diff --git a/reme2/component/file_graph/nx_file_graph.py b/reme2/component/file_graph/nx_file_graph.py
index 21198862..888cd7e4 100644
--- a/reme2/component/file_graph/nx_file_graph.py
+++ b/reme2/component/file_graph/nx_file_graph.py
@@ -95,8 +95,14 @@ class NxFileGraph(BaseFileGraph):
             if self._graph.in_degree(path) == 0:
                 self._graph.remove_node(path)
 
-    async def get_nodes(self, paths: list[str]) -> list[FileNode]:
+    async def get_nodes(self, paths: list[str] | None = None) -> list[FileNode]:
         nodes_view = self._graph.nodes
+        if paths is None:
+            return [
+                data["node"]
+                for _, data in nodes_view(data=True)
+                if "node" in data
+            ]
         return [
             nodes_view[path]["node"]
             for path in paths
diff --git a/reme2/config/obsidian.md b/reme2/config/obsidian.md
index 874c5bc1..3c47c6f6 100644
--- a/reme2/config/obsidian.md
+++ b/reme2/config/obsidian.md
@@ -8,14 +8,16 @@
 | 通用     | version |                                                           |
 | search | search  | query="search term" limit=10 tag="[]" score=0.1 copy=true |
 
-| 通用 | tags | |@sen
+@sen
+| tags | stat |  返回特定tag信息 |
+| tags | list |  返回所有tag列表 |
 | crud | upload/download | 其他文件                                                   |
-| stat/list | stat | path |
-| stat/list | list | path |
+| file | stat | path |
+| file | list | path |
 | property | property:read | |
 | property | property:update | path="My Note" status=done xx=xxx |
 | property | property:delete | keys="[xxxx, xxxx]"                                               |
-| link | walk | path="My Note"  directtion=forward/backward depth=1 predicat=xxx |                                                   
+| graph | traverse | path="My Note"  directtion=forward/backward depth=1 predicat=xxx |                                                   
 
 @wangce
 | crud | create | path="New Note" content="# Hello" title="xxx" tags="[]" status="" |
diff --git a/reme2/memory/agent_toolkit.py b/reme2/memory/agent_toolkit.py
deleted file mode 100644
index fd3c121c..00000000
--- a/reme2/memory/agent_toolkit.py
+++ /dev/null
@@ -1,78 +0,0 @@
-"""Agent toolkit — composition entry point for the agent's tool surface.
-
-The actual tool implementations are split by category across four
-modules, each cohesive and self-contained:
-
-    memory_toolkit.py   memory_get / memory_create / memory_update_body /
-                        memory_update_meta / memory_search
-                        + memory_graph_search (MCP-only)
-                        + schema policy helpers (path template, status
-                          state machine, create_with_schema, update_status)
-
-    file_toolkit.py     file_download / file_upload / file_delete /
-                        file_list / file_move
-                        + path resolution + session temp dir
-
-    graph_toolkit.py    graph_traverse + adjacency BFS
-
-    event_toolkit.py    event_open / event_complete
-                        — uses memory_toolkit's schema gates so the
-                        event index follows the memory schema
-
-This module just composes them into one ``Toolkit`` for hosts that
-want all 13 tools bound at once. Hosts that want a subset can import
-the per-category builders and pass through a shared ``Toolkit``.
-
-Lint tools (``check_dangling`` / ``check_orphans`` / ``check_collisions``
-/ ``check_schema``) live in ``lint_toolkit.py`` — separate factory,
-not bound here by default.
-"""
-
-from __future__ import annotations
-
-from agentscope.tool import Toolkit
-
-from ..component import R
-from ..enumeration import ComponentEnum
-from .event_toolkit import EVENT_TOOL_NAMES
-from .file_toolkit import FILE_TOOL_NAMES
-from .graph_toolkit import GRAPH_TOOL_NAMES
-from .memory_toolkit import MEMORY_TOOL_NAMES
-
-
-# Order: memory → file → graph → event. Within memory, search comes
-# last (retrieval is a separate concern from CRUD); the rest follow
-# their CRUD order. memory_graph_search stays out — it's MCP-only.
-AGENT_TOOL_NAMES: tuple[str, ...] = (
-    *MEMORY_TOOL_NAMES,
-    *FILE_TOOL_NAMES,
-    *GRAPH_TOOL_NAMES,
-    *EVENT_TOOL_NAMES,
-)
-
-
-def build_agent_toolkit(
-    app_context,
-    audit: list[dict] | None = None,
-    toolkit: Toolkit | None = None,
-) -> Toolkit:
-    """Bind every agent tool's method to an agentscope ``Toolkit``.
-
-    For each name in ``AGENT_TOOL_NAMES``, instantiates the registered
-    BaseStep against ``app_context``, attaches the shared ``audit``
-    list, and registers the same-named class method as a tool function.
-    agentscope introspects the method signature directly — no separate
-    JSON schema layer.
-    """
-    toolkit = toolkit or Toolkit()
-    for name in AGENT_TOOL_NAMES:
-        step_cls = R.get(ComponentEnum.STEP, name)
-        if step_cls is None:
-            continue
-        instance = step_cls(app_context=app_context)
-        instance.audit = audit  # type: ignore[attr-defined]
-        toolkit.register_tool_function(
-            getattr(instance, name),
-            namesake_strategy="override",
-        )
-    return toolkit
diff --git a/reme2/memory/event_toolkit.py b/reme2/memory/event_toolkit.py
index 7435d9ae..6f428f57 100644
--- a/reme2/memory/event_toolkit.py
+++ b/reme2/memory/event_toolkit.py
@@ -39,7 +39,7 @@ from ..component import R
 from ..component.base_step import BaseStep
 from .file_toolkit import resolve_vault_path
 from .memory_toolkit import create_with_schema, update_status
-from .runtime_response import _set_answer, _tool_response
+from ..steps.runtime_response import _set_answer, _tool_response
 
 
 # ===========================================================================
diff --git a/reme2/memory/file_toolkit.py b/reme2/memory/file_toolkit.py
deleted file mode 100644
index 332191a5..00000000
--- a/reme2/memory/file_toolkit.py
+++ /dev/null
@@ -1,248 +0,0 @@
-"""File category — type-agnostic vault transport + directory operations.
-
-Five tools:
-
-    file_download   copy vault file → session temp dir
-    file_upload     copy local file → vault path (any type)
-    file_delete     remove vault file (universal entry point)
-    file_list       enumerate vault files with optional filters
-    file_move       rename / relocate; optional inbound-link rewrite
-
-Session-scoped temp dir is lazy: created on first ``file_download``,
-auto-cleaned on process exit. Each download lands in a fresh
-sub-directory so concurrent agents can freely modify the temp file
-without trampling each other.
-"""
-
-from __future__ import annotations
-
-import mimetypes
-import shutil
-import tempfile
-from pathlib import Path
-
-from agentscope.tool import ToolResponse
-
-from . import memory_io
-from ..component import R
-from ..component.base_step import BaseStep
-from .runtime_response import _set_answer, _tool_response
-
-
-# ===========================================================================
-# Section 1 — Helpers (session temp dir + path resolution)
-# ===========================================================================
-
-
-_TEMP_ROOT: Path | None = None
-
-
-def _get_temp_root() -> Path:
-    """Lazy session-scoped temp dir. Auto-cleaned on process exit."""
-    global _TEMP_ROOT
-    if _TEMP_ROOT is None:
-        _TEMP_ROOT = Path(tempfile.mkdtemp(prefix="reme2-files-"))
-    return _TEMP_ROOT
-
-
-def resolve_vault_path(file_store, vault_path: str) -> Path:
-    """Compose the absolute on-disk path for a vault-relative entry.
-
-    Exposed (no underscore) so ``event_toolkit`` can reuse the same
-    path-resolution logic when computing event workspace locations.
-    """
-    working_dir = getattr(file_store, "working_dir", None) or "."
-    p = Path(vault_path)
-    if p.is_absolute():
-        return p.resolve()
-    return (Path(working_dir) / p).resolve()
-
-
-# ===========================================================================
-# Section 2 — File tools
-# ===========================================================================
-
-
-@R.register("file_download")
-class FileDownload(BaseStep):
-    """Copy a vault file to a session temp dir; return the local path."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        vault_path: str = self.context.get("vault_path", "") or ""
-        assert vault_path, "vault_path is required"
-        payload = self._download(vault_path)
-        self.context.response.success = "error" not in payload
-        _set_answer(self.context, payload)
-
-    async def file_download(self, vault_path: str) -> ToolResponse:
-        """Copy a vault file to session temp dir; return the local path."""
-        payload = self._download(vault_path)
-        ok = "error" not in payload
-        return _tool_response("file_download", ok, payload, audit=self.audit)
-
-    def _download(self, vault_path: str) -> dict:
-        src = resolve_vault_path(self.file_store, vault_path)
-        if not src.is_file():
-            return {"vault_path": vault_path, "error": "not found"}
-        dst_dir = Path(tempfile.mkdtemp(prefix="dl-", dir=_get_temp_root()))
-        dst = dst_dir / src.name
-        shutil.copy2(src, dst)
-        return {
-            "vault_path": vault_path,
-            "local_path": str(dst),
-            "size": dst.stat().st_size,
-        }
-
-
-@R.register("file_upload")
-class FileUpload(BaseStep):
-    """Copy a local file into the vault. Watcher / parser register the
-    FileNode asynchronously."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        local_path: str = self.context.get("local_path", "") or ""
-        vault_path: str = self.context.get("vault_path", "") or ""
-        overwrite: bool = bool(self.context.get("overwrite", True))
-        assert local_path and vault_path, "local_path and vault_path are required"
-        payload = self._upload(local_path, vault_path, overwrite)
-        self.context.response.success = "error" not in payload
-        _set_answer(self.context, payload)
-
-    async def file_upload(
-        self, local_path: str, vault_path: str, overwrite: bool = True,
-    ) -> ToolResponse:
-        """Copy local_path into the vault at vault_path."""
-        payload = self._upload(local_path, vault_path, overwrite)
-        ok = "error" not in payload
-        return _tool_response("file_upload", ok, payload, audit=self.audit)
-
-    def _upload(self, local_path: str, vault_path: str, overwrite: bool) -> dict:
-        src = Path(local_path)
-        if not src.is_file():
-            return {"local_path": local_path, "error": "source not found"}
-        dst = resolve_vault_path(self.file_store, vault_path)
-        if dst.exists() and not overwrite:
-            return {"vault_path": vault_path, "error": "destination exists; pass overwrite=True"}
-        dst.parent.mkdir(parents=True, exist_ok=True)
-        shutil.copy2(src, dst)
-        return {
-            "vault_path": vault_path,
-            "size": dst.stat().st_size,
-            "mime": mimetypes.guess_type(dst.name)[0] or "application/octet-stream",
-        }
-
-
-@R.register("file_delete")
-class FileDelete(BaseStep):
-    """Delete a vault file. Universal entry point for any file type."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        vault_path: str = self.context.get("vault_path", "") or ""
-        assert vault_path, "vault_path is required"
-        target = resolve_vault_path(self.file_store, vault_path)
-        ok, payload = memory_io.delete_file(target)
-        self.context.response.success = ok
-        _set_answer(self.context, payload)
-
-    async def file_delete(self, vault_path: str) -> ToolResponse:
-        """Delete a vault file."""
-        target = resolve_vault_path(self.file_store, vault_path)
-        ok, payload = memory_io.delete_file(target)
-        return _tool_response("file_delete", ok, payload, audit=self.audit)
-
-
-@R.register("file_list")
-class FileList(BaseStep):
-    """Enumerate vault files with optional frontmatter filters."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        result = memory_io.list_files(
-            self.file_store,
-            path_prefix=self.context.get("prefix") or self.context.get("path_prefix"),
-            tags=self.context.get("tags") or [],
-            metadata=self.context.get("metadata") or {},
-            limit=int(self.context.get("limit") or 100),
-        )
-        _set_answer(self.context, result)
-
-    async def file_list(
-        self,
-        prefix: str | None = None,
-        tags: list[str] | None = None,
-        metadata: dict | None = None,
-        limit: int = 100,
-    ) -> ToolResponse:
-        """List vault files. Filters: path prefix, frontmatter tags / fields."""
-        result = memory_io.list_files(
-            self.file_store,
-            path_prefix=prefix,
-            tags=tags or [],
-            metadata=metadata or {},
-            limit=limit,
-        )
-        return _tool_response("file_list", True, result, audit=self.audit)
-
-
-@R.register("file_move")
-class FileMove(BaseStep):
-    """Rename / relocate. Default leaves inbound wikilinks untouched
-    (maintainer cleans dangling refs); pass ``update_refs=True`` to
-    rewrite ``[[old]] → [[new]]`` in every referencing file."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        src: str = self.context.get("src") or self.context.get("old_path") or ""
-        dst: str = self.context.get("dst") or self.context.get("new_path") or ""
-        update_refs: bool = bool(self.context.get("update_refs", False))
-        assert src and dst, "src and dst are required"
-        payload = self._move(src, dst, update_refs)
-        self.context.response.success = payload.get("ok", False)
-        _set_answer(self.context, payload)
-
-    async def file_move(
-        self, src: str, dst: str, update_refs: bool = False,
-    ) -> ToolResponse:
-        """Rename / relocate. update_refs=True rewrites [[old]] → [[new]]."""
-        payload = self._move(src, dst, update_refs)
-        ok = payload.get("ok", False)
-        return _tool_response("file_move", ok, payload, audit=self.audit)
-
-    def _move(self, src: str, dst: str, update_refs: bool) -> dict:
-        src_abs = resolve_vault_path(self.file_store, src)
-        dst_abs = resolve_vault_path(self.file_store, dst)
-        if not src_abs.is_file():
-            return {"ok": False, "src": src, "error": "source not found"}
-        if update_refs:
-            working_dir = Path(getattr(self.file_store, "working_dir", None) or ".").resolve()
-            ok, payload = memory_io.rename_file(
-                self.file_store, working_dir,
-                old_path=src_abs, new_path=dst_abs,
-            )
-            payload["ok"] = ok
-            return payload
-        dst_abs.parent.mkdir(parents=True, exist_ok=True)
-        shutil.move(str(src_abs), str(dst_abs))
-        return {"ok": True, "src": str(src_abs), "dst": str(dst_abs), "refs_updated": 0}
-
-
-FILE_TOOL_NAMES: tuple[str, ...] = (
-    "file_download",
-    "file_upload",
-    "file_delete",
-    "file_list",
-    "file_move",
-)
diff --git a/reme2/memory/graph_toolkit.py b/reme2/memory/graph_toolkit.py
deleted file mode 100644
index 15522338..00000000
--- a/reme2/memory/graph_toolkit.py
+++ /dev/null
@@ -1,148 +0,0 @@
-"""Graph category — relationship exploration via BFS.
-
-Single tool: ``graph_traverse``. depth=1 covers the trivial outlinks /
-inlinks lookups (just set direction); multi-hop covers exploration.
-Output is one record per edge traversed (not per node), so the same
-node may appear multiple times if reached via different predicates or
-paths — agents dedupe at the call site if they want a flat node set.
-
-Adjacency source: walks ``file_store.file_nodes`` directly (each
-FileNode carries its ``links`` list). Inbound is a linear scan over
-all nodes; cheap for vault sizes. Swap for a precomputed reverse
-index on the file_graph component if it ever becomes a hot path.
-"""
-
-from __future__ import annotations
-
-from collections import deque
-
-from agentscope.tool import ToolResponse
-
-from ..component import R
-from ..component.base_step import BaseStep
-from .runtime_response import _set_answer, _tool_response
-
-
-# ===========================================================================
-# Section 1 — Adjacency lookups
-# ===========================================================================
-
-
-def _outlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]:
-    """Outgoing edges from ``path`` — [(target_path, predicate, anchor)]."""
-    node = file_store.file_nodes.get(path)
-    if node is None:
-        return []
-    return [(link.path, link.predicate, link.anchor) for link in node.links if link.path]
-
-
-def _inlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]:
-    """Incoming edges to ``path`` — linear scan over all nodes' links."""
-    out: list[tuple[str, str | None, str | None]] = []
-    for src_path, src_node in file_store.file_nodes.items():
-        if src_path == path:
-            continue
-        for link in src_node.links:
-            if link.path == path:
-                out.append((src_path, link.predicate, link.anchor))
-    return out
-
-
-def _bfs(
-    file_store,
-    seeds: list[str],
-    max_depth: int,
-    direction: str,
-    predicate: str | None,
-) -> list[dict]:
-    """BFS from each seed. One record per edge traversed."""
-    visited_edges: set[tuple[str, str, str | None]] = set()
-    results: list[dict] = []
-    queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds)
-    while queue:
-        current, depth = queue.popleft()
-        if depth >= max_depth:
-            continue
-        edges: list[tuple[str, str | None, str | None]] = []
-        if direction in ("out", "both"):
-            for tgt, pred, anchor in _outlinks(file_store, current):
-                if predicate is not None and pred != predicate:
-                    continue
-                edges.append((tgt, pred, anchor))
-        if direction in ("in", "both"):
-            for src, pred, anchor in _inlinks(file_store, current):
-                if predicate is not None and pred != predicate:
-                    continue
-                edges.append((src, pred, anchor))
-        for next_path, pred, anchor in edges:
-            edge_key = (current, next_path, pred)
-            if edge_key in visited_edges:
-                continue
-            visited_edges.add(edge_key)
-            results.append({
-                "path": next_path,
-                "depth": depth + 1,
-                "via": current,
-                "predicate": pred,
-                "anchor": anchor,
-            })
-            if depth + 1 < max_depth:
-                queue.append((next_path, depth + 1))
-    return results
-
-
-# ===========================================================================
-# Section 2 — Graph tool
-# ===========================================================================
-
-
-@R.register("graph_traverse")
-class GraphTraverse(BaseStep):
-    """BFS from seed(s) to explore relationships in the memory graph."""
-
-    audit: list[dict] | None = None
-
-    async def execute(self):
-        assert self.context is not None
-        seeds_raw = self.context.get("seeds") or []
-        if isinstance(seeds_raw, str):
-            seeds = [seeds_raw]
-        else:
-            seeds = list(seeds_raw)
-        max_depth: int = int(self.context.get("max_depth") or 1)
-        direction: str = self.context.get("direction", "out") or "out"
-        predicate = self.context.get("predicate")
-        assert seeds, "seeds is required (single path or list of paths)"
-        assert direction in ("out", "in", "both"), \
-            f"direction must be 'out' | 'in' | 'both', got {direction!r}"
-        results = _bfs(self.file_store, seeds, max_depth, direction, predicate)
-        _set_answer(self.context, results)
-
-    async def graph_traverse(
-        self,
-        seeds: str | list[str],
-        max_depth: int = 1,
-        direction: str = "out",
-        predicate: str | None = None,
-    ) -> ToolResponse:
-        """BFS from seed(s) to explore relationships.
-
-        Args:
-            seeds: single path or list of paths to start from.
-            max_depth: hops to expand (default 1 = immediate neighbors).
-            direction: "out" / "in" / "both".
-            predicate: filter edges by predicate (None = no filter).
-        """
-        if isinstance(seeds, str):
-            seeds_list = [seeds]
-        else:
-            seeds_list = list(seeds)
-        assert direction in ("out", "in", "both"), \
-            f"direction must be 'out' | 'in' | 'both', got {direction!r}"
-        results = _bfs(self.file_store, seeds_list, max_depth, direction, predicate)
-        return _tool_response("graph_traverse", True, results, audit=self.audit)
-
-
-GRAPH_TOOL_NAMES: tuple[str, ...] = (
-    "graph_traverse",
-)
diff --git a/reme2/memory/ingestor.py b/reme2/memory/ingestor.py
index 166c0c3d..11e47353 100644
--- a/reme2/memory/ingestor.py
+++ b/reme2/memory/ingestor.py
@@ -28,7 +28,7 @@ from agentscope.message import Msg
 from agentscope.tool import Toolkit
 from pydantic import BaseModel, Field
 
-from .runtime_response import _set_answer, _to_jsonable
+from ..steps.runtime_response import _set_answer, _to_jsonable
 from . import memory_io
 from .memory_io import create_file
 from .agent_toolkit import build_agent_toolkit
diff --git a/reme2/memory/lint_toolkit.py b/reme2/memory/lint_toolkit.py
index cfa2b057..dbcca658 100644
--- a/reme2/memory/lint_toolkit.py
+++ b/reme2/memory/lint_toolkit.py
@@ -28,7 +28,7 @@ from agentscope.tool import Toolkit, ToolResponse
 
 from ..component import R
 from ..component.base_step import BaseStep
-from .runtime_response import _set_answer, _tool_response
+from ..steps.runtime_response import _set_answer, _tool_response
 from ..enumeration import ComponentEnum
 
 
diff --git a/reme2/memory/maintainer.py b/reme2/memory/maintainer.py
index f87db16f..da53e5f6 100644
--- a/reme2/memory/maintainer.py
+++ b/reme2/memory/maintainer.py
@@ -69,7 +69,7 @@ from pydantic import BaseModel, Field
 
 from ..component import R
 from ..component.base_step import BaseStep
-from .runtime_response import _set_answer
+from ..steps.runtime_response import _set_answer
 from . import memory_io
 from .schema import parse_frontmatter
 
diff --git a/reme2/memory/memory_toolkit.py b/reme2/memory/memory_toolkit.py
index 99943247..c9fe7bf5 100644
--- a/reme2/memory/memory_toolkit.py
+++ b/reme2/memory/memory_toolkit.py
@@ -29,7 +29,7 @@ from ..component import R
 from ..component.base_step import BaseStep
 from ..enumeration import ComponentEnum
 from .retriever import BaseRetriever, HybridRetriever
-from .runtime_response import _set_answer, _tool_response
+from ..steps.runtime_response import _set_answer, _tool_response
 
 
 # ===========================================================================
diff --git a/reme2/steps/crud/download.py b/reme2/steps/crud/download.py
new file mode 100644
index 00000000..a79e9eb1
--- /dev/null
+++ b/reme2/steps/crud/download.py
@@ -0,0 +1,89 @@
+"""``file_download`` — copy a vault file to a session temp dir.
+
+Agent flow: agent calls ``file_download(path)``, gets back a
+local path under a fresh per-call temp directory, then opens / parses
+the file with whatever tooling it likes. The vault copy is untouched.
+
+The temp root is lazy and session-scoped — created on first download,
+left for the OS to clean up at process exit. Each download lands in
+its own subdirectory so concurrent agents don't trample each other.
+
+Also exports ``resolve_path`` — the shared helper for turning
+a vault-relative or absolute path into an absolute on-disk path.
+``upload`` and ``list`` import it from here.
+"""
+
+from __future__ import annotations
+
+import shutil
+import tempfile
+from pathlib import Path
+
+from agentscope.tool import ToolResponse
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+_TEMP_ROOT: Path | None = None
+
+
+def _get_temp_root() -> Path:
+    """Lazy session-scoped temp dir. Auto-cleaned on process exit."""
+    global _TEMP_ROOT
+    if _TEMP_ROOT is None:
+        _TEMP_ROOT = Path(tempfile.mkdtemp(prefix="reme2-files-"))
+    return _TEMP_ROOT
+
+
+def resolve_path(file_store, path: str) -> Path:
+    """Compose the absolute on-disk path for relative entry.
+
+    Public so sibling steps (``upload``, ``list``, event tools) can
+    reuse the same path-resolution rule. Absolute paths pass through;
+    relative paths join under ``file_store.working_dir``.
+    """
+    working_dir = getattr(file_store, "working_dir", None) or "."
+    p = Path(path)
+    if p.is_absolute():
+        return p.resolve()
+    return (Path(working_dir) / p).resolve()
+
+
+@R.register("file_download")
+class FileDownload(BaseStep):
+    """Copy a vault file to a session temp dir; return the local path."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path: str = self.context.get("path", "") or ""
+        assert path, "path is required"
+        payload = self._download(path)
+        self.context.response.success = "error" not in payload
+        _set_answer(self.context, payload)
+
+    async def file_download(self, path: str) -> ToolResponse:
+        """Copy a vault file to session temp dir; return the local path."""
+        payload = self._download(path)
+        ok = "error" not in payload
+        return _tool_response("file_download", ok, payload, audit=self.audit)
+
+    def _download(self, path: str) -> dict:
+        src = resolve_path(self.file_store, path)
+        if not src.is_file():
+            return {"path": path, "error": "not found"}
+        dst_dir = Path(tempfile.mkdtemp(prefix="dl-", dir=_get_temp_root()))
+        dst = dst_dir / src.name
+        shutil.copy2(src, dst)
+        return {
+            "path": path,
+            "local_path": str(dst),
+            "size": dst.stat().st_size,
+        }
diff --git a/reme2/steps/crud/list.py b/reme2/steps/crud/list.py
new file mode 100644
index 00000000..a6e920f9
--- /dev/null
+++ b/reme2/steps/crud/list.py
@@ -0,0 +1,101 @@
+"""``file_list`` — enumerate vault files with optional frontmatter filters.
+
+Walks the file_graph in one shot via ``get_nodes(None)`` — no filesystem
+scan, no per-file frontmatter parse, no per-file graph round-trip.
+
+Filters compose:
+    path_prefix    — prefix match against the indexed path string
+    tags           — every requested tag must be present in ``tags``
+    metadata       — frontmatter must equal each ``{key: value}`` pair
+
+The ``metadata`` filter sees the full frontmatter dict — typed fields
+(title / description / tags) merged with any ``extra=allow`` extras —
+so agents can filter on schema-known keys or arbitrary extras alike.
+"""
+
+from __future__ import annotations
+
+from agentscope.tool import ToolResponse
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+def _matches(
+    path: str,
+    md: dict,
+    *,
+    path_prefix: str | None,
+    tags: list[str],
+    metadata: dict,
+) -> bool:
+    if path_prefix and not path.startswith(path_prefix):
+        return False
+    if metadata and any(md.get(k) != v for k, v in metadata.items()):
+        return False
+    if tags:
+        file_tags = set(md.get("tags") or [])
+        if not all(t in file_tags for t in tags):
+            return False
+    return True
+
+
+async def _list(
+    file_store,
+    *,
+    path_prefix: str | None,
+    tags: list[str],
+    metadata: dict,
+    limit: int,
+) -> dict:
+    if not file_store.file_graph:
+        return {"items": [], "count": 0}
+    items: list[dict] = []
+    for node in await file_store.file_graph.get_nodes():
+        md = node.front_matter.model_dump()
+        if not _matches(node.path, md, path_prefix=path_prefix, tags=tags, metadata=metadata):
+            continue
+        items.append({"path": node.path, "metadata": md})
+        if len(items) >= limit:
+            break
+    return {"items": items, "count": len(items)}
+
+
+@R.register("file_list")
+class FileList(BaseStep):
+    """Enumerate vault files with optional frontmatter filters."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        result = await _list(
+            self.file_store,
+            path_prefix=self.context.get("prefix") or self.context.get("path_prefix"),
+            tags=self.context.get("tags") or [],
+            metadata=self.context.get("metadata") or {},
+            limit=int(self.context.get("limit") or 100),
+        )
+        _set_answer(self.context, result)
+
+    async def file_list(
+        self,
+        prefix: str | None = None,
+        tags: list[str] | None = None,
+        metadata: dict | None = None,
+        limit: int = 100,
+    ) -> ToolResponse:
+        """List vault files. Filters: path prefix, frontmatter tags / fields."""
+        result = await _list(
+            self.file_store,
+            path_prefix=prefix,
+            tags=tags or [],
+            metadata=metadata or {},
+            limit=limit,
+        )
+        return _tool_response("file_list", True, result, audit=self.audit)
diff --git a/reme2/steps/crud/property/__init__.py b/reme2/steps/crud/property/__init__.py
new file mode 100644
index 00000000..4c5db21b
--- /dev/null
+++ b/reme2/steps/crud/property/__init__.py
@@ -0,0 +1,17 @@
+"""Property steps — frontmatter-only CRUD on memory files.
+
+Three Steps:
+
+    property:read    — return the frontmatter dict
+    property:update  — merge a patch into the frontmatter
+    property:delete  — drop the listed keys
+
+Body content stays untouched; use the ``crud`` siblings (``read``,
+``edit``, ``append``, ``prepend``) for body-level operations. Each
+Step here is a pure disk read-modify-write — the watcher / parser
+notices the change and refreshes the projections asynchronously.
+"""
+
+from . import read  # noqa: F401  -- @R.register("property:read")
+from . import update  # noqa: F401  -- @R.register("property:update")
+from . import delete  # noqa: F401  -- @R.register("property:delete")
diff --git a/reme2/steps/crud/property/delete.py b/reme2/steps/crud/property/delete.py
new file mode 100644
index 00000000..c1fc9c6c
--- /dev/null
+++ b/reme2/steps/crud/property/delete.py
@@ -0,0 +1,79 @@
+"""``property:delete`` — remove keys from a memory file's frontmatter.
+
+Returns both ``deleted`` (keys that were present and removed) and
+``missing`` (keys that weren't there) so the agent can tell whether
+a no-op happened. The file is rewritten only when at least one key
+is actually removed — calling delete with all-missing keys is a
+zero-side-effect read.
+"""
+
+from __future__ import annotations
+
+import frontmatter
+from agentscope.tool import ToolResponse
+
+from ..download import resolve_path
+
+from ...base_step import BaseStep
+from ...runtime_response import _set_answer, _tool_response
+
+from ....component import R
+from ....enumeration import ComponentEnum
+
+
+def _delete(file_store, path: str, keys: list[str]) -> dict:
+    target = resolve_path(file_store, path)
+    if not target.is_file():
+        return {"path": path, "error": "not found"}
+    if not keys:
+        return {"path": path, "error": "keys is empty"}
+    raw = target.read_text(encoding="utf-8")
+    post = frontmatter.loads(raw)
+    deleted: list[str] = []
+    missing: list[str] = []
+    for k in keys:
+        if k in post.metadata:
+            del post.metadata[k]
+            deleted.append(k)
+        else:
+            missing.append(k)
+    if deleted:
+        target.write_text(frontmatter.dumps(post), encoding="utf-8")
+    return {
+        "path": path,
+        "deleted": deleted,
+        "missing": missing,
+        "frontmatter": dict(post.metadata),
+    }
+
+
+@R.register("property:delete")
+class PropertyDelete(BaseStep):
+    """Remove keys from a memory file's frontmatter."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path: str = self.context.get("path") or self.context.get("path") or ""
+        assert path, "path is required"
+        keys = self.context.get("keys") or []
+        if isinstance(keys, str):
+            keys = [keys]
+        payload = _delete(self.file_store, path, list(keys))
+        self.context.response.success = "error" not in payload
+        _set_answer(self.context, payload)
+
+    async def property_delete(
+        self,
+        path: str,
+        keys: list[str] | str,
+    ) -> ToolResponse:
+        """Remove the listed ``keys`` from the frontmatter at ``path``."""
+        if isinstance(keys, str):
+            keys = [keys]
+        payload = _delete(self.file_store, path, list(keys))
+        ok = "error" not in payload
+        return _tool_response("property:delete", ok, payload, audit=self.audit)
diff --git a/reme2/steps/crud/property/read.py b/reme2/steps/crud/property/read.py
new file mode 100644
index 00000000..57105e23
--- /dev/null
+++ b/reme2/steps/crud/property/read.py
@@ -0,0 +1,51 @@
+"""``property:read`` — return the frontmatter dict of a memory file.
+
+Cheap structured read — frontmatter only, no body. Use ``crud:read``
+when you need the body too. Returns ``{exists: false}`` when the
+target doesn't exist; otherwise ``{exists: true, frontmatter: {...}}``.
+"""
+
+from __future__ import annotations
+
+import frontmatter
+from agentscope.tool import ToolResponse
+
+from ..download import resolve_path
+
+from ...base_step import BaseStep
+from ...runtime_response import _set_answer, _tool_response
+
+from ....component import R
+from ....enumeration import ComponentEnum
+
+
+def _read(file_store, path: str) -> dict:
+    target = resolve_path(file_store, path)
+    if not target.is_file():
+        return {"path": path, "exists": False}
+    raw = target.read_text(encoding="utf-8")
+    meta = dict(frontmatter.loads(raw).metadata)
+    return {"path": path, "exists": True, "frontmatter": meta}
+
+
+@R.register("property:read")
+class PropertyRead(BaseStep):
+    """Read a memory file's frontmatter (YAML metadata only)."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path: str = self.context.get("path") or self.context.get("path") or ""
+        assert path, "path is required"
+        payload = _read(self.file_store, path)
+        self.context.response.success = payload.get("exists", False)
+        _set_answer(self.context, payload)
+
+    async def property_read(self, path: str) -> ToolResponse:
+        """Return the frontmatter dict of the memory file at ``path``."""
+        payload = _read(self.file_store, path)
+        ok = payload.get("exists", False)
+        return _tool_response("property:read", ok, payload, audit=self.audit)
diff --git a/reme2/steps/crud/property/update.py b/reme2/steps/crud/property/update.py
new file mode 100644
index 00000000..b85bd322
--- /dev/null
+++ b/reme2/steps/crud/property/update.py
@@ -0,0 +1,93 @@
+"""``property:update`` — merge a patch into a memory file's frontmatter.
+
+Patch semantics are merge-write: keys in ``patch`` overwrite existing
+keys, keys not in ``patch`` are left untouched. To remove a key,
+use ``property:delete`` — passing ``None`` here just sets the literal
+None (which is rarely what you want).
+
+Per the obsidian config convention the agent calls this with
+arbitrary keyword arguments — ``path="My Note" status=done xx=xxx``.
+The Step packs everything except ``path`` into the patch dict, so
+the tool method accepts ``**fields`` directly.
+"""
+
+from __future__ import annotations
+
+import frontmatter
+from agentscope.tool import ToolResponse
+
+from ..download import resolve_path
+
+from ...base_step import BaseStep
+from ...runtime_response import _set_answer, _tool_response
+
+from ....component import R
+from ....enumeration import ComponentEnum
+
+
+# Context keys that belong to plumbing (request envelope, response
+# slot, the path itself) and must never be promoted into a free-form
+# frontmatter patch.
+_RESERVED_CONTEXT_KEYS = {
+    "path", "path", "patch", "response", "request", "data",
+}
+
+
+def _update(file_store, path: str, patch: dict) -> dict:
+    target = resolve_path(file_store, path)
+    if not target.is_file():
+        return {"path": path, "error": "not found"}
+    if not patch:
+        return {"path": path, "error": "patch is empty"}
+    raw = target.read_text(encoding="utf-8")
+    post = frontmatter.loads(raw)
+    post.metadata.update(patch)
+    target.write_text(frontmatter.dumps(post), encoding="utf-8")
+    return {
+        "path": path,
+        "applied": dict(patch),
+        "frontmatter": dict(post.metadata),
+    }
+
+
+@R.register("property:update")
+class PropertyUpdate(BaseStep):
+    """Merge a patch into a memory file's frontmatter."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path: str = self.context.get("path") or self.context.get("path") or ""
+        assert path, "path is required"
+        patch = self.context.get("patch")
+        if patch is None:
+            # Free-form mode — every other context key is treated as a
+            # patch entry, matching the obsidian convention
+            # `path=... key=val key=val`.
+            patch = {
+                k: v for k, v in self.context.data.items()
+                if k not in _RESERVED_CONTEXT_KEYS
+            }
+        payload = _update(self.file_store, path, dict(patch or {}))
+        self.context.response.success = "error" not in payload
+        _set_answer(self.context, payload)
+
+    async def property_update(
+        self,
+        path: str,
+        patch: dict | None = None,
+        **fields,
+    ) -> ToolResponse:
+        """Merge ``patch`` (or free-form ``**fields``) into the
+        frontmatter at ``path``."""
+        merged: dict = {}
+        if patch:
+            merged.update(patch)
+        if fields:
+            merged.update(fields)
+        payload = _update(self.file_store, path, merged)
+        ok = "error" not in payload
+        return _tool_response("property:update", ok, payload, audit=self.audit)
diff --git a/reme2/steps/crud/stat.py b/reme2/steps/crud/stat.py
new file mode 100644
index 00000000..d27cfc10
--- /dev/null
+++ b/reme2/steps/crud/stat.py
@@ -0,0 +1,94 @@
+"""``file_stat`` — peek at vault file metadata without copying it.
+
+Cheap inspection alternative to ``file_download``: the agent gets
+size, mtime, mime type, and (for markdown files) the parsed
+frontmatter — enough to decide whether to download / parse / skip
+without paying the copy cost.
+
+Returns a uniform envelope:
+
+    {exists, type, size, mtime, ctime, mime, frontmatter}
+
+``exists=False`` short-circuits everything else to ``None``. ``type``
+is ``"file"`` / ``"dir"`` (covers event workspace probes too).
+``frontmatter`` is populated only for ``.md`` files and only when
+parsing succeeds — schema validity is a lint concern.
+"""
+
+from __future__ import annotations
+
+import mimetypes
+from datetime import datetime
+from pathlib import Path
+
+import frontmatter
+from agentscope.tool import ToolResponse
+
+from .download import resolve_path
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+def _iso(ts: float) -> str:
+    """Filesystem timestamp → ISO 8601 string."""
+    return datetime.fromtimestamp(ts).isoformat()
+
+
+def _stat(file_store, path: str) -> dict:
+    target = resolve_path(file_store, path)
+    if not target.exists():
+        return {"path": path, "exists": False}
+
+    st = target.stat()
+    out: dict = {
+        "path": path,
+        "path": str(target),
+        "exists": True,
+        "type": "dir" if target.is_dir() else "file",
+        "mtime": _iso(st.st_mtime),
+        "ctime": _iso(st.st_ctime),
+    }
+
+    if target.is_file():
+        out["size"] = st.st_size
+        out["mime"] = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
+        if target.suffix == ".md":
+            try:
+                meta = dict(frontmatter.loads(target.read_text(encoding="utf-8")).metadata)
+            except Exception:
+                meta = {}
+            out["frontmatter"] = meta
+
+    return out
+
+
+@R.register("file_stat")
+class FileStat(BaseStep):
+    """Return metadata for a vault file or directory."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path: str = self.context.get("path", "") or ""
+        assert path, "path is required"
+        payload = _stat(self.file_store, path)
+        self.context.response.success = payload.get("exists", False)
+        _set_answer(self.context, payload)
+
+    async def file_stat(self, path: str) -> ToolResponse:
+        """Return metadata for a vault file or directory.
+
+        Returns ``{exists, type, size, mtime, ctime, mime, frontmatter}``.
+        ``size`` / ``mime`` / ``frontmatter`` are file-only;
+        ``frontmatter`` is markdown-only.
+        """
+        payload = _stat(self.file_store, path)
+        ok = payload.get("exists", False)
+        return _tool_response("file_stat", ok, payload, audit=self.audit)
diff --git a/reme2/steps/crud/upload.py b/reme2/steps/crud/upload.py
new file mode 100644
index 00000000..551fc45d
--- /dev/null
+++ b/reme2/steps/crud/upload.py
@@ -0,0 +1,72 @@
+"""``file_upload`` — copy a local file into the vault.
+
+Agent flow: agent has a file at some local path (a download result, a
+freshly generated artifact, a user-supplied attachment), and wants it
+to live inside the vault at a known location. This is type-agnostic
+— text, binary, anything — so it's the entry point for non-markdown
+materials too.
+
+The watcher / parser pick up the new file asynchronously; this step
+just performs the copy. ``overwrite=True`` is the default since most
+upload flows are intentional replacements (re-uploading the same
+material after re-generation).
+"""
+
+from __future__ import annotations
+
+import mimetypes
+import shutil
+from pathlib import Path
+
+from agentscope.tool import ToolResponse
+
+from .download import resolve_path
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+@R.register("file_upload")
+class FileUpload(BaseStep):
+    """Copy a local file into the vault. Watcher / parser register the
+    FileNode asynchronously."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        local_path: str = self.context.get("local_path", "") or ""
+        path: str = self.context.get("path", "") or ""
+        overwrite: bool = bool(self.context.get("overwrite", True))
+        assert local_path and path, "local_path and path are required"
+        payload = self._upload(local_path, path, overwrite)
+        self.context.response.success = "error" not in payload
+        _set_answer(self.context, payload)
+
+    async def file_upload(
+        self, local_path: str, path: str, overwrite: bool = True,
+    ) -> ToolResponse:
+        """Copy ``local_path`` into the vault at ``path``."""
+        payload = self._upload(local_path, path, overwrite)
+        ok = "error" not in payload
+        return _tool_response("file_upload", ok, payload, audit=self.audit)
+
+    def _upload(self, local_path: str, path: str, overwrite: bool) -> dict:
+        src = Path(local_path)
+        if not src.is_file():
+            return {"local_path": local_path, "error": "source not found"}
+        dst = resolve_path(self.file_store, path)
+        if dst.exists() and not overwrite:
+            return {"path": path, "error": "destination exists; pass overwrite=True"}
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(src, dst)
+        return {
+            "path": path,
+            "size": dst.stat().st_size,
+            "mime": mimetypes.guess_type(dst.name)[0] or "application/octet-stream",
+        }
diff --git a/reme2/steps/graph/__init__.py b/reme2/steps/graph/__init__.py
new file mode 100644
index 00000000..f04f0f80
--- /dev/null
+++ b/reme2/steps/graph/__init__.py
@@ -0,0 +1,15 @@
+"""Graph steps — wikilink relationship exploration.
+
+Single Step:
+
+    graph:traverse  — BFS from a seed file over wikilink edges.
+
+Direction vocabulary matches the obsidian convention
+(``forward`` / ``backward`` / ``both``); the engine vocab
+(``out`` / ``in`` / ``both``) is accepted as alias. Inbound
+traversal walks the vault to reconstruct source paths since the
+file_graph contract's ``get_inlinks`` only returns target-shaped
+FileLinks.
+"""
+
+from . import traverse  # noqa: F401  -- @R.register("graph:traverse")
diff --git a/reme2/steps/graph/traverse.py b/reme2/steps/graph/traverse.py
new file mode 100644
index 00000000..95ce3f20
--- /dev/null
+++ b/reme2/steps/graph/traverse.py
@@ -0,0 +1,190 @@
+"""``graph:traverse`` — BFS over wikilink edges from a seed file.
+
+Single tool for relationship browsing. ``depth=1`` covers the trivial
+"what does this link to / what links here" lookups (set ``direction``
+accordingly); higher depth opens up multi-hop exploration.
+
+Output is one record per edge traversed (not per node), so the same
+target can appear multiple times if reached via different predicates
+or paths — agents dedupe at the call site if they want a flat node
+set. Each record carries ``via`` (the predecessor) and the link's
+``predicate`` / ``anchor`` so the agent can reconstruct the path.
+
+Adjacency is loaded once via ``file_graph.get_nodes(None)`` — every
+real node arrives with its full ``links`` payload, and we build both
+the outbound and the inbound index in a single pass. The BFS then
+runs purely in memory: no per-frontier-node graph round-trips, no
+filesystem walk. The ``get_inlinks`` / ``get_outlinks`` contract
+methods stay unused here because they'd add network round-trips for
+data we already have.
+
+Direction vocabulary accepts both the obsidian convention
+(``forward`` / ``backward`` / ``both``) and the engine convention
+(``out`` / ``in`` / ``both``).
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from pathlib import Path
+
+from agentscope.tool import ToolResponse
+
+from ..crud.download import resolve_path
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+from ...schema import FileLink
+
+
+_FORWARD = {"out", "forward"}
+_BACKWARD = {"in", "backward"}
+_BOTH = {"both"}
+_VALID_DIRECTIONS = _FORWARD | _BACKWARD | _BOTH
+
+
+async def _build_indexes(
+    file_store,
+) -> tuple[
+    dict[str, list[tuple[str, FileLink]]],
+    dict[str, list[tuple[str, FileLink]]],
+]:
+    """One ``get_nodes(None)`` call → (outbound, inbound) adjacency dicts.
+
+    Each dict is keyed by node path; values are ``(neighbor_path, link)``
+    tuples. Source paths land in the inbound index alongside the link
+    object — solving the contract gap where ``get_inlinks`` returns
+    target-shaped FileLinks without source attribution.
+    """
+    outbound: dict[str, list[tuple[str, FileLink]]] = {}
+    inbound: dict[str, list[tuple[str, FileLink]]] = {}
+    if not file_store.file_graph:
+        return outbound, inbound
+    for node in await file_store.file_graph.get_nodes():
+        for link in node.links:
+            if not link.path:
+                continue
+            outbound.setdefault(node.path, []).append((link.path, link))
+            inbound.setdefault(link.path, []).append((node.path, link))
+    return outbound, inbound
+
+
+def _bfs(
+    seeds: list[str],
+    max_depth: int,
+    direction: str,
+    predicate: str | None,
+    outbound: dict[str, list[tuple[str, FileLink]]],
+    inbound: dict[str, list[tuple[str, FileLink]]],
+) -> list[dict]:
+    """In-memory BFS. One record per edge traversed."""
+    walk_out = direction in _FORWARD or direction in _BOTH
+    walk_in = direction in _BACKWARD or direction in _BOTH
+
+    visited_edges: set[tuple[str, str, str | None]] = set()
+    results: list[dict] = []
+    queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds)
+
+    while queue:
+        current, depth = queue.popleft()
+        if depth >= max_depth:
+            continue
+
+        edges: list[tuple[str, str | None, str | None]] = []
+        if walk_out:
+            for tgt, link in outbound.get(current, ()):
+                if predicate is not None and link.predicate != predicate:
+                    continue
+                edges.append((tgt, link.predicate, link.anchor))
+        if walk_in:
+            for src, link in inbound.get(current, ()):
+                if predicate is not None and link.predicate != predicate:
+                    continue
+                edges.append((src, link.predicate, link.anchor))
+
+        for next_path, pred, anchor in edges:
+            edge_key = (current, next_path, pred)
+            if edge_key in visited_edges:
+                continue
+            visited_edges.add(edge_key)
+            results.append({
+                "path": next_path,
+                "depth": depth + 1,
+                "via": current,
+                "predicate": pred,
+                "anchor": anchor,
+            })
+            if depth + 1 < max_depth:
+                queue.append((next_path, depth + 1))
+
+    return results
+
+
+def _normalize_seeds(file_store, raw) -> list[str]:
+    """Coerce a single path or list of paths into resolved absolute strings."""
+    if isinstance(raw, (str, Path)):
+        items = [raw]
+    else:
+        items = list(raw or [])
+    return [str(resolve_path(file_store, str(p))) for p in items if p]
+
+
+@R.register("graph:traverse")
+class GraphTraverse(BaseStep):
+    """BFS from a seed file to explore wikilink relationships.
+
+    Parameters (per obsidian convention):
+        path       — single seed (str). ``seeds`` accepted as alias for batch mode.
+        direction  — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``).
+        depth      — hop limit (default 1 = immediate neighbors).
+        predicate  — optional edge-type filter; ``None`` = no filter.
+    """
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        path = self.context.get("path")
+        seeds_raw = self.context.get("seeds") if path is None else path
+        seeds = _normalize_seeds(self.file_store, seeds_raw)
+        depth = int(self.context.get("depth") or self.context.get("max_depth") or 1)
+        direction = (self.context.get("direction") or "forward").lower()
+        predicate = self.context.get("predicate")
+        assert seeds, "path (or seeds) is required"
+        assert direction in _VALID_DIRECTIONS, (
+            f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}"
+        )
+        outbound, inbound = await _build_indexes(self.file_store)
+        results = _bfs(seeds, depth, direction, predicate, outbound, inbound)
+        _set_answer(self.context, results)
+
+    async def graph_traverse(
+        self,
+        path: str | list[str],
+        direction: str = "forward",
+        depth: int = 1,
+        predicate: str | None = None,
+    ) -> ToolResponse:
+        """BFS from ``path`` over wikilink edges.
+
+        Args:
+            path: seed file (or list of seeds).
+            direction: ``forward`` (outbound) / ``backward`` (inbound) /
+                ``both``. Aliases ``out`` / ``in`` accepted.
+            depth: hops to expand (default 1 = immediate neighbors).
+            predicate: filter edges by predicate (None = no filter).
+        """
+        direction = direction.lower()
+        assert direction in _VALID_DIRECTIONS, (
+            f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}"
+        )
+        seeds = _normalize_seeds(self.file_store, path)
+        assert seeds, "path is required"
+        outbound, inbound = await _build_indexes(self.file_store)
+        results = _bfs(seeds, depth, direction, predicate, outbound, inbound)
+        return _tool_response("graph:traverse", True, results, audit=self.audit)
diff --git a/reme2/memory/runtime_response.py b/reme2/steps/runtime_response.py
similarity index 100%
rename from reme2/memory/runtime_response.py
rename to reme2/steps/runtime_response.py
diff --git a/reme2/steps/tags/__init__.py b/reme2/steps/tags/__init__.py
new file mode 100644
index 00000000..3b954bbc
--- /dev/null
+++ b/reme2/steps/tags/__init__.py
@@ -0,0 +1,15 @@
+"""Tags steps — frontmatter-tag enumeration and per-tag statistics.
+
+Two Steps:
+
+    tags:list  — distinct tags + document counts across the vault.
+    tags:stat  — count + file list for one specific tag.
+
+Both consume what ``file_store`` already has in memory: path
+discovery from ``file_store.file_chunks``, tag lookup from a single
+batched ``file_graph.get_nodes(paths)`` call. No filesystem walk —
+the cost is one dict iteration plus one graph round-trip.
+"""
+
+from . import list  # noqa: F401  -- @R.register("tags:list")
+from . import stat  # noqa: F401  -- @R.register("tags:stat")
diff --git a/reme2/steps/tags/list.py b/reme2/steps/tags/list.py
new file mode 100644
index 00000000..0a07a772
--- /dev/null
+++ b/reme2/steps/tags/list.py
@@ -0,0 +1,84 @@
+"""``tags:list`` — enumerate every tag declared in the vault.
+
+Single ``file_graph.get_nodes(None)`` call streams every real node;
+we accumulate tag counts in one pass. No filesystem walk, no
+per-file frontmatter parse.
+
+Returns:
+
+    {tags: [{tag, count}, ...], total: int}
+
+Per-tag file lists live on ``tags:stat`` so this listing stays cheap.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+
+from agentscope.tool import ToolResponse
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+async def _iter_tagged(file_store):
+    """Yield ``(path, tags_list)`` for every indexed node that has tags.
+
+    Owns the tag-extraction rule so ``tags:stat`` shares it via import —
+    single source of truth for "what counts as tagged".
+    """
+    if not file_store.file_graph:
+        return
+    for node in await file_store.file_graph.get_nodes():
+        tags = node.front_matter.tags or []
+        if tags:
+            yield node.path, [str(t) for t in tags if t]
+
+
+async def _list(file_store, sort: str, limit: int | None) -> dict:
+    counter: Counter[str] = Counter()
+    async for _, tags in _iter_tagged(file_store):
+        counter.update(tags)
+    if sort == "alpha":
+        items = [{"tag": t, "count": c} for t, c in sorted(counter.items())]
+    else:  # "count" (default) — desc count then alpha tiebreak
+        items = [
+            {"tag": t, "count": c}
+            for t, c in sorted(counter.items(), key=lambda kv: (-kv[1], kv[0]))
+        ]
+    if limit is not None and limit > 0:
+        items = items[:limit]
+    return {"tags": items, "total": len(counter)}
+
+
+@R.register("tags:list")
+class TagsList(BaseStep):
+    """List every distinct tag in the vault with its document count."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        sort = (self.context.get("sort") or "count").lower()
+        limit = self.context.get("limit")
+        payload = await _list(self.file_store, sort, int(limit) if limit else None)
+        _set_answer(self.context, payload)
+
+    async def tags_list(
+        self,
+        sort: str = "count",
+        limit: int | None = None,
+    ) -> ToolResponse:
+        """List every tag in the vault with its document count.
+
+        Args:
+            sort: ``count`` (default, descending) or ``alpha``.
+            limit: cap the number of returned tags; ``None`` = all.
+        """
+        payload = await _list(self.file_store, sort.lower(), limit)
+        return _tool_response("tags:list", True, payload, audit=self.audit)
diff --git a/reme2/steps/tags/stat.py b/reme2/steps/tags/stat.py
new file mode 100644
index 00000000..d812d81f
--- /dev/null
+++ b/reme2/steps/tags/stat.py
@@ -0,0 +1,77 @@
+"""``tags:stat`` — usage statistics for a single tag.
+
+Reuses ``_iter_tagged`` from ``tags:list`` so node discovery and tag
+extraction follow the same rule. Filters the stream by the requested
+tag and collects matching paths.
+
+Returns:
+
+    {tag, exists, count, paths, truncated}
+
+``exists=False`` short-circuits to ``count=0`` and ``paths=[]`` —
+useful for "does the agent need to introduce this tag?" probes.
+``limit`` caps the path list (default 100); ``count`` is always
+the true total.
+"""
+
+from __future__ import annotations
+
+from agentscope.tool import ToolResponse
+
+from .list import _iter_tagged
+
+from ..base_step import BaseStep
+from ..runtime_response import _set_answer, _tool_response
+
+from ...component import R
+from ...enumeration import ComponentEnum
+
+
+async def _stat(file_store, tag: str, limit: int) -> dict:
+    matches: list[str] = []
+    count = 0
+    async for path, tags in _iter_tagged(file_store):
+        if tag in tags:
+            count += 1
+            if len(matches) < limit:
+                matches.append(path)
+    return {
+        "tag": tag,
+        "exists": count > 0,
+        "count": count,
+        "paths": matches,
+        "truncated": count > len(matches),
+    }
+
+
+@R.register("tags:stat")
+class TagsStat(BaseStep):
+    """Usage statistics + file list for a single tag."""
+
+    component_type = ComponentEnum.STEP
+
+    audit: list[dict] | None = None
+
+    async def execute(self):
+        assert self.context is not None
+        tag: str = self.context.get("tag", "") or ""
+        assert tag, "tag is required"
+        limit = int(self.context.get("limit") or 100)
+        payload = await _stat(self.file_store, tag, limit)
+        self.context.response.success = True
+        _set_answer(self.context, payload)
+
+    async def tags_stat(
+        self,
+        tag: str,
+        limit: int = 100,
+    ) -> ToolResponse:
+        """Return usage statistics for ``tag``: count + file list.
+
+        Args:
+            tag: the tag to look up.
+            limit: cap the number of returned paths (count is always
+                the true total). Default 100.
+        """
+        payload = await _stat(self.file_store, tag, limit)
+        return _tool_response("tags:stat", True, payload, audit=self.audit)