feat(index): add file size limits and oversized file handling (#362)

* feat(index): add file size limits and oversized file handling

- Implement max_file_bytes configuration option for content processing jobs
- Add default 20MB file size limit for background processing in default config
- Skip oversized files during auto_resource step with appropriate metadata
- Clear stale index entries when oversized files are modified
- Add size-based filtering logic to update_changes step with skip reporting
- Include file size validation in UpdateIndexStep with proper response handling
- Add comprehensive tests for oversized file scenarios in auto_resource and update_index
- Document file size limits in constants with appropriate thresholds

* chore(version): bump version to 0.4.1.1

- Update __version__ from 0.4.1.0 to 0.4.1.1 in __init__.py

* fix(index): isolate batch metadata and handle file races

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

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]

reme/constants.py

@@ -14,3 +14,7 @@ TRUNCATION_NOTICE_MARKER = "<<TRUNCATION_NOTICE>>"
 # 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

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})

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

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)

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."""
 

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"):