From 30971af5ecd0f1d92c737d75dee53cddb914e1da Mon Sep 17 00:00:00 2001 From: huangsen Date: Fri, 15 May 2026 12:15:17 +0800 Subject: [PATCH] refactor(memory): restructure tool surface architecture BREAKING CHANGE: Split monolithic agent_toolkit into modular components: - memory_toolkit.py: memory_get, memory_create, memory_update_body, memory_update_meta, memory_search + schema policy helpers - file_toolkit.py: file_download, file_upload, file_delete, file_list, file_move + path resolution and session temp dir - graph_toolkit.py: graph_traverse + adjacency BFS - event_toolkit.py: event_open, event_complete (using memory schema gates) - lint_toolkit.py: check_* atomic primitives remain separate Agent toolkit now composes these modules instead of containing all logic. - Remove sync job from EXPECTED_JOBS since sync.py is no longer imported - Add event_open and event_complete jobs to test expectations - Update module imports to reference new toolkit structure - Maintain backward compatibility through composition layer --- reme2/mcp/test/test_expert.py | 4 +- reme2/memory/__init__.py | 58 ++- reme2/memory/agent_toolkit.py | 771 ++------------------------------- reme2/memory/event_toolkit.py | 309 +++++++++++++ reme2/memory/file_toolkit.py | 248 +++++++++++ reme2/memory/graph_toolkit.py | 148 +++++++ reme2/memory/memory_io.py | 663 ++++++++++------------------ reme2/memory/memory_toolkit.py | 434 +++++++++++++++++++ reme2/memory/sync.py | 493 --------------------- 9 files changed, 1432 insertions(+), 1696 deletions(-) create mode 100644 reme2/memory/event_toolkit.py create mode 100644 reme2/memory/file_toolkit.py create mode 100644 reme2/memory/graph_toolkit.py create mode 100644 reme2/memory/memory_toolkit.py delete mode 100644 reme2/memory/sync.py diff --git a/reme2/mcp/test/test_expert.py b/reme2/mcp/test/test_expert.py index 808b1a36..d45fc418 100644 --- a/reme2/mcp/test/test_expert.py +++ b/reme2/mcp/test/test_expert.py @@ -29,7 +29,6 @@ from ._helpers import AppContext, decode, wait_for_index # `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) @@ -45,6 +44,9 @@ EXPECTED_JOBS: tuple[str, ...] = ( "file_move", # graph (1) "graph_traverse", + # event (2) + "event_open", + "event_complete", # lint (4 atomic checks) "check_dangling", "check_orphans", diff --git a/reme2/memory/__init__.py b/reme2/memory/__init__.py index 49defe14..c925b3fa 100644 --- a/reme2/memory/__init__.py +++ b/reme2/memory/__init__.py @@ -16,27 +16,38 @@ reached via MCP, HTTP, or direct Python. atomic ``check_*`` primitives below. summarizer.py Auxiliary used by the services. - Tool surfaces (each ``@R.register`` exposes a step the agent - invokes by name). + Tool surfaces — split by category, each cohesive and self- + contained. ``agent_toolkit`` is just a thin orchestrator that + composes them into one Toolkit for hosts that want all 13 at once. - 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). + memory_toolkit.py memory_get / memory_create / + memory_update_body / memory_update_meta / + memory_search + + memory_graph_search (MCP-only) + + schema policy helpers (path template, + status state machine, + create_with_schema, update_status) - 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. + file_toolkit.py file_download / file_upload / + file_delete / file_list / file_move + + path resolution + session temp dir + + graph_toolkit.py graph_traverse + adjacency BFS + + event_toolkit.py event_open / event_complete + (uses memory schema gates so the event + index follows the memory schema) + + agent_toolkit.py AGENT_TOOL_NAMES + build_agent_toolkit + + 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). @@ -51,9 +62,14 @@ 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 agent_toolkit # noqa: F401 -- 11 agent tools + memory_graph_search + +# Tool surfaces — each module's @R.register decorators fire on import. +from . import memory_toolkit # noqa: F401 -- 5 memory_* tools + memory_graph_search +from . import file_toolkit # noqa: F401 -- 5 file_* tools +from . import graph_toolkit # noqa: F401 -- graph_traverse +from . import event_toolkit # noqa: F401 -- event_open / event_complete +from . import agent_toolkit # noqa: F401 -- composition: AGENT_TOOL_NAMES 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 diff --git a/reme2/memory/agent_toolkit.py b/reme2/memory/agent_toolkit.py index 1e746d9a..fd3c121c 100644 --- a/reme2/memory/agent_toolkit.py +++ b/reme2/memory/agent_toolkit.py @@ -1,758 +1,53 @@ -"""Agent toolkit — the 11 tools an agent uses to operate on the vault. +"""Agent toolkit — composition entry point for the agent's tool surface. -Three categories. Each tool is a single-purpose ``BaseStep`` exposing -two surfaces: +The actual tool implementations are split by category across four +modules, each cohesive and self-contained: - * ``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. + memory_toolkit.py memory_get / memory_create / memory_update_body / + memory_update_meta / memory_search + + memory_graph_search (MCP-only) + + schema policy helpers (path template, status + state machine, create_with_schema, update_status) -Categories: + file_toolkit.py file_download / file_upload / file_delete / + file_list / file_move + + path resolution + session temp dir - Memory (5) schema-bound markdown management - memory_get / memory_create / memory_update_body / - memory_update_meta / memory_search + graph_toolkit.py graph_traverse + adjacency BFS - File (5) type-agnostic vault transport + directory operations - file_download / file_upload / file_delete / file_list / - file_move + event_toolkit.py event_open / event_complete + — uses memory_toolkit's schema gates so the + event index follows the memory schema - Graph (1) relationship exploration via BFS - graph_traverse +This module just composes them into one ``Toolkit`` for hosts that +want all 13 tools bound at once. Hosts that want a subset can import +the per-category builders and pass through a shared ``Toolkit``. -`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. +Lint tools (``check_dangling`` / ``check_orphans`` / ``check_collisions`` +/ ``check_schema``) live in ``lint_toolkit.py`` — separate factory, +not bound here by default. """ from __future__ import annotations -import json -import mimetypes -import shutil -import tempfile -from collections import deque -from pathlib import Path -from typing import Any +from agentscope.tool import Toolkit -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 +from .event_toolkit import EVENT_TOOL_NAMES +from .file_toolkit import FILE_TOOL_NAMES +from .graph_toolkit import GRAPH_TOOL_NAMES +from .memory_toolkit import MEMORY_TOOL_NAMES -# =========================================================================== -# 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). +# Order: memory → file → graph → event. Within memory, search comes +# last (retrieval is a separate concern from CRUD); the rest follow +# their CRUD order. memory_graph_search stays out — it's MCP-only. AGENT_TOOL_NAMES: tuple[str, ...] = ( - # memory (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", + *MEMORY_TOOL_NAMES, + *FILE_TOOL_NAMES, + *GRAPH_TOOL_NAMES, + *EVENT_TOOL_NAMES, ) diff --git a/reme2/memory/event_toolkit.py b/reme2/memory/event_toolkit.py new file mode 100644 index 00000000..7435d9ae --- /dev/null +++ b/reme2/memory/event_toolkit.py @@ -0,0 +1,309 @@ +"""Event category — task workspace lifecycle. + +Two tools: + + event_open open or create today's event workspace; create the + folder-note index following the memory schema + event_complete close the workspace; mode='ingest' marks distilled + (and triggers the Ingestor service when wired); + mode='abandon' marks archived + +The event index file ``events/{date}/{name}/{name}.md`` is created +through the same ``create_with_schema`` gate as ``memory_create`` — it +follows memory's frontmatter schema, status state machine, and path +template. The body starts empty; the agent maintains it as the working +context summary (Plan / Progress / Findings / Decisions / Next Steps) +via ``memory_update_body`` over the lifetime of the task. + +Workspace materials (anything else under the event folder) accumulate +through ``file_upload``. By convention, agent puts user-supplied files +under ``user_uploads/`` so the Ingestor can distinguish ground truth +from agent-derived artifacts purely by path. + +Status lifecycle (reuses memory state machine): + + active — workspace open, agent writing; ingest gate accepts + distilled — ingest succeeded; topics updated, workspace frozen + archived — workspace abandoned (mode=abandon, force-set since + active → archived isn't a normal transition) +""" + +from __future__ import annotations + +from datetime import date as _date +from pathlib import Path + +from agentscope.tool import ToolResponse + +from ..component import R +from ..component.base_step import BaseStep +from .file_toolkit import resolve_vault_path +from .memory_toolkit import create_with_schema, update_status +from .runtime_response import _set_answer, _tool_response + + +# =========================================================================== +# Section 1 — Helpers +# =========================================================================== + + +def _today_str() -> str: + return _date.today().isoformat() + + +def _index_relpath(date: str, name: str) -> str: + """``events/{date}/{name}/{name}.md`` — folder-note convention.""" + return f"events/{date}/{name}/{name}.md" + + +def _resolve_event_index(file_store, name_or_path: str) -> Path: + """Accept either a bare event name or an absolute / vault-relative + path to the index file. Bare names are interpreted under today.""" + if "/" in name_or_path or name_or_path.endswith(".md"): + return resolve_vault_path(file_store, name_or_path) + return resolve_vault_path(file_store, _index_relpath(_today_str(), name_or_path)) + + +def _allocate_name(file_store, date: str, name: str, force_new: bool) -> tuple[str, bool]: + """Resolve the event name to use, handling collisions. + + Returns ``(final_name, created_now)``. When ``force_new`` is False + and the workspace already exists, returns the existing name with + ``created_now=False`` (idempotent open). When ``force_new`` is True, + suffixes ``-2``, ``-3``, ... until a free slot is found. + """ + candidate = name + base = resolve_vault_path(file_store, _index_relpath(date, candidate)) + if not base.exists(): + return candidate, True + if not force_new: + return candidate, False + suffix = 2 + while True: + candidate = f"{name}-{suffix}" + probe = resolve_vault_path(file_store, _index_relpath(date, candidate)) + if not probe.exists(): + return candidate, True + suffix += 1 + + +# Default frontmatter for the event index. Agent-curated workspaces +# default to lifecycle=streaming + role=observation + scope=instance + +# source=agent — these are the values memory_create's schema validator +# accepts for event-shaped files (see test fixtures in test_expert). +_EVENT_DEFAULTS = { + "lifecycle": "streaming", + "scope": "instance", + "source": "agent", + "role": "observation", + "status": "active", +} + + +def _build_metadata(name: str, intent: str, related_topics: list[str] | None) -> dict: + meta: dict = { + "title": name, + **_EVENT_DEFAULTS, + } + if intent: + meta["intent"] = intent + if related_topics: + meta["related_topics"] = list(related_topics) + return meta + + +# =========================================================================== +# Section 2 — Event tools +# =========================================================================== + + +@R.register("event_open") +class EventOpen(BaseStep): + """Open or create today's event workspace. + + Idempotent: same name returns the existing workspace with + ``created=False``. ``force_new=True`` allocates a fresh suffix + instead. The folder-note index is created through the memory + schema gate (path template + frontmatter validation).""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + name: str = self.context.get("name", "") or "" + intent: str = self.context.get("intent", "") or "" + related_topics = self.context.get("related_topics") or [] + date: str = self.context.get("date") or _today_str() + force_new: bool = bool(self.context.get("force_new", False)) + assert name, "name is required" + payload = self._open(name, intent, list(related_topics), date, force_new) + self.context.response.success = "error" not in payload + _set_answer(self.context, payload) + + async def event_open( + self, + name: str, + intent: str = "", + related_topics: list[str] | None = None, + date: str | None = None, + force_new: bool = False, + ) -> ToolResponse: + """Open or create today's event workspace. + + Args: + name: workspace name (event id within the daily bucket). + intent: short statement of the task purpose (lands in + index frontmatter for the Ingestor to read). + related_topics: optional list of topic paths the agent + expects this task to touch. + date: defaults to today; pass YYYY-MM-DD to target a + specific daily bucket. + force_new: if True and ``name`` is taken, allocate a fresh + ``name-N`` suffix; otherwise return the existing workspace. + + Returns ``{path, name, created, date}``. Agent then uploads + materials via ``file_upload`` under the returned path, with + ``user_uploads/`` reserved for user-supplied files. + """ + payload = self._open( + name, intent, list(related_topics or []), + date or _today_str(), force_new, + ) + ok = "error" not in payload + return _tool_response("event_open", ok, payload, audit=self.audit) + + def _open( + self, + name: str, + intent: str, + related_topics: list[str], + date: str, + force_new: bool, + ) -> dict: + final_name, created_now = _allocate_name(self.file_store, date, name, force_new) + index_relpath = _index_relpath(date, final_name) + if not created_now: + existing = resolve_vault_path(self.file_store, index_relpath) + return { + "path": str(existing.parent), + "index": str(existing), + "name": final_name, + "date": date, + "created": False, + } + target = resolve_vault_path(self.file_store, index_relpath) + metadata = _build_metadata(final_name, intent, related_topics) + ok, payload = create_with_schema( + self.file_store, target, + metadata=metadata, content="", + overwrite=False, force=False, + ) + if not ok: + return { + "name": final_name, + "date": date, + "error": payload.get("error", "create failed"), + "detail": payload, + } + return { + "path": str(target.parent), + "index": str(target), + "name": final_name, + "date": date, + "created": True, + } + + +@R.register("event_complete") +class EventComplete(BaseStep): + """Close the event workspace. + + ``mode="ingest"`` flips status active → distilled (and is the + hook for Ingestor invocation when wired). ``mode="abandon"`` + marks the workspace archived (force-set since active → archived + isn't a normal state machine transition). + + Returns the event index path and (when ingest is wired) the list + of topics updated by the Ingestor. + """ + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + name_or_path: str = ( + self.context.get("name") or self.context.get("path") or "" + ) + mode: str = self.context.get("mode", "ingest") or "ingest" + assert name_or_path, "name or path is required" + assert mode in ("ingest", "abandon"), \ + f"mode must be 'ingest' or 'abandon', got {mode!r}" + payload = await self._complete(name_or_path, mode) + self.context.response.success = "error" not in payload + _set_answer(self.context, payload) + + async def event_complete( + self, + name: str, + mode: str = "ingest", + ) -> ToolResponse: + """Close the event workspace. + + Args: + name: workspace name (under today) OR vault-relative / + absolute path to the index file. + mode: ``"ingest"`` (default) flips active → distilled and + triggers ingestion; ``"abandon"`` marks archived. + + Returns ``{event_path, status, distilled, updated_topics}``. + """ + assert mode in ("ingest", "abandon"), \ + f"mode must be 'ingest' or 'abandon', got {mode!r}" + payload = await self._complete(name, mode) + ok = "error" not in payload + return _tool_response("event_complete", ok, payload, audit=self.audit) + + async def _complete(self, name_or_path: str, mode: str) -> dict: + index_path = _resolve_event_index(self.file_store, name_or_path) + if not index_path.is_file(): + return { + "name": name_or_path, + "error": "event index not found", + "expected_at": str(index_path), + } + if mode == "abandon": + ok, payload = update_status(index_path, value="archived", force=True) + return { + "event_path": str(index_path), + "status": "archived" if ok else None, + "distilled": False, + "updated_topics": [], + **({"error": payload.get("error")} if not ok else {}), + } + # mode == "ingest" + # Ingestor invocation hook — when the Ingestor service exposes a + # callable interface for "ingest one workspace", trigger it here + # and capture updated_topics. Until then we just flip status to + # distilled so downstream tools see the workspace as closed. + ok, payload = update_status(index_path, value="distilled", force=False) + if not ok: + return { + "event_path": str(index_path), + "status": None, + "distilled": False, + "updated_topics": [], + "error": payload.get("error", "status transition failed"), + "detail": payload, + } + return { + "event_path": str(index_path), + "status": "distilled", + "distilled": True, + "updated_topics": [], # populated when Ingestor is wired in + } + + +EVENT_TOOL_NAMES: tuple[str, ...] = ( + "event_open", + "event_complete", +) diff --git a/reme2/memory/file_toolkit.py b/reme2/memory/file_toolkit.py new file mode 100644 index 00000000..332191a5 --- /dev/null +++ b/reme2/memory/file_toolkit.py @@ -0,0 +1,248 @@ +"""File category — type-agnostic vault transport + directory operations. + +Five tools: + + file_download copy vault file → session temp dir + file_upload copy local file → vault path (any type) + file_delete remove vault file (universal entry point) + file_list enumerate vault files with optional filters + file_move rename / relocate; optional inbound-link rewrite + +Session-scoped temp dir is lazy: created on first ``file_download``, +auto-cleaned on process exit. Each download lands in a fresh +sub-directory so concurrent agents can freely modify the temp file +without trampling each other. +""" + +from __future__ import annotations + +import mimetypes +import shutil +import tempfile +from pathlib import Path + +from agentscope.tool import ToolResponse + +from . import memory_io +from ..component import R +from ..component.base_step import BaseStep +from .runtime_response import _set_answer, _tool_response + + +# =========================================================================== +# Section 1 — Helpers (session temp dir + path resolution) +# =========================================================================== + + +_TEMP_ROOT: Path | None = None + + +def _get_temp_root() -> Path: + """Lazy session-scoped temp dir. Auto-cleaned on process exit.""" + global _TEMP_ROOT + if _TEMP_ROOT is None: + _TEMP_ROOT = Path(tempfile.mkdtemp(prefix="reme2-files-")) + return _TEMP_ROOT + + +def resolve_vault_path(file_store, vault_path: str) -> Path: + """Compose the absolute on-disk path for a vault-relative entry. + + Exposed (no underscore) so ``event_toolkit`` can reuse the same + path-resolution logic when computing event workspace locations. + """ + working_dir = getattr(file_store, "working_dir", None) or "." + p = Path(vault_path) + if p.is_absolute(): + return p.resolve() + return (Path(working_dir) / p).resolve() + + +# =========================================================================== +# Section 2 — File tools +# =========================================================================== + + +@R.register("file_download") +class FileDownload(BaseStep): + """Copy a vault file to a session temp dir; return the local path.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + vault_path: str = self.context.get("vault_path", "") or "" + assert vault_path, "vault_path is required" + payload = self._download(vault_path) + self.context.response.success = "error" not in payload + _set_answer(self.context, payload) + + async def file_download(self, vault_path: str) -> ToolResponse: + """Copy a vault file to session temp dir; return the local path.""" + payload = self._download(vault_path) + ok = "error" not in payload + return _tool_response("file_download", ok, payload, audit=self.audit) + + def _download(self, vault_path: str) -> dict: + src = resolve_vault_path(self.file_store, vault_path) + if not src.is_file(): + return {"vault_path": vault_path, "error": "not found"} + dst_dir = Path(tempfile.mkdtemp(prefix="dl-", dir=_get_temp_root())) + dst = dst_dir / src.name + shutil.copy2(src, dst) + return { + "vault_path": vault_path, + "local_path": str(dst), + "size": dst.stat().st_size, + } + + +@R.register("file_upload") +class FileUpload(BaseStep): + """Copy a local file into the vault. Watcher / parser register the + FileNode asynchronously.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + local_path: str = self.context.get("local_path", "") or "" + vault_path: str = self.context.get("vault_path", "") or "" + overwrite: bool = bool(self.context.get("overwrite", True)) + assert local_path and vault_path, "local_path and vault_path are required" + payload = self._upload(local_path, vault_path, overwrite) + self.context.response.success = "error" not in payload + _set_answer(self.context, payload) + + async def file_upload( + self, local_path: str, vault_path: str, overwrite: bool = True, + ) -> ToolResponse: + """Copy local_path into the vault at vault_path.""" + payload = self._upload(local_path, vault_path, overwrite) + ok = "error" not in payload + return _tool_response("file_upload", ok, payload, audit=self.audit) + + def _upload(self, local_path: str, vault_path: str, overwrite: bool) -> dict: + src = Path(local_path) + if not src.is_file(): + return {"local_path": local_path, "error": "source not found"} + dst = resolve_vault_path(self.file_store, vault_path) + if dst.exists() and not overwrite: + return {"vault_path": vault_path, "error": "destination exists; pass overwrite=True"} + dst.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(src, dst) + return { + "vault_path": vault_path, + "size": dst.stat().st_size, + "mime": mimetypes.guess_type(dst.name)[0] or "application/octet-stream", + } + + +@R.register("file_delete") +class FileDelete(BaseStep): + """Delete a vault file. Universal entry point for any file type.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + vault_path: str = self.context.get("vault_path", "") or "" + assert vault_path, "vault_path is required" + target = resolve_vault_path(self.file_store, vault_path) + ok, payload = memory_io.delete_file(target) + self.context.response.success = ok + _set_answer(self.context, payload) + + async def file_delete(self, vault_path: str) -> ToolResponse: + """Delete a vault file.""" + target = resolve_vault_path(self.file_store, vault_path) + ok, payload = memory_io.delete_file(target) + return _tool_response("file_delete", ok, payload, audit=self.audit) + + +@R.register("file_list") +class FileList(BaseStep): + """Enumerate vault files with optional frontmatter filters.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + result = memory_io.list_files( + self.file_store, + path_prefix=self.context.get("prefix") or self.context.get("path_prefix"), + tags=self.context.get("tags") or [], + metadata=self.context.get("metadata") or {}, + limit=int(self.context.get("limit") or 100), + ) + _set_answer(self.context, result) + + async def file_list( + self, + prefix: str | None = None, + tags: list[str] | None = None, + metadata: dict | None = None, + limit: int = 100, + ) -> ToolResponse: + """List vault files. Filters: path prefix, frontmatter tags / fields.""" + result = memory_io.list_files( + self.file_store, + path_prefix=prefix, + tags=tags or [], + metadata=metadata or {}, + limit=limit, + ) + return _tool_response("file_list", True, result, audit=self.audit) + + +@R.register("file_move") +class FileMove(BaseStep): + """Rename / relocate. Default leaves inbound wikilinks untouched + (maintainer cleans dangling refs); pass ``update_refs=True`` to + rewrite ``[[old]] → [[new]]`` in every referencing file.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + src: str = self.context.get("src") or self.context.get("old_path") or "" + dst: str = self.context.get("dst") or self.context.get("new_path") or "" + update_refs: bool = bool(self.context.get("update_refs", False)) + assert src and dst, "src and dst are required" + payload = self._move(src, dst, update_refs) + self.context.response.success = payload.get("ok", False) + _set_answer(self.context, payload) + + async def file_move( + self, src: str, dst: str, update_refs: bool = False, + ) -> ToolResponse: + """Rename / relocate. update_refs=True rewrites [[old]] → [[new]].""" + payload = self._move(src, dst, update_refs) + ok = payload.get("ok", False) + return _tool_response("file_move", ok, payload, audit=self.audit) + + def _move(self, src: str, dst: str, update_refs: bool) -> dict: + src_abs = resolve_vault_path(self.file_store, src) + dst_abs = resolve_vault_path(self.file_store, dst) + if not src_abs.is_file(): + return {"ok": False, "src": src, "error": "source not found"} + if update_refs: + working_dir = Path(getattr(self.file_store, "working_dir", None) or ".").resolve() + ok, payload = memory_io.rename_file( + self.file_store, working_dir, + old_path=src_abs, new_path=dst_abs, + ) + payload["ok"] = ok + return payload + dst_abs.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(src_abs), str(dst_abs)) + return {"ok": True, "src": str(src_abs), "dst": str(dst_abs), "refs_updated": 0} + + +FILE_TOOL_NAMES: tuple[str, ...] = ( + "file_download", + "file_upload", + "file_delete", + "file_list", + "file_move", +) diff --git a/reme2/memory/graph_toolkit.py b/reme2/memory/graph_toolkit.py new file mode 100644 index 00000000..15522338 --- /dev/null +++ b/reme2/memory/graph_toolkit.py @@ -0,0 +1,148 @@ +"""Graph category — relationship exploration via BFS. + +Single tool: ``graph_traverse``. depth=1 covers the trivial outlinks / +inlinks lookups (just set direction); multi-hop covers exploration. +Output is one record per edge traversed (not per node), so the same +node may appear multiple times if reached via different predicates or +paths — agents dedupe at the call site if they want a flat node set. + +Adjacency source: walks ``file_store.file_nodes`` directly (each +FileNode carries its ``links`` list). Inbound is a linear scan over +all nodes; cheap for vault sizes. Swap for a precomputed reverse +index on the file_graph component if it ever becomes a hot path. +""" + +from __future__ import annotations + +from collections import deque + +from agentscope.tool import ToolResponse + +from ..component import R +from ..component.base_step import BaseStep +from .runtime_response import _set_answer, _tool_response + + +# =========================================================================== +# Section 1 — Adjacency lookups +# =========================================================================== + + +def _outlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]: + """Outgoing edges from ``path`` — [(target_path, predicate, anchor)].""" + node = file_store.file_nodes.get(path) + if node is None: + return [] + return [(link.path, link.predicate, link.anchor) for link in node.links if link.path] + + +def _inlinks(file_store, path: str) -> list[tuple[str, str | None, str | None]]: + """Incoming edges to ``path`` — linear scan over all nodes' links.""" + out: list[tuple[str, str | None, str | None]] = [] + for src_path, src_node in file_store.file_nodes.items(): + if src_path == path: + continue + for link in src_node.links: + if link.path == path: + out.append((src_path, link.predicate, link.anchor)) + return out + + +def _bfs( + file_store, + seeds: list[str], + max_depth: int, + direction: str, + predicate: str | None, +) -> list[dict]: + """BFS from each seed. One record per edge traversed.""" + visited_edges: set[tuple[str, str, str | None]] = set() + results: list[dict] = [] + queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds) + while queue: + current, depth = queue.popleft() + if depth >= max_depth: + continue + edges: list[tuple[str, str | None, str | None]] = [] + if direction in ("out", "both"): + for tgt, pred, anchor in _outlinks(file_store, current): + if predicate is not None and pred != predicate: + continue + edges.append((tgt, pred, anchor)) + if direction in ("in", "both"): + for src, pred, anchor in _inlinks(file_store, current): + if predicate is not None and pred != predicate: + continue + edges.append((src, pred, anchor)) + for next_path, pred, anchor in edges: + edge_key = (current, next_path, pred) + if edge_key in visited_edges: + continue + visited_edges.add(edge_key) + results.append({ + "path": next_path, + "depth": depth + 1, + "via": current, + "predicate": pred, + "anchor": anchor, + }) + if depth + 1 < max_depth: + queue.append((next_path, depth + 1)) + return results + + +# =========================================================================== +# Section 2 — Graph tool +# =========================================================================== + + +@R.register("graph_traverse") +class GraphTraverse(BaseStep): + """BFS from seed(s) to explore relationships in the memory graph.""" + + audit: list[dict] | None = None + + async def execute(self): + assert self.context is not None + seeds_raw = self.context.get("seeds") or [] + if isinstance(seeds_raw, str): + seeds = [seeds_raw] + else: + seeds = list(seeds_raw) + max_depth: int = int(self.context.get("max_depth") or 1) + direction: str = self.context.get("direction", "out") or "out" + predicate = self.context.get("predicate") + assert seeds, "seeds is required (single path or list of paths)" + assert direction in ("out", "in", "both"), \ + f"direction must be 'out' | 'in' | 'both', got {direction!r}" + results = _bfs(self.file_store, seeds, max_depth, direction, predicate) + _set_answer(self.context, results) + + async def graph_traverse( + self, + seeds: str | list[str], + max_depth: int = 1, + direction: str = "out", + predicate: str | None = None, + ) -> ToolResponse: + """BFS from seed(s) to explore relationships. + + Args: + seeds: single path or list of paths to start from. + max_depth: hops to expand (default 1 = immediate neighbors). + direction: "out" / "in" / "both". + predicate: filter edges by predicate (None = no filter). + """ + if isinstance(seeds, str): + seeds_list = [seeds] + else: + seeds_list = list(seeds) + assert direction in ("out", "in", "both"), \ + f"direction must be 'out' | 'in' | 'both', got {direction!r}" + results = _bfs(self.file_store, seeds_list, max_depth, direction, predicate) + return _tool_response("graph_traverse", True, results, audit=self.audit) + + +GRAPH_TOOL_NAMES: tuple[str, ...] = ( + "graph_traverse", +) diff --git a/reme2/memory/memory_io.py b/reme2/memory/memory_io.py index 0c68afa0..7a760044 100644 --- a/reme2/memory/memory_io.py +++ b/reme2/memory/memory_io.py @@ -1,238 +1,110 @@ -"""Memory File System engine API — the core engine's outward surface. +"""Memory File System engine API — minimal surface for the agent toolkit. -The .md files are the SSOT (per `structure.md` §"核心引擎"). The engine -is layered: +The .md files are the SSOT. The engine layer is: Memory File System → Watcher & Parser → Projections (vector / FTS) (write entry) (incremental) (read entry, derived) -This module is the **single public API surface** over that engine. Every -consumer — agent-facing steps, the three memory services (Retriever, -Ingestor, Maintainer), and the toolkit (`memory_toolkit`) — talks to -the engine through these functions, not by reaching into -``BaseFileStore`` directly. +This module is the single public surface over that engine for the +agent toolkit. ``BaseFileGraph`` deliberately doesn't expose "scan +everything" (only ``get_nodes(paths)``); when we need to enumerate +(file_list, collisions check, filter resolution) we walk the +filesystem directly. The graph is consulted only for adjacency +lookups around a known path. -Layering note. The slim ``BaseFileStore`` interface only owns the -search projections (vector / FTS) plus atomic file upserts/deletes. -Iteration, single-node lookup, and the wikilink graph live HERE — we -walk ``LocalFileStore._nodes`` directly (engine-layer peer) and compute -links on-the-fly from each ``FileNode.links``. There is no precomputed -graph index; the graph is a derivation of the SSoT. +Public surface (everything else has been removed): -Naming convention: - - Verb-first: `get_file`, `create_file`, `search_vector`. - - `file_store` is always the first positional argument when the - function needs the engine handle; remaining args are keyword-only. - - Pure-disk writes (`update_body`, `update_meta`, `delete_file`, - `archive_file`) don't take `file_store` — they hit the filesystem - and the watcher picks them up. The asymmetry is honest. + Reads: get_file, list_files + Writes: create_file, update_body, update_meta, delete_file, + rename_file + Filter: make_filter + +Pure-disk writes (``update_body`` / ``update_meta`` / ``delete_file``) +don't take ``file_store`` — they hit the filesystem and the watcher +picks them up. Engine-aware writes (``create_file`` / ``rename_file``) +take ``file_store`` because they consult the vault index for +collision/adjacency gates before writing. """ from __future__ import annotations import re -import shutil -from collections import deque -from collections.abc import Iterable, Iterator +from collections.abc import Iterator from pathlib import Path import frontmatter from ..component.file_store.base_file_store import BaseFileStore -from ..schema import ChunkFilter, FileChunk, FileNode -from ..utils.wikilink_resolver import ( - _WIKILINK_RE, - extract_wikilinks, -) +from ..schema import ChunkFilter, FileNode +from ..utils.wikilink_resolver import _WIKILINK_RE # =========================================================================== -# Internal helpers — engine-layer access to the concrete node index -# =========================================================================== -# -# ``BaseFileStore`` only declares the search/upsert contract; iteration -# and single-node lookup live on the concrete impl (``LocalFileStore``). -# memory_io is a peer at the engine layer, so reaching into ``_nodes`` -# is intentional — every iteration / graph walk funnels through here so -# the day the engine grows a public ``iter_nodes()``, this is the only -# place to swap. - - -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 _filter_to_dict(chunk_filter: ChunkFilter | None) -> dict: - """Serialize ``ChunkFilter`` for the search engine's ``search_filter`` arg.""" - if chunk_filter is None: - return {} - return chunk_filter.model_dump(mode="json") - - -# =========================================================================== -# Section 1 — MFS Reads +# Internal helpers # =========================================================================== -async def get_file( - file_store: BaseFileStore, - path: str, - *, - include_chunks: bool = False, -) -> dict: - """Read frontmatter + body for one path. Optionally include parsed chunks. - - On-disk frontmatter is the source of truth — the file_store cache may - lag a write that hasn't been picked up by the watcher yet. - """ - node = await file_store.get_node_by_path(path) # type: ignore[attr-defined] - result: dict = {"path": path, "exists": False} - if node is not None: - result.update( - { - "exists": True, - "metadata": _meta(node), - "link": [link.model_dump(exclude_none=True) for link in node.links], - } - ) - - file_path = Path(path) - if file_path.is_file(): - raw = file_path.read_text(encoding="utf-8") - post = frontmatter.loads(raw) - result["exists"] = True - result["content"] = post.content - result["metadata"] = dict(post.metadata) - - if include_chunks: - chunks = await file_store.get_chunks_by_path(path) # type: ignore[attr-defined] - result["chunks"] = [c.model_dump(exclude_none=True) for c in chunks] - return result +def _vault_root(file_store: BaseFileStore) -> Path | None: + """Resolve the vault root from the file_store. ``working_dir`` is the + convention; falls back to ``working_path`` if set on the component.""" + wd = getattr(file_store, "working_dir", None) + if wd: + return Path(wd).resolve() + wp = getattr(file_store, "working_path", None) + if wp: + return Path(wp).resolve() + return None -def list_files( +_SKIP_DIRS = {".git", ".obsidian", ".reme", ".reme2", "__pycache__", "Archive"} + + +def _walk_vault( file_store: BaseFileStore, *, - path_prefix: str | None = None, - tags: list[str] | None = None, - metadata: dict | None = None, - limit: int = 100, -) -> dict: - """List indexed files filtered by frontmatter exact-match, tags, and prefix.""" - metadata_filter = metadata or {} - tag_filter = tags or [] - items: list[dict] = [] - for path, node in _nodes(file_store).items(): - if path_prefix and not path.startswith(path_prefix): - continue - md = _meta(node) - if metadata_filter and any(md.get(k) != v for k, v in metadata_filter.items()): - continue - if tag_filter: - file_tags = set(md.get("tags") or []) - if not all(t in file_tags for t in tag_filter): - continue - items.append({"path": path, "metadata": md}) - if len(items) >= limit: - break - return {"items": items, "count": len(items)} + suffix: str = ".md", +) -> Iterator[Path]: + """Yield every file under the vault matching ``suffix``. - -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, - "inlinks": file_store.file_graph.get_inlinks(path), - } - - -def get_outlinks(file_store: BaseFileStore, path: str) -> dict: - """Files that link TO `path`. Each entry carries the typed-link predicate.""" - return { - "path": path, - "outlinks": file_store.file_graph.get_outlinks(path), - } - - -def resolve_wikilink(file_store: BaseFileStore, wikilink: str) -> dict: - """Resolve a `[[target]]` wikilink with full ambiguity context. - - Returns: - unique resolution → {wikilink, path, exists: True, - ambiguous: False, candidates: [path]} - ambiguous → {wikilink, path: None, exists: False, - ambiguous: True, candidates: [...]} - dangling → {wikilink, path: None, exists: False, - ambiguous: False, candidates: []} + Filesystem-as-SSOT: the .md files are authoritative for "what + exists". Hidden directories and ``Archive/`` are skipped (Archive + lives in the vault but isn't part of the active set). """ - if "/" in wikilink or wikilink.endswith(".md"): - hit = _resolve_wikilink(file_store, wikilink) - return { - "wikilink": wikilink, - "path": hit, - "exists": hit is not None, - "ambiguous": False, - "candidates": [hit] if hit else [], - } - - candidates = wikilink_candidates(file_store, wikilink) - if len(candidates) == 1: - return { - "wikilink": wikilink, - "path": candidates[0], - "exists": True, - "ambiguous": False, - "candidates": candidates, - } - return { - "wikilink": wikilink, - "path": None, - "exists": False, - "ambiguous": len(candidates) > 1, - "candidates": candidates, - } + root = _vault_root(file_store) + if root is None or not root.is_dir(): + return + for path in root.rglob(f"*{suffix}"): + if any(part in _SKIP_DIRS for part in path.relative_to(root).parts[:-1]): + continue + if path.is_file(): + yield path.resolve() -def iter_files(file_store: BaseFileStore) -> Iterator[tuple[str, FileNode]]: - """Walk every indexed (path, FileNode). Used by Maintainer scans.""" - return iter(_nodes(file_store).items()) - - -async def count_tokens( - token_counter, +def _walk_vault_meta( + file_store: BaseFileStore, *, - path: str | None = None, - text: str | None = None, -) -> dict: - """Estimate tokens for a file body (frontmatter excluded) or raw text.""" - if path: - target = Path(path) - if not target.is_file(): - return {"path": str(target), "error": "file not found"} - raw = target.read_text(encoding="utf-8") - post = frontmatter.loads(raw) - body = post.content - tokens = await token_counter.count(messages=[], text=body) - return { - "source": "file", - "path": str(target.resolve()), - "tokens": tokens, - "body_chars": len(body), - } - if text: - tokens = await token_counter.count(messages=[], text=text) - return { - "source": "text", - "tokens": tokens, - "body_chars": len(text), - } - return {"error": "one of `path` or `text` is required"} + suffix: str = ".md", +) -> Iterator[tuple[str, dict]]: + """Walk the vault, yielding ``(absolute_path_str, frontmatter_dict)``. + + Files that fail to parse get an empty dict — that's a lint concern, + not the listing concern. + """ + for abs_path in _walk_vault(file_store, suffix=suffix): + try: + raw = abs_path.read_text(encoding="utf-8") + meta = dict(frontmatter.loads(raw).metadata) + except Exception: + meta = {} + yield str(abs_path), meta -# =========================================================================== -# Section 2 — MFS Writes -# =========================================================================== +async def _get_node(file_store: BaseFileStore, path: str) -> FileNode | None: + """Single-node fetch via the file_graph contract.""" + if not file_store.file_graph: + return None + nodes = await file_store.file_graph.get_nodes([path]) + return nodes[0] if nodes else None def _replace_wikilink_targets(text: str, mapping: dict[str, str]) -> str: @@ -250,6 +122,113 @@ def _replace_wikilink_targets(text: str, mapping: dict[str, str]) -> str: return _WIKILINK_RE.sub(sub, text) +def collisions_after_create( + file_store: BaseFileStore, + proposed_path: str | Path, +) -> list[str]: + """Existing paths that would conflict with adding `proposed_path`. + + Folder-note rule: if `proposed_path`'s parent dir name == its stem, + only colliding folder-notes are returned (siblings with the same + stem don't ambiguate). Otherwise both folder-notes AND stem hits + are returned. + + Walks the filesystem since the engine contract has no "scan all + paths" API on the graph. Public so other write paths (sync.py) + can reuse the same gate. + """ + p = Path(proposed_path) + stem = p.stem + proposed_abs = str(p.resolve()) + is_folder_note = p.parent.name == stem + + folder_hits: list[str] = [] + stem_hits: list[str] = [] + for abs_path in _walk_vault(file_store): + path = str(abs_path) + if path == proposed_abs: + continue + if abs_path.stem != stem: + continue + if abs_path.parent.name == stem: + folder_hits.append(path) + else: + stem_hits.append(path) + + if is_folder_note: + return folder_hits + return folder_hits + stem_hits + + +# =========================================================================== +# Reads +# =========================================================================== + + +async def get_file(file_store: BaseFileStore, path: str) -> dict: + """Read frontmatter + body from disk; attach links from the graph. + + On-disk frontmatter is the source of truth — the graph cache may + lag a write that the watcher hasn't picked up yet. The graph is + consulted only for the link list (adjacency). + """ + result: dict = {"path": path, "exists": False} + + file_path = Path(path) + if file_path.is_file(): + raw = file_path.read_text(encoding="utf-8") + post = frontmatter.loads(raw) + result["exists"] = True + result["content"] = post.content + result["metadata"] = dict(post.metadata) + + node = await _get_node(file_store, path) + if node is not None: + result["link"] = [link.model_dump(exclude_none=True) for link in node.links] + if not result["exists"]: + result["exists"] = True + result.setdefault("metadata", node.front_matter.model_dump()) + else: + result["link"] = [] + return result + + +def list_files( + file_store: BaseFileStore, + *, + path_prefix: str | None = None, + tags: list[str] | None = None, + metadata: dict | None = None, + limit: int = 100, +) -> dict: + """List vault files filtered by frontmatter exact-match, tags, and prefix. + + Filesystem walk — the graph contract has no enumeration API. + ``path_prefix`` matches the absolute path string. + """ + metadata_filter = metadata or {} + tag_filter = tags or [] + items: list[dict] = [] + for path, md in _walk_vault_meta(file_store): + if path_prefix and not path.startswith(path_prefix): + continue + if metadata_filter and any(md.get(k) != v for k, v in metadata_filter.items()): + continue + if tag_filter: + file_tags = set(md.get("tags") or []) + if not all(t in file_tags for t in tag_filter): + continue + items.append({"path": path, "metadata": md}) + if len(items) >= limit: + break + return {"items": items, "count": len(items)} + + +# =========================================================================== +# Writes +# =========================================================================== + + def create_file( file_store: BaseFileStore, path: Path, @@ -259,11 +238,10 @@ def create_file( overwrite: bool = False, force: bool = False, ) -> tuple[bool, dict]: - """Single L1 entry point for creating a markdown file. + """Create a markdown file. Refuses when: - Refuses when: - - file already exists (unless overwrite=True) - - creating it would make `[[stem]]` ambiguous (unless force=True) + - file already exists (unless ``overwrite=True``) + - creating it would make ``[[stem]]`` ambiguous (unless ``force=True``) """ if path.exists() and not overwrite: return False, {"path": str(path), "error": "file already exists"} @@ -299,7 +277,7 @@ def update_body( new_string: str, replace_all: bool = False, ) -> tuple[bool, dict]: - """Edit-style content update — replace `old_string` with `new_string`.""" + """Edit-style content update — replace ``old_string`` with ``new_string``.""" target = Path(path) if not target.is_file(): return False, {"path": str(target), "error": "file not found"} @@ -330,7 +308,7 @@ def update_body( def update_meta(path: Path | str, *, key: str, value) -> tuple[bool, dict]: - """Update a single YAML frontmatter key. value=None deletes the key.""" + """Update a single YAML frontmatter key. ``value=None`` deletes the key.""" target = Path(path) if not target.is_file(): return False, {"path": str(target), "error": "file not found"} @@ -344,6 +322,15 @@ def update_meta(path: Path | str, *, key: str, value) -> tuple[bool, dict]: return True, {"path": str(target), "key": key, "value": value} +def delete_file(path: Path | str) -> tuple[bool, dict]: + """Delete a file. Watcher removes from store + graph asynchronously.""" + target = Path(path) + if not target.exists(): + return False, {"path": str(target), "error": "not found"} + target.unlink() + return True, {"path": str(target), "deleted": True} + + def rename_file( file_store: BaseFileStore, working_dir: Path | str, @@ -351,7 +338,13 @@ def rename_file( old_path: Path | str, new_path: Path | str, ) -> tuple[bool, dict]: - """Rename a file and rewrite incoming wikilinks across the vault.""" + """Rename a file and rewrite incoming wikilinks across the vault. + + Walks the filesystem to find referencing files (the graph + contract has no source-path lookup for inbound edges, and the + rewrite is text-level anyway). Same-stem renames within the same + folder are no-op for link rewrite. + """ old_p = Path(old_path).resolve() new_p = Path(new_path).resolve() @@ -366,7 +359,8 @@ def rename_file( if conflicts: return False, { "error": ( - f"stem `[[{new_p.stem}]]` would resolve ambiguously " f"to {len(conflicts) + 1} paths after this rename" + f"stem `[[{new_p.stem}]]` would resolve ambiguously " + f"to {len(conflicts) + 1} paths after this rename" ), "conflicts": conflicts, "hint": ( @@ -391,255 +385,39 @@ def rename_file( except ValueError: pass - referring_paths = [m.path for m, _ in _get_inlinks(file_store, str(old_p))] - new_p.parent.mkdir(parents=True, exist_ok=True) old_p.rename(new_p) updated_files: list[str] = [] write_errors: list[dict] = [] - if replacements and referring_paths: - for path in referring_paths: - file_path = Path(path) - if not file_path.is_file(): + if replacements: + for abs_path in _walk_vault(file_store): + if abs_path == new_p: continue try: - raw = file_path.read_text(encoding="utf-8") + raw = abs_path.read_text(encoding="utf-8") new_raw = _replace_wikilink_targets(raw, replacements) if new_raw != raw: - file_path.write_text(new_raw, encoding="utf-8") - updated_files.append(path) + abs_path.write_text(new_raw, encoding="utf-8") + updated_files.append(str(abs_path)) except Exception as exc: - write_errors.append({"path": path, "error": str(exc)}) + write_errors.append({"path": str(abs_path), "error": str(exc)}) return True, { "old_path": str(old_p), "new_path": str(new_p), "stem_changed": old_stem != new_stem, "replacements": replacements, - "referring_count": len(referring_paths), "updated_files": updated_files, "write_errors": write_errors, } -def delete_file(path: Path | str) -> tuple[bool, dict]: - """Delete a file. Watcher removes from store + graph.""" - target = Path(path) - if not target.exists(): - return False, {"path": str(target), "error": "not found"} - target.unlink() - return True, {"path": str(target), "deleted": True} - - -def archive_file( - working_dir: Path | str, - path: Path | str, - *, - archive_dir: str = "Archive", -) -> tuple[bool, dict]: - """Archive a file: flip `status: archived`, then move under `//`.""" - src = Path(path).resolve() - if not src.is_file(): - return False, {"path": str(src), "error": "file not found"} - - vault = Path(working_dir).resolve() - try: - rel = src.relative_to(vault) - except ValueError: - return False, { - "path": str(src), - "error": f"path is outside working_dir {vault}", - } - - dst = vault / archive_dir / rel - if dst.exists(): - return False, { - "path": str(src), - "error": f"archive destination already exists: {dst}", - } - - ok, prop_payload = update_meta(src, key="status", value="archived") - if not ok: - return False, {**prop_payload, "stage": "update_meta"} - - dst.parent.mkdir(parents=True, exist_ok=True) - try: - shutil.move(str(src), str(dst)) - except OSError as exc: - return False, { - "path": str(src), - "error": f"move failed: {exc}", - "stage": "move", - } - return True, { - "old_path": str(src), - "new_path": str(dst), - "archived": True, - } - - # =========================================================================== -# Section 3 — Projection Queries +# Filter helper # =========================================================================== -async def search_vector( - file_store: BaseFileStore, - query: str, - *, - limit: int, - chunk_filter: ChunkFilter | None = None, -) -> list[FileChunk]: - """Vector similarity over the chunk-level Vector projection.""" - return await file_store.vector_search(query, limit, _filter_to_dict(chunk_filter)) - - -async def search_keyword( - file_store: BaseFileStore, - query: str, - *, - limit: int, - chunk_filter: ChunkFilter | None = None, -) -> list[FileChunk]: - """FTS5 keyword search over the chunk-level keyword projection.""" - return await file_store.keyword_search(query, limit, _filter_to_dict(chunk_filter)) - - -async def get_chunks( - file_store: BaseFileStore, - paths: Iterable[str], -) -> list[FileChunk]: - """Batch fetch chunks across many paths.""" - out: list[FileChunk] = [] - for p in paths: - out.extend(await file_store.get_chunks_by_path(p)) # type: ignore[attr-defined] - return out - - -# =========================================================================== -# Section 4 — Graph helpers (BFS / scoring / collisions) -# =========================================================================== -# -# All graph traversal walks ``FileNode.links`` directly via ``_get_outlinks`` -# / ``_get_inlinks``. There's no precomputed adjacency index — every BFS -# resolves wikilinks per call. Cheap enough for typical vault sizes; -# revisit if the maintainer's lint pass becomes a hotspot. - - -def expand_neighbors( - file_store: BaseFileStore, - seeds: Iterable[str], - *, - depth: int = 1, - direction: str = "both", - per_node_cap: int | None = 50, -) -> dict[str, int]: - """BFS over wikilink edges. Returns `{path: hop_distance}`.""" - if direction not in ("out", "in", "both"): - raise ValueError(f"direction must be one of out/in/both, got {direction!r}") - if depth < 0: - raise ValueError(f"depth must be >= 0, got {depth}") - - nodes = _nodes(file_store) - seen: dict[str, int] = {} - frontier: deque[tuple[str, int]] = deque() - for path in seeds: - if path in nodes and path not in seen: - seen[path] = 0 - frontier.append((path, 0)) - - while frontier: - path, dist = frontier.popleft() - if dist >= depth: - continue - - neighbors: list[str] = [] - if direction in ("out", "both"): - neighbors.extend(m.path for m, _ in _get_outlinks(file_store, path)) - if direction in ("in", "both"): - neighbors.extend(m.path for m, _ in _get_inlinks(file_store, path)) - - if per_node_cap is not None and len(neighbors) > per_node_cap: - neighbors = neighbors[:per_node_cap] - - for nb in neighbors: - if nb in seen: - continue - seen[nb] = dist + 1 - frontier.append((nb, dist + 1)) - - return seen - - -def subgraph_score( - file_store: BaseFileStore, - seeds: Iterable[str], - *, - decay: float = 0.5, - depth: int = 1, - direction: str = "both", - per_node_cap: int | None = 50, -) -> dict[str, float]: - """Decayed score per path: seed=1.0, 1-hop=decay, 2-hop=decay²...""" - if not (0.0 <= decay <= 1.0): - raise ValueError(f"decay must be in [0, 1], got {decay}") - hops = expand_neighbors( - file_store, - seeds, - depth=depth, - direction=direction, - per_node_cap=per_node_cap, - ) - return {path: decay**hop for path, hop in hops.items()} - - -def extract_anchors(file_store: BaseFileStore, text: str) -> list[str]: - """Pull `[[X]]` anchors from `text` and resolve each (deduped).""" - seen: set[str] = set() - out: list[str] = [] - for raw in extract_wikilinks(text): - hit = _resolve_wikilink(file_store, raw) - if hit is not None and hit not in seen: - seen.add(hit) - out.append(hit) - return out - - -def collisions_after_create( - file_store: BaseFileStore, - proposed_path: str | Path, -) -> list[str]: - """Existing paths that would conflict with adding `proposed_path`. - - Folder-note rule: if `proposed_path`'s parent dir name == its stem, - only colliding folder-notes are returned (siblings with the same - stem don't ambiguate). Otherwise both folder-notes AND stem hits - are returned. - """ - p = Path(proposed_path) - stem = p.stem - proposed_abs = str(p.resolve()) - is_folder_note = p.parent.name == stem - - folder_hits: list[str] = [] - stem_hits: list[str] = [] - for path in _nodes(file_store): - if path == proposed_abs: - continue - path_obj = Path(path) - if path_obj.stem != stem: - continue - if path_obj.parent.name == stem: - folder_hits.append(path) - else: - stem_hits.append(path) - - if is_folder_note: - return folder_hits - return folder_hits + stem_hits - - def make_filter( file_store: BaseFileStore, *, @@ -647,17 +425,16 @@ def make_filter( tags: list[str] | None = None, exclude_paths: list[str] | None = None, ) -> ChunkFilter | None: - """Compile a ChunkFilter against the file_store's path/tag indexes.""" + """Compile a ChunkFilter and resolve it against the vault. + + Walks the filesystem to determine which paths match the filter's + metadata clauses (the engine contract has no path enumeration on + the graph). Empty filters skip the walk. + """ cf = ChunkFilter(paths=paths, tags=tags, exclude_paths=exclude_paths) if cf.is_empty(): return cf - cf.resolved_paths = {p for p, n in _nodes(file_store).items() if cf.match_metadata(p, _meta(n))} + cf.resolved_paths = { + p for p, md in _walk_vault_meta(file_store) if cf.match_metadata(p, md) + } return cf - - -def find_collisions(file_store: BaseFileStore) -> dict[str, list[str]]: - """Every stem that resolves to >1 path. Used by Maintainer.lint.""" - by_stem: dict[str, list[str]] = {} - for path in _nodes(file_store): - by_stem.setdefault(Path(path).stem, []).append(path) - return {s: ps for s, ps in by_stem.items() if len(ps) > 1} diff --git a/reme2/memory/memory_toolkit.py b/reme2/memory/memory_toolkit.py new file mode 100644 index 00000000..99943247 --- /dev/null +++ b/reme2/memory/memory_toolkit.py @@ -0,0 +1,434 @@ +"""Memory category — schema-bound markdown management + hybrid retrieval. + +Five tools: + + memory_get read structured (frontmatter + body + links) + memory_create schema-validated create + memory_update_body Edit-style body replacement + memory_update_meta frontmatter patch (status state machine enforced) + memory_search V+K hybrid retrieval + +Plus ``memory_graph_search`` (MCP-only — no agent toolkit method, but +registered so HTTP/MCP callers can still tune graph fusion knobs). + +Schema policy lives here — ``validate_status_transition``, +``validate_path_template``, ``_create_with_schema``, ``_update_status`` +— and is reused by ``event_toolkit`` so event index files go through +the same gates as memory_create. +""" + +from __future__ import annotations + +from pathlib import Path + +import frontmatter +from agentscope.tool import ToolResponse + +from . import memory_io +from ..component import R +from ..component.base_step import BaseStep +from ..enumeration import ComponentEnum +from .retriever import BaseRetriever, HybridRetriever +from .runtime_response import _set_answer, _tool_response + + +# =========================================================================== +# Section 1 — Schema policy (status state machine + path templates) +# =========================================================================== +# +# Used by memory_create (path template), memory_update_meta (status +# state machine), and event_open (both, since the event index file +# follows the same memory schema). 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 | str | 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 + vault = Path(working_dir).resolve() + try: + rel = path.resolve().relative_to(vault) + except ValueError: + return f"path {path} is outside working_dir {vault}" + 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 — Memory 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 + return all_ok, {"path": path, "applied": results} + + +# =========================================================================== +# Section 3 — Retrieval (memory_search + memory_graph_search) +# =========================================================================== + + +_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) + + +@R.register("memory_graph_search") +class MemoryGraphSearch(BaseStep): + """V + K + graph BFS fusion. MCP-only — not bound to agent toolkit. + + Per-call overrides for fusion knobs are forwarded if present in + the RuntimeContext.""" + + _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, + ) + overrides = {k: ctx.get(k) for k in self._OVERRIDE_KEYS if ctx.get(k) is not None} + retriever = _resolve_retriever(self) + results = await retriever.search( + query=query, max_results=max_results, min_score=min_score, + chunk_filter=chunk_filter, seeds=explicit_seeds or None, **overrides, + ) + payload = [_serialize_chunk(r, self.file_store) for r in results] + _set_answer(self.context, payload) + + +MEMORY_TOOL_NAMES: tuple[str, ...] = ( + "memory_get", + "memory_create", + "memory_update_body", + "memory_update_meta", + "memory_search", +) diff --git a/reme2/memory/sync.py b/reme2/memory/sync.py deleted file mode 100644 index 5ffd6ab7..00000000 --- a/reme2/memory/sync.py +++ /dev/null @@ -1,493 +0,0 @@ -"""sync — continuously sync key materials into an event folder. - -Layout (one folder per logical thread): - - events/{date}/{name}/ - ├── {name}.md # index: frontmatter + narrative + Materials footer - ├── {material_filename} # raw artifact 1 - ├── ... - └── {material_filename} # raw artifact N - -Hot-path write entry. The agent picks a stable `name` per logical -thread and calls `sync` repeatedly through the task — each call -extends the same event folder rather than creating a new one. This -turns discrete writes into a coherent stream and lets PreCompact / -SessionEnd hooks treat sync as the last-chance flush before context -loss. - -Behavior: - * If `events/{date}/{name}/` does NOT exist → create new folder, - write index `{name}.md` (status=active), write materials. - * If it exists with `status: active` → APPEND: - - new content → `## Update — {iso}` section appended to the body - - new materials → siblings (auto-suffix on filename collision) - - Materials footer regenerated as the trailing section, listing - every artifact actually present in the folder - - frontmatter `topics` + `tags` unioned, `updated` set to today - * If it exists with `status: distilled` / `archived` → REFUSE, - return suggested_name so the agent can start a new thread. - -Zero LLM cost. -""" - -import json -import re -from collections.abc import Iterable -from datetime import date as date_type, datetime, timezone -from pathlib import Path - -import frontmatter -from pydantic import ValidationError - -from reme2.component import R -from reme2.component.base_step import BaseStep -from .memory_io import collisions_after_create, create_file -from .schema import EVENT_PRESET, MemoryFileNode - - -_SAFE_FILENAME_RE = re.compile(r"^[a-zA-Z0-9._-]+$") -_MATERIALS_HEADER = "## Materials" - - -def _event_path( - working_dir: str | Path, - name: str, - on_date: date_type | str | None = None, - events_dir: str = "events", -) -> Path: - """events/{YYYY-MM-DD}/{name}/{name}.md under the given vault root.""" - if on_date is None: - on_date = date_type.today() - if isinstance(on_date, date_type): - on_date = on_date.isoformat() - return Path(working_dir) / events_dir / on_date / name / f"{name}.md" - - -def _next_suffixed_stem(taken: Iterable[str], base: str) -> str: - """Lowest unused `-N` (N≥2). Returns `base` itself if not taken. - - Why advisory: when a same-name-on-same-day collision is detected, the - suggested suffix nudges the agent to pick a domain-specific qualifier - (e.g. `Apple-Inc` vs `Apple-Fruit`) over a numeric one. - """ - taken_set = set(taken) - if base not in taken_set: - return base - pattern = re.compile(rf"^{re.escape(base)}-(\d+)$") - used: set[int] = set() - for s in taken_set: - m = pattern.match(s) - if m: - try: - used.add(int(m.group(1))) - except ValueError: - continue - n = 2 - while n in used: - n += 1 - return f"{base}-{n}" - - -@R.register("sync") -class Sync(BaseStep): - """Upsert into an event folder under events/{date}/{name}/. - - Inputs (RuntimeContext): - name (str, required): kebab-case event identifier; becomes both - the parent dir and the index filename stem. - Reuse the same name across calls in one - thread to keep extending the same folder. - description (str): one-line summary for index frontmatter - (only set on initial create). - content (str): markdown body. On create it's the initial - body; on append it's added under a new - `## Update — {iso}` section. - topics (list[str]): related topic wikilinks; merged (union) - into frontmatter on append. - tags (list[str]): free-form tags; merged (union) on append. - materials (list[dict]): [{filename, content}, ...] — raw artifacts - written as siblings of the index. Filename - collisions auto-suffix (foo.txt → foo-2.txt). - Index body's Materials footer regenerated - each call from actual folder contents. - on_date (str | None): ISO date for the events/{date}/ bucket. - Defaults to today. - origin_session_id (str): optional source session identifier (only - set on initial create). - - Output (context.response.answer): - JSON {path, materials: [paths of NEW materials this call], - created: bool, action: "created"|"appended"} on success; - {error, ...} on failure (including refusal when existing event - has status != "active"). - """ - - def __init__(self, events_dir: str = "events", **kwargs): - super().__init__(**kwargs) - self.events_dir = events_dir - - def _root(self) -> Path: - vr = getattr(self.file_store, "working_dir", None) - if vr is not None: - return Path(vr) - raise RuntimeError("sync requires file_store.working_dir to be configured") - - @staticmethod - def _validate_materials(materials: list, index_filename: str) -> tuple[list[dict], str | None]: - """Sanity-check the materials list. Returns (cleaned, error_message).""" - cleaned: list[dict] = [] - seen_in_call: set[str] = set() - for i, m in enumerate(materials): - if not isinstance(m, dict): - return [], f"materials[{i}] must be an object {{filename, content}}" - fname = m.get("filename") - if not isinstance(fname, str) or not fname: - return [], f"materials[{i}].filename is required" - if not _SAFE_FILENAME_RE.match(fname): - return [], ( - f"materials[{i}].filename {fname!r} is unsafe — only " - f"letters / digits / dot / underscore / dash allowed" - ) - if fname == index_filename: - return [], f"materials[{i}].filename {fname!r} collides with the index file" - if fname in seen_in_call: - return [], f"materials[{i}].filename {fname!r} duplicated within the same call" - seen_in_call.add(fname) - content = m.get("content", "") - if not isinstance(content, str): - return [], f"materials[{i}].content must be a string" - cleaned.append({"filename": fname, "content": content}) - return cleaned, None - - @staticmethod - def _strip_materials_footer(body: str) -> str: - """Drop our trailing `## Materials` footer if present; return narrative.""" - if not body: - return "" - # Match the footer at end-of-doc: `## Materials\n\n- [...](./...)\n` repeated. - # Cheaper rule: find the LAST `## Materials` heading at line start; strip - # from there to EOF. Whatever the user wrote above stays intact. - m = re.search(r"(?:\A|\n)##\s+Materials[ \t]*\n", body) - if m is None: - return body.rstrip() - # Find the LAST such heading by scanning all matches. - last = None - for hit in re.finditer(r"(?:\A|\n)##\s+Materials[ \t]*\n", body): - last = hit - assert last is not None - cut = last.start() - # If the heading was at offset 0 (no leading \n), keep nothing before; - # otherwise keep up to (but not including) the leading \n. - return body[:cut].rstrip() - - @staticmethod - def _emit_body(narrative: str, material_filenames: list[str]) -> str: - """Assemble body = narrative (possibly empty) + Materials footer.""" - narrative = (narrative or "").rstrip() - if not material_filenames: - return f"{narrative}\n" if narrative else "" - listing = "\n".join(f"- [{f}](./{f})" for f in material_filenames) - if narrative: - return f"{narrative}\n\n{_MATERIALS_HEADER}\n\n{listing}\n" - return f"{_MATERIALS_HEADER}\n\n{listing}\n" - - @staticmethod - def _resolve_filename(existing: set[str], requested: str) -> str: - """Auto-suffix `foo.txt` → `foo-2.txt` (then -3, -4, …) on collision.""" - if requested not in existing: - return requested - if "." in requested: - stem, _, ext = requested.rpartition(".") - n = 2 - while f"{stem}-{n}.{ext}" in existing: - n += 1 - return f"{stem}-{n}.{ext}" - n = 2 - while f"{requested}-{n}" in existing: - n += 1 - return f"{requested}-{n}" - - @staticmethod - def _list_existing_materials(folder: Path, index_filename: str) -> list[str]: - """Filenames in `folder` excluding the index, sorted for stability.""" - if not folder.is_dir(): - return [] - return sorted(entry.name for entry in folder.iterdir() if entry.is_file() and entry.name != index_filename) - - @staticmethod - def _union(prior: list, incoming: list) -> list: - """Order-preserving union: keep prior order, append new items in input order.""" - out = list(prior) - seen = set(prior) - for item in incoming: - if item not in seen: - out.append(item) - seen.add(item) - return out - - def _set_error(self, payload: dict) -> None: - assert self.context is not None - self.context.response.success = False - self.context.response.answer = json.dumps(payload, ensure_ascii=False) - - def _set_ok(self, payload: dict) -> None: - assert self.context is not None - self.context.response.success = True - self.context.response.answer = json.dumps(payload, ensure_ascii=False) - - async def execute(self): - assert self.context is not None - name: str = self.context.get("name", "") or "" - description: str = self.context.get("description", "") or "" - content: str = self.context.get("content", "") or "" - topics: list[str] = list(self.context.get("topics") or []) - tags: list[str] = list(self.context.get("tags") or []) - materials_in = list(self.context.get("materials") or []) - on_date = self.context.get("on_date") - origin_session_id = self.context.get("origin_session_id") - - assert name, "name is required" - - target = _event_path(self._root(), name, on_date, self.events_dir) - materials, mat_err = self._validate_materials(materials_in, target.name) - if mat_err is not None: - self._set_error({"error": mat_err}) - return - - if target.exists(): - await self._append(target, content, materials, topics, tags) - else: - await self._create( - target, - name, - description, - content, - materials, - topics, - tags, - on_date, - origin_session_id, - ) - - async def _create( - self, - target: Path, - name: str, - description: str, - content: str, - materials: list[dict], - topics: list[str], - tags: list[str], - on_date, - origin_session_id, - ) -> None: - today = date_type.today().isoformat() - on_date_str = on_date.isoformat() if isinstance(on_date, date_type) else (on_date or today) - # Start from EVENT_PRESET (4 axes + status + legacy `category`), - # layer caller-supplied identity fields on top. - metadata: dict = { - **EVENT_PRESET, - "title": name, - "description": description, - "tags": tags, - "topics": topics, - "created": on_date_str, - "updated": today, - } - if origin_session_id: - metadata["originSessionId"] = origin_session_id - - try: - MemoryFileNode.model_validate( - { - "path": str(target.resolve()), - "st_mtime": 0.0, - **metadata, - } - ) - except ValidationError as e: - self._set_error( - { - "error": "MemoryFileNode schema validation failed", - "details": e.errors(include_context=False, include_url=False), - } - ) - return - - graph = self.file_store - conflicts = collisions_after_create(graph, target) - if conflicts: - taken = {Path(p).stem for p in graph.nodes} - suggested_name = _next_suffixed_stem(taken, name) - self._set_error( - { - "error": ( - f"stem `[[{name}]]` would resolve ambiguously " - f"to {len(conflicts) + 1} paths after this create" - ), - "conflicts": conflicts, - "suggested_name": suggested_name, - "hint": ( - f"retry with name='{suggested_name}', or pick a " - f"semantic qualifier (e.g. '{name}-followup')." - ), - } - ) - return - - material_filenames = [m["filename"] for m in materials] - index_body = self._emit_body(content, material_filenames) - ok, payload = create_file( - self.file_store, - target, - metadata=metadata, - content=index_body, - ) - if not ok: - self._set_error( - { - "path": str(target.resolve()), - "error": payload.get("error", "create failed"), - "details": payload, - } - ) - return - - material_paths: list[str] = [] - for m in materials: - material_path = target.parent / m["filename"] - try: - material_path.write_text(m["content"], encoding="utf-8") - except Exception as e: - self.logger.warning( - f"sync: failed to write material {m['filename']}: {e}", - ) - continue - material_paths.append(str(material_path.resolve())) - - self._set_ok( - { - "path": str(target.resolve()), - "category": "event", - "status": "active", - "topics": topics, - "materials": material_paths, - "created": True, - "action": "created", - } - ) - - async def _append( - self, - target: Path, - content: str, - materials: list[dict], - topics: list[str], - tags: list[str], - ) -> None: - # Read current frontmatter + body. - try: - raw = target.read_text(encoding="utf-8") - except Exception as e: - self._set_error({"path": str(target.resolve()), "error": f"read failed: {e}"}) - return - post = frontmatter.loads(raw) - meta = dict(post.metadata) - - status = meta.get("status") - if status != "active": - # Don't extend a distilled / archived thread — make the agent pick - # a new name so the prior cognition isn't silently mutated. - graph = self.file_store - taken = {Path(p).stem for p in graph.nodes} - base = target.stem - suggested = _next_suffixed_stem(taken, base) - self._set_error( - { - "path": str(target.resolve()), - "error": ( - f"event `{base}` already exists with status={status!r}; " - f"pick a new name to start a fresh thread" - ), - "status": status, - "suggested_name": suggested, - } - ) - return - - folder = target.parent - existing_filenames = self._list_existing_materials(folder, target.name) - existing_set = set(existing_filenames) - - # Resolve filename collisions for new materials. - resolved: list[tuple[str, str]] = [] # (filename_on_disk, content) - all_taken = set(existing_set) - for m in materials: - fname = self._resolve_filename(all_taken, m["filename"]) - all_taken.add(fname) - resolved.append((fname, m["content"])) - - # Write new materials to disk. - new_material_paths: list[str] = [] - for fname, mcontent in resolved: - material_path = folder / fname - try: - material_path.write_text(mcontent, encoding="utf-8") - except Exception as e: - self.logger.warning(f"sync: failed to write material {fname}: {e}") - continue - new_material_paths.append(str(material_path.resolve())) - existing_filenames.append(fname) - - # Rebuild the index body: narrative (existing + optional new Update - # section) + Materials footer (regenerated from disk). - narrative = self._strip_materials_footer(post.content) - if content.strip(): - ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") - update_section = f"## Update — {ts}\n\n{content.rstrip()}\n" - narrative = f"{narrative.rstrip()}\n\n{update_section}" if narrative else update_section - - all_filenames_sorted = sorted(set(existing_filenames)) - new_body = self._emit_body(narrative, all_filenames_sorted) - - # Update frontmatter: union topics/tags, bump updated. - meta["topics"] = self._union(list(meta.get("topics") or []), topics) - meta["tags"] = self._union(list(meta.get("tags") or []), tags) - meta["updated"] = date_type.today().isoformat() - - try: - MemoryFileNode.model_validate( - { - "path": str(target.resolve()), - "st_mtime": 0.0, - **meta, - } - ) - except ValidationError as e: - self._set_error( - { - "path": str(target.resolve()), - "error": "MemoryFileNode schema validation failed on append", - "details": e.errors(include_context=False, include_url=False), - } - ) - return - - new_post = frontmatter.Post(new_body, **meta) - try: - target.write_text(frontmatter.dumps(new_post), encoding="utf-8") - except Exception as e: - self._set_error({"path": str(target.resolve()), "error": f"write failed: {e}"}) - return - - self._set_ok( - { - "path": str(target.resolve()), - "category": "event", - "status": "active", - "topics": meta["topics"], - "materials": new_material_paths, - "created": False, - "action": "appended", - } - )