feat: add frontend-ready wikilink graph APIs (#414)

reme/config/default.yaml

@@ -258,19 +258,29 @@ jobs:
 
   traverse:
     backend: base
-    description: "Walk the wikilink graph from a path."
+    description: >-
+      Return a bounded wikilink graph with frontend-ready nodes and directed edges.
     parameters:
       type: object
       properties:
         path:
-          type: string
-          description: "path"
+          oneOf:
+            - type: string
+            - type: array
+              items:
+                type: string
+              minItems: 1
+          description: "workspace-relative path or paths used as graph seeds"
         depth:
           type: integer
-          description: "hop limit"
+          minimum: 0
+          description: "maximum number of wikilink hops"
           default: 1
         direction:
           type: string
+          description: >-
+            traversal direction; returned edges always preserve the original
+            wikilink source-to-target direction
           enum:
             - forward
             - backward
@@ -281,6 +291,15 @@ jobs:
     steps:
       - backend: traverse_step
 
+  graph_snapshot:
+    backend: base
+    description: "Return the category-rooted digest wikilink graph with daily-note leaves."
+    parameters:
+      type: object
+      properties: { }
+    steps:
+      - backend: graph_snapshot_step
+
   reindex:
     backend: base
     description: "wipe the file store and rebuild it from the existing files"

reme/schema/__init__.py

@@ -35,12 +35,14 @@ from .dream import (
 from .emb_node import EmbNode
 from .file_chunk import FileChunk
 from .file_front_matter import FileFrontMatter
+from .graph_snapshot import GraphSnapshot, GraphSnapshotEdge, GraphSnapshotNode
 from .file_link import FileLink
 from .file_node import FileNode
 from .request import Request
 from .response import Response
 from .stream_chunk import StreamChunk
 from .token_usage import TokenUsage
+from .traverse_graph import TraverseGraph, TraverseGraphEdge, TraverseGraphNode
 
 __all__ = [
     "ApplicationConfig",
@@ -74,6 +76,9 @@ __all__ = [
     "FileFrontMatter",
     "FileLink",
     "FileNode",
+    "GraphSnapshot",
+    "GraphSnapshotEdge",
+    "GraphSnapshotNode",
     "IntegrateOutcome",
     "JobConfig",
     "PaperInfo",
@@ -86,4 +91,7 @@ __all__ = [
     "StreamChunk",
     "TokenUsage",
     "TopicSelectionOutput",
+    "TraverseGraph",
+    "TraverseGraphEdge",
+    "TraverseGraphNode",
 ]

reme/schema/graph_snapshot.py

@@ -0,0 +1,32 @@
+"""Public response schemas for a complete indexed wikilink graph snapshot."""
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class GraphSnapshotNode(BaseModel):
+    """One virtual category or indexed Markdown file in a graph snapshot."""
+
+    id: str = Field(description="Stable node identifier; documents use their workspace-relative path")
+    path: str = Field(description="Workspace-relative file or virtual category path")
+    name: str = Field(default="", description="Document name from frontmatter")
+    description: str = Field(default="", description="Document description from frontmatter")
+    indexed: bool = Field(description="Whether the target has an indexed FileNode")
+    virtual: bool = Field(default=False, description="Whether this is a generated category node")
+
+
+class GraphSnapshotEdge(BaseModel):
+    """One directed category or wikilink edge in a graph snapshot."""
+
+    source: str = Field(description="Source node identifier")
+    target: str = Field(description="Target node identifier")
+    target_anchor: str | None = Field(default=None, description="Optional heading, block, or line anchor")
+
+
+class GraphSnapshot(BaseModel):
+    """A category-rooted snapshot of digest wikilinks and their daily-note leaves."""
+
+    version: Literal[1] = 1
+    nodes: list[GraphSnapshotNode]
+    edges: list[GraphSnapshotEdge]

reme/schema/traverse_graph.py

@@ -0,0 +1,36 @@
+"""Public response schemas for wikilink graph traversal."""
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class TraverseGraphNode(BaseModel):
+    """One indexed file or unresolved wikilink target in a traversal result."""
+
+    id: str = Field(description="Stable node identifier; equal to the workspace-relative path")
+    path: str = Field(description="Workspace-relative file path")
+    name: str = Field(default="", description="Document name from frontmatter")
+    description: str = Field(default="", description="Document description from frontmatter")
+    depth: int = Field(ge=0, description="Shortest hop distance from any seed")
+    indexed: bool = Field(description="Whether the target has an indexed FileNode")
+
+
+class TraverseGraphEdge(BaseModel):
+    """One directed wikilink, preserving its original source and target."""
+
+    source: str = Field(description="Workspace-relative source file path")
+    target: str = Field(description="Workspace-relative target file path")
+    target_anchor: str | None = Field(default=None, description="Optional heading, block, or line anchor")
+    depth: int = Field(ge=1, description="Traversal depth at which this edge was first reached")
+
+
+class TraverseGraph(BaseModel):
+    """A bounded wikilink graph rooted at one or more workspace paths."""
+
+    version: Literal[1] = 1
+    seeds: list[str] = Field(description="Normalized workspace-relative traversal roots")
+    depth: int = Field(ge=0, description="Requested hop limit")
+    direction: Literal["forward", "backward", "both"]
+    nodes: list[TraverseGraphNode]
+    edges: list[TraverseGraphEdge]

reme/steps/file_io/list.py

@@ -68,7 +68,7 @@ class ListStep(BaseStep):
         out: list[str] = []
         for entry in files:
             try:
-                out.append(str(entry.relative_to(workspace_dir)))
+                out.append(entry.relative_to(workspace_dir).as_posix())
             except ValueError:
                 out.append(str(entry))
         return out

reme/steps/index/__init__.py

@@ -4,6 +4,7 @@ from .bm25_search import Bm25SearchStep
 from .clear_paths import ClearPathsStep
 from .clear_store import ClearStoreStep
 from .draft import AddDraftStep, ReadAllDraftStep
+from .graph_snapshot import GraphSnapshotStep
 from .log_changes import LogChangesStep
 from .node_search import NodeSearchStep
 from .init_changes import InitChangesStep
@@ -30,6 +31,7 @@ __all__ = [
     "DEFAULT_LOW_POWER_POLL_MS",
     "DEFAULT_WATCH_DEBOUNCE_MS",
     "DEFAULT_WATCH_STEP_MS",
+    "GraphSnapshotStep",
     "InitChangesStep",
     "LogChangesStep",
     "NodeSearchStep",

reme/steps/index/graph_snapshot.py

@@ -0,0 +1,86 @@
+"""Return the digest graph, rooted by its three memory categories."""
+
+from ..base_step import BaseStep
+from ...components import R
+from ...enumeration import DreamBucketEnum
+from ...schema import GraphSnapshot, GraphSnapshotEdge, GraphSnapshotNode
+
+_CATEGORY_BUCKETS = (
+    DreamBucketEnum.WIKI,
+    DreamBucketEnum.PERSONAL,
+    DreamBucketEnum.PROCEDURE,
+)
+
+
+@R.register("graph_snapshot_step")
+class GraphSnapshotStep(BaseStep):
+    """Build the frontend digest graph through the file-store contract.
+
+    Category nodes connect to every indexed Markdown file in their bucket.
+    Digest files retain wikilinks to other digest files and to daily notes. Daily
+    notes are leaves: their own outgoing links are intentionally not returned.
+    """
+
+    async def execute(self):
+        assert self.context is not None
+        indexed_nodes = await self.file_store.get_nodes()
+        node_by_path = {node.path: node for node in indexed_nodes if node.path.lower().endswith(".md")}
+
+        digest_dir = str(self.config_value("digest_dir")).strip("/")
+        daily_dir = str(self.config_value("daily_dir")).strip("/")
+        category_paths = {
+            bucket: f"{digest_dir}/{bucket.value}" if digest_dir else bucket.value for bucket in _CATEGORY_BUCKETS
+        }
+        daily_prefix = f"{daily_dir}/" if daily_dir else ""
+
+        digest_paths_by_category = {
+            label: sorted(path for path in node_by_path if path.startswith(f"{category_path}/"))
+            for label, category_path in category_paths.items()
+        }
+        digest_paths = {path for paths in digest_paths_by_category.values() for path in paths}
+
+        edge_keys: set[tuple[str, str, str | None]] = set()
+        for source in digest_paths:
+            for link in node_by_path[source].links:
+                target = link.target_path
+                if target in digest_paths or (target in node_by_path and target.startswith(daily_prefix)):
+                    edge_keys.add((source, target, link.target_anchor))
+
+        daily_paths = {target for _source, target, _anchor in edge_keys if target.startswith(daily_prefix)}
+
+        nodes = [
+            GraphSnapshotNode(
+                id=f"virtual:{bucket.value}",
+                path=category_paths[bucket],
+                name=bucket.value,
+                indexed=False,
+                virtual=True,
+            )
+            for bucket in _CATEGORY_BUCKETS
+        ]
+        for bucket in _CATEGORY_BUCKETS:
+            for path in digest_paths_by_category[bucket]:
+                edge_keys.add((f"virtual:{bucket.value}", path, None))
+
+        for path in sorted(digest_paths | daily_paths):
+            node = node_by_path[path]
+            nodes.append(
+                GraphSnapshotNode(
+                    id=path,
+                    path=path,
+                    name=node.front_matter.name,
+                    description=node.front_matter.description,
+                    indexed=True,
+                ),
+            )
+
+        edges = [
+            GraphSnapshotEdge(source=source, target=target, target_anchor=anchor)
+            for source, target, anchor in sorted(edge_keys, key=lambda edge: (edge[0], edge[1], edge[2] or ""))
+        ]
+        graph = GraphSnapshot(nodes=nodes, edges=edges)
+
+        self.context.response.success = True
+        self.context.response.answer = graph.model_dump()
+        self.logger.info(f"[{self.name}] nodes={len(nodes)} edges={len(edges)}")
+        return self.context.response

reme/steps/index/traverse.py

@@ -1,127 +1,181 @@
-"""BFS over wikilink edges from one or more seed files.
-
-One record per traversed *edge* (not per node): the same target can repeat
-if reached via different anchors or paths. Each record carries the
-predecessor and anchor so callers can reconstruct the path. Adjacency is
-built once via a single ``file_store.get_nodes()`` call — BFS then runs
-purely in memory with no per-frontier round-trips.
-"""
+"""Build a bounded, frontend-ready graph from indexed Markdown wikilinks."""
 
 from collections import deque
+from dataclasses import dataclass
 from pathlib import Path
 
 from ..base_step import BaseStep
 from ...components import R
-from ...schema import FileLink
+from ...schema import FileNode, TraverseGraph, TraverseGraphEdge, TraverseGraphNode
 
-_OUT = {"out", "forward", "both"}
-_IN = {"in", "backward", "both"}
-_VALID = _OUT | _IN
-
-# source path -> list of (neighbor path, link)
-Adjacency = dict[str, list[tuple[str, FileLink]]]
+_DIRECTION_ALIASES = {
+    "out": "forward",
+    "forward": "forward",
+    "in": "backward",
+    "backward": "backward",
+    "both": "both",
+}
 
 
-async def _build_adjacency(file_store) -> tuple[Adjacency, Adjacency]:
-    """Single ``get_nodes()`` pass → (outbound, inbound) adjacency maps.
+@dataclass(frozen=True, slots=True)
+class _TraversalLink:
+    """An adjacency entry carrying both traversal and original edge direction."""
 
-    Inbound stores the source path next to each link so BFS can attribute
-    inbound edges back to their origin — ``get_inlinks`` alone returns
-    target-shaped FileLinks without source attribution.
-    """
-    outbound: Adjacency = {}
-    inbound: Adjacency = {}
-    for node in await file_store.get_nodes():
+    neighbor: str
+    source: str
+    target: str
+    target_anchor: str | None
+
+
+Adjacency = dict[str, list[_TraversalLink]]
+EdgeKey = tuple[str, str, str | None]
+
+
+def _build_adjacency(nodes: list[FileNode]) -> tuple[Adjacency, Adjacency]:
+    """Build forward/backward adjacency while preserving actual wikilink direction."""
+    forward: Adjacency = {}
+    backward: Adjacency = {}
+    for node in nodes:
         for link in node.links:
-            if link.target_path:
-                outbound.setdefault(node.path, []).append((link.target_path, link))
-                inbound.setdefault(link.target_path, []).append((node.path, link))
-    return outbound, inbound
+            if not link.target_path:
+                continue
+            item = _TraversalLink(
+                neighbor=link.target_path,
+                source=node.path,
+                target=link.target_path,
+                target_anchor=link.target_anchor,
+            )
+            forward.setdefault(node.path, []).append(item)
+            backward.setdefault(link.target_path, []).append(
+                _TraversalLink(
+                    neighbor=node.path,
+                    source=item.source,
+                    target=item.target,
+                    target_anchor=item.target_anchor,
+                ),
+            )
+    return forward, backward
 
 
-def _bfs(
+def _traverse(
     seeds: list[str],
     max_depth: int,
     direction: str,
-    outbound: Adjacency,
-    inbound: Adjacency,
-) -> list[dict]:
-    """In-memory BFS; emits one record per unique (src, dst, anchor) edge."""
-    sources: list[Adjacency] = []
-    if direction in _OUT:
-        sources.append(outbound)
-    if direction in _IN:
-        sources.append(inbound)
+    forward: Adjacency,
+    backward: Adjacency,
+) -> tuple[dict[str, int], dict[EdgeKey, int]]:
+    """Return shortest node depths and directed edges reached by bounded BFS."""
+    adjacency_maps = []
+    if direction in {"forward", "both"}:
+        adjacency_maps.append(forward)
+    if direction in {"backward", "both"}:
+        adjacency_maps.append(backward)
 
-    visited: set[tuple[str, str, str | None]] = set()
-    results: list[dict] = []
-    queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds)
+    node_depths = {seed: 0 for seed in seeds}
+    edge_depths: dict[EdgeKey, int] = {}
+    queue = deque(seeds)
 
     while queue:
-        current, depth = queue.popleft()
-        if depth >= max_depth:
+        current = queue.popleft()
+        current_depth = node_depths[current]
+        if current_depth >= max_depth:
             continue
-        for src in sources:
-            for next_path, link in src.get(current, ()):
-                key = (current, next_path, link.target_anchor)
-                if key in visited:
-                    continue
-                visited.add(key)
-                results.append(
-                    {
-                        "path": next_path,
-                        "depth": depth + 1,
-                        "via": current,
-                        "anchor": link.target_anchor,
-                    },
-                )
-                if depth + 1 < max_depth:
-                    queue.append((next_path, depth + 1))
-    return results
+
+        next_depth = current_depth + 1
+        for adjacency in adjacency_maps:
+            for link in adjacency.get(current, ()):
+                edge_key = (link.source, link.target, link.target_anchor)
+                previous_edge_depth = edge_depths.get(edge_key)
+                if previous_edge_depth is None or next_depth < previous_edge_depth:
+                    edge_depths[edge_key] = next_depth
+
+                previous_node_depth = node_depths.get(link.neighbor)
+                if previous_node_depth is None or next_depth < previous_node_depth:
+                    node_depths[link.neighbor] = next_depth
+                    queue.append(link.neighbor)
+
+    return node_depths, edge_depths
+
+
+def _build_graph(
+    *,
+    seeds: list[str],
+    max_depth: int,
+    direction: str,
+    indexed_nodes: list[FileNode],
+    node_depths: dict[str, int],
+    edge_depths: dict[EdgeKey, int],
+) -> TraverseGraph:
+    """Materialize deterministic public graph schemas from traversal state."""
+    node_by_path = {node.path: node for node in indexed_nodes}
+    graph_nodes = []
+    for path, node_depth in sorted(node_depths.items(), key=lambda item: (item[1], item[0])):
+        node = node_by_path.get(path)
+        graph_nodes.append(
+            TraverseGraphNode(
+                id=path,
+                path=path,
+                name=node.front_matter.name if node is not None else "",
+                description=node.front_matter.description if node is not None else "",
+                depth=node_depth,
+                indexed=node is not None,
+            ),
+        )
+
+    graph_edges = [
+        TraverseGraphEdge(source=source, target=target, target_anchor=anchor, depth=edge_depth)
+        for (source, target, anchor), edge_depth in sorted(
+            edge_depths.items(),
+            key=lambda item: (item[1], item[0][0], item[0][1], item[0][2] or ""),
+        )
+    ]
+    return TraverseGraph(
+        seeds=seeds,
+        depth=max_depth,
+        direction=direction,
+        nodes=graph_nodes,
+        edges=graph_edges,
+    )
 
 
 @R.register("traverse_step")
 class TraverseStep(BaseStep):
-    """BFS from one or more seed files to explore wikilink relationships.
-
-    Parameters:
-        path       — single seed (str) or list of seeds (workspace-relative).
-        direction  — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``).
-        depth      — hop limit (default 1 = immediate neighbors).
-    """
+    """Return a bounded wikilink graph rooted at one or more workspace paths."""
 
     async def execute(self):
         assert self.context is not None
-        raw = self.context.get("path")
-        items = [raw] if isinstance(raw, (str, Path)) else list(raw or [])
-        seeds = [str(p) for p in items if p]
-        assert seeds, "path is required"
-        depth = int(self.context.get("depth") or 1)
-        direction = (self.context.get("direction") or "both").lower()
-        assert direction in _VALID, f"direction must be one of {sorted(_VALID)}, got {direction!r}"
+        raw_paths = self.context.get("path")
+        items = [raw_paths] if isinstance(raw_paths, (str, Path)) else list(raw_paths or [])
+        seeds = list(dict.fromkeys(str(path).replace("\\", "/") for path in items if path))
+        if not seeds:
+            raise ValueError("path is required")
 
-        outbound, inbound = await _build_adjacency(self.file_store)
-        results = _bfs(seeds, depth, direction, outbound, inbound)
+        raw_depth = self.context.get("depth")
+        max_depth = 1 if raw_depth is None else int(raw_depth)
+        if max_depth < 0:
+            raise ValueError("depth must be greater than or equal to 0")
 
-        self.logger.info(
-            f"[{self.name}] seeds={seeds!r} depth={depth} direction={direction} "
-            f"nodes={len(outbound) + len(inbound)} edges={len(results)}",
+        raw_direction = str(self.context.get("direction") or "both").lower()
+        direction = _DIRECTION_ALIASES.get(raw_direction)
+        if direction is None:
+            raise ValueError(f"direction must be one of {sorted(_DIRECTION_ALIASES)}, got {raw_direction!r}")
+
+        indexed_nodes = await self.file_store.get_nodes()
+        forward, backward = _build_adjacency(indexed_nodes)
+        node_depths, edge_depths = _traverse(seeds, max_depth, direction, forward, backward)
+        graph = _build_graph(
+            seeds=seeds,
+            max_depth=max_depth,
+            direction=direction,
+            indexed_nodes=indexed_nodes,
+            node_depths=node_depths,
+            edge_depths=edge_depths,
         )
 
-        label = seeds[0] if len(seeds) == 1 else f"{len(seeds)} seeds"
-        if not results:
-            answer = f"No edges found from {label}"
-        else:
-            header = f"Traversed {len(results)} edge(s) from {label}"
-            lines = [header, ""]
-            for r in results:
-                target = r["path"]
-                if r["anchor"]:
-                    target = f"{target}#{r['anchor']}"
-                lines.append(f"[depth={r['depth']}] {r['via']} --> {target}")
-            answer = "\n".join(lines)
-
         self.context.response.success = True
-        self.context.response.answer = answer
-        self.context.response.metadata.update({"edges": results, "count": len(results)})
+        self.context.response.answer = graph.model_dump()
+        self.logger.info(
+            f"[{self.name}] seeds={seeds!r} depth={max_depth} direction={direction} "
+            f"nodes={len(graph.nodes)} edges={len(graph.edges)}",
+        )
         return self.context.response

tests/unit/test_common_steps.py

@@ -10,7 +10,7 @@ import warnings
 from reme.components.agent_wrapper import BaseAgentWrapper
 from reme.components.application_context import ApplicationContext
 from reme.components.file_store import LocalFileStore
-from reme.schema import FileLink, FileNode
+from reme.schema import FileFrontMatter, FileLink, FileNode, TraverseGraph
 from reme.steps.common.add import AddStep
 from reme.steps.common.health_check import _file_graph_status
 from reme.steps.common.llm_demo import LLMDemoStep
@@ -42,7 +42,13 @@ def _run(coro):
     asyncio.run(coro)
 
 
-def _node(path: str, links: list[tuple[str, str | None]] | None = None) -> FileNode:
+def _node(
+    path: str,
+    links: list[tuple[str, str | None]] | None = None,
+    *,
+    name: str = "",
+    description: str = "",
+) -> FileNode:
     """Build a FileNode with (target_path, target_anchor) outgoing edges."""
     return FileNode(
         path=path,
@@ -50,6 +56,7 @@ def _node(path: str, links: list[tuple[str, str | None]] | None = None) -> FileN
         links=[
             FileLink(source_path=path, target_path=target, target_anchor=anchor) for target, anchor in (links or [])
         ],
+        front_matter=FileFrontMatter(name=name, description=description),
     )
 
 
@@ -63,7 +70,7 @@ async def _make_store(nodes: list[FileNode]) -> LocalFileStore:
 
 
 def _edges(step) -> list[dict]:
-    return step.context.response.metadata.get("edges", [])
+    return step.context.response.answer["edges"]
 
 
 def test_add_step_coerces_numeric_inputs():
@@ -207,12 +214,11 @@ def test_traverse_forward_depth_1():
             step = traverse_mod.TraverseStep(file_store=store)
             await step(path="a.md", direction="forward", depth=1)
             results = _edges(step)
-            paths = {r["path"] for r in results}
+            paths = {r["target"] for r in results}
             assert paths == {"b.md", "c.md"}
-            # Anchors remain part of traversal metadata.
-            c_edge = next(r for r in results if r["path"] == "c.md")
-            assert c_edge["anchor"] == "intro"
-            assert c_edge["via"] == "a.md"
+            c_edge = next(r for r in results if r["target"] == "c.md")
+            assert c_edge["target_anchor"] == "intro"
+            assert c_edge["source"] == "a.md"
             assert c_edge["depth"] == 1
             await store.close()
         print("✓ test_traverse_forward_depth_1 passed")
@@ -220,6 +226,25 @@ def test_traverse_forward_depth_1():
     asyncio.run(run())
 
 
+def test_traverse_normalizes_windows_seed_path():
+    """Windows-style seeds match the graph's portable POSIX path keys."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
+            store = await _make_store(
+                [
+                    _node("topics/a.md", [("topics/b.md", None)]),
+                    _node("topics/b.md"),
+                ],
+            )
+            step = traverse_mod.TraverseStep(file_store=store)
+            await step(path=r"topics\a.md", direction="forward", depth=1)
+            assert [edge["target"] for edge in _edges(step)] == ["topics/b.md"]
+            await store.close()
+
+    asyncio.run(run())
+
+
 def test_traverse_backward_returns_inlinks():
     """direction=backward walks inbound edges."""
 
@@ -235,7 +260,10 @@ def test_traverse_backward_returns_inlinks():
             step = traverse_mod.TraverseStep(file_store=store)
             await step(path="b.md", direction="backward", depth=1)
             results = _edges(step)
-            assert {r["path"] for r in results} == {"a.md", "c.md"}
+            assert {(r["source"], r["target"]) for r in results} == {
+                ("a.md", "b.md"),
+                ("c.md", "b.md"),
+            }
             await store.close()
         print("✓ test_traverse_backward_returns_inlinks passed")
 
@@ -257,7 +285,7 @@ def test_traverse_depth_2_expands():
             step = traverse_mod.TraverseStep(file_store=store)
             await step(path="a.md", direction="forward", depth=2)
             results = _edges(step)
-            depth_map = {r["path"]: r["depth"] for r in results}
+            depth_map = {r["target"]: r["depth"] for r in results}
             assert depth_map.get("b.md") == 1
             assert depth_map.get("c.md") == 2
             await store.close()
@@ -320,13 +348,96 @@ def test_traverse_both_directions():
             step = traverse_mod.TraverseStep(file_store=store)
             await step(path="center.md", direction="both", depth=1)
             results = _edges(step)
-            assert {r["path"] for r in results} == {"upstream.md", "downstream.md"}
+            assert {(r["source"], r["target"]) for r in results} == {
+                ("upstream.md", "center.md"),
+                ("center.md", "downstream.md"),
+            }
             await store.close()
         print("✓ test_traverse_both_directions passed")
 
     asyncio.run(run())
 
 
+def test_traverse_both_preserves_reciprocal_edge_directions():
+    """Opposite wikilinks remain two distinct directed graph edges."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
+            store = await _make_store(
+                [
+                    _node("a.md", [("b.md", None)]),
+                    _node("b.md", [("a.md", None)]),
+                ],
+            )
+            step = traverse_mod.TraverseStep(file_store=store)
+            response = await step(path="a.md", direction="both", depth=1)
+            graph = TraverseGraph.model_validate(response.answer)
+
+            assert {(edge.source, edge.target) for edge in graph.edges} == {
+                ("a.md", "b.md"),
+                ("b.md", "a.md"),
+            }
+            assert response.metadata == {}
+            await store.close()
+
+    asyncio.run(run())
+
+
+def test_traverse_returns_frontmatter_and_unresolved_nodes():
+    """Graph nodes expose labels and distinguish indexed files from dangling targets."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
+            store = await _make_store(
+                [
+                    _node(
+                        "a.md",
+                        [("missing.md", "details")],
+                        name="Alpha",
+                        description="Root note",
+                    ),
+                ],
+            )
+            step = traverse_mod.TraverseStep(file_store=store)
+            response = await step(path="a.md", direction="forward", depth=1)
+            graph = TraverseGraph.model_validate(response.answer)
+            nodes = {node.path: node for node in graph.nodes}
+
+            assert graph.version == 1
+            assert graph.seeds == ["a.md"]
+            assert nodes["a.md"].model_dump() == {
+                "id": "a.md",
+                "path": "a.md",
+                "name": "Alpha",
+                "description": "Root note",
+                "depth": 0,
+                "indexed": True,
+            }
+            assert nodes["missing.md"].indexed is False
+            assert nodes["missing.md"].depth == 1
+            assert graph.edges[0].target_anchor == "details"
+            await store.close()
+
+    asyncio.run(run())
+
+
+def test_traverse_depth_zero_returns_only_seed_nodes():
+    """A zero-hop traversal is valid and emits no edges."""
+
+    async def run():
+        with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
+            store = await _make_store([_node("a.md", [("b.md", None)]), _node("b.md")])
+            step = traverse_mod.TraverseStep(file_store=store)
+            response = await step(path="a.md", direction="forward", depth=0)
+            graph = TraverseGraph.model_validate(response.answer)
+
+            assert [node.path for node in graph.nodes] == ["a.md"]
+            assert graph.edges == []
+            await store.close()
+
+    asyncio.run(run())
+
+
 if __name__ == "__main__":
     print("\n=== traverse step tests ===")
     test_traverse_forward_depth_1()
@@ -335,4 +446,7 @@ if __name__ == "__main__":
     test_traverse_short_seed_yields_empty()
     test_traverse_not_found_seed()
     test_traverse_both_directions()
+    test_traverse_both_preserves_reciprocal_edge_directions()
+    test_traverse_returns_frontmatter_and_unresolved_nodes()
+    test_traverse_depth_zero_returns_only_seed_nodes()
     print("\n所有测试通过!")

tests/unit/test_crud_steps.py

@@ -30,7 +30,7 @@ import asyncio
 import os
 import tempfile
 import warnings
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
 
 from reme.components.file_store import LocalFileStore
 from reme.schema import FileNode
@@ -177,6 +177,14 @@ def test_list_lists_files():
     asyncio.run(run())
 
 
+def test_list_formats_workspace_relative_windows_paths_as_posix():
+    """Workspace-relative list results use the graph's portable path format."""
+    workspace = PureWindowsPath("C:/workspace")
+    files = [workspace / "topics" / "a.md"]
+
+    assert crud_list.ListStep._format_relative(files, workspace) == ["topics/a.md"]
+
+
 def test_list_empty_directory_has_explicit_answer():
     """list tells an LLM explicitly when the target directory has no files."""
 

tests/unit/test_graph_snapshot_step.py

@@ -0,0 +1,94 @@
+"""Tests for the frontend-ready category-rooted digest graph snapshot."""
+
+import asyncio
+
+from reme.components.file_store import LocalFileStore
+from reme.schema import FileFrontMatter, FileLink, FileNode, GraphSnapshot
+from reme.steps.index.graph_snapshot import GraphSnapshotStep
+
+
+def _node(
+    path: str,
+    links: list[tuple[str, str | None]] | None = None,
+    *,
+    name: str = "",
+    description: str = "",
+) -> FileNode:
+    return FileNode(
+        path=path,
+        st_mtime=1.0,
+        links=[
+            FileLink(source_path=path, target_path=target, target_anchor=anchor) for target, anchor in (links or [])
+        ],
+        front_matter=FileFrontMatter(name=name, description=description),
+    )
+
+
+def test_graph_snapshot_returns_category_roots_digest_links_and_daily_leaves(tmp_path, monkeypatch):
+    """The snapshot is rooted at digest categories and stops at indexed daily notes."""
+    monkeypatch.chdir(tmp_path)
+
+    async def run():
+        store = LocalFileStore(name="snapshot", embedding_store="")
+        await store.start()
+        try:
+            await store.file_graph.upsert_nodes(
+                [
+                    _node(
+                        "digest/wiki/alpha.md",
+                        [
+                            ("digest/personal/beta.md", "intro"),
+                            ("daily/2026-08-05/event.md", None),
+                            ("daily/2026-08-05/missing.md", None),
+                            ("resource/ignored.md", None),
+                        ],
+                        name="Alpha",
+                        description="Root note",
+                    ),
+                    _node("digest/personal/beta.md", [("digest/wiki/alpha.md", None)], name="Beta"),
+                    _node("digest/procedure/how-to.md", name="How to"),
+                    _node(
+                        "daily/2026-08-05/event.md",
+                        [("digest/procedure/how-to.md", None), ("daily/2026-08-05/hidden.md", None)],
+                        name="Event",
+                    ),
+                    _node("daily/2026-08-05/isolated.md", name="Isolated daily"),
+                    _node("resource/ignored.md", name="Ignored resource"),
+                ],
+            )
+
+            response = await GraphSnapshotStep(file_store=store)()
+            graph = GraphSnapshot.model_validate(response.answer)
+
+            assert response.success is True
+            assert response.metadata == {}
+            assert [node.id for node in graph.nodes[:3]] == [
+                "virtual:wiki",
+                "virtual:personal",
+                "virtual:procedure",
+            ]
+            assert [node.path for node in graph.nodes[3:]] == [
+                "daily/2026-08-05/event.md",
+                "digest/personal/beta.md",
+                "digest/procedure/how-to.md",
+                "digest/wiki/alpha.md",
+            ]
+            nodes = {node.id: node for node in graph.nodes}
+            assert nodes["virtual:wiki"].path == "digest/wiki"
+            assert nodes["virtual:procedure"].path == "digest/procedure"
+            assert nodes["virtual:wiki"].virtual is True
+            assert nodes["digest/wiki/alpha.md"].name == "Alpha"
+            assert nodes["digest/wiki/alpha.md"].description == "Root note"
+            assert nodes["digest/wiki/alpha.md"].virtual is False
+            assert {(edge.source, edge.target, edge.target_anchor) for edge in graph.edges} == {
+                ("virtual:wiki", "digest/wiki/alpha.md", None),
+                ("virtual:personal", "digest/personal/beta.md", None),
+                ("virtual:procedure", "digest/procedure/how-to.md", None),
+                ("digest/wiki/alpha.md", "digest/personal/beta.md", "intro"),
+                ("digest/wiki/alpha.md", "daily/2026-08-05/event.md", None),
+                ("digest/personal/beta.md", "digest/wiki/alpha.md", None),
+            }
+        finally:
+            await store.close()
+
+    asyncio.run(run())