scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-09-21 00:22:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

feat: add Neo4j backend and refactor file graph architecture

Browse source 
- add neo4j dependency to project requirements
- introduce NetworkXFileGraph to replace LocalFileGraph implementation
- rename local_file_graph.py to networkx_file_graph.py with updated
  component registration as 'networkx'
- remove pickle persistence logic from NetworkX backend, simplify
  initialization
- update Neo4jFileGraph to return FileLink objects instead of
  (FileNode, FileLink) tuples from get_inlinks/get_outlinks methods
- remove unused AsyncIterator import and adjust method signatures
- add BareFileParser for handling binary/attachment files without
  content parsing
- move wikilink resolution utilities to dedicated utility module
- refactor memory I/O to use file graph's link resolution methods
  directly
- remove link extraction utilities from schema module, consolidate
  in utils.wikilink_resolver
This commit is contained in:
huangsen 2026-05-14 19:46:18 +08:00
parent d6c51e2173
commit b72690d954
22 changed files with 1969 additions and 1671 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
pyproject.toml
View file

@ -59,6 +59,7 @@ dependencies = [
"watchfiles>=1.1.1",
"pyyaml>=6.0.3",
"mistletoe",
"neo4j",
]
[project.optional-dependencies]

2
reme2/component/file_graph/__init__.py
View file

@ -1,13 +1,11 @@
"""File graph module."""
from .base_file_graph import BaseFileGraph
from .local_file_graph import LocalFileGraph
from .nx_file_graph import NxFileGraph
from .neo4j_file_graph import Neo4jFileGraph
__all__ = [
"BaseFileGraph",
"LocalFileGraph",
"NxFileGraph",
"Neo4jFileGraph",
]

128
reme2/component/file_graph/local_file_graph.py
View file

@ -1,128 +0,0 @@
"""Pure-Python file-graph backend (no external deps)."""
from pathlib import Path
from .base_file_graph import BaseFileGraph
from ..component_registry import R
from ...schema import FileLink, FileNode
@R.register("local")
class LocalFileGraph(BaseFileGraph):
"""Dict-backed file graph; trusts ``FileLink.path`` for adjacency."""
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._nodes: dict[str, FileNode] = {}
self._inverse: dict[str, set[str]] = {}
# Virtual edges: src→target where target isn't in ``_nodes`` yet.
self._pending: dict[str, set[str]] = {}
self._graph_file: Path = self.graph_path / f"{self.graph_name}.jsonl"
# -- Lifecycle ---------------------------------------------------------
async def _start(self) -> None:
await super()._start()
self._load()
await self.rebuild_links()
edges = sum(len(s) for s in self._inverse.values())
pending = sum(len(s) for s in self._pending.values())
self.logger.info(
f"LocalFileGraph '{self.graph_name}' ready: "
f"{len(self._nodes)} nodes, {edges} edges, {pending} pending",
)
async def _close(self) -> None:
self._dump()
await super()._close()
def _load(self) -> None:
if not self._graph_file.exists():
return
with open(self._graph_file, "r", encoding="utf-8") as f:
self._nodes.update(
(n.path, n)
for n in (FileNode.model_validate_json(line) for line in f if line.strip())
)
def _dump(self) -> None:
tmp = self._graph_file.with_suffix(".tmp")
with open(tmp, "w", encoding="utf-8") as f:
f.writelines(f"{n.model_dump_json()}\n" for n in self._nodes.values())
tmp.replace(self._graph_file)
# -- Edge bookkeeping --------------------------------------------------
def _add_edge(self, src: str, target: str) -> None:
bucket = self._inverse if target in self._nodes else self._pending
bucket.setdefault(target, set()).add(src)
def _remove_edge(self, src: str, target: str) -> None:
for bucket in (self._inverse, self._pending):
srcs = bucket.get(target)
if srcs is None or src not in srcs:
continue
srcs.discard(src)
if not srcs:
del bucket[target]
# -- Node CRUD ---------------------------------------------------------
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
for node in nodes:
path = node.path
old = self._nodes.get(path)
if old is not None:
for link in old.links:
if link.path:
self._remove_edge(path, link.path)
self._nodes[path] = node
for link in node.links:
if link.path:
self._add_edge(path, link.path)
# Promote virtual edges aimed at this newly-arrived target.
promoted = self._pending.pop(path, None)
if promoted:
self._inverse.setdefault(path, set()).update(promoted)
async def delete_nodes(self, paths: list[str]) -> None:
for path in paths:
node = self._nodes.pop(path, None)
if node is None:
continue
for link in node.links:
if link.path:
self._remove_edge(path, link.path)
# Demote inbound edges to virtual; sources still link here.
demoted = self._inverse.pop(path, None)
if demoted:
self._pending.setdefault(path, set()).update(demoted)
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
return [self._nodes[p] for p in paths if p in self._nodes]
async def rebuild_links(self) -> None:
self._inverse.clear()
self._pending.clear()
for src, node in self._nodes.items():
for link in node.links:
if link.path:
self._add_edge(src, link.path)
# -- Link access -------------------------------------------------------
async def get_outlinks(self, path: str) -> list[FileLink]:
node = self._nodes.get(path)
if node is None:
return []
return [link for link in node.links if link.path and link.path in self._nodes]
async def get_inlinks(self, path: str) -> list[FileLink]:
if path not in self._nodes:
return []
return [
link
for src in self._inverse.get(path, ())
for link in self._nodes[src].links
if link.path == path
]

361
reme2/component/file_graph/neo4j_file_graph.py
View file

@ -2,30 +2,39 @@
Property-graph mapping:
(:File {path, st_mtime, title, description, tags, links_json,
extra_json})
-[:LINKS {idx, anchor, predicate}]->(:File)
Real node: (:File {path, st_mtime, title, description, tags,
links_json, extra_json})
Virtual node: (:File {path}) — placeholder created when something
links to a path that hasn't been upserted yet.
Edge: (:File)-[:LINKS {idx, anchor, predicate}]->(:File)
The ``links_json`` property doubles as the "is real" marker — its
presence means the node was upserted with a payload; its absence
means the node exists only because some edge points at it. This
mirrors ``NxFileGraph`` exactly: ``upsert_nodes`` promotes virtuals
in place, ``delete_nodes`` demotes back to virtual (or fully removes
if nothing points here), and ``get_outlinks`` excludes edges into
virtuals so the agent never sees dangling pointers.
``path`` is the unique key (constraint enforced on ``_start``).
Frontmatter goes into flat properties; arbitrary extras land in
``extra_json``. The full ``FileLink[]`` payload is also stored as
``links_json`` so we can recover the node losslessly even for links
whose target wasn't indexed at upsert time (will be linked later by
``_restore_inlinks``).
``links_json`` so ``rebuild_links`` can rebuild the relationship
graph from per-node payloads after backend repair / migration.
Adjacency policy: file_graph trusts ``FileLink.path`` directly — no
internal wikilink resolution. The parser pipeline (with the external
Adjacency policy: trusts ``FileLink.path`` directly — no internal
wikilink resolution. The parser pipeline (with the external
resolver) produces safe links where ``link.path`` is already a
vault-relative target path.
vault-relative target.
Conditional dependency: the ``neo4j`` driver is loaded lazily; the
Conditional dependency: the ``neo4j`` driver loads lazily; the
import error fires at ``_start`` (boot), not at first call.
"""
from __future__ import annotations
import json
from collections.abc import AsyncIterator
from typing import Any
from .base_file_graph import BaseFileGraph
@ -37,6 +46,12 @@ from ...schema.file_node import FileFrontMatter
_TYPED_FRONTMATTER_FIELDS = {"title", "description", "tags"}
_LINK_FIELDS = {"path", "anchor", "predicate"}
# Properties that distinguish a "real" node from a virtual placeholder.
# Listed for the demote query (delete_nodes) so we can REMOVE them all.
_REAL_PROPS = (
"st_mtime", "title", "description", "tags", "links_json", "extra_json",
)
@R.register("neo4j")
class Neo4jFileGraph(BaseFileGraph):
@ -72,7 +87,8 @@ class Neo4jFileGraph(BaseFileGraph):
from neo4j import AsyncGraphDatabase
except ImportError as e:
raise ImportError(
"Neo4jFileGraph requires the neo4j driver. " "Install with `pip install neo4j`.",
"Neo4jFileGraph requires the neo4j driver. "
"Install with `pip install neo4j`.",
) from e
self._driver = AsyncGraphDatabase.driver(
self._uri,
@ -80,10 +96,14 @@ class Neo4jFileGraph(BaseFileGraph):
)
async with self._session() as session:
await session.run(
"CREATE CONSTRAINT file_path_unique IF NOT EXISTS " "FOR (f:File) REQUIRE f.path IS UNIQUE",
"CREATE CONSTRAINT file_path_unique IF NOT EXISTS "
"FOR (f:File) REQUIRE f.path IS UNIQUE",
)
real, virtual, edges = await self._counts(session)
self.logger.info(
f"Neo4jFileGraph '{self.graph_name}' connected at " f"{self._uri}/{self._database}",
f"Neo4jFileGraph '{self.graph_name}' connected at "
f"{self._uri}/{self._database}: "
f"{real} nodes, {edges} edges, {virtual} virtual",
)
async def _close(self) -> None:
@ -96,144 +116,252 @@ class Neo4jFileGraph(BaseFileGraph):
assert self._driver is not None, "Neo4jFileGraph not started"
return self._driver.session(database=self._database)
@staticmethod
async def _counts(session) -> tuple[int, int, int]:
rec = await session.run(
"""
MATCH (f:File)
WITH count(CASE WHEN f.links_json IS NOT NULL THEN 1 END) AS real,
count(CASE WHEN f.links_json IS NULL THEN 1 END) AS virtual
OPTIONAL MATCH ()-[r:LINKS]->()
RETURN real, virtual, count(r) AS edges
""",
)
row = await rec.single()
if row is None:
return 0, 0, 0
return int(row["real"] or 0), int(row["virtual"] or 0), int(row["edges"] or 0)
# -- Node CRUD ---------------------------------------------------------
async def upsert_node(self, node: FileNode) -> None:
props = self._node_props(node)
# Trust link.path; only emit LINKS for links with non-empty path.
link_payload = [
async def upsert_nodes(self, nodes: list[FileNode]) -> None:
"""Upsert in one tx: SET props (promotes virtual to real), drop
existing outgoing edges, re-emit edges (auto-creating virtual
nodes for unindexed targets)."""
if not nodes:
return
payload = [
{
"idx": i,
"anchor": link.anchor,
"predicate": link.predicate,
"target": link.path,
"path": node.path,
"props": self._node_props(node),
"links": [
{
"idx": i,
"anchor": link.anchor,
"predicate": link.predicate,
"target": link.path,
}
for i, link in enumerate(node.links)
if link.path
],
}
for i, link in enumerate(node.links)
if link.path
for node in nodes
]
async with self._session() as session:
await session.execute_write(
self._upsert_node_tx,
node.path,
props,
link_payload,
)
# Late-arriving target: restore in-links from other nodes whose
# links resolve here. Separate session call (not in tx) so it
# doesn't block the upsert ack.
await self._restore_inlinks(node.path)
await session.execute_write(self._upsert_nodes_tx, payload)
@staticmethod
async def _upsert_node_tx(tx, path, props, links):
async def _upsert_nodes_tx(tx, payload):
# 1. Upsert node props (promotes virtual → real where necessary).
await tx.run(
"MERGE (f:File {path: $path}) SET f += $props",
path=path,
props=props,
"""
UNWIND $items AS n
MERGE (f:File {path: n.path})
SET f += n.props
""",
items=payload,
)
# 2. Drop existing outgoing edges from these sources.
await tx.run(
"MATCH (f:File {path: $path})-[r:LINKS]->() DELETE r",
path=path,
"""
UNWIND $paths AS p
MATCH (f:File {path: p})-[r:LINKS]->()
DELETE r
""",
paths=[item["path"] for item in payload],
)
# 3. Re-emit edges; MERGE on target auto-creates virtual nodes
# for unindexed targets.
await tx.run(
"""
UNWIND $items AS n
MATCH (s:File {path: n.path})
UNWIND n.links AS link
MERGE (t:File {path: link.target})
MERGE (s)-[r:LINKS {idx: link.idx}]->(t)
SET r.anchor = link.anchor, r.predicate = link.predicate
""",
items=payload,
)
# MERGE only when the target node exists. ``OPTIONAL MATCH`` +
# ``WHERE t IS NOT NULL`` lets us batch this in one pass per link.
for link in links:
await tx.run(
"""
MATCH (s:File {path: $src})
OPTIONAL MATCH (t:File {path: $dst})
WITH s, t WHERE t IS NOT NULL
MERGE (s)-[r:LINKS {idx: $idx}]->(t)
SET r.anchor = $anchor, r.predicate = $predicate
""",
src=path,
dst=link["target"],
idx=link["idx"],
anchor=link["anchor"],
predicate=link["predicate"],
)
async def _restore_inlinks(self, path: str) -> None:
"""Walk other nodes' links, add LINKS to ``path`` where ``link.path``
matches. Uses ``links_json`` on each node — no resolution logic."""
async def delete_nodes(self, paths: list[str]) -> None:
"""Demote real → virtual to preserve inbound visibility; fully
remove the (now-virtual) node only if no edge points at it."""
if not paths:
return
async with self._session() as session:
await session.execute_write(self._delete_nodes_tx, list(paths))
@staticmethod
async def _delete_nodes_tx(tx, paths):
# 1. Drop outgoing edges, then strip "real" properties (demote).
# Building the REMOVE clause from _REAL_PROPS keeps the list of
# properties in one place (top of module).
remove_clause = ", ".join(f"f.{name}" for name in _REAL_PROPS)
await tx.run(
f"""
UNWIND $paths AS p
MATCH (f:File {{path: p}})
OPTIONAL MATCH (f)-[r:LINKS]->()
DELETE r
WITH DISTINCT f
REMOVE {remove_clause}
""",
paths=paths,
)
# 2. Garbage-collect: drop the virtual node entirely if nothing
# points at it anymore.
await tx.run(
"""
UNWIND $paths AS p
MATCH (f:File {path: p})
WHERE f.links_json IS NULL AND NOT (f)<-[:LINKS]-()
DELETE f
""",
paths=paths,
)
async def get_nodes(self, paths: list[str]) -> list[FileNode]:
"""Return only real nodes (virtual placeholders are filtered)."""
if 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),
)
rows = [row["f"] async for row in rec]
return [self._row_to_node(row) for row in rows]
async def rebuild_links(self) -> None:
"""Defensive full rebuild from each real node's ``links_json``.
Three steps in one tx: drop all LINKS edges; drop all virtual
nodes; re-emit edges from per-node link payloads (re-creating
virtual targets as needed). Useful after manual repair or
schema migration.
"""
async with self._session() as session:
rec = await session.run(
"""
MATCH (f:File)
WHERE f.path <> $path
WHERE f.links_json IS NOT NULL
RETURN f.path AS p, f.links_json AS l
""",
path=path,
)
rows = [dict(r) async for r in rec]
payload: list[dict] = []
for row in rows:
try:
links = json.loads(row.get("l") or "[]")
except json.JSONDecodeError:
continue
for i, link in enumerate(links):
if not isinstance(link, dict) or link.get("path") != path:
continue
async with self._session() as session:
await session.run(
"""
MATCH (s:File {path: $src})
MATCH (t:File {path: $dst})
MERGE (s)-[r:LINKS {idx: $idx}]->(t)
SET r.anchor = $anchor, r.predicate = $predicate
""",
src=row["p"],
dst=path,
idx=i,
anchor=link.get("anchor"),
predicate=link.get("predicate"),
)
items = [
{
"idx": i,
"anchor": link.get("anchor"),
"predicate": link.get("predicate"),
"target": link.get("path"),
}
for i, link in enumerate(links)
if isinstance(link, dict) and link.get("path")
]
payload.append({"path": row["p"], "links": items})
async def delete_node(self, path: str) -> FileNode | None:
node = await self.get_node(path)
if node is None:
return None
async with self._session() as session:
await session.run(
"MATCH (f:File {path: $path}) DETACH DELETE f",
path=path,
)
return node
await session.execute_write(self._rebuild_links_tx, payload)
async def get_node(self, path: str) -> FileNode | None:
async with self._session() as session:
rec = await session.run(
"MATCH (f:File {path: $path}) RETURN f LIMIT 1",
path=path,
)
row = await rec.single()
return self._row_to_node(row["f"]) if row else None
@staticmethod
async def _rebuild_links_tx(tx, payload):
# 1. Wipe all edges and all virtual nodes.
await tx.run("MATCH ()-[r:LINKS]->() DELETE r")
await tx.run("MATCH (f:File) WHERE f.links_json IS NULL DELETE f")
if not payload:
return
# 2. Re-emit edges; virtual targets reappear via MERGE.
await tx.run(
"""
UNWIND $items AS n
MATCH (s:File {path: n.path})
UNWIND n.links AS link
MERGE (t:File {path: link.target})
MERGE (s)-[r:LINKS {idx: link.idx}]->(t)
SET r.anchor = link.anchor, r.predicate = link.predicate
""",
items=payload,
)
# -- Link access -------------------------------------------------------
async def get_outlinks(self, path: str) -> list[tuple[FileNode, FileLink]]:
async def get_outlinks(self, path: str) -> list[FileLink]:
"""Outgoing links from ``path``. Source must be real; targets
into virtual nodes are excluded so dangling refs are invisible."""
async with self._session() as session:
rec = await session.run(
"""
MATCH (s:File {path: $path})-[r:LINKS]->(t:File)
RETURN t, r ORDER BY r.idx ASC
MATCH (s:File {path: $path})
WHERE s.links_json IS NOT NULL
MATCH (s)-[r:LINKS]->(t:File)
WHERE t.links_json IS NOT NULL
RETURN t.path AS target, r.anchor AS anchor,
r.predicate AS predicate, r.idx AS idx
ORDER BY r.idx ASC
""",
path=path,
)
rows = [dict(row) async for row in rec]
return [(self._row_to_node(r["t"]), self._rel_to_link(r["r"], r["t"])) for r in rows]
return [
FileLink(
path=row["target"],
anchor=row.get("anchor"),
predicate=row.get("predicate"),
)
for row in rows
]
async def get_inlinks(self, path: str) -> list[tuple[FileNode, FileLink]]:
async def get_inlinks(self, path: str) -> list[FileLink]:
"""Incoming links to ``path`` (must be real). Sources are always
real because virtual nodes never have outgoing edges."""
async with self._session() as session:
rec = await session.run(
"""
MATCH (s:File)-[r:LINKS]->(t:File {path: $path})
RETURN s, r ORDER BY s.path ASC, r.idx ASC
MATCH (t:File {path: $path})
WHERE t.links_json IS NOT NULL
MATCH (s:File)-[r:LINKS]->(t)
RETURN r.anchor AS anchor, r.predicate AS predicate,
r.idx AS idx, s.path AS source
ORDER BY s.path ASC, r.idx ASC
""",
path=path,
)
rows = [dict(row) async for row in rec]
# ``rel_to_link`` needs the target's path to populate FileLink.path.
return [(self._row_to_node(r["s"]), self._rel_to_link(r["r"], target_path=path)) for r in rows]
# FileLink.path is the *target* path (the one we queried for) so
# the link stays "safe by construction" regardless of which side
# is asking. The source path is implicit in the query context.
return [
FileLink(
path=path,
anchor=row.get("anchor"),
predicate=row.get("predicate"),
)
for row in rows
]
# -- Internal: row ↔ schema marshaling ---------------------------------
@ -291,26 +419,3 @@ class Neo4jFileGraph(BaseFileGraph):
chunk_ids=[],
front_matter=FileFrontMatter(**fm_kwargs),
)
@staticmethod
def _rel_to_link(rel, target_path: Any = None) -> FileLink:
"""Reconstitute a ``FileLink`` from a Neo4j relationship.
The relationship row carries ``anchor`` and ``predicate``; the
target's ``path`` comes from the matched ``File`` node (passed
in by callers that have it on hand). For ``get_outlinks`` it's
the row's target node; for ``get_inlinks`` it's the path the
caller queried for. Either way, FileLink.path is set so the
link stays "safe by construction".
"""
d = dict(rel)
if hasattr(target_path, "get"):
# neo4j Node passed in
path = target_path.get("path", "")
else:
path = target_path or ""
return FileLink(
path=path,
anchor=d.get("anchor"),
predicate=d.get("predicate"),
)

2
reme2/component/file_parser/__init__.py
View file

@ -1,10 +1,12 @@
"""File parser implementations for different file formats."""
from .bare_file_parser import BareFileParser
from .base_file_parser import BaseFileParser
from .default_file_parser import DefaultFileParser
from .linked_file_parser import LinkedFileParser
__all__ = [
"BareFileParser",
"BaseFileParser",
"DefaultFileParser",
"LinkedFileParser",

31
reme2/component/file_parser/bare_file_parser.py Normal file
View file

@ -0,0 +1,31 @@
"""
BareFileParser — stat-only node for non-text files (attachments).
"""
from pathlib import Path
from .base_file_parser import BaseFileParser
from ..component_registry import R
from ...schema import FileChunk, FileNode
@R.register("bare")
class BareFileParser(BaseFileParser):
"""Stat-only parser for attachment/binary files.
No content read, no chunking, no link extraction. The resulting
``FileNode`` has empty ``links`` and ``chunk_ids``; ``front_matter``
carries ``mime`` and ``size`` as extras so retrieval can filter by
file type without reopening the file.
"""
async def parse(self, path: str | Path) -> tuple[FileNode, list[FileChunk]]:
file_path = Path(path)
stat = file_path.stat()
rel_path = self._get_relative_path(path)
node = FileNode(
path=rel_path,
st_mtime=stat.st_mtime,
links=[],
chunk_ids=[],
)
return node, []

313
reme2/mcp/test/test_expert.py
View file

@ -1,14 +1,16 @@
"""Expert-profile MCP tests.
Covers all 16 jobs registered by `reme2/config/expert.yaml`:
Covers the post-refactor 11-tool surface (5 memory + 5 file + 1 graph)
plus the auxiliary services exposed by `reme2/config/expert.yaml`:
Hot-write sync
Read memory_search, memory_graph_search,
memory_get, memory_list, memory_links,
memory_backlinks, memory_resolve_wikilink,
memory_count_tokens, memory_lint
Raw write memory_create, memory_update, memory_property_update,
memory_rename, memory_delete, memory_archive
Read memory_search, memory_graph_search, memory_get
Memory memory_create, memory_update_body, memory_update_meta
File file_download, file_upload, file_delete, file_list,
file_move
Graph graph_traverse
Lint check_dangling, check_orphans, check_collisions,
check_schema
Each `check_*` is an `async def` that takes a populated `AppContext`,
runs one MCP job (or a small sequence), asserts on the response, and
@ -26,22 +28,28 @@ from ._helpers import AppContext, decode, wait_for_index
# Manifest of every tool the expert profile must expose. Used by
# `check_registry` and as the upper bound for `wait_for_index` budgets.
EXPECTED_JOBS: tuple[str, ...] = (
# services
"sync",
"memory_search",
"memory_graph_search",
# memory primitives (5 — search counted above)
"memory_get",
"memory_list",
"memory_backlinks",
"memory_links",
"memory_resolve_wikilink",
"memory_count_tokens",
"memory_lint",
"memory_create",
"memory_update",
"memory_property_update",
"memory_rename",
"memory_delete",
"memory_archive",
"memory_update_body",
"memory_update_meta",
# file primitives (5)
"file_download",
"file_upload",
"file_delete",
"file_list",
"file_move",
# graph (1)
"graph_traverse",
# lint (4 atomic checks)
"check_dangling",
"check_orphans",
"check_collisions",
"check_schema",
)
@ -66,14 +74,6 @@ async def check_memory_get(ctx: AppContext) -> str:
return f"exists=True, edges={len(r.get('link', []))}"
async def check_memory_list(ctx: AppContext) -> str:
r = decode(await ctx.app.run_job("memory_list", tags=["person"]))
assert isinstance(r, dict) and r.get("count", 0) >= 2, r
paths = {item.get("path") for item in r.get("items", [])}
assert any("Alice.md" in p for p in paths), paths
return f"count={r['count']}"
async def check_memory_search(ctx: AppContext) -> str:
r = decode(
await ctx.app.run_job(
@ -100,55 +100,86 @@ async def check_memory_graph_search(ctx: AppContext) -> str:
)
hits = r if isinstance(r, list) else (r.get("chunks") if isinstance(r, dict) else [])
assert len(hits) > 0, r
# At least one result should carry a graph_hop annotation
hops = {h.get("graph_hop") for h in hits if isinstance(h, dict)}
return f"{len(hits)} hits, hops seen={sorted(h for h in hops if h is not None)}"
async def check_memory_links(ctx: AppContext) -> str:
async def check_lint_dangling(ctx: AppContext) -> str:
"""Atomic lint primitive — list FileLinks pointing at non-existent nodes.
Seeded vault is healthy, so we just verify the envelope shape."""
r = decode(await ctx.app.run_job("check_dangling"))
assert isinstance(r, dict) and "count" in r and "findings" in r, r
assert isinstance(r["findings"], list), r
return f"dangling={r['count']}"
async def check_lint_orphans(ctx: AppContext) -> str:
"""Atomic lint primitive — list nodes with no inlinks AND no outlinks."""
r = decode(await ctx.app.run_job("check_orphans"))
assert isinstance(r, dict) and "count" in r and "paths" in r, r
assert isinstance(r["paths"], list), r
return f"orphans={r['count']}"
async def check_lint_collisions(ctx: AppContext) -> str:
"""Atomic lint primitive — basenames resolving to >1 path."""
r = decode(await ctx.app.run_job("check_collisions"))
assert isinstance(r, dict) and "count" in r and "groups" in r, r
assert isinstance(r["groups"], dict), r
return f"collisions={r['count']}"
async def check_lint_schema(ctx: AppContext) -> str:
"""Atomic lint primitive — frontmatter schema violations."""
r = decode(await ctx.app.run_job("check_schema"))
assert isinstance(r, dict) and "count" in r and "findings" in r, r
assert isinstance(r["findings"], list), r
return f"schema_violations={r['count']}"
# ---------- file primitives -------------------------------------------
async def check_file_list(ctx: AppContext) -> str:
"""list_files projection: filter by tag returns indexed memories."""
r = decode(await ctx.app.run_job("file_list", tags=["person"]))
assert isinstance(r, dict) and r.get("count", 0) >= 2, r
paths = {item.get("path") for item in r.get("items", [])}
assert any("Alice.md" in p for p in paths), paths
return f"count={r['count']}"
# ---------- graph -----------------------------------------------------
async def check_graph_traverse_out(ctx: AppContext) -> str:
"""Outgoing traversal from Alice — matches the prior `memory_links` check."""
alice = ctx.abs_path("topics", "Alice", "Alice.md")
r = decode(await ctx.app.run_job("memory_links", path=alice))
assert isinstance(r, dict) and len(r.get("links", [])) >= 2, r
return f"{len(r['links'])} resolved outgoing links"
async def check_memory_backlinks(ctx: AppContext) -> str:
alice = ctx.abs_path("topics", "Alice", "Alice.md")
r = decode(await ctx.app.run_job("memory_backlinks", path=alice))
# Bob.md and Project X.md both link to [[Alice]]
assert isinstance(r, dict) and len(r.get("backlinks", [])) >= 2, r
return f"{len(r['backlinks'])} incoming backlinks"
async def check_memory_resolve_wikilink(ctx: AppContext) -> str:
r = decode(await ctx.app.run_job("memory_resolve_wikilink", wikilink="Alice"))
assert isinstance(r, dict) and r.get("exists") is True, r
assert "Alice.md" in (r.get("path") or ""), r
return "stem 'Alice' → Alice.md"
async def check_memory_count_tokens(ctx: AppContext) -> str:
r = decode(
await ctx.app.run_job(
"memory_count_tokens",
text="hello world from the smoke test",
"graph_traverse",
seeds=[alice],
max_depth=1,
direction="out",
),
)
assert isinstance(r, dict) and isinstance(r.get("tokens"), int), r
assert r["tokens"] > 0, r
return f"text → {r['tokens']} tokens"
assert isinstance(r, list) and len(r) >= 2, r
return f"{len(r)} outgoing edges"
async def check_memory_lint(ctx: AppContext) -> str:
"""Maintainer lint pass — read-only. The seeded vault is healthy
(no broken wikilinks / schema violations), so we just verify the
shell wires through and returns the expected envelope."""
r = decode(await ctx.app.run_job("memory_lint"))
assert isinstance(r, dict), r
assert "scanned" in r and "findings" in r, r
assert isinstance(r["findings"], list), r
assert r["scanned"] >= len(ctx.file_store), r
return f"scanned={r['scanned']}, findings={len(r['findings'])}"
async def check_graph_traverse_in(ctx: AppContext) -> str:
"""Incoming traversal to Alice — matches the prior `memory_backlinks` check."""
alice = ctx.abs_path("topics", "Alice", "Alice.md")
r = decode(
await ctx.app.run_job(
"graph_traverse",
seeds=[alice],
max_depth=1,
direction="in",
),
)
assert isinstance(r, list) and len(r) >= 2, r
return f"{len(r)} incoming edges"
# ---------- hot-write: sync (create / append / refusal) ---------------
@ -179,11 +210,8 @@ async def check_sync_create(ctx: AppContext) -> str:
index_text = (event_dir / "suite-event.md").read_text(encoding="utf-8")
assert "## Materials" in index_text, "Materials footer absent"
assert "raw-prompt.md" in index_text, "Materials footer missing raw-prompt link"
# 4-axis schema must be on disk
assert "lifecycle: streaming" in index_text, "schema axis 'lifecycle' missing"
assert "role: observation" in index_text, "schema axis 'role' missing"
# Stash for downstream checks via the context (small mutation pattern).
ctx.abs_path("__suite_event_dir__") # noop; readable side-effect below
setattr(ctx, "_suite_event_dir", event_dir)
setattr(ctx, "_suite_event_index", event_dir / "suite-event.md")
await wait_for_index(ctx.watcher, expected_min=len(ctx.file_store))
@ -196,10 +224,10 @@ async def check_sync_append(ctx: AppContext) -> str:
"sync",
name="suite-event",
content="## follow-up\n- second pass\n",
topics=["[[Bob]]"], # union with [[Alice]]
topics=["[[Bob]]"],
tags=["follow-up"],
materials=[
{"filename": "tool-output.txt", "content": "second run\n"}, # collision
{"filename": "tool-output.txt", "content": "second run\n"},
{"filename": "summary.md", "content": "# summary\n"},
],
),
@ -219,12 +247,12 @@ async def check_sync_append(ctx: AppContext) -> str:
async def check_sync_refuse_distilled(ctx: AppContext) -> str:
"""Flip status via memory_update_meta (patch dict), then sync should refuse."""
index_path = str(getattr(ctx, "_suite_event_index"))
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=index_path,
key="status",
value="distilled",
patch={"status": "distilled"},
)
r = decode(
await ctx.app.run_job(
@ -267,11 +295,11 @@ async def check_memory_create(ctx: AppContext) -> str:
return f"created {Path(target).name}"
async def check_memory_update(ctx: AppContext) -> str:
async def check_memory_update_body(ctx: AppContext) -> str:
carol = ctx.abs_path("topics", "Carol", "Carol.md")
r = decode(
await ctx.app.run_job(
"memory_update",
"memory_update_body",
path=carol,
old_string="Knows [[Alice]].",
new_string="Knows [[Alice]] and [[Bob]].",
@ -282,65 +310,87 @@ async def check_memory_update(ctx: AppContext) -> str:
return f"body edit applied (replaced={r['replaced']})"
async def check_memory_property_update(ctx: AppContext) -> str:
async def check_memory_update_meta(ctx: AppContext) -> str:
carol = ctx.abs_path("topics", "Carol", "Carol.md")
r = decode(
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=carol,
key="confidence",
value="✅",
patch={"confidence": "✅"},
),
)
assert isinstance(r, dict) and "error" not in r, r
assert r.get("key") == "confidence" and r.get("value") == "✅", r
applied = r.get("applied") or {}
assert applied.get("confidence", {}).get("value") == "✅", r
assert "confidence: ✅" in Path(carol).read_text(encoding="utf-8"), "property write not on disk"
return "set confidence=✅"
async def check_memory_rename(ctx: AppContext) -> str:
# ---------- file_move / file_delete -----------------------------------
async def check_file_move(ctx: AppContext) -> str:
"""Move (rename) Carol.md inside its folder. Default update_refs=False."""
src = ctx.abs_path("topics", "Carol", "Carol.md")
dst = ctx.abs_path("topics", "Carol", "Carol-renamed.md")
r = decode(
await ctx.app.run_job(
"memory_rename",
old_path=src,
new_path=dst,
"file_move",
src=src,
dst=dst,
),
)
assert isinstance(r, dict) and "error" not in r, r
assert r.get("new_path") and Path(r["new_path"]).is_file(), r
assert isinstance(r, dict) and r.get("ok") is True, r
assert Path(dst).is_file(), dst
assert not Path(src).exists(), f"old path still on disk: {src}"
setattr(ctx, "_carol_path", r["new_path"])
setattr(ctx, "_carol_path", dst)
return "Carol.md → Carol-renamed.md"
async def check_memory_archive(ctx: AppContext) -> str:
carol = getattr(ctx, "_carol_path", ctx.abs_path("topics", "Carol", "Carol-renamed.md"))
r = decode(await ctx.app.run_job("memory_archive", path=carol))
assert isinstance(r, dict) and r.get("archived") is True, r
archived_path = r.get("new_path")
assert archived_path and "Archive" in archived_path, r
assert Path(archived_path).is_file(), archived_path
setattr(ctx, "_carol_archived", archived_path)
return f"→ {Path(archived_path).resolve().relative_to(ctx.vault.resolve())}"
async def check_memory_delete(ctx: AppContext) -> str:
target = getattr(ctx, "_carol_archived", None) or ctx.abs_path("topics", "Carol", "Carol-renamed.md")
r = decode(await ctx.app.run_job("memory_delete", path=target))
async def check_file_delete(ctx: AppContext) -> str:
target = getattr(ctx, "_carol_path", None) or ctx.abs_path("topics", "Carol", "Carol-renamed.md")
r = decode(await ctx.app.run_job("file_delete", vault_path=target))
assert isinstance(r, dict) and r.get("deleted") is True, r
assert not Path(target).exists(), target
return "removed"
# ---------- file_download / file_upload -------------------------------
async def check_file_upload_then_download(ctx: AppContext) -> str:
"""Upload a non-md attachment under topics/Alice/, then download it back."""
import tempfile
payload = b"ATTACHMENT-BYTES-FROM-SUITE"
with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as tmp:
tmp.write(payload)
local_src = tmp.name
vault_path = ctx.abs_path("topics", "Alice", "alice-attachment.bin")
up = decode(
await ctx.app.run_job(
"file_upload",
local_path=local_src,
vault_path=vault_path,
),
)
assert isinstance(up, dict) and up.get("size") == len(payload), up
assert Path(vault_path).is_file(), vault_path
down = decode(
await ctx.app.run_job("file_download", vault_path=vault_path),
)
assert isinstance(down, dict) and down.get("local_path"), down
assert Path(down["local_path"]).read_bytes() == payload, "round-trip bytes mismatch"
return f"upload+download {len(payload)}B"
# ---------- schema gates (P4) -----------------------------------------
async def check_schema_path_template_refuses(ctx: AppContext) -> str:
"""memory_create rejects paths outside topics/{X}/{Y}.md,
events/{date}/{name}/..., or Archive/..."""
target = ctx.abs_path("notes", "freeform.md") # outside any template
target = ctx.abs_path("notes", "freeform.md")
r = decode(
await ctx.app.run_job(
"memory_create",
@ -391,16 +441,16 @@ async def check_schema_status_skip_refused(ctx: AppContext) -> str:
assert event_index is not None, "suite-event index not staged"
r = decode(
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=str(event_index),
key="status",
value="active",
patch={"status": "active"},
),
)
assert isinstance(r, dict), r
assert "error" in r and "transition" in r["error"].lower(), r
assert r.get("prior") == "distilled", r
return f"refused {r.get('prior')!r} → {r.get('requested')!r}"
applied = r.get("applied") or {}
err = (applied.get("status") or {}).get("error", "")
assert "transition" in err.lower(), r
return f"refused {applied['status'].get('prior')!r} → {applied['status'].get('requested')!r}"
async def check_schema_status_invalid_value(ctx: AppContext) -> str:
@ -409,40 +459,37 @@ async def check_schema_status_invalid_value(ctx: AppContext) -> str:
assert event_index is not None, "suite-event index not staged"
r = decode(
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=str(event_index),
key="status",
value="bogus",
patch={"status": "bogus"},
),
)
assert isinstance(r, dict), r
assert "error" in r and "invalid" in r["error"].lower(), r
applied = r.get("applied") or {}
err = (applied.get("status") or {}).get("error", "")
assert "invalid" in err.lower(), r
return "refused status='bogus'"
async def check_schema_status_force_bypass(ctx: AppContext) -> str:
"""force=True bypasses the state machine — useful when the agent
intentionally needs to step outside conventions."""
"""force=True bypasses the state machine."""
event_index = getattr(ctx, "_suite_event_index", None)
assert event_index is not None, "suite-event index not staged"
r = decode(
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=str(event_index),
key="status",
value="active",
patch={"status": "active"},
force=True,
),
)
assert isinstance(r, dict), r
assert "error" not in r, r
# restore for any downstream checks
decode(
await ctx.app.run_job(
"memory_property_update",
"memory_update_meta",
path=str(event_index),
key="status",
value="distilled",
patch={"status": "distilled"},
force=True,
),
)
@ -453,28 +500,28 @@ async def check_schema_status_force_bypass(ctx: AppContext) -> str:
# (label, async fn) — runner executes top-to-bottom; later checks may
# rely on side effects from earlier ones (e.g. sync.append needs
# sync.create to have run).
# rely on side effects from earlier ones.
CHECKS: list[tuple[str, callable]] = [
("registry", check_registry),
("memory_get", check_memory_get),
("memory_list", check_memory_list),
("file_list", check_file_list),
("memory_search", check_memory_search),
("memory_graph_search", check_memory_graph_search),
("memory_links", check_memory_links),
("memory_backlinks", check_memory_backlinks),
("memory_resolve_wikilink", check_memory_resolve_wikilink),
("memory_count_tokens", check_memory_count_tokens),
("memory_lint", check_memory_lint),
("graph_traverse.out", check_graph_traverse_out),
("graph_traverse.in", check_graph_traverse_in),
("lint.dangling", check_lint_dangling),
("lint.orphans", check_lint_orphans),
("lint.collisions", check_lint_collisions),
("lint.schema", check_lint_schema),
("sync.create", check_sync_create),
("sync.append", check_sync_append),
("sync.refuse_distilled", check_sync_refuse_distilled),
("memory_create", check_memory_create),
("memory_update", check_memory_update),
("memory_property_update", check_memory_property_update),
("memory_rename", check_memory_rename),
("memory_archive", check_memory_archive),
("memory_delete", check_memory_delete),
("memory_update_body", check_memory_update_body),
("memory_update_meta", check_memory_update_meta),
("file_upload+download", check_file_upload_then_download),
("file_move", check_file_move),
("file_delete", check_file_delete),
("schema.path_template_refuses", check_schema_path_template_refuses),
("schema.path_template_force", check_schema_path_template_force_bypass),
("schema.status_skip_refused", check_schema_status_skip_refused),

2
reme2/mcp/test/test_service.py
View file

@ -152,7 +152,7 @@ async def check_remember_log_append(ctx: AppContext) -> str:
async def check_remember_log_refuse_distilled(ctx: AppContext) -> str:
"""Curated profile lacks `memory_property_update`, so we flip status by
"""Curated profile lacks `memory_update_meta`, so we flip status by
rewriting the file directly — same observable effect on remember(mode=log)."""
index_path = Path(getattr(ctx, "_event_index"))
text = index_path.read_text(encoding="utf-8")

66
reme2/memory/__init__.py
View file

@ -1,32 +1,48 @@
"""Memory subsystem — agent-facing services + tools on top of the core engine.
Three services (read / cold-write / treatment) plus the agent-callable
write/read tools that operate on the Memory File System. Every
``@R.register`` step a host agent might invoke lives here, regardless
of whether it's reached via MCP, HTTP, or direct Python.
tools that operate on the Memory File System. Every ``@R.register``
step a host agent might invoke lives here, regardless of whether it's
reached via MCP, HTTP, or direct Python.
Services
retriever.py Read service — V + K + Graph BFS fusion + intent
routing. Registered as ``hybrid``; consumed by
the search shells below.
``memory_search`` / ``memory_graph_search``.
ingestor.py Cold-write service — LLM-driven R-M-W curator
(also registers an ``ingest`` MCP face).
maintainer.py Treatment service — Merge / Split / Decay / Lint,
woken by cron or thresholds.
woken by cron or thresholds. Composes the
atomic ``check_*`` primitives below.
summarizer.py Auxiliary used by the services.
Tools (each ``@R.register`` exposes a step the agent invokes by name)
memory_io.py Pure engine API (no schema, no Step boilerplate).
memory_toolkit.py 12 ``memory_*`` primitives — get / list / links /
backlinks / resolve_wikilink / count_tokens +
create / update / property_update / rename /
delete / archive.
memory_search.py ``memory_search`` + ``memory_graph_search`` —
shells over the Retriever service.
memory_lint.py ``memory_lint`` — narrowed projection of the
Maintainer's lint pass.
sync.py ``sync`` — hot-path event-folder upsert
(deterministic, no LLM).
Tool surfaces (each ``@R.register`` exposes a step the agent
invokes by name).
agent_toolkit.py The 11 agent tools across three categories
(build via ``build_agent_toolkit``):
memory_* get / create / update_body /
update_meta / search
file_* download / upload / delete /
list / move
graph_* traverse
Plus ``memory_graph_search`` (MCP-only,
not in the agent toolkit binding).
lint_toolkit.py Atomic vault-health checks (build via
``build_lint_toolkit``):
check_dangling
check_orphans
check_collisions
check_schema
Read-only, separate category — for maintainer
/ CLI / scheduled-job use.
sync.py ``sync`` — hot-path event-folder upsert
(deterministic, no LLM).
memory_io.py Pure engine API — no schema, no Step
boilerplate. The toolkits wrap it.
Importing this package triggers every ``@R.register`` so configs that
name these tools resolve at boot.
@ -35,7 +51,17 @@ name these tools resolve at boot.
from . import retriever # noqa: F401 -- @R.register("hybrid")
from . import ingestor # noqa: F401 -- @R.register("ingestor")
from . import maintainer # noqa: F401 -- @R.register("maintainer")
from . import memory_toolkit # noqa: F401 -- @R.register("memory_*") x12
from . import memory_search # noqa: F401 -- @R.register("memory_search", "memory_graph_search")
from . import memory_lint # noqa: F401 -- @R.register("memory_lint")
from . import agent_toolkit # noqa: F401 -- 11 agent tools + memory_graph_search
from . import lint_toolkit # noqa: F401 -- 4 check_* atomic primitives
from . import sync # noqa: F401 -- @R.register("sync")
from .agent_toolkit import AGENT_TOOL_NAMES, build_agent_toolkit
from .lint_toolkit import LINT_TOOL_NAMES, build_lint_toolkit
__all__ = [
"AGENT_TOOL_NAMES",
"LINT_TOOL_NAMES",
"build_agent_toolkit",
"build_lint_toolkit",
]

783
reme2/memory/agent_toolkit.py Normal file
View file

@ -0,0 +1,783 @@
"""Agent toolkit — the 11 tools an agent uses to operate on the vault.
Three categories. Each tool is a single-purpose ``BaseStep`` exposing
two surfaces:
* ``execute()`` — the MCP transport surface (reads
``RuntimeContext`` parameters, writes via ``_set_answer``).
* a method named after the tool (e.g. ``memory_get``) — the
agentscope toolkit surface; agentscope introspects the signature
directly, no separate JSON schema.
Categories:
Memory (5) schema-bound markdown management
memory_get / memory_create / memory_update_body /
memory_update_meta / memory_search
File (5) type-agnostic vault transport + directory operations
file_download / file_upload / file_delete / file_list /
file_move
Graph (1) relationship exploration via BFS
graph_traverse
`memory_graph_search` is also defined here as an MCP-only tool (no
agent toolkit method); it stays out of the 11-tool agent surface but
is registered for MCP/HTTP callers that want graph-aware retrieval.
Atomic maintenance/check tools live in ``lint_toolkit.py`` —
separate category, separate factory, NOT bound to the agent toolkit
by default.
"""
from __future__ import annotations
import json
import mimetypes
import shutil
import tempfile
from collections import deque
from pathlib import Path
from typing import Any
import frontmatter
from agentscope.tool import Toolkit, ToolResponse
from . import memory_io
from ..component import R
from ..component.base_step import BaseStep
from .retriever import BaseRetriever, HybridRetriever
from .runtime_response import _set_answer, _tool_response, _to_jsonable
from ..enumeration import ComponentEnum
# ===========================================================================
# Section 1 — Schema policy (status state machine + path templates)
# ===========================================================================
#
# Used by memory_create (path template) and memory_update_meta (status
# state machine). Pure helpers; the gates fire only when force=False.
_STATUS_STATES = ("active", "distilled", "archived")
_STATUS_TRANSITIONS: dict[str, set[str]] = {
"active": {"active", "distilled"},
"distilled": {"distilled", "archived"},
"archived": {"archived"},
}
def validate_status_transition(prior, requested) -> str | None:
"""Return error string if the requested status transition is invalid."""
if requested is None:
return None
if requested not in _STATUS_STATES:
return f"invalid status {requested!r}; must be one of {list(_STATUS_STATES)}"
if prior in _STATUS_STATES and requested not in _STATUS_TRANSITIONS[prior]:
return (
f"status transition {prior!r} → {requested!r} not allowed; "
f"state machine is single-direction "
f"active → distilled → archived"
)
return None
def validate_path_template(path: Path, working_dir: Path | None) -> str | None:
"""Return error string if `path` doesn't match an agent-facing template.
Allowed templates (relative to working_dir):
topics/{folder}/{name}.md — topic file
events/{date}/{name}/{filename} — event index OR sibling material
Archive/... — archive moves can land anywhere
"""
if working_dir is None:
return None
try:
rel = path.resolve().relative_to(working_dir)
except ValueError:
return f"path {path} is outside working_dir {working_dir}"
parts = rel.parts
if not parts:
return "path has no components relative to working_dir"
head = parts[0]
if head == "Archive":
return None
if head == "topics" and len(parts) >= 3:
return None
if head == "events" and len(parts) >= 4:
return None
return (
f"path {rel} doesn't match a known template — expected one of: "
f"topics/{{folder}}/{{name}}.md, "
f"events/{{date}}/{{name}}/{{filename}}, or Archive/..."
)
def _update_status(path: Path | str, *, value, force: bool = False) -> tuple[bool, dict]:
"""Schema-aware status flip. Reads current status, validates the
transition, then delegates to ``memory_io.update_meta``."""
target = Path(path)
if not force:
prior = None
if target.is_file():
try:
prior = frontmatter.loads(
target.read_text(encoding="utf-8"),
).metadata.get("status")
except Exception:
prior = None
err = validate_status_transition(prior, value)
if err is not None:
return False, {
"path": str(target),
"key": "status",
"error": err,
"prior": prior,
"requested": value,
}
return memory_io.update_meta(target, key="status", value=value)
def _create_with_schema(
file_store,
path: Path,
*,
metadata: dict,
content: str,
overwrite: bool = False,
force: bool = False,
) -> tuple[bool, dict]:
"""Schema-aware file create — path template gate then engine."""
if not force:
working_dir = getattr(file_store, "working_dir", None)
template_err = validate_path_template(path, working_dir)
if template_err is not None:
return False, {
"path": str(path),
"error": template_err,
"hint": (
"place topics under topics/{folder}/{name}.md and "
"events under events/{date}/{name}/...; pass "
"force=true only if you intentionally need a "
"non-template path"
),
}
return memory_io.create_file(
file_store, path,
metadata=metadata, content=content,
overwrite=overwrite, force=force,
)
# ===========================================================================
# Section 2 — File-IO support (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."""
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 3 — Memory category (5 tools)
# ===========================================================================
@R.register("memory_get")
class MemoryGet(BaseStep):
"""Read a single memory file (frontmatter + body, optional chunks)."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
include_chunks: bool = bool(self.context.get("include_chunks", False))
assert path, "path is required"
result = await memory_io.get_file(self.file_store, path, include_chunks=include_chunks)
_set_answer(self.context, result)
async def memory_get(self, path: str, include_chunks: bool = False) -> ToolResponse:
"""Read a single memory file (frontmatter + body, optional chunks)."""
result = await memory_io.get_file(self.file_store, path, include_chunks=include_chunks)
return _tool_response("memory_get", True, result, audit=self.audit)
@R.register("memory_create")
class MemoryCreate(BaseStep):
"""Create a markdown file. Path-template gate + wikilink-uniqueness
gate fire unless ``force=True``."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
metadata: dict = dict(self.context.get("metadata") or {})
content: str = self.context.get("content", "") or ""
overwrite: bool = bool(self.context.get("overwrite", False))
force: bool = bool(self.context.get("force", False))
assert path, "path is required"
target = Path(path)
ok, payload = _create_with_schema(
self.file_store, target,
metadata=metadata, content=content,
overwrite=overwrite, force=force,
)
self.context.response.success = ok
if ok:
payload = {**payload, "path": str(target.resolve())}
_set_answer(self.context, payload)
async def memory_create(
self,
path: str,
metadata: dict | None = None,
content: str = "",
overwrite: bool = False,
force: bool = False,
) -> ToolResponse:
"""Create a markdown file. Path template + wikilink uniqueness
gates fire unless ``force=True``."""
target = Path(path)
ok, payload = _create_with_schema(
self.file_store, target,
metadata=dict(metadata or {}), content=content,
overwrite=overwrite, force=force,
)
if ok:
payload = {**payload, "path": str(target.resolve())}
return _tool_response("memory_create", ok, payload, audit=self.audit)
@R.register("memory_update_body")
class MemoryUpdateBody(BaseStep):
"""Edit-style body update: replace ``old_string`` with ``new_string``.
Frontmatter is preserved verbatim."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
old_string: str = self.context.get("old_string", "") or ""
new_string: str = self.context.get("new_string", "") or ""
replace_all: bool = bool(self.context.get("replace_all", False))
assert path, "path is required"
ok, payload = memory_io.update_body(
path, old_string=old_string, new_string=new_string, replace_all=replace_all,
)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_update_body(
self,
path: str,
old_string: str,
new_string: str,
replace_all: bool = False,
) -> ToolResponse:
"""Edit-style body update: replace ``old_string`` with ``new_string``."""
ok, payload = memory_io.update_body(
path, old_string=old_string, new_string=new_string, replace_all=replace_all,
)
return _tool_response("memory_update_body", ok, payload, audit=self.audit)
@R.register("memory_update_meta")
class MemoryUpdateMeta(BaseStep):
"""Frontmatter patch (merge). value=None deletes the key.
``status`` transitions go through the state-machine validator
unless ``force=True``."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
patch: dict = dict(self.context.get("patch") or {})
force: bool = bool(self.context.get("force", False))
assert path, "path is required"
ok, payload = await self._apply(path, patch, force)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_update_meta(
self,
path: str,
patch: dict,
force: bool = False,
) -> ToolResponse:
"""Frontmatter patch (merge). value=None deletes the key."""
ok, payload = await self._apply(path, dict(patch or {}), force)
return _tool_response("memory_update_meta", ok, payload, audit=self.audit)
async def _apply(self, path: str, patch: dict, force: bool) -> tuple[bool, dict]:
results: dict[str, dict] = {}
all_ok = True
for key, value in patch.items():
if key == "status":
ok, payload = _update_status(path, value=value, force=force)
else:
ok, payload = memory_io.update_meta(path, key=key, value=value)
results[key] = payload
if not ok:
all_ok = False
break # stop on first failure; partial state already on disk
return all_ok, {"path": path, "applied": results}
# ----- memory_search (retrieval) ------------------------------------------
_RETRIEVER_CACHE: dict[int, BaseRetriever] = {}
def _resolve_retriever(step: BaseStep) -> BaseRetriever:
"""Get (or build) the retriever instance for this step."""
cached = _RETRIEVER_CACHE.get(id(step))
if cached is not None:
return cached
retriever = R.get(ComponentEnum.RETRIEVER, "hybrid")
if retriever is None:
retriever = HybridRetriever(app_context=step.app_context)
elif isinstance(retriever, type):
retriever = retriever(app_context=step.app_context)
_RETRIEVER_CACHE[id(step)] = retriever
return retriever
def _serialize_chunk(chunk, file_store, extras: dict | None = None) -> dict:
"""Flatten a FileChunk into a dict, joining file metadata."""
item = chunk.model_dump() if hasattr(chunk, "model_dump") else dict(chunk)
node = file_store.file_nodes.get(item.get("path"))
if node is not None:
meta = node.front_matter.model_dump()
item["file_metadata"] = meta
item["file_st_mtime"] = node.st_mtime
else:
item["file_metadata"] = {}
item["file_st_mtime"] = None
if extras:
item.update(extras)
return item
@R.register("memory_search")
class MemorySearch(BaseStep):
"""Pure-relevance retrieval (V + K hybrid). Delegates to the Retriever."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
query: str = self.context.get("query", "").strip()
min_score: float = self.context.get("min_score", 0.1)
max_results: int = self.context.get("max_results", 5)
assert query, "Query cannot be empty"
assert 0.0 <= min_score <= 1.0, f"min_score must be in [0,1], got {min_score}"
assert max_results > 0, f"max_results must be positive, got {max_results}"
chunk_filter = memory_io.make_filter(
self.file_store,
paths=self.context.get("paths") or None,
tags=self.context.get("tags") or None,
exclude_paths=self.context.get("exclude_paths") or None,
)
retriever = _resolve_retriever(self)
results = await retriever.search(
query=query, max_results=max_results, min_score=min_score, chunk_filter=chunk_filter,
)
payload = [_serialize_chunk(r, self.file_store) for r in results]
_set_answer(self.context, payload)
async def memory_search(
self,
query: str,
max_results: int = 5,
min_score: float = 0.1,
paths: list[str] | None = None,
tags: list[str] | None = None,
exclude_paths: list[str] | None = None,
) -> ToolResponse:
"""Pure-relevance retrieval (V + K hybrid)."""
chunk_filter = memory_io.make_filter(
self.file_store, paths=paths, tags=tags, exclude_paths=exclude_paths,
)
retriever = _resolve_retriever(self)
results = await retriever.search(
query=query, max_results=max_results, min_score=min_score, chunk_filter=chunk_filter,
)
payload = [_serialize_chunk(r, self.file_store) for r in results]
return _tool_response("memory_search", True, payload, audit=self.audit)
# ===========================================================================
# Section 4 — File category (5 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}
# ===========================================================================
# Section 5 — Graph category (1 tool)
# ===========================================================================
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.
Cheap for vault sizes; if it ever becomes hot, swap for a precomputed
reverse index on the file_graph component.
"""
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_traverse(
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
@R.register("graph_traverse")
class GraphTraverse(BaseStep):
"""BFS from seed(s) to explore relationships in the memory graph.
Output: one record per edge traversed (same node may appear
multiple times if reached via different predicates or paths).
"""
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_traverse(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). Args:
seeds: single path or list 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_traverse(self.file_store, seeds_list, max_depth, direction, predicate)
return _tool_response("graph_traverse", True, results, audit=self.audit)
# ===========================================================================
# Section 6 — Toolkit factory
# ===========================================================================
# The 11 tools the agent gets bound to. memory_graph_search stays
# registered for MCP/HTTP but is intentionally NOT in the agent surface
# (per-call retrieval-knob tuning is internal).
AGENT_TOOL_NAMES: tuple[str, ...] = (
# memory (5)
"memory_get",
"memory_create",
"memory_update_body",
"memory_update_meta",
"memory_search",
# file (5)
"file_download",
"file_upload",
"file_delete",
"file_list",
"file_move",
# graph (1)
"graph_traverse",
)
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

10
reme2/memory/ingestor.py
View file

@ -5,7 +5,7 @@ write entry point** to the markdown vault. Every mutation (create, body
edit, frontmatter flip, rename, delete, archive) flows through here.
Mirrors `Summarizer`'s pattern — drives a `ReActAgent` whose toolkit is
built by `memory_toolkit.build_memory_toolkit`. The agent runs its own
built by `agent_toolkit.build_agent_toolkit`. The agent runs its own
R-M-W loop: read related files via tools, decide which ones to mutate,
call the right write tool. Every write tool records into an audit list,
so the caller gets a deterministic mutation trail regardless of how the
@ -28,14 +28,14 @@ from agentscope.message import Msg
from agentscope.tool import Toolkit
from pydantic import BaseModel, Field
from ..component.runtime_response import _set_answer, _to_jsonable
from .runtime_response import _set_answer, _to_jsonable
from . import memory_io
from .memory_io import create_file
from .memory_toolkit import build_memory_toolkit
from .agent_toolkit import build_agent_toolkit
from ..component import R
from ..component.base_step import BaseStep
from ..enumeration import ComponentEnum
from ..schema import extract_wikilinks
from ..utils.wikilink_resolver import extract_wikilinks
class IngestResult(BaseModel):
@ -138,7 +138,7 @@ class Ingestor(BaseStep):
working_dir = self._working_dir()
audit: list[dict] = []
toolkit = build_memory_toolkit(self.app_context, audit=audit, toolkit=self.toolkit)
toolkit = build_agent_toolkit(self.app_context, audit=audit, toolkit=self.toolkit)
agent = ReActAgent(
name="reme_ingestor",

248
reme2/memory/lint_toolkit.py Normal file
View file

@ -0,0 +1,248 @@
"""Lint toolkit — atomic vault-health checks.
Separate from the agent toolkit. These are read-only diagnostics for
maintainer / CLI / scheduled-job use; agents typically don't need
them in their per-call working set.
Four atomic checks, each does one thing and returns pure data:
check_dangling — FileLinks pointing to non-existent nodes
check_orphans — nodes with no inlinks AND no outlinks
check_collisions — basenames resolving to >1 path (short-link
ambiguity)
check_schema — nodes violating frontmatter schema
(missing required fields, invalid status)
Maintainer compositions live in ``maintainer.py``; this file is the
underlying primitives. Each tool is also independently MCP/agent
callable for ad-hoc checks.
Bind via ``build_lint_toolkit`` (parallel to ``build_agent_toolkit``).
"""
from __future__ import annotations
from pathlib import Path
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 ..enumeration import ComponentEnum
# Frontmatter schema — required keys for a well-formed memory file.
# Mirrors the validator that fires inside `memory_create`. Keep in sync
# with the path-template + status-state-machine rules in
# ``agent_toolkit``.
_REQUIRED_META_KEYS = ("title", "lifecycle", "scope", "source", "role")
_VALID_STATUS = {"active", "distilled", "archived"}
# ===========================================================================
# Section 1 — Atomic check primitives (pure functions; no side effects)
# ===========================================================================
def _scan_dangling(file_store) -> list[dict]:
"""Find FileLinks pointing to nodes that don't exist in the index.
Returns one entry per dangling edge:
{"source": <path>, "target": <broken_link_path>,
"predicate": <str|None>, "anchor": <str|None>}
"""
nodes = file_store.file_nodes
out: list[dict] = []
for src_path, node in nodes.items():
for link in node.links:
if not link.path:
continue
if link.path not in nodes:
out.append({
"source": src_path,
"target": link.path,
"predicate": link.predicate,
"anchor": link.anchor,
})
return out
def _scan_orphans(file_store) -> list[str]:
"""Find nodes with no outlinks AND no inlinks.
O(N + E): one pass to mark which paths are referenced by anyone.
"""
nodes = file_store.file_nodes
referenced: set[str] = set()
for node in nodes.values():
for link in node.links:
if link.path:
referenced.add(link.path)
out: list[str] = []
for path, node in nodes.items():
has_out = any(link.path for link in node.links)
has_in = path in referenced
if not has_out and not has_in:
out.append(path)
return sorted(out)
def _scan_collisions(file_store) -> dict[str, list[str]]:
"""Find basenames (filename + ext) that resolve to >1 path.
Mirrors ``utils.wikilink_resolver.collisions`` but operates over
the local file_store index (no async iteration needed)."""
by_name: dict[str, list[str]] = {}
for path in file_store.file_nodes:
by_name.setdefault(Path(path).name, []).append(path)
return {name: sorted(paths) for name, paths in by_name.items() if len(paths) > 1}
def _scan_schema(file_store) -> list[dict]:
"""Find nodes whose frontmatter violates schema.
Each entry: {"path": <path>, "errors": [<error_msg>, ...]}.
"""
out: list[dict] = []
for path, node in file_store.file_nodes.items():
meta = node.front_matter.model_dump()
errs: list[str] = []
missing = [k for k in _REQUIRED_META_KEYS if not meta.get(k)]
if missing:
errs.append(f"missing required: {missing}")
status = meta.get("status")
if status is not None and status not in _VALID_STATUS:
errs.append(f"invalid status: {status!r} (expected one of {sorted(_VALID_STATUS)})")
if errs:
out.append({"path": path, "errors": errs})
return out
# ===========================================================================
# Section 2 — Step wrappers (one per check)
# ===========================================================================
@R.register("check_dangling")
class CheckDangling(BaseStep):
"""List every FileLink whose target is not in the graph."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
findings = _scan_dangling(self.file_store)
_set_answer(self.context, {"count": len(findings), "findings": findings})
async def check_dangling(self) -> ToolResponse:
"""List every FileLink whose target is not in the graph."""
findings = _scan_dangling(self.file_store)
return _tool_response(
"check_dangling", True,
{"count": len(findings), "findings": findings},
audit=self.audit,
)
@R.register("check_orphans")
class CheckOrphans(BaseStep):
"""List nodes with no inlinks AND no outlinks."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
orphans = _scan_orphans(self.file_store)
_set_answer(self.context, {"count": len(orphans), "paths": orphans})
async def check_orphans(self) -> ToolResponse:
"""List nodes with no inlinks AND no outlinks."""
orphans = _scan_orphans(self.file_store)
return _tool_response(
"check_orphans", True,
{"count": len(orphans), "paths": orphans},
audit=self.audit,
)
@R.register("check_collisions")
class CheckCollisions(BaseStep):
"""List basenames resolving to >1 path (short-link ambiguity)."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
groups = _scan_collisions(self.file_store)
_set_answer(self.context, {"count": len(groups), "groups": groups})
async def check_collisions(self) -> ToolResponse:
"""List basenames resolving to >1 path (short-link ambiguity)."""
groups = _scan_collisions(self.file_store)
return _tool_response(
"check_collisions", True,
{"count": len(groups), "groups": groups},
audit=self.audit,
)
@R.register("check_schema")
class CheckSchema(BaseStep):
"""List nodes whose frontmatter violates the memory schema."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
findings = _scan_schema(self.file_store)
_set_answer(self.context, {"count": len(findings), "findings": findings})
async def check_schema(self) -> ToolResponse:
"""List nodes whose frontmatter violates the memory schema."""
findings = _scan_schema(self.file_store)
return _tool_response(
"check_schema", True,
{"count": len(findings), "findings": findings},
audit=self.audit,
)
# ===========================================================================
# Section 3 — Toolkit factory
# ===========================================================================
LINT_TOOL_NAMES: tuple[str, ...] = (
"check_dangling",
"check_orphans",
"check_collisions",
"check_schema",
)
def build_lint_toolkit(
app_context,
audit: list[dict] | None = None,
toolkit: Toolkit | None = None,
) -> Toolkit:
"""Bind every check_* step's tool method to an agentscope ``Toolkit``.
Parallel to ``build_agent_toolkit`` but separate — these are
maintainer/CLI tools and aren't bound into the agent's working set
by default. Hosts that want a single toolkit with everything can
pass the result of ``build_agent_toolkit`` as the ``toolkit``
argument here.
"""
toolkit = toolkit or Toolkit()
for name in LINT_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

2
reme2/memory/maintainer.py
View file

@ -69,7 +69,7 @@ from pydantic import BaseModel, Field
from ..component import R
from ..component.base_step import BaseStep
from ..component.runtime_response import _set_answer
from .runtime_response import _set_answer
from . import memory_io
from .schema import parse_frontmatter

63
reme2/memory/memory_io.py
View file

@ -39,11 +39,10 @@ from pathlib import Path
import frontmatter
from ..component.file_store.base_file_store import BaseFileStore
from ..schema import ChunkFilter, FileChunk, FileLink, FileNode, extract_wikilinks
from ..schema.file_link import _WIKILINK_RE
from ..schema import ChunkFilter, FileChunk, FileNode
from ..utils.wikilink_resolver import (
resolve_wikilink as _resolve_wikilink,
wikilink_candidates,
_WIKILINK_RE,
extract_wikilinks,
)
@ -59,50 +58,11 @@ from ..utils.wikilink_resolver import (
# place to swap.
def _nodes(file_store: BaseFileStore) -> dict[str, FileNode]:
"""The concrete in-memory ``{path: FileNode}`` index."""
return file_store._nodes # type: ignore[attr-defined]
def _meta(node: FileNode) -> dict:
"""Full frontmatter dict for a node — typed fields (title/description/
tags) merged with any ``extra=allow`` extras."""
return node.front_matter.model_dump()
def _get_outlinks(
file_store: BaseFileStore,
path: str,
) -> list[tuple[FileNode, FileLink]]:
"""Resolved outgoing links from ``path`` — ``[(target_node, link), ...]``."""
node = _nodes(file_store).get(path)
if node is None:
return []
out: list[tuple[FileNode, FileLink]] = []
for link in node.links:
target = _resolve_wikilink(file_store, link.path)
if target is None:
continue
target_node = _nodes(file_store).get(target)
if target_node is not None:
out.append((target_node, link))
return out
def _get_inlinks(
file_store: BaseFileStore,
path: str,
) -> list[tuple[FileNode, FileLink]]:
"""Resolved incoming links to ``path`` — linear scan over all nodes."""
out: list[tuple[FileNode, FileLink]] = []
for src_node in _nodes(file_store).values():
for link in src_node.links:
target = _resolve_wikilink(file_store, link.path)
if target == path:
out.append((src_node, link))
return out
def _filter_to_dict(chunk_filter: ChunkFilter | None) -> dict:
"""Serialize ``ChunkFilter`` for the search engine's ``search_filter`` arg."""
if chunk_filter is None:
@ -179,28 +139,19 @@ def list_files(
return {"items": items, "count": len(items)}
def _link_to_dict(node: FileNode, link: FileLink) -> dict:
return {
"path": node.path,
"metadata": _meta(node),
"predicate": link.predicate,
"anchor": link.anchor,
}
def get_links(file_store: BaseFileStore, path: str) -> dict:
def get_inlinks(file_store: BaseFileStore, path: str) -> dict:
"""Files that `path` links TO (resolved). Each entry carries the typed-link predicate."""
return {
"path": path,
"links": [_link_to_dict(m, link) for m, link in _get_outlinks(file_store, path)],
"inlinks": file_store.file_graph.get_inlinks(path),
}
def get_backlinks(file_store: BaseFileStore, path: str) -> dict:
def get_outlinks(file_store: BaseFileStore, path: str) -> dict:
"""Files that link TO `path`. Each entry carries the typed-link predicate."""
return {
"path": path,
"backlinks": [_link_to_dict(m, link) for m, link in _get_inlinks(file_store, path)],
"outlinks": file_store.file_graph.get_outlinks(path),
}

73
reme2/memory/memory_lint.py
View file

@ -1,73 +0,0 @@
"""memory_lint — read-only projection of the Maintainer's lint findings.
Tier A surface for the Maintainer service. The host agent calls this
to discover what's wrong with the vault (broken wikilinks, schema
violations, stem collisions) and decides what to do with each finding
using the existing memory_* write primitives.
Equivalent to invoking `Maintainer.execute(ops=["lint"], dry_run=True)`
but with a focused response shape and a tighter parameter surface — the
agent doesn't need to know about decay/merge/split knobs.
"""
from __future__ import annotations
from ..component import R
from ..component.base_step import BaseStep
from ..component.runtime_response import _set_answer
from ..enumeration import ComponentEnum
@R.register("memory_lint")
class MemoryLint(BaseStep):
"""Run the Maintainer's lint pass and surface findings only.
Inputs (RuntimeContext, all optional):
target_prefix (str): restrict scan to relpaths starting with
this prefix (e.g. "events/2026-05-09/").
Output (context.response.answer):
{
"scanned": int, # files inspected
"findings": [LintFinding, ...], # each {path, kind, detail}
"target_prefix": str,
"ran_at": iso,
}
"""
async def execute(self):
assert self.context is not None
target_prefix = str(self.context.get("target_prefix") or "")
# Delegate to a Maintainer step. Force ops=["lint"] + dry_run so
# we never mutate. The Maintainer reads these from the context
# and produces a full audit; we narrow the response shape below.
if getattr(self, "_maintainer", None) is None:
self._maintainer = R.get(ComponentEnum.STEP, "maintainer")(
app_context=self.app_context,
)
self.context["ops"] = ["lint"]
self.context["dry_run"] = True
self.context["target_prefix"] = target_prefix
await self._maintainer(self.context)
# The Maintainer wrote a full audit to context.response.answer
# (proposed/plan/applied/skipped/failed/...). For lint, all the
# action lives in `proposed` (LintFindings never get applied or
# dropped). Reshape into a focused response.
import json
raw = self.context.response.answer
audit = json.loads(raw) if isinstance(raw, str) else (raw or {})
findings = audit.get("proposed") or []
_set_answer(
self.context,
{
"scanned": audit.get("scanned", 0),
"findings": findings,
"target_prefix": target_prefix,
"ran_at": audit.get("ran_at", ""),
},
)
self.context.response.success = True

206
reme2/memory/memory_search.py
View file

@ -1,206 +0,0 @@
"""Memory search step shells — thin wrappers over the Retriever service.
Per the architecture blueprint, retrieval policy (V+K hybrid + graph
BFS fusion + ranking + intent routing) lives in the **Retriever
service** (`reme2.memory.retriever`), registered as
``ComponentEnum.RETRIEVER``. The two steps here are the agent-facing
projection: they translate ``RuntimeContext`` (paths/tags/exclude_paths
filter, per-call knob overrides) into the retriever's call surface,
then serialize results for the response (joining file metadata onto
each chunk so callers don't have to fire a ``memory_get`` per hit).
Direct primary-key file ops (read/list/links/backlinks) live in
``memory_io`` — those aren't retrieval.
"""
from __future__ import annotations
from ..component import R
from ..component.base_step import BaseStep
from ..component.runtime_response import _set_answer
from ..enumeration import ComponentEnum
from . import memory_io
from .retriever import BaseRetriever, HybridRetriever
# Per-shell singleton cache: instantiating the retriever is cheap (it
# just stashes constructor knobs), but doing it once per call would
# still allocate every request. Keyed by step instance so each MCP
# job's overrides stay isolated.
_RETRIEVER_CACHE: dict[int, BaseRetriever] = {}
def _resolve_retriever(step: BaseStep) -> BaseRetriever:
"""Get (or build) the retriever instance for this MCP step.
The retriever is a registered Step (`@R.register("hybrid")`), but
it isn't pre-instantiated as a singleton component — there's no
RETRIEVER enum slot. Instead, each MCP shell builds its own
HybridRetriever the first time it's called, sharing the calling
step's `app_context` (so the lookup of `file_store`/`as_llm` works)
and forwarding any retriever knobs (`vector_weight`, `graph_*`, …)
from the step's kwargs as constructor defaults.
Callers can also pass a pre-built `BaseRetriever` instance via
`kwargs["retriever"]` — useful for tests / Python callers that want
to inject a stub.
"""
injected = step.kwargs.get("retriever")
if isinstance(injected, BaseRetriever):
return injected
cached = _RETRIEVER_CACHE.get(id(step))
if cached is not None:
return cached
backend = injected if isinstance(injected, str) else "hybrid"
cls = R.get(ComponentEnum.STEP, backend)
if cls is None or not (isinstance(cls, type) and issubclass(cls, BaseRetriever)):
# Fall back to the canonical implementation; lets configs that
# don't override `retriever` work out of the box.
cls = HybridRetriever
knob_keys = (
"vector_weight",
"graph_weight",
"graph_depth",
"graph_decay",
"graph_direction",
"graph_mode",
"graph_per_path_cap",
"candidate_multiplier",
"anchor_expand",
"file_store",
)
init_kwargs = {k: step.kwargs[k] for k in knob_keys if k in step.kwargs}
init_kwargs["app_context"] = step.app_context
instance = cls(**init_kwargs)
_RETRIEVER_CACHE[id(step)] = instance
return instance
def _serialize_chunk(chunk, file_store, extras: dict | None = None) -> dict:
"""Serialize a FileChunk for search-result payloads.
Joins the owning file's metadata (frontmatter + st_mtime) so callers
don't have to fire a `memory_get` per result just to read `category`,
`status`, dates, etc. The join is a single in-memory dict lookup —
cost is negligible vs. the round-trip we save.
`extras` lets the caller attach step-specific fields (e.g. `graph_hop`).
"""
item = chunk.model_dump(exclude_none=True, exclude={"embedding"})
node = file_store._nodes.get(chunk.path) # engine-layer peer access
if node is not None:
item["file_metadata"] = node.front_matter.model_dump()
item["file_st_mtime"] = node.st_mtime
else:
item["file_metadata"] = None
item["file_st_mtime"] = None
if extras:
item.update(extras)
return item
@R.register("memory_search")
class MemorySearch(BaseStep):
"""Pure-relevance retrieval (V + K hybrid). Delegates to the Retriever service."""
async def execute(self):
assert self.context is not None, "Context is not set"
query: str = self.context.get("query", "").strip()
min_score: float = self.context.get("min_score", 0.1)
max_results: int = self.context.get("max_results", 5)
assert query, "Query cannot be empty"
assert (
isinstance(min_score, float | int) and 0.0 <= min_score <= 1.0
), f"min_score must be between 0 and 1, got {min_score}"
assert (
isinstance(max_results, int) and max_results > 0
), f"max_results must be a positive integer, got {max_results}"
chunk_filter = memory_io.make_filter(
self.file_store,
paths=self.context.get("paths") or None,
tags=self.context.get("tags") or None,
exclude_paths=self.context.get("exclude_paths") or None,
)
retriever = _resolve_retriever(self)
results = await retriever.search(
query=query,
max_results=max_results,
min_score=min_score,
chunk_filter=chunk_filter,
)
payload = [_serialize_chunk(r, self.file_store) for r in results]
_set_answer(self.context, payload)
@R.register("memory_graph_search")
class MemoryGraphSearch(BaseStep):
"""V + K + graph BFS fusion. Delegates to the Retriever service.
Per-call overrides for fusion knobs (vector_weight, graph_weight,
graph_depth, graph_decay, graph_direction, graph_mode,
graph_per_path_cap, anchor_expand) are forwarded if present in the
RuntimeContext; otherwise the retriever falls back to its own
constructor defaults. These knobs are intentionally NOT exposed via
MCP — they're internal-Python-caller tuning.
"""
_OVERRIDE_KEYS = (
"vector_weight",
"graph_weight",
"graph_depth",
"graph_decay",
"graph_direction",
"graph_mode",
"graph_per_path_cap",
"anchor_expand",
)
async def execute(self):
assert self.context is not None
ctx = self.context
query: str = ctx.get("query", "").strip()
max_results: int = int(ctx.get("max_results", 5))
min_score: float = float(ctx.get("min_score", 0.0))
explicit_seeds: list[str] = list(ctx.get("seeds") or [])
assert query or explicit_seeds, "query or seeds must be provided"
assert max_results > 0
chunk_filter = memory_io.make_filter(
self.file_store,
paths=ctx.get("paths") or None,
tags=ctx.get("tags") or None,
exclude_paths=ctx.get("exclude_paths") or None,
)
# Forward per-call overrides only if the caller actually set them;
# the retriever falls back to its own defaults for missing keys.
overrides = {k: ctx.get(k) for k in self._OVERRIDE_KEYS if ctx.get(k) is not None}
retriever = _resolve_retriever(self)
results, hops = await retriever.graph_search(
query=query,
seeds=explicit_seeds,
max_results=max_results,
min_score=min_score,
chunk_filter=chunk_filter,
**overrides,
)
payload = []
for c in results:
extras = {}
hop = hops.get(c.path)
if hop is not None:
extras["graph_hop"] = hop
payload.append(_serialize_chunk(c, self.file_store, extras))
_set_answer(ctx, payload)

607
reme2/memory/memory_toolkit.py
View file

@ -1,607 +0,0 @@
"""Memory toolkit — schema-bound projection of the Memory File System.
Layered on top of `reme2.memory.memory_io` (the pure core engine). One
`BaseStep` subclass per memory_* tool, each exposing TWO class methods:
* `execute()` — the MCP surface. Reads parameters from
`RuntimeContext` and writes the result through `_set_answer`.
* a method named after the tool (e.g. `memory_get`) — the agent
toolkit surface. Takes explicit parameters, returns a
`ToolResponse`. agentscope's `Toolkit.register_tool_function`
introspects the signature directly — no hand-authored JSON schema.
`build_memory_toolkit(app_context, audit, toolkit)` instantiates every
registered memory_* step and binds its tool method to a `Toolkit`.
Each instance carries an `audit` list so the host can surface what the
agent actually called.
Schema policy (status state machine, path templates) lives at the top
of this file as pure helpers; the relevant write tools (`memory_create`
/ `memory_property_update`) call them. `force=True` is the single
escape hatch — bypasses BOTH the policy gates here and the
wikilink-uniqueness gate downstream in `memory_io.create_file`.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import frontmatter
from agentscope.message import TextBlock
from agentscope.tool import Toolkit, ToolResponse
from . import memory_io
from ..component import R
from ..component.base_step import BaseStep
from ..component.runtime_response import _set_answer, _to_jsonable
from ..enumeration import ComponentEnum
# ===========================================================================
# Section 1 — Schema policy
# ===========================================================================
_STATUS_STATES = ("active", "distilled", "archived")
_STATUS_TRANSITIONS: dict[str, set[str]] = {
"active": {"active", "distilled"},
"distilled": {"distilled", "archived"},
"archived": {"archived"},
}
def validate_status_transition(prior, requested) -> str | None:
"""Return error string if the requested status transition is invalid,
else None. Files without a prior status accept any initial value
(so first-write doesn't get blocked)."""
if requested is None:
return None # delete operation
if requested not in _STATUS_STATES:
return f"invalid status {requested!r}; must be one of " f"{list(_STATUS_STATES)}"
if prior in _STATUS_STATES and requested not in _STATUS_TRANSITIONS[prior]:
return (
f"status transition {prior!r} → {requested!r} not allowed; "
f"state machine is single-direction "
f"active → distilled → archived"
)
return None
def validate_path_template(path: Path, working_dir: Path | None) -> str | None:
"""Return error string if `path` doesn't match a known agent-facing
template, else None. When `working_dir` is unknown, skip the check.
Allowed templates (relative to working_dir):
topics/{folder}/{name}.md — topic file
events/{date}/{name}/{filename} — event index OR sibling material
Archive/... — archive moves can land anywhere
"""
if working_dir is None:
return None
try:
rel = path.resolve().relative_to(working_dir)
except ValueError:
return f"path {path} is outside working_dir {working_dir}"
parts = rel.parts
if not parts:
return "path has no components relative to working_dir"
head = parts[0]
if head == "Archive":
return None
if head == "topics" and len(parts) >= 3:
return None
if head == "events" and len(parts) >= 4:
return None
return (
f"path {rel} doesn't match a known template — expected one of: "
f"topics/{{folder}}/{{name}}.md, "
f"events/{{date}}/{{name}}/{{filename}}, or Archive/..."
)
def update_status(
path: Path | str,
*,
value,
force: bool = False,
) -> tuple[bool, dict]:
"""Schema-aware status flip. Reads current status from disk, validates
the transition, then delegates to `memory_io.update_meta`."""
target = Path(path)
if not force:
prior = None
if target.is_file():
try:
prior = frontmatter.loads(
target.read_text(encoding="utf-8"),
).metadata.get("status")
except Exception:
prior = None
err = validate_status_transition(prior, value)
if err is not None:
return False, {
"path": str(target),
"key": "status",
"error": err,
"prior": prior,
"requested": value,
}
return memory_io.update_meta(target, key="status", value=value)
def create_file_with_schema(
file_store,
path: Path,
*,
metadata: dict,
content: str,
overwrite: bool = False,
force: bool = False,
) -> tuple[bool, dict]:
"""Schema-aware file create. Validates the path template (unless
`force=True`), then delegates to `memory_io.create_file` (which
still enforces the wikilink-uniqueness graph invariant — that gate
lives in the engine because it's structural, not business policy).
`force=True` bypasses BOTH the template gate here and the wikilink
gate downstream — it's the single escape hatch for any caller that
intentionally needs to step outside conventions.
"""
if not force:
working_dir = getattr(file_store, "working_dir", None)
template_err = validate_path_template(path, working_dir)
if template_err is not None:
return False, {
"path": str(path),
"error": template_err,
"hint": (
"place topics under topics/{folder}/{name}.md and "
"events under events/{date}/{name}/...; pass "
"force=true only if you intentionally need a "
"non-template path"
),
}
return memory_io.create_file(
file_store,
path,
metadata=metadata,
content=content,
overwrite=overwrite,
force=force,
)
# ===========================================================================
# Section 2 — Tool response helper
# ===========================================================================
def _tool_response(
op: str,
ok: bool,
payload: Any,
audit: list[dict] | None = None,
) -> ToolResponse:
"""Wrap a tool-method result as a `ToolResponse` and optionally
append an audit row."""
if audit is not None:
entry = {"op": op, "ok": ok}
if isinstance(payload, dict):
entry.update(payload)
else:
entry["result"] = payload
audit.append(entry)
text = json.dumps(_to_jsonable(payload), ensure_ascii=False, indent=2)
return ToolResponse(content=[TextBlock(type="text", text=text)])
# ===========================================================================
# Section 3 — Memory steps (one BaseStep per tool, two surfaces each)
# ===========================================================================
@R.register("memory_get")
class MemoryGet(BaseStep):
"""Read a single memory file (frontmatter + body, optional chunks)."""
audit: list[dict] | None = None # set by build_memory_toolkit
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
include_chunks: bool = bool(self.context.get("include_chunks", False))
assert path, "path is required"
result = await memory_io.get_file(self.file_store, path, include_chunks=include_chunks)
_set_answer(self.context, result)
async def memory_get(self, path: str, include_chunks: bool = False) -> ToolResponse:
"""Read a single memory file (frontmatter + body, optional chunks)."""
result = await memory_io.get_file(self.file_store, path, include_chunks=include_chunks)
return _tool_response("memory_get", True, result, audit=self.audit)
@R.register("memory_list")
class MemoryList(BaseStep):
"""List indexed files filtered by frontmatter fields, tags, or path prefix."""
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("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 memory_list(
self,
path_prefix: str | None = None,
tags: list[str] | None = None,
metadata: dict | None = None,
limit: int = 100,
) -> ToolResponse:
"""List indexed files filtered by frontmatter fields, tags, or path prefix."""
result = memory_io.list_files(
self.file_store,
path_prefix=path_prefix,
tags=tags or [],
metadata=metadata or {},
limit=limit,
)
return _tool_response("memory_list", True, result, audit=self.audit)
@R.register("memory_backlinks")
class MemoryBacklinks(BaseStep):
"""Files linking TO a given path. Each entry carries the typed-edge predicate."""
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"
_set_answer(self.context, memory_io.get_backlinks(self.file_store, path))
async def memory_backlinks(self, path: str) -> ToolResponse:
"""Files linking TO a given path. Each entry carries the typed-edge predicate."""
result = memory_io.get_backlinks(self.file_store, path)
return _tool_response("memory_backlinks", True, result, audit=self.audit)
@R.register("memory_links")
class MemoryLinks(BaseStep):
"""Files a given path links to (resolved). Each entry carries the typed-edge predicate."""
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"
_set_answer(self.context, memory_io.get_links(self.file_store, path))
async def memory_links(self, path: str) -> ToolResponse:
"""Files a given path links to (resolved). Each entry carries the typed-edge predicate."""
result = memory_io.get_links(self.file_store, path)
return _tool_response("memory_links", True, result, audit=self.audit)
@R.register("memory_resolve_wikilink")
class MemoryResolveWikilink(BaseStep):
"""Resolve a `[[target]]` wikilink with full ambiguity context."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
wikilink: str = self.context.get("wikilink", "") or ""
assert wikilink, "wikilink is required"
payload = memory_io.resolve_wikilink(self.file_store, wikilink)
self.context.response.success = bool(payload.get("exists"))
_set_answer(self.context, payload)
async def memory_resolve_wikilink(self, wikilink: str) -> ToolResponse:
"""Resolve a `[[target]]` wikilink with full ambiguity context."""
payload = memory_io.resolve_wikilink(self.file_store, wikilink)
return _tool_response(
"memory_resolve_wikilink",
bool(payload.get("exists")),
payload,
audit=self.audit,
)
@R.register("memory_create")
class MemoryCreate(BaseStep):
"""Create a markdown file. Path-template gate + wikilink-uniqueness
gate both fire unless `force=True`."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
metadata: dict = dict(self.context.get("metadata") or {})
content: str = self.context.get("content", "") or ""
overwrite: bool = bool(self.context.get("overwrite", False))
force: bool = bool(self.context.get("force", False))
assert path, "path is required"
target = Path(path)
ok, payload = create_file_with_schema(
self.file_store,
target,
metadata=metadata,
content=content,
overwrite=overwrite,
force=force,
)
self.context.response.success = ok
if ok:
payload = {**payload, "path": str(target.resolve())}
_set_answer(self.context, payload)
async def memory_create(
self,
path: str,
metadata: dict | None = None,
content: str = "",
overwrite: bool = False,
force: bool = False,
) -> ToolResponse:
"""Create a markdown file. Path-template gate + wikilink-uniqueness
gate both fire unless `force=True`."""
target = Path(path)
ok, payload = create_file_with_schema(
self.file_store,
target,
metadata=dict(metadata or {}),
content=content,
overwrite=overwrite,
force=force,
)
if ok:
payload = {**payload, "path": str(target.resolve())}
return _tool_response("memory_create", ok, payload, audit=self.audit)
@R.register("memory_delete")
class MemoryDelete(BaseStep):
"""Delete a file. Watcher removes from store + graph."""
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"
ok, payload = memory_io.delete_file(path)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_delete(self, path: str) -> ToolResponse:
"""Delete a file. Watcher removes from store + graph."""
ok, payload = memory_io.delete_file(path)
return _tool_response("memory_delete", ok, payload, audit=self.audit)
@R.register("memory_rename")
class MemoryRename(BaseStep):
"""Rename a file and rewrite incoming wikilinks across the vault."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
old_path: str = self.context.get("old_path", "") or ""
new_path: str = self.context.get("new_path", "") or ""
assert old_path and new_path, "old_path and new_path are required"
working_dir = Path(self.file_store.working_dir or Path.cwd()).resolve()
ok, payload = memory_io.rename_file(
self.file_store,
working_dir,
old_path=old_path,
new_path=new_path,
)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_rename(self, old_path: str, new_path: str) -> ToolResponse:
"""Rename a file and rewrite incoming wikilinks across the vault."""
vr = getattr(self.file_store, "working_dir", None)
working_dir = Path(vr).resolve() if vr else Path.cwd()
ok, payload = memory_io.rename_file(
self.file_store,
working_dir,
old_path=old_path,
new_path=new_path,
)
return _tool_response("memory_rename", ok, payload, audit=self.audit)
@R.register("memory_property_update")
class MemoryPropertyUpdate(BaseStep):
"""Update a single YAML frontmatter key. value=null deletes the key.
When key=='status', enforces the active → distilled → archived
state machine via `update_status`. Pass `force=True` to bypass.
Other keys go through the bare engine."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
key: str = self.context.get("key", "") or ""
value = self.context.get("value")
force: bool = bool(self.context.get("force", False))
assert path and key, "path and key are required"
if key == "status":
ok, payload = update_status(path, value=value, force=force)
else:
ok, payload = memory_io.update_meta(path, key=key, value=value)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_property_update(
self,
path: str,
key: str,
value: Any = None,
force: bool = False,
) -> ToolResponse:
"""Update a single YAML frontmatter key. value=null deletes the key.
When key=='status', enforces the active → distilled → archived
state machine. Pass `force=True` to bypass."""
if key == "status":
ok, payload = update_status(path, value=value, force=force)
else:
ok, payload = memory_io.update_meta(path, key=key, value=value)
return _tool_response("memory_property_update", ok, payload, audit=self.audit)
@R.register("memory_update")
class MemoryUpdate(BaseStep):
"""Edit-style content update: replace `old_string` with `new_string`."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
old_string: str = self.context.get("old_string", "") or ""
new_string: str = self.context.get("new_string", "") or ""
replace_all: bool = bool(self.context.get("replace_all", False))
assert path, "path is required"
ok, payload = memory_io.update_body(
path,
old_string=old_string,
new_string=new_string,
replace_all=replace_all,
)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_update(
self,
path: str,
old_string: str,
new_string: str,
replace_all: bool = False,
) -> ToolResponse:
"""Edit-style content update: replace `old_string` with `new_string`."""
ok, payload = memory_io.update_body(
path,
old_string=old_string,
new_string=new_string,
replace_all=replace_all,
)
return _tool_response("memory_update", ok, payload, audit=self.audit)
@R.register("memory_archive")
class MemoryArchive(BaseStep):
"""Archive a file: flip `status: archived` then move to `<vault>/Archive/`."""
audit: list[dict] | None = None
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
archive_dir_name: str = self.context.get("archive_dir", "Archive") or "Archive"
assert path, "path is required"
vr = getattr(self.file_store, "working_dir", None)
working_dir = Path(vr).resolve() if vr else Path.cwd()
ok, payload = memory_io.archive_file(working_dir, path, archive_dir=archive_dir_name)
self.context.response.success = ok
_set_answer(self.context, payload)
async def memory_archive(self, path: str, archive_dir: str = "Archive") -> ToolResponse:
"""Archive a file: flip `status: archived` then move to `<vault>/<archive_dir>/`."""
vr = getattr(self.file_store, "working_dir", None)
working_dir = Path(vr).resolve() if vr else Path.cwd()
ok, payload = memory_io.archive_file(working_dir, path, archive_dir=archive_dir)
return _tool_response("memory_archive", ok, payload, audit=self.audit)
@R.register("memory_count_tokens")
class MemoryCountTokens(BaseStep):
"""Estimate tokens for a file body or raw text. One of `path`/`text` required.
MCP-only — token counting is an editor concern, not part of the
R-M-W loop. No agent toolkit projection."""
async def execute(self):
assert self.context is not None
path: str = self.context.get("path", "") or ""
text: str = self.context.get("text", "") or ""
result = await memory_io.count_tokens(
self.as_token_counter,
path=path or None,
text=text or None,
)
self.context.response.success = "error" not in result
_set_answer(self.context, result)
# ===========================================================================
# Section 4 — Toolkit factory
# ===========================================================================
# The 11 memory_* tools the Ingestor's ReActAgent gets bound to.
# Each name doubles as the BaseStep registration key AND the tool method
# name on that step.
MEMORY_TOOL_NAMES: tuple[str, ...] = (
"memory_get",
"memory_list",
"memory_resolve_wikilink",
"memory_backlinks",
"memory_links",
"memory_create",
"memory_update",
"memory_property_update",
"memory_rename",
"memory_delete",
"memory_archive",
)
def build_memory_toolkit(
app_context,
audit: list[dict] | None = None,
toolkit: Toolkit | None = None,
) -> Toolkit:
"""Bind every memory_* step's tool method to an agentscope `Toolkit`.
For each name in `MEMORY_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 — there is no
separate JSON schema layer.
"""
toolkit = toolkit or Toolkit()
for name in MEMORY_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

31
reme2/memory/runtime_response.py
View file

@ -1,15 +1,20 @@
"""Helpers for serializing Step output onto `RuntimeContext.response`.
"""Helpers for serializing Step output onto `RuntimeContext.response`
and into `agentscope.tool.ToolResponse`.
Lives next to `runtime_context.py` because both are about the BaseStep
interface — the response side, specifically. Used by every Step that
returns a JSON-shaped payload (memory_*, sync, the three memory
services).
returns a JSON-shaped payload (agent toolkit, lint toolkit, the three
memory services).
"""
from __future__ import annotations
import json
from datetime import date, datetime
from typing import Any
from agentscope.message import TextBlock
from agentscope.tool import ToolResponse
def _to_jsonable(value):
@ -24,3 +29,23 @@ def _to_jsonable(value):
def _set_answer(context, payload) -> None:
context.response.answer = json.dumps(_to_jsonable(payload), ensure_ascii=False, indent=2)
def _tool_response(
op: str,
ok: bool,
payload: Any,
audit: list[dict] | None = None,
) -> ToolResponse:
"""Wrap a tool-method result as `ToolResponse` and optionally
append an audit row. Shared by every BaseStep's tool-method
surface so each toolkit doesn't reinvent serialization."""
if audit is not None:
entry = {"op": op, "ok": ok}
if isinstance(payload, dict):
entry.update(payload)
else:
entry["result"] = payload
audit.append(entry)
text = json.dumps(_to_jsonable(payload), ensure_ascii=False, indent=2)
return ToolResponse(content=[TextBlock(type="text", text=text)])

4
reme2/schema/__init__.py
View file

@ -5,7 +5,7 @@ from .as_msg_stat import AsBlockStat, AsMsgStat
from .emb_node import EmbNode
from .chunk_filter import ChunkFilter
from .file_chunk import FileChunk
from .file_link import FileLink, extract_wikilinks, iter_links
from .file_link import FileLink
from .file_node import FileFrontMatter, FileNode
from .request import Request
from .response import Response
@ -26,6 +26,4 @@ __all__ = [
"Request",
"Response",
"StreamChunk",
"extract_wikilinks",
"iter_links",
]

147
reme2/schema/file_link.py
View file

@ -3,111 +3,26 @@
One type, two states (the value of ``path`` distinguishes them):
* **pre-resolution** — ``path`` holds the raw wikilink target as
written, e.g. ``"Foo"`` or ``"topics/Bar"``. Produced by
``iter_links(text)``: a pure regex pass over body text, no graph
access.
written, e.g. ``"Foo"`` or ``"topics/Bar"``.
* **resolved** — ``path`` holds the vault-relative path
file_graph stores, e.g. ``"topics/Foo/Foo.md"``. Produced by
``utils.wikilink_resolver.resolve_links(graph, links)`` (or the
one-shot ``text_to_links(graph, text)``), which expands stem
ambiguity into one ``FileLink`` per candidate path.
file_graph stores, e.g. ``"topics/Foo/Foo.md"``.
The extractor (``utils.wikilink_resolver.iter_links``) produces the
pre-resolution form from body text. The resolver
(``utils.wikilink_resolver.resolve_links``, or one-shot
``text_to_links``) rewrites ``path`` to the resolved form, expanding
stem ambiguity into one ``FileLink`` per candidate.
file_graph trusts ``link.path`` directly for adjacency: it only ever
stores resolved links. The pre-resolution form is internal pipeline
plumbing.
## Inline forms recognised by ``iter_links``
[[X]] bare wikilink → predicate=None
extends:: [[X]] line-level Dataview → predicate="extends"
[extends:: [[X]]] inline-bracketed → predicate="extends"
Multi-target — every wikilink under one typed context inherits its
predicate (any separator works, not just commas):
extends:: [[A]], [[B]] line-level multi → 2 links, both "extends"
extends:: [[A]] and [[B]] prose-style multi → 2 links, both "extends"
[concerns:: [[A]], [[B]]] inline multi → 2 links, both "concerns"
extends:: [[A#s1]], [[B#s2]] multi w/ anchors → anchors preserved per link
Context precedence is **inline-bracketed > line-level > bare** —
a wikilink inside a ``[predicate:: …]`` envelope is typed by that
envelope even if the line happens to start ``predicate:: …``.
"""
from __future__ import annotations
import re
from pydantic import BaseModel, ConfigDict, Field
# -- Regexes (module-private) ---------------------------------------------
_WIKILINK_RE = re.compile(
r"""
(?:!)?
\[\[
(?P<target>[^\]\|\#\n]+?)
(?:\#(?P<anchor>[^\]\|\n]+))?
(?:\|[^\]\n]+)?
\]\]
""",
re.VERBOSE,
)
_DATAVIEW_LINE_RE = re.compile(
r"^[ \t]*(?:[-*+][ \t]+)?(?P<predicate>[A-Za-z][A-Za-z0-9_]*)\s*::\s*(?P<value>.+?)\s*$",
re.MULTILINE,
)
_INLINE_FIELD_OPEN_RE = re.compile(r"\[(?P<predicate>[A-Za-z][A-Za-z0-9_]*)\s*::\s*")
def _iter_inline_fields(text: str) -> list[tuple[int, int, str]]:
"""Find inline-bracketed ``[predicate:: …]`` field spans by depth scan."""
out: list[tuple[int, int, str]] = []
for m in _INLINE_FIELD_OPEN_RE.finditer(text):
depth = 1
i = m.end()
n = len(text)
while i < n:
c = text[i]
if c == "\n":
break
if c == "[":
depth += 1
elif c == "]":
depth -= 1
if depth == 0:
out.append((m.start(), i + 1, m.group("predicate")))
break
i += 1
return out
def _predicate_for(
text: str,
pos: int,
inline_spans: list[tuple[int, int, str]],
) -> str | None:
"""Resolve the predicate governing a wikilink at offset ``pos``."""
for field_start, field_end, predicate in inline_spans:
if field_start <= pos < field_end:
return predicate
line_start = text.rfind("\n", 0, pos) + 1
line_end = text.find("\n", pos)
if line_end == -1:
line_end = len(text)
m = _DATAVIEW_LINE_RE.match(text[line_start:line_end])
if m and line_start + m.start("value") <= pos:
return m.group("predicate")
return None
# -- Schema ---------------------------------------------------------------
class FileLink(BaseModel):
"""Typed wikilink — ``(path, anchor, predicate)``.
@ -144,49 +59,3 @@ class FileLink(BaseModel):
default=None,
description="Typed-link predicate (Dataview-style). None for bare [[X]].",
)
# -- Public extraction utilities ------------------------------------------
def iter_links(text: str) -> list[FileLink]:
"""Extract every wikilink in ``text`` as a pre-resolution ``FileLink``.
Each emitted ``FileLink`` has ``path`` set to the raw wikilink
target as written (the file portion only — the ``#anchor`` lives
in its own field). Predicate is decided by surrounding context
with precedence **inline-bracketed > line-level > bare**.
Pure function: no graph access. Pass the result through
``utils.wikilink_resolver.resolve_links`` (or one-shot
``text_to_links``) to get the resolved form file_graph stores.
"""
if not text:
return []
inline_spans = _iter_inline_fields(text)
links: list[FileLink] = []
for wm in _WIKILINK_RE.finditer(text):
target = wm.group("target").strip()
anchor_raw = wm.group("anchor")
anchor = anchor_raw.strip() if anchor_raw else ""
links.append(
FileLink(
path=target,
anchor=anchor or None,
predicate=_predicate_for(text, wm.start(), inline_spans),
),
)
return links
def extract_wikilinks(text: str) -> list[str]:
"""Flat list of wikilink **file targets** in body text (no dedup).
Returns just the file part of each wikilink (before any ``#anchor``).
Single regex pass — cheaper than ``iter_links`` when callers don't
need predicates (e.g. ``extract_anchors`` for query seeding in the
retriever).
"""
if not text:
return []
return [m.group("target").strip() for m in _WIKILINK_RE.finditer(text)]

427
reme2/utils/wikilink_resolver.py
View file

@ -1,52 +1,91 @@
"""Wikilink resolver — vault convention over ``BaseFileGraph``.
"""Wikilink syntax + resolver — vault convention over ``BaseFileGraph``.
Stem-form wikilinks like ``[[Foo]]`` are a vault convention. The
file_graph engine doesn't know about them — it only stores nodes and
trusts ``FileLink.path`` for adjacency. This module bridges the gap,
applying:
This module is the single home for wikilink **syntax** (regex
extraction, predicate detection) and wikilink **resolution** (mapping
raw targets to vault-relative paths via the file_graph).
* **folder-note rule** — ``[[X]]`` prefers ``topics/X/X.md`` over a
sibling ``topics/X.md`` when both exist
* **stem ambiguity** — ``[[Foo]]`` matching multiple paths is handled
by **emitting multiple ``FileLink`` records, one per candidate**
(file_graph then has unambiguous adjacency)
* **path-form passthrough** — ``[[foo/bar]]`` and ``[[foo/bar.md]]``
resolve directly against the graph (vault-relative paths; no
working_dir absolutization)
The ``FileLink`` schema (see ``schema.file_link``) is just a typed
record; here we define how it's produced from text and resolved
against the graph.
## Two-layer link semantics
1. Implicit extension — ``[[Foo]]`` has no extension on its last
segment, so it's completed to ``Foo.md`` (markdown is the
default vault content type). ``[[image.png]]`` already has an
extension; left as-is. Done in ``iter_links`` at extraction
time, so all FileLinks emerge with extension-bearing paths.
2. Short link — a path with no ``/`` (after implicit completion)
is matched against the basename of every node in the graph.
The folder-note rule applies: when both ``X.md`` and
``X/X.md`` exist, the folder-note wins. Ambiguity expands into
multiple FileLink records (one per candidate path). Targets
containing ``/`` are treated as literal paths and looked up
directly.
So ``[[Foo]]`` → ``Foo.md`` (implicit) → search basename ``Foo.md``
across the vault, returning e.g. ``topics/Foo/Foo.md``;
``[[topics/Bar]]`` → ``topics/Bar.md`` (implicit) → literal lookup;
``[[image.png]]`` → ``image.png`` (no completion needed) → search
basename ``image.png``; ``[[topics/image.png]]`` → literal lookup.
All paths are vault-relative — ``graph.iter_nodes()`` returns the
key form file_graph stores, and emitted ``FileLink.path`` matches.
All functions are stateless — they walk ``graph.iter_nodes()`` per
call. For batch operations (``resolve_links`` over many links) the
stem index is built once and reused.
All resolver functions are stateless — they walk ``graph.iter_nodes()``
per call. For batch operations (``resolve_links``) the basename index
is built once and reused.
Seven entry points mapped to call sites:
## Inline forms recognised by ``iter_links``
resolve single link → path | None
used by: ``extract_anchors``, ``memory_resolve_wikilink``
candidates stem → [path] (folder-note ordered first)
used by: ``memory_resolve_wikilink`` (ambiguity report)
collisions all stems with >1 path
used by: ``maintainer.lint``
collisions_for paths conflicting with a proposed new path
used by: ``memory_create``, ``sync`` (preflight)
extract_anchors parse [[X]] from text + resolve, dedup
used by: ``retriever`` (query anchor seeds)
resolve_links ``[FileLink]`` (raw path) → ``[FileLink]`` (resolved)
core resolution; stem ambiguity expands
text_to_links one-shot: ``iter_links(text)`` + ``resolve_links``
used by: parser pipeline (before ``upsert_node``)
[[X]] bare wikilink → predicate=None
extends:: [[X]] line-level Dataview → predicate="extends"
[extends:: [[X]]] inline-bracketed → predicate="extends"
Multi-target — every wikilink under one typed context inherits its
predicate (any separator works, not just commas):
extends:: [[A]], [[B]] line-level multi → 2 links, both "extends"
extends:: [[A]] and [[B]] prose-style multi → 2 links, both "extends"
[concerns:: [[A]], [[B]]] inline multi → 2 links, both "concerns"
extends:: [[A#s1]], [[B#s2]] multi w/ anchors → anchors preserved per link
Context precedence is **inline-bracketed > line-level > bare** —
a wikilink inside a ``[predicate:: …]`` envelope is typed by that
envelope even if the line happens to start ``predicate:: …``.
## Entry points
Extraction (no graph)
iter_links text → [FileLink] (path = target after implicit .md)
extract_wikilinks text → [str] (raw target list, no completion)
Resolution (graph-backed)
resolve single link → path | None
used by: ``extract_anchors``, ``memory_resolve_wikilink``
candidates target → [path] (folder-note ordered first)
used by: ``memory_resolve_wikilink`` (ambiguity report)
collisions all basenames with >1 path
used by: ``maintainer.lint``
collisions_for paths conflicting with a proposed new path
used by: ``memory_create``, ``sync`` (preflight)
extract_anchors parse [[X]] from text + resolve, dedup
used by: ``retriever`` (query anchor seeds)
resolve_links ``[FileLink]`` (raw path) → ``[FileLink]`` (resolved)
core resolution; short-link ambiguity expands
text_to_links one-shot: ``iter_links(text)`` + ``resolve_links``
used by: parser pipeline (before ``upsert_node``)
"""
from __future__ import annotations
import re
from collections.abc import Iterable
from pathlib import Path
from ..component.file_graph.base_file_graph import BaseFileGraph
from ..schema import FileLink
from ..schema.file_link import extract_wikilinks, iter_links
from .logger_utils import get_logger
@ -54,15 +93,141 @@ _logger = get_logger()
# =========================================================================
# Internal helpers
# Wikilink syntax — regex extraction + predicate detection
# =========================================================================
async def _build_stem_index(graph: BaseFileGraph) -> dict[str, list[str]]:
"""Walk once, group paths by stem. Used by batch hot paths."""
_WIKILINK_RE = re.compile(
r"""
(?:!)?
\[\[
(?P<target>[^\]\|\#\n]+?)
(?:\#(?P<anchor>[^\]\|\n]+))?
(?:\|[^\]\n]+)?
\]\]
""",
re.VERBOSE,
)
_DATAVIEW_LINE_RE = re.compile(
r"^[ \t]*(?:[-*+][ \t]+)?(?P<predicate>[A-Za-z][A-Za-z0-9_]*)\s*::\s*(?P<value>.+?)\s*$",
re.MULTILINE,
)
_INLINE_FIELD_OPEN_RE = re.compile(r"\[(?P<predicate>[A-Za-z][A-Za-z0-9_]*)\s*::\s*")
def _iter_inline_fields(text: str) -> list[tuple[int, int, str]]:
"""Find inline-bracketed ``[predicate:: …]`` field spans by depth scan."""
out: list[tuple[int, int, str]] = []
for m in _INLINE_FIELD_OPEN_RE.finditer(text):
depth = 1
i = m.end()
n = len(text)
while i < n:
c = text[i]
if c == "\n":
break
if c == "[":
depth += 1
elif c == "]":
depth -= 1
if depth == 0:
out.append((m.start(), i + 1, m.group("predicate")))
break
i += 1
return out
def _predicate_for(
text: str,
pos: int,
inline_spans: list[tuple[int, int, str]],
) -> str | None:
"""Resolve the predicate governing a wikilink at offset ``pos``."""
for field_start, field_end, predicate in inline_spans:
if field_start <= pos < field_end:
return predicate
line_start = text.rfind("\n", 0, pos) + 1
line_end = text.find("\n", pos)
if line_end == -1:
line_end = len(text)
m = _DATAVIEW_LINE_RE.match(text[line_start:line_end])
if m and line_start + m.start("value") <= pos:
return m.group("predicate")
return None
def _complete_path(target: str) -> str:
"""Append ``.md`` if the last path segment has no extension.
Implements the implicit-markdown rule: ``[[Foo]]`` → ``Foo.md``,
``[[topics/Bar]]`` → ``topics/Bar.md``, but ``[[image.png]]`` is
left alone. ``"."`` in the last segment counts as "has extension".
"""
if not target:
return target
last = target.rsplit("/", 1)[-1]
if "." in last:
return target
return target + ".md"
def iter_links(text: str) -> list[FileLink]:
"""Extract every wikilink in ``text`` as a pre-resolution ``FileLink``.
Each emitted ``FileLink`` has ``path`` set to the wikilink target
with the implicit ``.md`` rule applied (so ``[[Foo]]`` emerges as
``path="Foo.md"``); the ``#anchor`` lives in its own field.
Predicate is decided by surrounding context with precedence
**inline-bracketed > line-level > bare**.
Pure function: no graph access. Pass the result through
``resolve_links`` (or one-shot ``text_to_links``) to apply
short-link resolution and get the form file_graph stores.
"""
if not text:
return []
inline_spans = _iter_inline_fields(text)
links: list[FileLink] = []
for wm in _WIKILINK_RE.finditer(text):
target = wm.group("target").strip()
anchor_raw = wm.group("anchor")
anchor = anchor_raw.strip() if anchor_raw else ""
links.append(FileLink(
path=_complete_path(target),
anchor=anchor or None,
predicate=_predicate_for(text, wm.start(), inline_spans),
))
return links
def extract_wikilinks(text: str) -> list[str]:
"""Flat list of wikilink **file targets** in body text (no dedup).
Returns just the file part of each wikilink **as written** —
no implicit ``.md`` completion (callers like ``resolve`` apply
completion themselves). Single regex pass — cheaper than
``iter_links`` when callers don't need predicates.
"""
if not text:
return []
return [m.group("target").strip() for m in _WIKILINK_RE.finditer(text)]
# =========================================================================
# Resolution helpers (internal)
# =========================================================================
async def _build_basename_index(graph: BaseFileGraph) -> dict[str, list[str]]:
"""Walk once, group paths by basename (file name with extension).
Used by short-link resolution batch hot paths.
"""
out: dict[str, list[str]] = {}
async for path, _ in graph.iter_nodes():
out.setdefault(Path(path).stem, []).append(path)
out.setdefault(Path(path).name, []).append(path)
return out
@ -81,15 +246,23 @@ def _split_link(link: str) -> tuple[str, str]:
return target_raw.strip(), anchor_raw.strip()
def _is_path_form(target: str) -> bool:
return "/" in target or target.endswith(".md")
def _has_dir(target: str) -> bool:
"""``True`` for literal-path targets (containing ``/``);
``False`` for short links (basename only).
"""
return "/" in target
def _filter_stem_candidates(stem: str, paths: list[str]) -> list[str]:
"""Apply folder-note rule: if any folder-note exists, ONLY folder-notes
are candidates; otherwise all sibling-stems are candidates."""
def _filter_short_link_candidates(basename: str, paths: list[str]) -> list[str]:
"""Apply folder-note rule to short-link basename matches.
If any path is a folder-note (parent dir name == file stem) the
folder-notes win as the only candidates; otherwise all matches
are returned. Sorted for determinism.
"""
if not paths:
return []
stem = Path(basename).stem
folder_hits = sorted(p for p in paths if Path(p).parent.name == stem)
if folder_hits:
return folder_hits
@ -104,91 +277,115 @@ def _filter_stem_candidates(stem: str, paths: list[str]) -> list[str]:
async def resolve(graph: BaseFileGraph, link: str) -> str | None:
"""Resolve a single wikilink to **one** vault-relative path, or None.
Applies implicit ``.md`` completion, then dispatches:
* literal path (contains ``/``) → direct ``get_node`` lookup
* short link (no ``/``) → basename match + folder-note rule
Returns None if dangling or ambiguous (with warning on ambiguity).
Use ``resolve_links`` for the multi-link expansion semantics.
"""
target, _ = _split_link(link)
if not target:
return None
target = _complete_path(target)
if _is_path_form(target):
candidate = target if target.endswith(".md") else f"{target}.md"
return candidate if await graph.get_node(candidate) else None
if _has_dir(target):
return target if await graph.get_node(target) else None
paths = [p async for p, _ in graph.iter_nodes() if Path(p).stem == target]
candidates_for_stem = _filter_stem_candidates(target, paths)
if len(candidates_for_stem) == 1:
return candidates_for_stem[0]
if len(candidates_for_stem) > 1:
paths = [
p async for p, _ in graph.iter_nodes() if Path(p).name == target
]
candidates_for = _filter_short_link_candidates(target, paths)
if len(candidates_for) == 1:
return candidates_for[0]
if len(candidates_for) > 1:
_logger.warning(
f"Wikilink [[{target}]] is ambiguous, " f"candidates: {candidates_for_stem}",
f"Wikilink [[{target}]] is ambiguous, "
f"candidates: {candidates_for}",
)
return None
async def candidates(graph: BaseFileGraph, stem: str) -> list[str]:
"""All paths a ``[[stem]]`` could match. Folder-note hits ordered first."""
async def candidates(graph: BaseFileGraph, target: str) -> list[str]:
"""All vault paths a ``[[target]]`` could match.
Applies implicit ``.md`` completion. For literal paths (with ``/``)
returns ``[target]`` if it exists else ``[]``. For short links
returns every node whose basename matches, with folder-note hits
ordered first.
"""
target = _complete_path(target)
if _has_dir(target):
return [target] if await graph.get_node(target) else []
stem = Path(target).stem
folder_hits: list[str] = []
stem_hits: list[str] = []
name_hits: list[str] = []
async for path, _ in graph.iter_nodes():
p = Path(path)
if p.stem != stem:
if Path(path).name != target:
continue
if p.parent.name == stem:
if Path(path).parent.name == stem:
folder_hits.append(path)
else:
stem_hits.append(path)
name_hits.append(path)
if folder_hits:
return sorted(folder_hits)
return sorted(stem_hits)
return sorted(name_hits)
async def collisions(graph: BaseFileGraph) -> dict[str, list[str]]:
"""Every stem that resolves to >1 path. Used by maintainer.lint."""
stem_index = await _build_stem_index(graph)
return {stem: sorted(paths) for stem, paths in stem_index.items() if len(paths) > 1}
"""Every basename that resolves to >1 path. Used by maintainer.lint.
Reflects short-link ambiguity: ``[[X.md]]`` (or ``[[X]]``) hitting
multiple files in different directories.
"""
basename_index = await _build_basename_index(graph)
return {
name: sorted(paths)
for name, paths in basename_index.items()
if len(paths) > 1
}
async def collisions_for(
graph: BaseFileGraph,
proposed_path: str | Path,
graph: BaseFileGraph, proposed_path: str | Path,
) -> list[str]:
"""Existing paths that would conflict with adding ``proposed_path``.
``proposed_path`` is treated as vault-relative (matching the form
``graph.iter_nodes()`` returns). Folder-note rule: when the
proposed path is itself a folder-note (parent dir name == stem),
only colliding folder-notes are returned. Otherwise all paths
sharing the stem are returned.
``proposed_path`` is vault-relative (matches ``graph.iter_nodes()``).
Folder-note rule: when the proposed path is itself a folder-note
(parent dir name == file stem), only colliding folder-notes are
returned. Otherwise all paths sharing the basename are returned.
"""
p = Path(proposed_path)
name = p.name
stem = p.stem
proposed_str = str(p)
is_folder_note = p.parent.name == stem
folder_hits: list[str] = []
stem_hits: list[str] = []
name_hits: list[str] = []
async for path, _ in graph.iter_nodes():
if path == proposed_str:
continue
path_obj = Path(path)
if path_obj.stem != stem:
if path_obj.name != name:
continue
if path_obj.parent.name == stem:
folder_hits.append(path)
else:
stem_hits.append(path)
name_hits.append(path)
if is_folder_note:
return sorted(folder_hits)
return sorted(folder_hits) + sorted(stem_hits)
return sorted(folder_hits) + sorted(name_hits)
async def extract_anchors(graph: BaseFileGraph, text: str) -> list[str]:
"""Pull ``[[X]]`` from ``text``, resolve each, dedup in source order.
Uses single-target ``resolve`` semantics — ambiguous stems return
no anchor. (For multi-link expansion at *write* time, see
Uses single-target ``resolve`` semantics — ambiguous short links
return no anchor. (For multi-link expansion at *write* time, see
``resolve_links``; ``extract_anchors`` is for read-time seeding
where a single deterministic target is wanted.)
"""
@ -210,72 +407,72 @@ async def extract_anchors(graph: BaseFileGraph, text: str) -> list[str]:
async def resolve_links(
graph: BaseFileGraph,
links: Iterable[FileLink],
graph: BaseFileGraph, links: Iterable[FileLink],
) -> list[FileLink]:
"""Resolve pre-resolution ``FileLink`` records against the graph.
Each input link has ``path`` holding a raw wikilink target (the
extractor form). Returns FileLinks with ``path`` rewritten to the
vault-relative resolved path file_graph stores. ``anchor`` and
``predicate`` pass through unchanged.
Each input link has ``path`` already passed through
``_complete_path`` (so it has an extension). Returns FileLinks
with ``path`` rewritten to the vault-relative resolved form
file_graph stores. ``anchor`` and ``predicate`` pass through
unchanged.
Stem expansion semantics — one input link produces zero, one, or
many output links:
Resolution dispatch:
* literal path (contains ``/``) → direct ``get_node`` lookup;
keep if exists, drop if dangling.
* short link (no ``/``) → basename match + folder-note
rule. Ambiguity expands into multiple FileLinks (one per
candidate path); dangling produces zero.
* path-form, target indexed → 1 link
* path-form, dangling → 0 links
* stem-form, 1 folder-note hit → 1 link
* stem-form, N folder-note hits → N links (one per)
* stem-form, 0 folder-notes, 1 sibling → 1 link
* stem-form, 0 folder-notes, N siblings → N links (one per)
* stem-form, dangling → 0 links
Output cardinality per input link:
Build the stem index lazily: only paid if at least one input is
stem-form.
* literal, target indexed → 1 link
* literal, dangling → 0 links
* short, 1 folder-note hit → 1 link
* short, N folder-note hits → N links (one per)
* short, 0 folder-notes, 1 basename hit → 1 link
* short, 0 folder-notes, N basename hits → N links (one per)
* short, dangling → 0 links
Build the basename index lazily: only paid if at least one input
is a short link.
"""
link_list = list(links)
if not link_list:
return []
stem_index: dict[str, list[str]] | None = None
basename_index: dict[str, list[str]] | None = None
out: list[FileLink] = []
for link in link_list:
target = link.path
if not target:
continue
if _is_path_form(target):
candidate = target if target.endswith(".md") else f"{target}.md"
if await graph.get_node(candidate) is None:
if _has_dir(target):
if await graph.get_node(target) is None:
continue
out.append(
FileLink(
path=candidate,
anchor=link.anchor,
predicate=link.predicate,
),
)
out.append(FileLink(
path=target,
anchor=link.anchor,
predicate=link.predicate,
))
continue
# Stem form — may expand into multiple links.
if stem_index is None:
stem_index = await _build_stem_index(graph)
for chosen in _filter_stem_candidates(target, stem_index.get(target, [])):
out.append(
FileLink(
path=chosen,
anchor=link.anchor,
predicate=link.predicate,
),
)
# Short link — may expand into multiple links.
if basename_index is None:
basename_index = await _build_basename_index(graph)
for chosen in _filter_short_link_candidates(target, basename_index.get(target, [])):
out.append(FileLink(
path=chosen,
anchor=link.anchor,
predicate=link.predicate,
))
return out
async def text_to_links(
graph: BaseFileGraph,
text: str,
graph: BaseFileGraph, text: str,
) -> list[FileLink]:
"""One-shot: extract wikilinks from ``text`` and resolve to safe links.

133
tests/test_file_link.py
View file

@ -6,15 +6,19 @@ Covers:
* dedup against typed wrappers (no double-emit when a wikilink lives
inside a typed envelope)
* open-vocabulary predicate pass-through and the identifier-shape gate
* the implicit-markdown rule: ``[[Foo]]`` (no extension) emerges as
``path="Foo.md"``; ``[[image.png]]`` (has extension) emerges as-is.
Resolution of short links (basename match + folder-note rule) is
tested separately against a ``BaseFileGraph``.
* the explicit decision that frontmatter is no longer walked for links
(extraction takes body text only)
* the 3-field schema: ``(path, anchor, predicate)`` — all real fields,
extra='forbid'
Pre-resolution form: ``iter_links(text)`` returns ``FileLink`` records with
``path`` set to the raw wikilink target as written. The resolver
(``utils.wikilink_resolver``) rewrites ``path`` to the vault-relative
resolved path; here we only test the extractor.
``path`` set to the wikilink target with implicit ``.md`` applied. The
resolver (``utils.wikilink_resolver``) then maps short links to the
vault-relative full path; here we only test the extractor.
"""
import inspect
@ -23,41 +27,62 @@ import pytest
from pydantic import ValidationError
from reme2.schema import FileLink
from reme2.schema.file_link import extract_wikilinks, iter_links
from reme2.utils.wikilink_resolver import extract_wikilinks, iter_links
# --------------------------------------------------------------------------
# Bare wikilinks
# Bare wikilinks — implicit ``.md`` completion at extraction time
# --------------------------------------------------------------------------
def test_bare_wikilink():
def test_bare_wikilink_gets_implicit_md():
links = iter_links("see [[X]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].predicate is None
assert links[0].anchor is None
def test_anchor_split_from_target():
"""``[[X#sec]]`` → path='X', anchor='sec' (separate field, not derived)."""
"""``[[X#sec]]`` → path='X.md' (implicit), anchor='sec'."""
links = iter_links("![[X#sec|Alias]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].anchor == "sec"
def test_alias_only_dropped():
links = iter_links("[[X|Alias]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].anchor is None
def test_embed_only_dropped():
links = iter_links("![[X]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
def test_explicit_extension_kept_as_is():
"""``[[image.png]]`` already has an extension — no completion."""
assert iter_links("![[image.png]]")[0].path == "image.png"
assert iter_links("[[notes.txt]]")[0].path == "notes.txt"
def test_explicit_md_extension_kept_as_is():
"""``[[Foo.md]]`` already has the ``.md`` extension — no double-append."""
assert iter_links("[[Foo.md]]")[0].path == "Foo.md"
def test_dir_path_no_extension_gets_md():
"""``[[topics/Bar]]`` → ``topics/Bar.md`` (last segment lacks extension)."""
assert iter_links("[[topics/Bar]]")[0].path == "topics/Bar.md"
def test_dir_path_with_extension_kept():
"""``[[topics/image.png]]`` → kept literal."""
assert iter_links("[[topics/image.png]]")[0].path == "topics/image.png"
# --------------------------------------------------------------------------
@ -68,24 +93,24 @@ def test_embed_only_dropped():
def test_line_level_field():
links = iter_links("extends:: [[Source Topic]]")
assert len(links) == 1
assert links[0].path == "Source Topic"
assert links[0].path == "Source Topic.md"
assert links[0].predicate == "extends"
def test_line_level_multi_target():
links = iter_links("concerns:: [[A]], [[B]], [[C]]")
assert [(link.path, link.predicate) for link in links] == [
("A", "concerns"),
("B", "concerns"),
("C", "concerns"),
("A.md", "concerns"),
("B.md", "concerns"),
("C.md", "concerns"),
]
def test_line_level_with_bullet():
links = iter_links("- extends:: [[X]]\n * concerns:: [[Y]]")
assert [(link.path, link.predicate) for link in links] == [
("X", "extends"),
("Y", "concerns"),
("X.md", "extends"),
("Y.md", "concerns"),
]
@ -98,9 +123,9 @@ def test_multi_target_with_anchors_preserves_each():
"""Each comma-separated target keeps its own anchor as a separate field."""
links = iter_links("extends:: [[A#sec1]], [[B#sec2]], [[C]]")
assert [(link.path, link.anchor, link.predicate) for link in links] == [
("A", "sec1", "extends"),
("B", "sec2", "extends"),
("C", None, "extends"),
("A.md", "sec1", "extends"),
("B.md", "sec2", "extends"),
("C.md", None, "extends"),
]
@ -109,8 +134,8 @@ def test_multi_target_non_comma_separator_still_typed():
inherit the line's predicate. Useful for prose-style fields."""
links = iter_links("extends:: [[A]] and also [[B]]")
assert [(link.path, link.predicate) for link in links] == [
("A", "extends"),
("B", "extends"),
("A.md", "extends"),
("B.md", "extends"),
]
@ -121,10 +146,10 @@ def test_multi_dataview_lines_each_multi_target():
"extends:: [[A]], [[B]]\nrelates:: [[C]], [[D]]",
)
assert [(link.path, link.predicate) for link in links] == [
("A", "extends"),
("B", "extends"),
("C", "relates"),
("D", "relates"),
("A.md", "extends"),
("B.md", "extends"),
("C.md", "relates"),
("D.md", "relates"),
]
@ -133,8 +158,8 @@ def test_inline_bracketed_then_bare_on_same_line():
a trailing bare wikilink on the same line stays bare."""
links = iter_links("[ext:: [[A]]] then [[B]]")
assert [(link.path, link.predicate) for link in links] == [
("A", "ext"),
("B", None),
("A.md", "ext"),
("B.md", None),
]
@ -143,8 +168,8 @@ def test_mid_line_dataview_like_not_typed():
a ``predicate::`` mid-line is just prose, so its wikilinks are bare."""
links = iter_links("[ext:: [[A]]] and concerns:: [[B]]")
assert [(link.path, link.predicate) for link in links] == [
("A", "ext"),
("B", None), # `concerns::` mid-line is not Dataview
("A.md", "ext"),
("B.md", None), # `concerns::` mid-line is not Dataview
]
@ -156,15 +181,15 @@ def test_mid_line_dataview_like_not_typed():
def test_inline_bracketed():
links = iter_links("This [extends:: [[Y]]] something else.")
assert len(links) == 1
assert links[0].path == "Y"
assert links[0].path == "Y.md"
assert links[0].predicate == "extends"
def test_inline_bracketed_multi_target():
links = iter_links("[concerns:: [[A]], [[B]]]")
assert [(link.path, link.predicate) for link in links] == [
("A", "concerns"),
("B", "concerns"),
("A.md", "concerns"),
("B.md", "concerns"),
]
@ -173,7 +198,7 @@ def test_inline_bracketed_skips_cross_line():
# wikilink falls back to bare; the unmatched `[` does not eat tail text.
links = iter_links("[extends:: [[X]]\nbad]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].predicate is None
@ -196,7 +221,7 @@ def test_line_level_value_does_not_double_emit():
def test_typed_and_bare_coexist_for_same_target():
links = iter_links("extends:: [[X]]\nFree text mentioning [[X]] again.")
paths_preds = sorted(((link.path, link.predicate or "") for link in links))
assert paths_preds == [("X", ""), ("X", "extends")]
assert paths_preds == [("X.md", ""), ("X.md", "extends")]
# --------------------------------------------------------------------------
@ -207,7 +232,7 @@ def test_typed_and_bare_coexist_for_same_target():
def test_arbitrary_predicate_preserved():
links = iter_links("anything_goes:: [[X]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].predicate == "anything_goes"
@ -222,12 +247,12 @@ def test_predicate_must_be_identifier_shaped():
# not recognised as a Dataview field — the wikilink falls back to bare.
links = iter_links("123bad:: [[X]]")
assert len(links) == 1
assert links[0].path == "X"
assert links[0].path == "X.md"
assert links[0].predicate is None
def test_file_link_accepts_any_predicate_string():
link = FileLink(path="X", predicate="totally_made_up")
link = FileLink(path="X.md", predicate="totally_made_up")
assert link.predicate == "totally_made_up"
@ -262,20 +287,20 @@ def test_no_frontmatter_walker_exported():
def test_file_link_extra_forbid():
with pytest.raises(ValidationError):
FileLink(path="X", target="X") # type: ignore[call-arg]
FileLink(path="X.md", target="X.md") # type: ignore[call-arg]
def test_file_link_field_set():
"""Stored fields are ``(path, anchor, predicate)`` — no others."""
link = FileLink(path="X")
link = FileLink(path="X.md")
dumped = link.model_dump()
assert set(dumped.keys()) == {"path", "anchor", "predicate"}
assert dumped == {"path": "X", "anchor": None, "predicate": None}
assert dumped == {"path": "X.md", "anchor": None, "predicate": None}
def test_file_link_dump_excludes_none_when_asked():
link = FileLink(path="X")
assert link.model_dump(exclude_none=True) == {"path": "X"}
link = FileLink(path="X.md")
assert link.model_dump(exclude_none=True) == {"path": "X.md"}
def test_file_link_full_construction():
@ -291,8 +316,7 @@ def test_file_link_path_required():
# --------------------------------------------------------------------------
# Anchor extraction edge cases — anchor is a real field, captured by the
# regex's named group (not derived from path string-splitting)
# Anchor extraction edge cases
# --------------------------------------------------------------------------
@ -313,12 +337,14 @@ def test_anchor_extraction_edge_cases():
def test_anchor_inside_pipe_alias_still_extracted():
"""Anchor is captured before the alias pipe."""
links = iter_links("[[topics/Foo#sec|Display]]")
assert links[0].path == "topics/Foo"
assert links[0].path == "topics/Foo.md"
assert links[0].anchor == "sec"
# --------------------------------------------------------------------------
# Back-compat: extract_wikilinks returns flat target list (used by retriever)
# Back-compat: extract_wikilinks returns flat target list (used by retriever).
# Note: extract_wikilinks does NOT apply implicit-md completion — callers
# (resolve, extract_anchors) apply it themselves at resolve time.
# --------------------------------------------------------------------------
@ -328,18 +354,23 @@ def test_extract_wikilinks_flat_targets():
def test_extract_wikilinks_strips_anchor():
"""``extract_wikilinks`` returns just the file part — anchor stripped
so callers can feed it to ``resolve``."""
"""``extract_wikilinks`` returns just the file part — anchor stripped."""
targets = extract_wikilinks("see [[X#sec]] and ![[Y#a|alias]]")
assert targets == ["X", "Y"]
def test_extract_wikilinks_does_not_complete_md():
"""Raw form — no implicit ``.md`` (that's a resolution-stage concern)."""
targets = extract_wikilinks("[[Foo]] [[image.png]] [[topics/Bar]]")
assert targets == ["Foo", "image.png", "topics/Bar"]
# --------------------------------------------------------------------------
# Source ordering stability
# --------------------------------------------------------------------------
def test_links_sorted_by_source_position():
body = "intro [[First]] then\n" "extends:: [[Second]]\n" "tail [[Third]]\n"
body = "intro [[First]] then\nextends:: [[Second]]\ntail [[Third]]\n"
links = iter_links(body)
assert [link.path for link in links] == ["First", "Second", "Third"]
assert [link.path for link in links] == ["First.md", "Second.md", "Third.md"]