From b3031c6df1983bc43153a632b1cd8e70a547e453 Mon Sep 17 00:00:00 2001 From: "jinli.yl" Date: Wed, 27 May 2026 01:10:58 +0800 Subject: [PATCH] up --- docs4/reme4_report.md | 4 +- reme4/application.py | 31 +++- reme4/components/base_component.py | 5 +- reme4/components/component_registry.py | 4 +- .../embedding/base_embedding_model.py | 3 + .../file_catalog/base_file_catalog.py | 2 + .../file_catalog/local_file_catalog.py | 2 + .../components/file_graph/local_file_graph.py | 5 +- .../keyword_index/base_keyword_index.py | 15 +- reme4/components/keyword_index/bm25_index.py | 25 +-- reme4/components/prompt_handler.py | 4 +- reme4/components/tokenizer/base_tokenizer.py | 1 + reme4/components/tokenizer/jieba_tokenizer.py | 2 +- reme4/config/demo.yaml | 23 --- reme4/config/qwenpaw.yaml | 129 ++++--------- reme4/steps/__init__.py | 33 ++-- reme4/steps/background/__init__.py | 2 - reme4/steps/background/watch_changes.py | 4 +- reme4/steps/common/__init__.py | 6 +- reme4/steps/common/health_check.py | 129 +++++++------ reme4/steps/common/help.py | 41 +++-- reme4/steps/common/traverse.py | 163 +++++++---------- reme4/steps/crud/__init__.py | 13 +- reme4/steps/index/__init__.py | 13 ++ reme4/steps/{common => index}/reindex.py | 19 +- reme4/steps/{common => index}/search.py | 0 reme4/steps/index/update_catalog.py | 94 ++++++++++ .../update_index.py} | 6 +- reme4/steps/transfer/__init__.py | 20 +++ reme4/steps/{crud => transfer}/download.py | 0 reme4/steps/{crud => transfer}/upload.py | 0 .../{crud => transfer}/upload_resource.py | 0 tests4/unittest/test_crud_steps.py | 2 +- tests4/unittest/test_file_catalog.py | 1 + tests4/unittest/test_keyword_index.py | 170 ++++++++++-------- tests4/unittest/test_resource_steps.py | 4 +- 36 files changed, 523 insertions(+), 452 deletions(-) create mode 100644 reme4/steps/index/__init__.py rename reme4/steps/{common => index}/reindex.py (59%) rename reme4/steps/{common => index}/search.py (100%) create mode 100644 reme4/steps/index/update_catalog.py rename reme4/steps/{background/update_store_index.py => index/update_index.py} (96%) create mode 100644 reme4/steps/transfer/__init__.py rename reme4/steps/{crud => transfer}/download.py (100%) rename reme4/steps/{crud => transfer}/upload.py (100%) rename reme4/steps/{crud => transfer}/upload_resource.py (100%) diff --git a/docs4/reme4_report.md b/docs4/reme4_report.md index b5bdb29e..9b66aa28 100644 --- a/docs4/reme4_report.md +++ b/docs4/reme4_report.md @@ -119,13 +119,13 @@ skills 建议使用index 看 whole picture~ ├── daily/ │ ├── 20260518.md # 当天主索引(兼容 write/edit) │ └── 20260518/ -│ ├── meeting-with-alice.md +│ ├── meeting-with-alice.md │ ├── debug-login.md │ └── reading-paper.md └── digest/ # 固定四个子目录:personal / knowledge / procedural / proactive ├── personal/ # 个性化记忆:偏好、习惯、个人事件 .change_log.md difference - _moc.md + _moc.md │ └── 用户偏好.md ├── knowledge/ # 知识类记忆:用户自定义二级目录(work / financial / ...) │ ├── work/ diff --git a/reme4/application.py b/reme4/application.py index 08125351..fb610868 100644 --- a/reme4/application.py +++ b/reme4/application.py @@ -54,7 +54,10 @@ class Application(BaseComponent): def _init_service(self) -> None: """Instantiate the single service backend declared in config.service.""" self.context.service = self._instantiate( - ComponentEnum.SERVICE, self.config.service, label="Service", expected_type=BaseService, + ComponentEnum.SERVICE, + self.config.service, + label="Service", + expected_type=BaseService, ) def _init_components(self) -> None: @@ -63,19 +66,32 @@ class Application(BaseComponent): self.context.components[ctype] = {} for name, cfg in group.items(): self.context.components[ctype][name] = self._instantiate( - ctype, cfg, label=f"Component '{name}'", expected_type=BaseComponent, name=name, + ctype, + cfg, + label=f"Component '{name}'", + expected_type=BaseComponent, + name=name, ) def _init_jobs(self) -> None: """Instantiate every job declared under config.jobs.""" for name, cfg in self.config.jobs.items(): self.context.jobs[name] = self._instantiate( - ComponentEnum.JOB, cfg, label=f"Job '{name}'", expected_type=BaseJob, name=name, + ComponentEnum.JOB, + cfg, + label=f"Job '{name}'", + expected_type=BaseJob, + name=name, ) def _instantiate( - self, ctype: ComponentEnum, cfg: ComponentConfig, *, - label: str, expected_type: type[T], name: str | None = None, + self, + ctype: ComponentEnum, + cfg: ComponentConfig, + *, + label: str, + expected_type: type[T], + name: str | None = None, ) -> T: """Resolve cfg.backend through the registry and construct the instance. @@ -194,7 +210,10 @@ class Application(BaseComponent): stream_queue: asyncio.Queue = asyncio.Queue() task = asyncio.create_task(self.context.jobs[name](stream_queue=stream_queue, **kwargs)) async for chunk in execute_stream_task( - stream_queue=stream_queue, task=task, task_name=name, output_format="chunk", + stream_queue=stream_queue, + task=task, + task_name=name, + output_format="chunk", ): assert isinstance(chunk, StreamChunk) yield chunk diff --git a/reme4/components/base_component.py b/reme4/components/base_component.py index 9d8bf50e..bc96c64b 100644 --- a/reme4/components/base_component.py +++ b/reme4/components/base_component.py @@ -42,8 +42,7 @@ class Dependency: def __getattr__(self, item: str) -> Any: # Catches accidental use of the placeholder before start() resolves it. raise RuntimeError( - f"Dependency {self.ctype.value}:{self.name} accessed before start() " - f"(attribute '{item}')", + f"Dependency {self.ctype.value}:{self.name} accessed before start() " f"(attribute '{item}')", ) @@ -99,7 +98,7 @@ class BaseComponent(ABC): ctype = getattr(base_cls, "component_type", None) if not isinstance(ctype, ComponentEnum) or ctype is ComponentEnum.BASE: raise TypeError( - f"{base_cls.__name__} must declare a non-BASE ComponentEnum 'component_type'" + f"{base_cls.__name__} must declare a non-BASE ComponentEnum 'component_type'", ) return cast(T, Dependency(ctype, name, default_factory, optional)) diff --git a/reme4/components/component_registry.py b/reme4/components/component_registry.py index 3e59b88a..8f141d43 100644 --- a/reme4/components/component_registry.py +++ b/reme4/components/component_registry.py @@ -25,7 +25,7 @@ class ComponentRegistry: component_type = getattr(cls, "component_type", None) if not isinstance(component_type, ComponentEnum): raise TypeError( - f"{cls.__name__} must have a ComponentEnum 'component_type' attribute" + f"{cls.__name__} must have a ComponentEnum 'component_type' attribute", ) if not name: raise ValueError("Component name cannot be empty") @@ -33,7 +33,7 @@ class ComponentRegistry: group = self._registry.setdefault(component_type, {}) if name in group: self.logger.warning( - f"Component '{name}' already registered for {component_type}, overwriting" + f"Component '{name}' already registered for {component_type}, overwriting", ) group[name] = cls return cls diff --git a/reme4/components/embedding/base_embedding_model.py b/reme4/components/embedding/base_embedding_model.py index 3f4356cc..a149cfcf 100644 --- a/reme4/components/embedding/base_embedding_model.py +++ b/reme4/components/embedding/base_embedding_model.py @@ -56,6 +56,7 @@ class BaseEmbeddingModel(BaseComponent): @property def cache_path(self) -> Path: + """Path of the persisted embedding cache, namespaced by name and version.""" return self.vault_metadata_path / "embedding_cache" / f"{self.name}_{self.cache_version}.npz" async def _start(self) -> None: @@ -84,6 +85,7 @@ class BaseEmbeddingModel(BaseComponent): # -- Public API -- async def get_embedding(self, input_text: str, **kwargs) -> np.ndarray | None: + """Embed a single text; returns None if the provider yields nothing.""" results = await self.get_embeddings([input_text], **kwargs) return results[0] if results else None @@ -96,6 +98,7 @@ class BaseEmbeddingModel(BaseComponent): return results async def get_node_embeddings(self, nodes: list[EmbNode], **kwargs) -> list[EmbNode]: + """Embed each node's text in-place and return the same list.""" embeddings = await self.get_embeddings([n.text for n in nodes], **kwargs) if len(embeddings) == len(nodes): for node, vec in zip(nodes, embeddings): diff --git a/reme4/components/file_catalog/base_file_catalog.py b/reme4/components/file_catalog/base_file_catalog.py index 661af8d9..4b6612f5 100644 --- a/reme4/components/file_catalog/base_file_catalog.py +++ b/reme4/components/file_catalog/base_file_catalog.py @@ -1,3 +1,5 @@ +"""Abstract base class for file catalog backends.""" + from abc import abstractmethod from ..base_component import BaseComponent diff --git a/reme4/components/file_catalog/local_file_catalog.py b/reme4/components/file_catalog/local_file_catalog.py index 9f35364d..150be1bb 100644 --- a/reme4/components/file_catalog/local_file_catalog.py +++ b/reme4/components/file_catalog/local_file_catalog.py @@ -1,3 +1,5 @@ +"""Local file catalog backend: in-memory dict persisted as JSONL.""" + import aiofiles from .base_file_catalog import BaseFileCatalog diff --git a/reme4/components/file_graph/local_file_graph.py b/reme4/components/file_graph/local_file_graph.py index 6ec573d1..1a67c8a6 100644 --- a/reme4/components/file_graph/local_file_graph.py +++ b/reme4/components/file_graph/local_file_graph.py @@ -133,8 +133,5 @@ class LocalFileGraph(BaseFileGraph): if scope in (LinkScopeEnum.VIRTUAL, LinkScopeEnum.ALL): sources |= self._pending.get(path, set()) return [ - link - for src in sources if src in self._nodes - for link in self._nodes[src].links - if link.target_path == path + link for src in sources if src in self._nodes for link in self._nodes[src].links if link.target_path == path ] diff --git a/reme4/components/keyword_index/base_keyword_index.py b/reme4/components/keyword_index/base_keyword_index.py index d0012f10..466f4c66 100644 --- a/reme4/components/keyword_index/base_keyword_index.py +++ b/reme4/components/keyword_index/base_keyword_index.py @@ -1,3 +1,5 @@ +"""Abstract base class for keyword indexes (BM25 and other lexical backends).""" + from abc import abstractmethod from ..base_component import BaseComponent @@ -30,16 +32,20 @@ class BaseKeywordIndex(BaseComponent): return self.tokenizer.tokenize([text])[0] @abstractmethod - async def add_docs(self, docs_dict: dict[str, str]) -> None: ... + async def add_docs(self, docs_dict: dict[str, str]) -> None: + """Add or replace documents keyed by id.""" @abstractmethod - async def delete_docs(self, doc_ids: list[str]) -> None: ... + async def delete_docs(self, doc_ids: list[str]) -> None: + """Delete documents by id; missing ids are skipped.""" @abstractmethod - async def retrieve(self, query: str, limit: int = 3) -> dict[str, float]: ... + async def retrieve(self, query: str, limit: int = 3) -> dict[str, float]: + """Return top-`limit` doc_id → score for the given query.""" @abstractmethod - async def clear(self) -> None: ... + async def clear(self) -> None: + """Wipe in-memory state and remove any persisted artifacts.""" async def reset_index(self, docs_dict: dict[str, str]) -> None: """Wipe the index, rebuild it from `docs_dict`, and persist the result.""" @@ -49,4 +55,3 @@ class BaseKeywordIndex(BaseComponent): async def optimize_index(self) -> None: """Compact or rebuild the index. No-op by default; override as needed.""" - pass diff --git a/reme4/components/keyword_index/bm25_index.py b/reme4/components/keyword_index/bm25_index.py index c2f402ef..81c5de7a 100644 --- a/reme4/components/keyword_index/bm25_index.py +++ b/reme4/components/keyword_index/bm25_index.py @@ -28,6 +28,7 @@ from ..component_registry import R @R.register("bm25") class BM25Index(BaseKeywordIndex): + """BM25 inverted index with lazy deletion and on-disk persistence.""" def __init__(self, k1: float = 1.5, b: float = 0.75, index_version: str = "v1", **kwargs): super().__init__(**kwargs) @@ -92,11 +93,7 @@ class BM25Index(BaseKeywordIndex): out: dict[int, dict[str, int]] = {} for tid, doc_idxs in self._posting_doc_idxs.items(): tfs = self._posting_tfs[tid] - posting = { - self._doc_ids[int(i)]: int(tf) - for i, tf in zip(doc_idxs, tfs) - if not self._deleted[int(i)] - } + posting = {self._doc_ids[int(i)]: int(tf) for i, tf in zip(doc_idxs, tfs) if not self._deleted[int(i)]} if posting: out[tid] = posting return out @@ -153,7 +150,10 @@ class BM25Index(BaseKeywordIndex): return unique_tids, len(token_ids), counts def _append_doc_arrays( - self, new_doc_ids: list[str], new_doc_lens: list[int], new_doc_token_ids: list[np.ndarray] + self, + new_doc_ids: list[str], + new_doc_lens: list[int], + new_doc_token_ids: list[np.ndarray], ) -> None: """Append metadata for a batch of new docs to the doc-level arrays.""" if not new_doc_ids: @@ -352,10 +352,7 @@ class BM25Index(BaseKeywordIndex): def _compact_vocab(self, active_mask: np.ndarray) -> tuple[dict[str, int], dict[int, int]]: """Keep only tokens still referenced by a live doc; renumber contiguously.""" - used_tids = { - tid for tid, doc_idxs in self._posting_doc_idxs.items() - if active_mask[doc_idxs].any() - } + used_tids = {tid for tid, doc_idxs in self._posting_doc_idxs.items() if active_mask[doc_idxs].any()} new_vocab: dict[str, int] = {} old_to_new: dict[int, int] = {} for token, old_tid in self.vocab.items(): @@ -384,7 +381,9 @@ class BM25Index(BaseKeywordIndex): return new_idxs, new_tfs def _compact_docs( - self, active_mask: np.ndarray, old_tid_to_new: dict[int, int] + self, + active_mask: np.ndarray, + old_tid_to_new: dict[int, int], ) -> tuple[list[str], list[np.ndarray]]: """Rebuild doc_id list and unique-token arrays under the new vocab.""" active_old_idxs = np.where(active_mask)[0] @@ -410,7 +409,9 @@ class BM25Index(BaseKeywordIndex): old_to_new_idx, n_active = self._build_idx_remap(active_mask) new_vocab, old_tid_to_new = self._compact_vocab(active_mask) new_posting_idxs, new_posting_tfs = self._compact_postings( - active_mask, old_to_new_idx, old_tid_to_new + active_mask, + old_to_new_idx, + old_tid_to_new, ) new_doc_ids, new_doc_token_ids = self._compact_docs(active_mask, old_tid_to_new) diff --git a/reme4/components/prompt_handler.py b/reme4/components/prompt_handler.py index db97a963..bbc4a046 100644 --- a/reme4/components/prompt_handler.py +++ b/reme4/components/prompt_handler.py @@ -92,7 +92,7 @@ class PromptHandler: if key in self.data: return self.data[key].strip() raise KeyError( - f"Prompt '{prompt_name}' not found. Available: {list(self.data.keys())[:10]}" + f"Prompt '{prompt_name}' not found. Available: {list(self.data.keys())[:10]}", ) def has_prompt(self, prompt_name: str) -> bool: @@ -142,7 +142,7 @@ class PromptHandler: required = {f for _, f, _, _ in Formatter().parse(prompt) if f is not None} if missing := required - set(formats.keys()): raise ValueError( - f"Missing format variables for '{prompt_name}': {sorted(missing)}" + f"Missing format variables for '{prompt_name}': {sorted(missing)}", ) def __repr__(self) -> str: diff --git a/reme4/components/tokenizer/base_tokenizer.py b/reme4/components/tokenizer/base_tokenizer.py index 55a8c500..8795f4a4 100644 --- a/reme4/components/tokenizer/base_tokenizer.py +++ b/reme4/components/tokenizer/base_tokenizer.py @@ -45,6 +45,7 @@ class BaseTokenizer(BaseComponent): @property def stopwords(self) -> set[str]: + """Loaded stopwords (empty set if none were loaded).""" return self._stopwords def tokenize(self, texts: list[str], lower: bool = True, **kwargs) -> list[list[str]]: diff --git a/reme4/components/tokenizer/jieba_tokenizer.py b/reme4/components/tokenizer/jieba_tokenizer.py index e63c7a71..5e35fd23 100644 --- a/reme4/components/tokenizer/jieba_tokenizer.py +++ b/reme4/components/tokenizer/jieba_tokenizer.py @@ -21,7 +21,7 @@ class JiebaTokenizer(BaseTokenizer): super().__init__(**kwargs) if backend not in self.SUPPORTED_BACKENDS: raise ValueError( - f"Unknown jieba backend {backend!r}; expected one of {self.SUPPORTED_BACKENDS}" + f"Unknown jieba backend {backend!r}; expected one of {self.SUPPORTED_BACKENDS}", ) self.backend = backend self._cut: Callable[[str], list[str]] | None = None diff --git a/reme4/config/demo.yaml b/reme4/config/demo.yaml index 51e6bd5d..f765a194 100644 --- a/reme4/config/demo.yaml +++ b/reme4/config/demo.yaml @@ -2,24 +2,6 @@ service: backend: http jobs: - version: - backend: base - description: "return reme4 package version" - parameters: - type: object - properties: { } - steps: - - backend: version_step - - help: - backend: base - description: "list all registered jobs with their metadata" - parameters: - type: object - properties: { } - steps: - - backend: help_step - demo: backend: base description: "demo job description" @@ -61,8 +43,3 @@ jobs: steps: - backend: stream_demo_step1 - backend: stream_demo_step2 - -components: - tokenizer: - default: - backend: regex diff --git a/reme4/config/qwenpaw.yaml b/reme4/config/qwenpaw.yaml index 95bb8369..c8f7c445 100644 --- a/reme4/config/qwenpaw.yaml +++ b/reme4/config/qwenpaw.yaml @@ -1,22 +1,14 @@ service: backend: http -# backend: mcp -# Default dev config points vault_dir at ./.reme so `python -m reme4 start` -# can be run from the repo root and exercise the full atomic-tool surface -# against the seeded test data. Override the `vault_dir=` CLI arg. -vault_dir: .reme daily_dir: daily digest_dir: digest -resource_dir: resource +resource_dir: "" jobs: - # ════════════════════════════════════════════════════════════════════ - # UTILITY — service introspection - # ════════════════════════════════════════════════════════════════════ version: backend: base - description: "return reme4 package version" + description: "return reme package version" parameters: type: object properties: {} @@ -25,7 +17,7 @@ jobs: health_check: backend: base - description: "return a concise health-check snapshot of reme4 components" + description: "return a concise health-check snapshot of reme components" parameters: type: object properties: {} @@ -41,6 +33,33 @@ jobs: steps: - backend: help_step + traverse: + backend: base + description: "Walk the wikilink graph from a path." + parameters: + type: object + properties: + path: + type: string + description: "path" + depth: + type: integer + description: "hop limit" + default: 1 + direction: + type: string + enum: + - forward + - backward + - both + default: both + required: + - path + steps: + - backend: traverse_step + + + reindex: backend: base description: "wipe the file store and rebuild it from the watcher's tracked files" @@ -108,27 +127,7 @@ jobs: expand_links: true max_links_per_direction: 10 - traverse: - backend: base - description: "Walk the wikilink graph from a seed path." - parameters: - type: object - properties: - path: - type: string - description: "seed path (vault-relative)" - depth: - type: integer - description: "hop limit" - default: 1 - direction: - type: string - description: "forward / backward / both" - default: both - required: - - path - steps: - - backend: traverse_step + # ── Read Operations ─────────────────────────────────────────────────────────── list: @@ -307,33 +306,6 @@ jobs: steps: - backend: frontmatter_delete_step - # ── File Operations (relocate / cross-realm) ────────────────────────────── - move: - backend: base - description: "Move / rename a vault file; rewrites inbound wikilinks by default." - parameters: - type: object - properties: - src_path: - type: string - description: "vault-relative source" - dst_path: - type: string - description: "vault-relative destination" - overwrite: - type: boolean - description: "overwrite if dst exists" - default: false - retarget: - type: boolean - description: "rewrite [[src]] → [[dst]] across the vault" - default: true - required: - - src_path - - dst_path - steps: - - backend: move_step - delete: backend: base description: "Delete a vault file or folder; returns surviving inbound wikilinks." @@ -348,28 +320,6 @@ jobs: steps: - backend: delete_step - upload: - backend: base - description: "Copy a host file into the vault at an explicit destination." - parameters: - type: object - properties: - src_path: - type: string - description: "host absolute path" - dst_path: - type: string - description: "vault-relative destination" - overwrite: - type: boolean - description: "overwrite if dst exists" - default: false - required: - - src_path - - dst_path - steps: - - backend: upload_step - upload_resource: backend: base description: "Ingest an external-channel asset into resource// with provenance." @@ -516,7 +466,7 @@ components: embedding_model: default: backend: ${EMBEDDING_BACKEND:-openai} - api_key: ${EMBEDDING_API_KEY:-} + api_key: ${EMBEDDING_API_KEY} base_url: ${EMBEDDING_BASE_URL:-https://api.openai.com/v1} model_name: ${EMBEDDING_MODEL_NAME:-text-embedding-v4} dimensions: 1024 @@ -553,18 +503,3 @@ components: embedding_model: default keyword_index: default file_graph: default - -# as_llm / formatter aren't required for atomic primitives; configure -# only if you'll invoke digester/synchronizer or other LLM-driven -# paths from the dev server. -# as_llm: -# default: -# backend: ${LLM_BACKEND:-openai} -# api_key: ${LLM_API_KEY:-} -# model_name: ${LLM_MODEL_NAME:-gpt-4o-mini} -# client_kwargs: -# base_url: ${LLM_BASE_URL:-https://api.openai.com/v1} -# -# as_llm_formatter: -# default: -# backend: ${LLM_BACKEND:-openai} diff --git a/reme4/steps/__init__.py b/reme4/steps/__init__.py index b039ba08..614fa2b3 100644 --- a/reme4/steps/__init__.py +++ b/reme4/steps/__init__.py @@ -6,30 +6,35 @@ step backend (e.g. ``graph_traverse_step``, ``write``, ``digester``) will find it in the registry without the caller having to remember which submodule it lives in. -File-I/O is split by blast radius. The ``crud`` package covers both -opaque-byte ops (list / stat / move / delete / upload / download) and -whole-file text ops (read / write / append / edit) — they share the -same path-resolution helpers, so they live in one package. -``frontmatter`` is the one sliced surface that earns its own RUD -package (YAML is structured data — surgical key edits cannot be safely -emulated with string-substitution on the body). For mid-file body -edits, use ``edit`` (exact string replacement) or do a read + write -round-trip. +File-I/O is split by blast radius. The ``crud`` package covers +single-resource ops within the vault — both opaque-byte ops (list / +stat / move / delete) and whole-file text ops (read / write / append +/ edit), which share the same path-resolution helpers. The +``transfer`` package handles cross-domain bridges (vault ↔ local fs: +upload / download / upload_resource). ``frontmatter`` is the one +sliced surface that earns its own RUD package (YAML is structured +data — surgical key edits cannot be safely emulated with +string-substitution on the body). For mid-file body edits, use +``edit`` (exact string replacement) or do a read + write round-trip. -* ``common`` — search / health_check / help / reindex / version / graph_traverse -* ``crud`` — list / stat / move / delete / upload / download / read / write / append / edit +* ``common`` — health_check / help / version / traverse +* ``crud`` — list / stat / move / delete / read / write / append / edit +* ``transfer`` — upload / download / upload_resource (vault ↔ local fs) +* ``index`` — search / reindex / update_catalog / update_index * ``frontmatter`` — markdown frontmatter slice RUD (frontmatter_read_step / update / delete) * ``daily`` — note genesis / list / day-index reindex * ``jobs`` — synchronizer / digester (LLM-driven orchestrators) """ -from . import common # noqa: F401 -- registers common steps (search, version, graph_traverse, ...) -from . import crud # noqa: F401 -- registers list/stat/upload/download/move/delete/read/write/append/edit +from . import common # noqa: F401 -- registers common steps (health_check, help, version, traverse, ...) +from . import crud # noqa: F401 -- registers list/stat/move/delete/read/write/append/edit +from . import transfer # noqa: F401 -- registers upload/download/upload_resource from . import frontmatter # noqa: F401 -- registers frontmatter_read_step/update/delete from . import ( daily, ) # noqa: F401 -- registers daily_read_step / daily_write_step / daily_list_step / daily_reindex_step from . import background # noqa: F401 +from . import index # noqa: F401 -- registers update_catalog_step / update_index_step # from . import jobs # noqa: F401 -- registers synchronizer / digester from .base_step import BaseStep @@ -39,7 +44,9 @@ __all__ = [ "background", "common", "crud", + "transfer", "graph", + "index", "frontmatter", "daily", "BaseStep", diff --git a/reme4/steps/background/__init__.py b/reme4/steps/background/__init__.py index 82dfb3e5..9ecf0853 100644 --- a/reme4/steps/background/__init__.py +++ b/reme4/steps/background/__init__.py @@ -1,11 +1,9 @@ """Background steps.""" from .scan_changes import ScanChangesStep -from .update_store_index import UpdateStoreIndexStep from .watch_changes import WatchChangesStep __all__ = [ "ScanChangesStep", - "UpdateStoreIndexStep", "WatchChangesStep", ] diff --git a/reme4/steps/background/watch_changes.py b/reme4/steps/background/watch_changes.py index 26fd3e30..4fe4874f 100644 --- a/reme4/steps/background/watch_changes.py +++ b/reme4/steps/background/watch_changes.py @@ -1,4 +1,4 @@ -"""Long-running awatch loop: convert raw changes into update_store_index calls.""" +"""Long-running awatch loop: convert raw changes into update_index calls.""" import asyncio @@ -10,7 +10,7 @@ from ...components import R @R.register("watch_changes_step") class WatchChangesStep(BaseStep): - """Watch files and forward each batch of raw changes to the update_store_index job.""" + """Watch files and forward each batch of raw changes to the update_index job.""" def __init__( self, diff --git a/reme4/steps/common/__init__.py b/reme4/steps/common/__init__.py index 73248fd5..ea265351 100644 --- a/reme4/steps/common/__init__.py +++ b/reme4/steps/common/__init__.py @@ -1,17 +1,13 @@ -"""Common steps.""" +"""Common steps: health, help, traverse, version, demo.""" from .health_check import HealthCheckStep from .help import HelpStep -from .reindex import ReindexStep -from .search import SearchStep from .traverse import TraverseStep from .version import VersionStep __all__ = [ "HealthCheckStep", "HelpStep", - "ReindexStep", - "SearchStep", "TraverseStep", "VersionStep", ] diff --git a/reme4/steps/common/health_check.py b/reme4/steps/common/health_check.py index 2fefd826..7bdf182d 100644 --- a/reme4/steps/common/health_check.py +++ b/reme4/steps/common/health_check.py @@ -1,76 +1,86 @@ -"""Return a concise health check snapshot of ReMe runtime components.""" +"""Concise health snapshot of ReMe runtime components.""" import sys from collections.abc import Mapping import numpy as np -from ...enumeration.component_enum import ComponentEnum - from ..base_step import BaseStep from ... import __version__ from ...components import R +from ...enumeration import ComponentEnum + + +# --------------------------------------------------------------------------- +# Memory accounting +# --------------------------------------------------------------------------- def _deep_size(obj, _seen: set | None = None) -> int: - """Recursive sizeof; uses ndarray.nbytes for numpy and walks containers / __dict__.""" + """Recursive sizeof. Uses ndarray.nbytes; walks Mappings, sequences, __dict__.""" if _seen is None: _seen = set() - obj_id = id(obj) - if obj_id in _seen: + if id(obj) in _seen: return 0 - _seen.add(obj_id) + _seen.add(id(obj)) if isinstance(obj, np.ndarray): return int(obj.nbytes) + sys.getsizeof(obj) size = sys.getsizeof(obj) if isinstance(obj, (str, bytes, bytearray, int, float, bool, type(None))): - return size - if isinstance(obj, Mapping): - size += sum(_deep_size(k, _seen) + _deep_size(v, _seen) for k, v in obj.items()) + extra = 0 + elif isinstance(obj, Mapping): + extra = sum(_deep_size(k, _seen) + _deep_size(v, _seen) for k, v in obj.items()) elif isinstance(obj, (list, tuple, set, frozenset)): - size += sum(_deep_size(item, _seen) for item in obj) + extra = sum(_deep_size(item, _seen) for item in obj) elif hasattr(obj, "__dict__"): - size += _deep_size(vars(obj), _seen) + extra = _deep_size(vars(obj), _seen) elif hasattr(obj, "__slots__"): - for slot in obj.__slots__: - if hasattr(obj, slot): - size += _deep_size(getattr(obj, slot), _seen) - return size + extra = sum(_deep_size(getattr(obj, s), _seen) for s in obj.__slots__ if hasattr(obj, s)) + else: + extra = 0 + return size + extra def _mb_str(*objs) -> str: - """Return summed deep size of objs formatted as 'X.XX MB'.""" + """Sum deep size of objs and format as 'X.XX MB'.""" seen: set = set() total = sum(_deep_size(o, seen) for o in objs) return f"{total / (1024 * 1024):.2f} MB" +# --------------------------------------------------------------------------- +# Per-component status collectors +# --------------------------------------------------------------------------- + + def _embedding_status(comp) -> dict: + cache = getattr(comp, "_embedding_cache", {}) or {} return { "is_started": comp.is_started, "is_healthy": getattr(comp, "is_healthy", None), "model_name": getattr(comp, "model_name", None), "dimensions": getattr(comp, "dimensions", None), - "cache_size": len(getattr(comp, "_embedding_cache", {}) or {}), - "memory": _mb_str(getattr(comp, "_embedding_cache", {}) or {}), + "cache_size": len(cache), + "memory": _mb_str(cache), } -def _file_graph_status(comp) -> dict: - # Nx backend: single _graph attribute holds nodes/edges, virtuals are nodes without "node" payload. - g = getattr(comp, "_graph", None) - if g is not None: - n_real = sum(1 for _, d in g.nodes(data=True) if "node" in d) - return { - "is_started": comp.is_started, - "n_nodes": n_real, - "n_edges": g.number_of_edges(), - "n_virtual": g.number_of_nodes() - n_real, - "memory": _mb_str(g), - } - # Local backend: separate dicts for nodes, resolved inverse edges, and pending edges. +def _file_graph_nx_status(comp, graph) -> dict: + """Networkx backend: virtuals are nodes without a 'node' payload.""" + n_real = sum(1 for _, d in graph.nodes(data=True) if "node" in d) + return { + "is_started": comp.is_started, + "n_nodes": n_real, + "n_edges": graph.number_of_edges(), + "n_virtual": graph.number_of_nodes() - n_real, + "memory": _mb_str(graph), + } + + +def _file_graph_local_status(comp) -> dict: + """Local backend: nodes / inverse edges / pending edges held as separate dicts.""" nodes = getattr(comp, "_nodes", {}) or {} inverse = getattr(comp, "_inverse", {}) or {} pending = getattr(comp, "_pending", {}) or {} @@ -83,6 +93,11 @@ def _file_graph_status(comp) -> dict: } +def _file_graph_status(comp) -> dict: + graph = getattr(comp, "_graph", None) + return _file_graph_nx_status(comp, graph) if graph is not None else _file_graph_local_status(comp) + + def _file_store_status(comp) -> dict: chunks = getattr(comp, "file_chunks", {}) or {} return { @@ -94,12 +109,13 @@ def _file_store_status(comp) -> dict: def _keyword_index_status(comp) -> dict: + vocab = getattr(comp, "vocab", {}) or {} return { "is_started": comp.is_started, "n_docs": getattr(comp, "n_docs", None), - "vocab_size": len(getattr(comp, "vocab", {}) or {}), + "vocab_size": len(vocab), "memory": _mb_str( - getattr(comp, "vocab", {}) or {}, + vocab, getattr(comp, "inverted_index", {}) or {}, getattr(comp, "doc_meta", {}) or {}, getattr(comp, "_idf_cache", {}) or {}, @@ -115,8 +131,13 @@ _HANDLERS = { } -def _is_status_healthy(ctype: ComponentEnum, status: dict) -> bool: - """Per-component health rule. Unstarted = unhealthy; type-specific extras checked.""" +# --------------------------------------------------------------------------- +# Health rules and step entry point +# --------------------------------------------------------------------------- + + +def _is_healthy(ctype: ComponentEnum, status: dict) -> bool: + """Unstarted = unhealthy; embedding model also requires is_healthy != False.""" if not status.get("is_started"): return False if ctype is ComponentEnum.EMBEDDING_MODEL and status.get("is_healthy") is False: @@ -124,30 +145,38 @@ def _is_status_healthy(ctype: ComponentEnum, status: dict) -> bool: return True +def _collect_components(app_context) -> tuple[dict, bool]: + """Walk every registered component type and produce {type: {name: status}}, plus overall flag.""" + components: dict = {} + healthy = True + for ctype, handler in _HANDLERS.items(): + bucket = {} + for name, comp in app_context.components.get(ctype, {}).items(): + status = handler(comp) + bucket[name] = status + if not _is_healthy(ctype, status): + healthy = False + components[ctype.value] = bucket + return components, healthy + + @R.register("health_check_step") class HealthCheckStep(BaseStep): - """Collect a concise health check snapshot of the relevant components.""" + """Collect a concise health snapshot of the relevant components.""" async def execute(self): assert self.context is not None - components: dict = {} - healthy = True if self.app_context is not None: - for ctype, handler in _HANDLERS.items(): - comp_map = self.app_context.components.get(ctype, {}) - bucket = {} - for name, comp in comp_map.items(): - s = handler(comp) - bucket[name] = s - if not _is_status_healthy(ctype, s): - healthy = False - components[ctype.value] = bucket + components, healthy = _collect_components(self.app_context) + else: + components, healthy = {}, True health = {"version": __version__, "healthy": healthy, "components": components} self.logger.info(f"[{self.name}] health collected: {health}") - status_emoji = "✅" if healthy else "❌" - self.context.response.answer = f"{status_emoji} ReMe v{__version__} - {'healthy' if healthy else 'unhealthy'}" + emoji = "✅" if healthy else "❌" + label = "healthy" if healthy else "unhealthy" + self.context.response.answer = f"{emoji} ReMe v{__version__} - {label}" self.context.response.metadata["health"] = health return self.context.response diff --git a/reme4/steps/common/help.py b/reme4/steps/common/help.py index 6c850772..3ccf94d1 100644 --- a/reme4/steps/common/help.py +++ b/reme4/steps/common/help.py @@ -4,26 +4,26 @@ from ..base_step import BaseStep from ...components import R +def _format_params(parameters: dict) -> str: + props = (parameters or {}).get("properties") or {} + if not props: + return "no args" + required = set((parameters or {}).get("required") or []) + parts = [] + for name, schema in props.items(): + ptype = schema.get("type", "any") + if name in required: + parts.append(f"{name}:{ptype}*") + elif "default" in schema: + parts.append(f"{name}:{ptype}={schema['default']}") + else: + parts.append(f"{name}:{ptype}") + return ", ".join(parts) + + @R.register("help_step") class HelpStep(BaseStep): - """List all registered jobs (excluding self) as compact one-liners for an LLM.""" - - @staticmethod - def _format_params(parameters: dict) -> str: - props = (parameters or {}).get("properties") or {} - if not props: - return "no args" - required = set((parameters or {}).get("required") or []) - parts = [] - for pname, pschema in props.items(): - ptype = pschema.get("type", "any") - if pname in required: - parts.append(f"{pname}:{ptype}*") - elif "default" in pschema: - parts.append(f"{pname}:{ptype}={pschema['default']}") - else: - parts.append(f"{pname}:{ptype}") - return ", ".join(parts) + """List all registered jobs (excluding self and non-servable) as compact one-liners for an LLM.""" async def execute(self): assert self.context is not None @@ -31,12 +31,11 @@ class HelpStep(BaseStep): lines = [] if self.app_context is not None: for name, job in self.app_context.jobs.items(): - if name == "help": + if name == "help" or not getattr(job, "enable_serve", True): continue - lines.append(f"🛠️ `{name}` — {job.description} 📥 {self._format_params(job.parameters)}") + lines.append(f"🛠️ `{name}` — {job.description} 📥 {_format_params(job.parameters)}") self.logger.info(f"[{self.name}] returning {len(lines)} jobs") - self.context.response.answer = "\n".join(lines) self.context.response.metadata["job_count"] = len(lines) return self.context.response diff --git a/reme4/steps/common/traverse.py b/reme4/steps/common/traverse.py index ae955acf..ac5f1ed4 100644 --- a/reme4/steps/common/traverse.py +++ b/reme4/steps/common/traverse.py @@ -1,71 +1,41 @@ -"""``traverse_step`` — BFS over wikilink edges from a seed file. +"""BFS over wikilink edges from one or more seed files. -Single tool for relationship browsing. ``depth=1`` covers the trivial -"what does this link to / what links here" lookups (set ``direction`` -accordingly); higher depth opens up multi-hop exploration. - -Output is one record per edge traversed (not per node), so the same -target can appear multiple times if reached via different predicates -or paths — agents dedupe at the call site if they want a flat node -set. Each record carries ``via`` (the predecessor) and the link's -``predicate`` / ``anchor`` so the agent can reconstruct the path. - -Adjacency is loaded once via ``file_graph.get_nodes(None)`` — every -real node arrives with its full ``links`` payload, and we build both -the outbound and the inbound index in a single pass. The BFS then -runs purely in memory: no per-frontier-node graph round-trips, no -filesystem walk. The ``get_inlinks`` / ``get_outlinks`` contract -methods stay unused here because they'd add network round-trips for -data we already have. - -Direction vocabulary accepts both the standard convention -(``forward`` / ``backward`` / ``both``) and the engine convention -(``out`` / ``in`` / ``both``). - -Seeds are paths relative to the vault used as-is — short-form resolution -is no longer attempted. Seeds that don't match any graph node yield -empty BFS results (no error). +One record per traversed *edge* (not per node): the same target can repeat +if reached via different predicates or paths. Each record carries the +predecessor plus the link's predicate/anchor so callers can reconstruct +the path. Adjacency is built once via a single ``file_store.get_nodes()`` +call — BFS then runs purely in memory with no per-frontier round-trips. """ from collections import deque from pathlib import Path from ..base_step import BaseStep - from ...components import R - from ...schema import FileLink +_OUT = {"out", "forward", "both"} +_IN = {"in", "backward", "both"} +_VALID = _OUT | _IN -_FORWARD = {"out", "forward"} -_BACKWARD = {"in", "backward"} -_BOTH = {"both"} -_VALID_DIRECTIONS = _FORWARD | _BACKWARD | _BOTH +# source path -> list of (neighbor path, link) +Adjacency = dict[str, list[tuple[str, FileLink]]] -async def _build_indexes( - file_store, -) -> tuple[ - dict[str, list[tuple[str, FileLink]]], - dict[str, list[tuple[str, FileLink]]], -]: - """One ``get_nodes(None)`` call → (outbound, inbound) adjacency dicts. +async def _build_adjacency(file_store) -> tuple[Adjacency, Adjacency]: + """Single ``get_nodes()`` pass → (outbound, inbound) adjacency maps. - Each dict is keyed by node path; values are ``(neighbor_path, link)`` - tuples. Source paths land in the inbound index alongside the link - object — solving the contract gap where ``get_inlinks`` returns + Inbound stores the source path next to each link so BFS can attribute + inbound edges back to their origin — ``get_inlinks`` alone returns target-shaped FileLinks without source attribution. """ - outbound: dict[str, list[tuple[str, FileLink]]] = {} - inbound: dict[str, list[tuple[str, FileLink]]] = {} - if not file_store.file_graph: - return outbound, inbound - for node in await file_store.file_graph.get_nodes(): + outbound: Adjacency = {} + inbound: Adjacency = {} + for node in await file_store.get_nodes(): for link in node.links: - if not link.target_path: - continue - outbound.setdefault(node.path, []).append((link.target_path, link)) - inbound.setdefault(link.target_path, []).append((node.path, link)) + if link.target_path: + outbound.setdefault(node.path, []).append((link.target_path, link)) + inbound.setdefault(link.target_path, []).append((node.path, link)) return outbound, inbound @@ -73,14 +43,17 @@ def _bfs( seeds: list[str], max_depth: int, direction: str, - outbound: dict[str, list[tuple[str, FileLink]]], - inbound: dict[str, list[tuple[str, FileLink]]], + outbound: Adjacency, + inbound: Adjacency, ) -> list[dict]: - """In-memory BFS. One record per edge traversed.""" - walk_out = direction in _FORWARD or direction in _BOTH - walk_in = direction in _BACKWARD or direction in _BOTH + """In-memory BFS; emits one record per unique (src, dst, predicate) edge.""" + sources: list[Adjacency] = [] + if direction in _OUT: + sources.append(outbound) + if direction in _IN: + sources.append(inbound) - visited_edges: set[tuple[str, str, str | None]] = set() + visited: set[tuple[str, str, str | None]] = set() results: list[dict] = [] queue: deque[tuple[str, int]] = deque((s, 0) for s in seeds) @@ -88,67 +61,51 @@ def _bfs( current, depth = queue.popleft() if depth >= max_depth: continue - - edges: list[tuple[str, str | None, str | None]] = [] - if walk_out: - for tgt, link in outbound.get(current, ()): - edges.append((tgt, link.predicate, link.target_anchor)) - if walk_in: - for src, link in inbound.get(current, ()): - edges.append((src, link.predicate, link.target_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)) - + for src in sources: + for next_path, link in src.get(current, ()): + key = (current, next_path, link.predicate) + if key in visited: + continue + visited.add(key) + results.append( + { + "path": next_path, + "depth": depth + 1, + "via": current, + "predicate": link.predicate, + "anchor": link.target_anchor, + }, + ) + if depth + 1 < max_depth: + queue.append((next_path, depth + 1)) return results -def _normalize_seeds(raw) -> list[str]: - """Coerce raw seed input to a non-empty list of strings relative to the vault.""" - if isinstance(raw, (str, Path)): - items = [raw] - else: - items = list(raw or []) - return [str(p) for p in items if p] - - @R.register("traverse_step") class TraverseStep(BaseStep): - """BFS from a seed file to explore wikilink relationships. + """BFS from one or more seed files to explore wikilink relationships. Parameters: - path — single seed (str) or a list of seeds. + path — single seed (str) or list of seeds (vault-relative). direction — ``forward`` / ``backward`` / ``both`` (or ``out`` / ``in`` / ``both``). depth — hop limit (default 1 = immediate neighbors). """ async def execute(self): assert self.context is not None - seeds_raw = self.context.get("path") + raw = self.context.get("path") + items = [raw] if isinstance(raw, (str, Path)) else list(raw or []) + seeds = [str(p) for p in items if p] + assert seeds, "path is required" depth = int(self.context.get("depth") or 1) direction = (self.context.get("direction") or "both").lower() - assert ( - direction in _VALID_DIRECTIONS - ), f"direction must be one of {sorted(_VALID_DIRECTIONS)}, got {direction!r}" - seeds = _normalize_seeds(seeds_raw) - assert seeds, "path is required" - outbound, inbound = await _build_indexes(self.file_store) + assert direction in _VALID, f"direction must be one of {sorted(_VALID)}, got {direction!r}" + + outbound, inbound = await _build_adjacency(self.file_store) results = _bfs(seeds, depth, direction, outbound, inbound) + + label = seeds[0] if len(seeds) == 1 else f"{len(seeds)} seeds" self.context.response.success = True - seed_label = seeds[0] if len(seeds) == 1 else f"{len(seeds)} seeds" - self.context.response.answer = f"Traversed {len(results)} edge(s) from {seed_label}" + self.context.response.answer = f"Traversed {len(results)} edge(s) from {label}" self.context.response.metadata.update({"edges": results, "count": len(results)}) + return self.context.response diff --git a/reme4/steps/crud/__init__.py b/reme4/steps/crud/__init__.py index 78356a47..4d94825f 100644 --- a/reme4/steps/crud/__init__.py +++ b/reme4/steps/crud/__init__.py @@ -3,14 +3,15 @@ The package covers two related surfaces: * **Opaque-byte ops** (don't care about file type): ``delete``, - ``download``, ``list``, ``move``, ``stat``, ``upload``, - ``upload_resource``. + ``list``, ``move``, ``stat``. * **Text-content ops** (markdown-aware; layered on the same path- resolution helpers in ``_file_io.py``): ``read``, ``write``, ``append``, ``edit``. For frontmatter slice RUD (YAML structured-data semantics) see -``reme4.steps.frontmatter``. +``reme4.steps.frontmatter``. For vault ↔ local-fs bridge ops +(``download`` / ``upload`` / ``upload_resource``) see +``reme4.steps.transfer``. """ from .read import ReadStep @@ -20,10 +21,7 @@ from .write import WriteStep from .append import AppendStep from .move import MoveStep from .stat import StatStep -from .download import DownloadStep from .list import ListStep -from .upload import UploadStep -from .upload_resource import UploadResourceStep __all__ = [ "DeleteStep", @@ -31,10 +29,7 @@ __all__ = [ "AppendStep", "MoveStep", "StatStep", - "DownloadStep", "ListStep", - "UploadStep", - "UploadResourceStep", "ReadStep", "EditStep", ] diff --git a/reme4/steps/index/__init__.py b/reme4/steps/index/__init__.py new file mode 100644 index 00000000..23e939fd --- /dev/null +++ b/reme4/steps/index/__init__.py @@ -0,0 +1,13 @@ +"""Index steps: write-side updates, full reindex, and read-side hybrid search.""" + +from .reindex import ReindexStep +from .search import SearchStep +from .update_catalog import UpdateCatalogStep +from .update_index import UpdateIndexStep + +__all__ = [ + "ReindexStep", + "SearchStep", + "UpdateCatalogStep", + "UpdateIndexStep", +] diff --git a/reme4/steps/common/reindex.py b/reme4/steps/index/reindex.py similarity index 59% rename from reme4/steps/common/reindex.py rename to reme4/steps/index/reindex.py index 84f86939..3e4d6f39 100644 --- a/reme4/steps/common/reindex.py +++ b/reme4/steps/index/reindex.py @@ -6,23 +6,18 @@ from ...components import R @R.register("reindex_step") class ReindexStep(BaseStep): - """Full re-index: clear store, walk vault, hand the file list to index_changes.""" + """Full re-index: clear the store, walk the vault, hand the file list to ``index_changes``.""" async def execute(self): assert self.context is not None - - suffix_filters: list[str] = self.context.get("suffix_filters", ["md"]) - suffixes = tuple("." + s.strip(".") for s in suffix_filters) if suffix_filters else None + suffixes = tuple("." + s.strip(".") for s in self.context.get("suffix_filters", ["md"])) await self.file_store.clear() - - paths: list[str] = [] - for p in self.vault_path.rglob("*"): - if not p.is_file(): - continue - if suffixes and not str(p).endswith(suffixes): - continue - paths.append(str(p.absolute())) + paths = [ + str(p.absolute()) + for p in self.vault_path.rglob("*") + if p.is_file() and (not suffixes or str(p).endswith(suffixes)) + ] if paths: await self.run_job("index_changes", changes=[{"change": "added", "path": p} for p in paths]) diff --git a/reme4/steps/common/search.py b/reme4/steps/index/search.py similarity index 100% rename from reme4/steps/common/search.py rename to reme4/steps/index/search.py diff --git a/reme4/steps/index/update_catalog.py b/reme4/steps/index/update_catalog.py new file mode 100644 index 00000000..c94d97ad --- /dev/null +++ b/reme4/steps/index/update_catalog.py @@ -0,0 +1,94 @@ +"""Update file catalog with a batch of file changes.""" + +from pathlib import Path + +from watchfiles import Change + +from ..base_step import BaseStep +from ...components import R +from ...components.file_catalog import BaseFileCatalog +from ...enumeration import ComponentEnum +from ...schema import FileNode + + +@R.register("update_catalog_step") +class UpdateCatalogStep(BaseStep): + """Classify raw watcher changes and update the file_catalog.""" + + @property + def file_catalog(self) -> BaseFileCatalog: + """Return the file catalog component.""" + return self._resolve("file_catalog", BaseFileCatalog, ComponentEnum.FILE_CATALOG) + + def _to_vault_relative(self, path: str | Path) -> str: + abs_path = Path(path).absolute() + try: + return str(abs_path.relative_to(self.vault_path)) + except ValueError: + return str(abs_path) + + async def execute(self): + assert self.context is not None + # Each item: {"change": Change | "added"|"modified"|"deleted", "path": absolute path} + changes: list[dict] = self.context.get("changes") or [] + persist: bool = bool(self.context.get("persist", False)) + + buckets: dict[Change, list[str]] = {Change.added: [], Change.modified: [], Change.deleted: []} + for item in changes: + c = item["change"] + if isinstance(c, str): + c = Change.__members__.get(c) + if isinstance(c, Change) and c in buckets: + buckets[c].append(item["path"]) + + results: list[dict] = [] + + for change, action in ((Change.added, "Adding"), (Change.modified, "Updating")): + paths = buckets[change] + if not paths: + continue + self.logger.info(f"Detected {len(paths)} {change.name} file(s)") + nodes: list[FileNode] = [] + ok_paths: list[str] = [] + for path in paths: + abs_path = Path(path) + if not abs_path.is_file(): + results.append({"change": change.name, "path": path, "success": False, "error": "not a file"}) + continue + self.logger.info(f"{action} file: {path}") + try: + stat = abs_path.stat() + nodes.append(FileNode(path=self._to_vault_relative(abs_path), st_mtime=stat.st_mtime)) + ok_paths.append(path) + except Exception as e: + self.logger.exception(f"Failed to stat {path}") + results.append({"change": change.name, "path": path, "success": False, "error": str(e)}) + if nodes: + try: + await self.file_catalog.delete([n.path for n in nodes]) + await self.file_catalog.upsert(nodes) + results.extend({"change": change.name, "path": p, "success": True} for p in ok_paths) + except Exception as e: + self.logger.exception(f"Failed to upsert {len(nodes)} {change.name} file(s)") + results.extend( + {"change": change.name, "path": p, "success": False, "error": str(e)} for p in ok_paths + ) + + if deleted := buckets[Change.deleted]: + if self.file_catalog is None: + raise RuntimeError("file_catalog is not initialized!") + self.logger.info(f"Detected {len(deleted)} deleted file(s)") + rel_deleted = [self._to_vault_relative(p) for p in deleted] + try: + await self.file_catalog.delete(rel_deleted) + results.extend({"change": "deleted", "path": p, "success": True} for p in deleted) + except Exception as e: + self.logger.exception(f"Failed to delete {len(deleted)} file(s)") + results.extend({"change": "deleted", "path": p, "success": False, "error": str(e)} for p in deleted) + + if persist and results: + await self.file_catalog.dump() + + self.context.response.answer = results + self.context.response.success = all(r["success"] for r in results) if results else True + return self.context.response diff --git a/reme4/steps/background/update_store_index.py b/reme4/steps/index/update_index.py similarity index 96% rename from reme4/steps/background/update_store_index.py rename to reme4/steps/index/update_index.py index 031f0617..95357cb6 100644 --- a/reme4/steps/background/update_store_index.py +++ b/reme4/steps/index/update_index.py @@ -1,4 +1,4 @@ -"""Update store index with a batch of file changes.""" +"""Update index with a batch of file changes.""" from pathlib import Path @@ -9,8 +9,8 @@ from ...components import R from ...schema import FileChunk, FileNode -@R.register("update_store_index_step") -class UpdateStoreIndexStep(BaseStep): +@R.register("update_index_step") +class UpdateIndexStep(BaseStep): """Classify raw watcher changes and update the file_store index.""" async def execute(self): diff --git a/reme4/steps/transfer/__init__.py b/reme4/steps/transfer/__init__.py new file mode 100644 index 00000000..63a818cf --- /dev/null +++ b/reme4/steps/transfer/__init__.py @@ -0,0 +1,20 @@ +"""Bridge ops between vault_dir and the local filesystem. + +Unlike ``crud`` (single-resource ops within the vault), these steps +cross two domains: vault ↔ host fs. + +* ``download`` — vault → local fs (export; vault is read-only). +* ``upload`` — local fs → vault (raw copy to a caller-supplied path). +* ``upload_resource`` — local fs → ``resource//`` (passive + ingest with provenance metadata + day-view rendering). +""" + +from .download import DownloadStep +from .upload import UploadStep +from .upload_resource import UploadResourceStep + +__all__ = [ + "DownloadStep", + "UploadStep", + "UploadResourceStep", +] diff --git a/reme4/steps/crud/download.py b/reme4/steps/transfer/download.py similarity index 100% rename from reme4/steps/crud/download.py rename to reme4/steps/transfer/download.py diff --git a/reme4/steps/crud/upload.py b/reme4/steps/transfer/upload.py similarity index 100% rename from reme4/steps/crud/upload.py rename to reme4/steps/transfer/upload.py diff --git a/reme4/steps/crud/upload_resource.py b/reme4/steps/transfer/upload_resource.py similarity index 100% rename from reme4/steps/crud/upload_resource.py rename to reme4/steps/transfer/upload_resource.py diff --git a/tests4/unittest/test_crud_steps.py b/tests4/unittest/test_crud_steps.py index 0f170de6..eea683e8 100644 --- a/tests4/unittest/test_crud_steps.py +++ b/tests4/unittest/test_crud_steps.py @@ -36,11 +36,11 @@ from reme4.components.file_store import LocalFileStore from reme4.schema import FileNode from reme4.steps.crud import ( delete as crud_delete, - download as crud_download, list as crud_list, move as crud_move, stat as crud_stat, ) +from reme4.steps.transfer import download as crud_download from reme4.utils import call_action, call_and_check, mock_reme_server from reme4.utils.wikilink_handler import WikilinkHandler diff --git a/tests4/unittest/test_file_catalog.py b/tests4/unittest/test_file_catalog.py index 71a3cc8f..68e50b5d 100644 --- a/tests4/unittest/test_file_catalog.py +++ b/tests4/unittest/test_file_catalog.py @@ -29,6 +29,7 @@ class temp_chdir: def make_node(path: str, mtime: float = 1.0) -> FileNode: + """Build a FileNode with the given path and mtime for fixture use.""" return FileNode(path=path, st_mtime=mtime) diff --git a/tests4/unittest/test_keyword_index.py b/tests4/unittest/test_keyword_index.py index be374198..311d550b 100644 --- a/tests4/unittest/test_keyword_index.py +++ b/tests4/unittest/test_keyword_index.py @@ -41,8 +41,11 @@ class temp_chdir: os.chdir(self.old) -async def create_bm25(k1: float = 1.5, b: float = 0.75, - filter_stopwords: bool = False) -> BM25Index: +async def create_bm25( + k1: float = 1.5, + b: float = 0.75, + filter_stopwords: bool = False, +) -> BM25Index: """Create and start a BM25Index in cwd with a non-filtering RegexTokenizer. Stopword filtering is off so short test words ("hello", "我", "的") survive. @@ -75,7 +78,7 @@ def test_basic_init(): assert bm25.b == 0.75 assert bm25.index_version == "v1" assert bm25.vocab == {} - assert bm25.inverted_index == {} + assert not bm25.inverted_index assert bm25.doc_meta == {} assert bm25.n_docs == 0 assert bm25.total_len == 0 @@ -183,11 +186,13 @@ def test_add_multiple_docs_and_inverted_index(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "hello world", - "d2": "hello python", - "d3": "world python", - }) + await bm25.add_docs( + { + "d1": "hello world", + "d2": "hello python", + "d3": "world python", + }, + ) assert bm25.n_docs == 3 inv = bm25.inverted_index tid_hello = bm25.vocab["hello"] @@ -359,11 +364,13 @@ def test_retrieve_score_ordering_by_tf(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "high": "python python python", - "mid": "python python other", - "low": "python alpha beta", - }) + await bm25.add_docs( + { + "high": "python python python", + "mid": "python python other", + "low": "python alpha beta", + }, + ) results = await bm25.retrieve("python", limit=3) assert list(results.keys()) == ["high", "mid", "low"] scores = list(results.values()) @@ -398,10 +405,12 @@ def test_retrieve_length_normalization(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "short": "python", - "long": "python " + " ".join(f"w{i}" for i in range(50)), - }) + await bm25.add_docs( + { + "short": "python", + "long": "python " + " ".join(f"w{i}" for i in range(50)), + }, + ) results = await bm25.retrieve("python", limit=2) assert results["short"] > results["long"] await bm25.close() @@ -435,11 +444,13 @@ def test_chinese_only_corpus(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "我爱北京天安门", - "d2": "北京是中国的首都", - "d3": "上海的天气很好", - }) + await bm25.add_docs( + { + "d1": "我爱北京天安门", + "d2": "北京是中国的首都", + "d3": "上海的天气很好", + }, + ) # Regex tokenizer splits Chinese per character. assert "北" in bm25.vocab assert "京" in bm25.vocab @@ -458,10 +469,12 @@ def test_english_only_corpus_is_lowercased(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "Python Programming Language", - "d2": "Java Programming Language", - }) + await bm25.add_docs( + { + "d1": "Python Programming Language", + "d2": "Java Programming Language", + }, + ) assert "python" in bm25.vocab assert "Python" not in bm25.vocab @@ -498,11 +511,13 @@ def test_mixed_doc_chinese_query_matches(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "Python 是一种编程语言", - "d2": "Java 编程语言", - "d3": "Python 数据分析", - }) + await bm25.add_docs( + { + "d1": "Python 是一种编程语言", + "d2": "Java 编程语言", + "d3": "Python 数据分析", + }, + ) results = await bm25.retrieve("编程", limit=3) assert set(results) >= {"d1", "d2"} assert "d3" not in results @@ -517,11 +532,13 @@ def test_mixed_doc_english_query_matches(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "Python 是一种编程语言", - "d2": "Java 编程语言", - "d3": "Python 数据分析", - }) + await bm25.add_docs( + { + "d1": "Python 是一种编程语言", + "d2": "Java 编程语言", + "d3": "Python 数据分析", + }, + ) results = await bm25.retrieve("python", limit=3) assert set(results) == {"d1", "d3"} assert "d2" not in results @@ -536,14 +553,16 @@ def test_mixed_query_combines_chinese_and_english_signal(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "py_cn": "Python 编程", # matches both 'python' and '编','程' - "py_only": "Python tutorial", # matches only 'python' - "cn_only": "编程入门", # matches only '编','程' - # Avoid Chinese chars that the query splits into ('编','程') — '教程' would - # leak '程' into 'other' and pollute IDF, so use unrelated chars only. - "other": "Java 教学", - }) + await bm25.add_docs( + { + "py_cn": "Python 编程", # matches both 'python' and '编','程' + "py_only": "Python tutorial", # matches only 'python' + "cn_only": "编程入门", # matches only '编','程' + # Avoid Chinese chars that the query splits into ('编','程') — '教程' would + # leak '程' into 'other' and pollute IDF, so use unrelated chars only. + "other": "Java 教学", + }, + ) results = await bm25.retrieve("Python 编程", limit=4) # py_cn should rank highest because it matches both branches. assert next(iter(results)) == "py_cn" @@ -562,11 +581,13 @@ def test_mixed_doc_more_matches_outrank_fewer(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "full": "machine learning 机器 学习", - "en_only": "machine learning algorithm", - "cn_only": "机器 学习 算法", - }) + await bm25.add_docs( + { + "full": "machine learning 机器 学习", + "en_only": "machine learning algorithm", + "cn_only": "机器 学习 算法", + }, + ) results = await bm25.retrieve("machine 机器", limit=3) # full has both English and Chinese hits → highest score. assert next(iter(results)) == "full" @@ -581,10 +602,12 @@ def test_unicode_word_with_digits_preserved(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "iPhone15 Pro 售价 9999 元", - "d2": "Android 旗舰 999 元", - }) + await bm25.add_docs( + { + "d1": "iPhone15 Pro 售价 9999 元", + "d2": "Android 旗舰 999 元", + }, + ) assert "iphone15" in bm25.vocab assert "9999" in bm25.vocab assert "元" in bm25.vocab @@ -620,11 +643,13 @@ def test_mixed_persistence_roundtrip(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "Python 编程语言", - "d2": "Java 编程", - "d3": "数据分析 with Python", - }) + await bm25.add_docs( + { + "d1": "Python 编程语言", + "d2": "Java 编程", + "d3": "数据分析 with Python", + }, + ) before = await bm25.retrieve("Python 编程", limit=3) await bm25.close() # close triggers dump @@ -648,11 +673,13 @@ def test_dump_load_roundtrip_preserves_state(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25(k1=2.0, b=0.4) - await bm25.add_docs({ - "d1": "hello world", - "d2": "hello python", - "d3": "programming language", - }) + await bm25.add_docs( + { + "d1": "hello world", + "d2": "hello python", + "d3": "programming language", + }, + ) old_vocab = dict(bm25.vocab) old_meta = {k: dict(v) for k, v in bm25.doc_meta.items()} await bm25.dump() @@ -738,10 +765,12 @@ def test_optimize_drops_deleted_only_terms(): async def go(): with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp): bm25 = await create_bm25() - await bm25.add_docs({ - "d1": "alpha unique_to_d1", - "d2": "alpha beta", - }) + await bm25.add_docs( + { + "d1": "alpha unique_to_d1", + "d2": "alpha beta", + }, + ) assert "unique_to_d1" in bm25.vocab await bm25.delete_docs(["d1"]) @@ -853,7 +882,7 @@ def test_avg_len_tracks_live_docs_only(): bm25 = await create_bm25() assert bm25.avg_len == 0.0 await bm25.add_docs({"d1": "alpha beta gamma delta"}) # 4 tokens - await bm25.add_docs({"d2": "alpha beta"}) # 2 tokens + await bm25.add_docs({"d2": "alpha beta"}) # 2 tokens assert bm25.avg_len == 3.0 await bm25.delete_docs(["d1"]) @@ -909,10 +938,7 @@ if __name__ == "__main__": import sys mod = sys.modules[__name__] - tests = [ - (name, obj) for name, obj in inspect.getmembers(mod, inspect.isfunction) - if name.startswith("test_") - ] + tests = [(name, obj) for name, obj in inspect.getmembers(mod, inspect.isfunction) if name.startswith("test_")] print(f"\n=== BaseKeywordIndex / BM25Index Tests ({len(tests)}) ===\n") failed = [] for name, fn in tests: diff --git a/tests4/unittest/test_resource_steps.py b/tests4/unittest/test_resource_steps.py index a0c4ef5f..51acfa9b 100644 --- a/tests4/unittest/test_resource_steps.py +++ b/tests4/unittest/test_resource_steps.py @@ -24,8 +24,8 @@ from pathlib import Path from reme4.components.file_store import LocalFileStore from reme4.schema import FileFrontMatter, FileNode -from reme4.steps.crud import upload_resource as crud_upload -from reme4.steps.crud.upload_resource import _assemble_day_md +from reme4.steps.transfer import upload_resource as crud_upload +from reme4.steps.transfer.upload_resource import _assemble_day_md warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba") warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")