diff --git a/reme/__init__.py b/reme/__init__.py index 0ca5442f..049fd30f 100644 --- a/reme/__init__.py +++ b/reme/__init__.py @@ -1,6 +1,6 @@ """ReMe CLI package.""" -__version__ = "0.4.1.0" +__version__ = "0.4.1.1" from . import config from . import constants diff --git a/reme/config/default.yaml b/reme/config/default.yaml index 48df45fd..843981f0 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -4,6 +4,7 @@ service: jobs: index_update_loop: backend: background + max_file_bytes: 20971520 watch_dirs: [daily_dir, digest_dir, resource_dir] # watch_dirs: [daily_dir, digest_dir] watch_suffixes: [md, jsonl] @@ -20,6 +21,7 @@ jobs: resource_watch_loop: backend: background + max_file_bytes: 20971520 watch_dirs: [resource_dir] watch_suffixes: [md, txt, json, jsonl, csv, yaml, html] steps: @@ -38,6 +40,7 @@ jobs: digest_watch_loop: backend: background + max_file_bytes: 20971520 watch_dirs: [daily_dir, digest_dir] watch_suffixes: [md] steps: @@ -279,6 +282,7 @@ jobs: reindex: backend: base + max_file_bytes: 20971520 description: "wipe the file store and rebuild it from the existing files" watch_dirs: [daily_dir, digest_dir, resource_dir] watch_suffixes: [md, jsonl] diff --git a/reme/constants.py b/reme/constants.py index ee4c1149..c9566a7a 100644 --- a/reme/constants.py +++ b/reme/constants.py @@ -14,3 +14,7 @@ TRUNCATION_NOTICE_MARKER = "<>" # read_image step: oversized images above this threshold return path & metadata # only (no base64) to keep LLM context budgets safe. DEFAULT_MAX_IMAGE_BYTES = 5 * 1024 * 1024 + +# Background content-processing jobs skip files above this size. File watchers +# and catalogs still track them so deletes and later size reductions are seen. +DEFAULT_MAX_FILE_BYTES = 20 * 1024 * 1024 diff --git a/reme/steps/base_step.py b/reme/steps/base_step.py index d9435786..e2ea5a44 100644 --- a/reme/steps/base_step.py +++ b/reme/steps/base_step.py @@ -15,6 +15,7 @@ from ..components.prompt_handler import PromptHandler from ..components.runtime_context import RuntimeContext from ..enumeration import ComponentEnum from ..schema import ApplicationConfig, Response +from ..constants import DEFAULT_MAX_FILE_BYTES if TYPE_CHECKING: from ..components import ApplicationContext @@ -164,6 +165,13 @@ class BaseStep(ComponentMixin, ABC): value = getattr(cfg, key) return getattr(defaults, key) if value in (None, "") else value + def max_file_bytes(self) -> int: + """Return the content-processing size limit from Step or Job context.""" + value = self.kwargs.get("max_file_bytes") + if value is None and self.context is not None: + value = self.context.get("max_file_bytes") + return int(value) if value is not None else DEFAULT_MAX_FILE_BYTES + def copy(self, **kwargs) -> "BaseStep": """Construct a new instance from the original init args, applying overrides.""" return self.__class__(*self._init_args, **{**self._init_kwargs, **kwargs}) diff --git a/reme/steps/evolve/auto_resource.py b/reme/steps/evolve/auto_resource.py index 2fc3c88b..5264b5c5 100644 --- a/reme/steps/evolve/auto_resource.py +++ b/reme/steps/evolve/auto_resource.py @@ -340,6 +340,48 @@ class AutoResourceStep(BaseStep): self.logger.warning(f"[{self.name}] resource missing file_path={file_path}") return + skip_read = False + try: + size_bytes = abs_path.stat().st_size + except OSError as exc: + self.context.response.success = False + self.context.response.answer = f"Failed to inspect resource file: {file_path}: {exc}" + self.context.response.metadata.update( + { + "path": file_path, + "action": "failed", + "error": str(exc), + "modified": False, + }, + ) + self.logger.warning(f"[{self.name}] resource stat failed file_path={file_path} error={exc}") + skip_read = True + if not skip_read: + max_file_bytes = self.max_file_bytes() + if size_bytes > max_file_bytes: + self.context.response.success = True + self.context.response.answer = ( + f"Skipped oversized resource file: {file_path} ({size_bytes} > {max_file_bytes} bytes)" + ) + self.context.response.metadata.update( + { + "path": file_path, + "action": "skipped", + "reason": "file_too_large", + "oversized": True, + "size_bytes": size_bytes, + "max_file_bytes": max_file_bytes, + "modified": False, + }, + ) + self.logger.warning( + f"[{self.name}] skip oversized resource file_path={file_path} " + f"size_bytes={size_bytes} max_file_bytes={max_file_bytes}", + ) + skip_read = True + if skip_read: + return + self.logger.info(f"[{self.name}] read resource start file_path={file_path}") async with aiofiles.open(abs_path, encoding="utf-8", errors="replace") as f: file_content = await f.read() @@ -433,6 +475,9 @@ class AutoResourceStep(BaseStep): async def _handle_change(self, file_path: str, raw_change) -> dict: assert self.context is not None + # Handlers write item-scoped fields into the shared response. Start each + # change with a fresh mapping so one result cannot inherit another's metadata. + self.context.response.metadata = {} file_path = self.to_workspace_relative(file_path) if file_path and Path(file_path).is_absolute() else file_path if not file_path: self.context.response.success = False diff --git a/reme/steps/index/update_changes.py b/reme/steps/index/update_changes.py index f288c362..ea4fcaf2 100644 --- a/reme/steps/index/update_changes.py +++ b/reme/steps/index/update_changes.py @@ -18,6 +18,7 @@ class ChangeApplyStep(BaseStep): """Shared added/modified/deleted handling for index update targets.""" target_name = "target" + reads_file_content = False def __init__(self, persist: bool | None = None, **kwargs): super().__init__(**kwargs) @@ -75,14 +76,33 @@ class ChangeApplyStep(BaseStep): async def _try_build_item(self, change: Change, action: str, path: str, results: list[dict]): abs_path = self._to_abs_path(path) - if not abs_path.is_file(): - results.append({"change": change.name, "path": path, "success": False, "error": "not a file"}) - return None - self.logger.info(f"{action} file: {path}") try: + if not abs_path.is_file(): + results.append({"change": change.name, "path": path, "success": False, "error": "not a file"}) + return None + size_bytes = abs_path.stat().st_size + max_file_bytes = self.max_file_bytes() + if self.reads_file_content and size_bytes > max_file_bytes: + await self.delete_paths([self.to_workspace_relative(abs_path)]) + self.logger.warning( + f"Skipping oversized file: {path} size_bytes={size_bytes} max_file_bytes={max_file_bytes}", + ) + results.append( + { + "change": change.name, + "path": path, + "success": True, + "skipped": True, + "reason": "file_too_large", + "size_bytes": size_bytes, + "max_file_bytes": max_file_bytes, + }, + ) + return None + self.logger.info(f"{action} file: {path}") return await self.build_item(abs_path) except Exception as e: - self.logger.exception(f"Failed to parse {path}") + self.logger.exception(f"Failed to process {path}") results.append({"change": change.name, "path": path, "success": False, "error": str(e)}) return None @@ -144,6 +164,7 @@ class UpdateIndexStep(ChangeApplyStep): """Update file_store with a batch of file changes.""" target_name = "file_store" + reads_file_content = True async def build_item(self, path: Path) -> tuple[FileNode, list[FileChunk]]: return await self.chunk_file(path) diff --git a/tests/unit/test_background_steps.py b/tests/unit/test_background_steps.py index 08aa4901..698770f8 100644 --- a/tests/unit/test_background_steps.py +++ b/tests/unit/test_background_steps.py @@ -8,7 +8,7 @@ InitChangesStep writes its result into ``context["changes"]`` for a downstream ``update_index_step`` to consume; tests assert against that key. """ -# pylint: disable=protected-access +# pylint: disable=protected-access,too-many-lines import asyncio import datetime @@ -16,7 +16,7 @@ import os import tempfile import warnings from pathlib import Path -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from watchfiles import Change @@ -39,6 +39,7 @@ from reme.steps.index import ( InitChangesStep, LogChangesStep, UpdateCatalogStep, + UpdateIndexStep, WatchChangesStep, ) from reme.steps.index._change_batch import bucket_changes @@ -461,6 +462,79 @@ def test_update_catalog_relative_path_uses_workspace(): asyncio.run(run()) +def test_update_index_skips_oversized_file_and_clears_stale_index(): + """Oversized content is not read and any previous index entry is removed.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + source = write_file(cwd / "daily" / "a.md", "small") + fs = LocalFileStore(name="default", embedding_store="") + chunker = DefaultFileChunker() + await fs.start() + await chunker.start() + try: + app_ctx = _make_app_context(cwd) + app_ctx.components = { + ComponentEnum.FILE_CHUNKER: {"default": chunker}, + } + step = UpdateIndexStep(file_store=fs, persist=False, app_context=app_ctx) + + added = await step( + RuntimeContext( + changes=[{"change": "added", "path": str(source)}], + max_file_bytes=10, + ), + ) + assert added.success is True + assert {node.path for node in await fs.get_nodes()} == {"daily/a.md"} + + source.write_text("now too large", encoding="utf-8") + modified = await step( + RuntimeContext( + changes=[{"change": "modified", "path": str(source)}], + max_file_bytes=10, + ), + ) + + assert modified.success is True + assert modified.answer[0]["skipped"] is True + assert modified.answer[0]["reason"] == "file_too_large" + assert await fs.get_nodes() == [] + finally: + await chunker.close() + await fs.close() + + asyncio.run(run()) + + +def test_update_index_handles_file_removed_before_stat(): + """A file disappearing after is_file is reported without aborting the batch.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + step = UpdateIndexStep(app_context=_make_app_context(Path.cwd())) + results = [] + + with ( + patch.object(Path, "is_file", return_value=True), + patch.object(Path, "stat", side_effect=FileNotFoundError("file disappeared")), + ): + item = await step._try_build_item(Change.modified, "Updating", "daily/a.md", results) + + assert item is None + assert results == [ + { + "change": "modified", + "path": "daily/a.md", + "success": False, + "error": "file disappeared", + }, + ] + + asyncio.run(run()) + + def test_index_update_loop_init_dispatch_updates_store_across_batches(): """index_update_loop init scan dispatches to update_index_step and preserves final store state.""" @@ -717,6 +791,123 @@ def test_auto_resource_batch_deleted_changes(): asyncio.run(run()) +def test_auto_resource_skips_oversized_file_before_reading(): + """Oversized resources are reported as successful skips without an agent call.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + wrapper = _FakeAgentWrapper() + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + source = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large") + step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + resp = await step( + RuntimeContext( + changes=[{"change": "added", "path": str(source)}], + max_file_bytes=4, + ), + ) + + result = resp.metadata["results"][0] + assert resp.success is True + assert result["metadata"]["oversized"] is True + assert result["metadata"]["reason"] == "file_too_large" + assert resp.metadata["modified"] is False + assert wrapper.inputs == "" + finally: + await fs.close() + + asyncio.run(run()) + + +def test_auto_resource_batch_keeps_result_metadata_isolated(): + """Oversized metadata from one resource does not leak into the next result.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + wrapper = _FakeAgentWrapper() + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + large = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large") + small = write_file(cwd / "resource" / "2026-01-01" / "small.txt", "ok") + second_large = write_file(cwd / "resource" / "2026-01-01" / "second-large.txt", "also large") + step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) + resp = await step( + RuntimeContext( + changes=[ + {"change": "added", "path": str(large)}, + {"change": "added", "path": str(small)}, + {"change": "added", "path": str(second_large)}, + ], + max_file_bytes=4, + ), + ) + + large_metadata = resp.metadata["results"][0]["metadata"] + small_metadata = resp.metadata["results"][1]["metadata"] + second_large_metadata = resp.metadata["results"][2]["metadata"] + assert large_metadata["oversized"] is True + assert large_metadata["reason"] == "file_too_large" + assert "oversized" not in small_metadata + assert "reason" not in small_metadata + assert "size_bytes" not in small_metadata + assert second_large_metadata["oversized"] is True + assert "created" not in second_large_metadata + assert "agent_session_id" not in second_large_metadata + finally: + await fs.close() + + asyncio.run(run()) + + +def test_auto_resource_handles_file_removed_before_stat(): + """A resource disappearing after is_file becomes one failed batch result.""" + + async def run(): + with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): + cwd = Path.cwd() + app_ctx = _make_app_context(cwd) + fs = LocalFileStore(name="test_store", embedding_store="") + await fs.start() + _install_file_jobs(app_ctx, fs) + try: + source = write_file(cwd / "resource" / "2026-01-01" / "vanishing.txt", "content") + original_stat = Path.stat + source_stat_calls = 0 + + def disappearing_stat(path, *args, **kwargs): + nonlocal source_stat_calls + if path == source: + source_stat_calls += 1 + if source_stat_calls > 1: + raise FileNotFoundError("file disappeared") + return original_stat(path, *args, **kwargs) + + step = AutoResourceStep(app_context=app_ctx, file_store=fs) + with patch.object(Path, "stat", disappearing_stat): + resp = await step( + RuntimeContext(changes=[{"change": "added", "path": str(source)}]), + ) + + result = resp.metadata["results"][0] + assert resp.success is False + assert result["success"] is False + assert result["metadata"]["action"] == "failed" + assert result["metadata"]["error"] == "file disappeared" + finally: + await fs.close() + + asyncio.run(run()) + + def test_auto_resource_accepts_loose_root_resource(): """Root-level resource files use today's date without moving the source.""" diff --git a/tests/unit/test_config_parser.py b/tests/unit/test_config_parser.py index 7e0ed024..95838397 100644 --- a/tests/unit/test_config_parser.py +++ b/tests/unit/test_config_parser.py @@ -72,6 +72,14 @@ def test_default_config_keeps_frontmatter_chunk_metadata_opt_in(): ) in (None, []) +def test_default_config_limits_background_and_reindex_file_processing(): + """All built-in content-processing entry points use the 20 MiB limit.""" + cfg = _load_config("default.yaml") + + for job_name in ("index_update_loop", "resource_watch_loop", "digest_watch_loop", "reindex"): + assert cfg["jobs"][job_name]["max_file_bytes"] == 20 * 1024 * 1024 + + def test_parse_args_rejects_non_key_value_extra_argument(): """Extra CLI arguments must use key=value syntax.""" with pytest.raises(ValueError, match="expected key=value"):