"""Tests for background steps: scan/watch/dispatch steps. Both scan steps are subclasses of BaseStep. To exercise them without spinning up the full ApplicationContext, we pass real (started) file_store/file_chunker via the step's kwargs (so the BaseStep _resolve() machinery returns them). InitChangesStep writes its result into ``context["changes"]`` for a downstream ``update_index_step`` to consume; tests assert against that key. """ # pylint: disable=protected-access,too-many-lines import asyncio import datetime import os import tempfile import warnings import weakref from pathlib import Path from unittest.mock import MagicMock, patch import numpy as np import pytest from watchfiles import Change from reme.components import R from reme.components.agent_wrapper import BaseAgentWrapper from reme.components.file_chunker import DefaultFileChunker from reme.components.file_catalog import LocalFileCatalog from reme.components.file_store import LocalFileStore from reme.components.runtime_context import RuntimeContext from reme.enumeration import ComponentEnum from reme.steps.evolve.auto_memory import AutoMemoryStep from reme.steps.evolve.auto_resource import AutoResourceStep, _compute_note_stem from reme.steps.file_io.daily_list import DailyListStep from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep from reme.steps.file_io.move import MoveStep from reme.steps.index import ( DEFAULT_LOW_POWER_POLL_MS, DEFAULT_WATCH_DEBOUNCE_MS, DEFAULT_WATCH_STEP_MS, ClearStoreStep, InitChangesStep, LogChangesStep, UpdateCatalogStep, UpdateIndexStep, WatchChangesStep, ) from reme.steps.index._change_batch import bucket_changes from reme.steps.index._watch_rules import WatchRule, build_watch_rules, collect_existing, match_file warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba") warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources") class temp_chdir: """Context manager to temporarily chdir into a path and restore on exit.""" def __init__(self, path): self.path = path self.old = None def __enter__(self): self.old = os.getcwd() os.chdir(self.path) return self def __exit__(self, *exc): os.chdir(self.old) def write_file(path: Path, content: str = "x") -> Path: """Create parent dirs and write `content` to `path`; return the path.""" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content, encoding="utf-8") return path class _FakeAgentWrapper(BaseAgentWrapper): """Capture agent calls without invoking a real model.""" def __init__(self): super().__init__() self.inputs = "" self.kwargs = {} self.on_reply = None async def reply(self, inputs, **kwargs) -> dict: self.inputs = inputs self.kwargs = kwargs if self.on_reply is not None: self.on_reply(inputs, kwargs) return {"result": "ok"} class _StepJob: """Tiny job adapter for unit tests that need BaseStep.run_job.""" def __init__(self, step_cls, app_context, file_store): self.step_cls = step_cls self.app_context = app_context self.file_store = file_store async def __call__(self, **kwargs): step = self.step_cls(app_context=self.app_context, file_store=self.file_store) result = await step(**kwargs) return result or step.context.response def _make_app_context(workspace_path: Path, daily_dir="daily", digest_dir="digest", resource_dir="resource"): """Create a mock app_context with app_config pointing to the given workspace.""" ctx = MagicMock() ctx.app_config.workspace_dir = str(workspace_path) ctx.app_config.daily_dir = daily_dir ctx.app_config.digest_dir = digest_dir ctx.app_config.resource_dir = resource_dir ctx.app_config.session_dir = "session" ctx.app_config.timezone = None ctx.registry = R.copy() return ctx def _install_file_jobs(app_context, file_store) -> None: app_context.jobs = { "daily_list": _StepJob(DailyListStep, app_context, file_store), "frontmatter_update": _StepJob(FrontmatterUpdateStep, app_context, file_store), "move": _StepJob(MoveStep, app_context, file_store), } # --------------------------------------------------------------------------- # _watch_rules module tests # --------------------------------------------------------------------------- def test_build_watch_rules_basic(): """Build rules from watch_dirs and watch_suffixes.""" app_config = MagicMock() app_config.daily_dir = "daily" app_config.digest_dir = "digest" app_config.resource_dir = "resource" workspace = Path("/fake/workspace") rules = build_watch_rules(app_config, workspace, watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"]) assert len(rules) == 2 assert rules[0].path == workspace / "daily" assert rules[0].suffixes == ["md"] assert rules[1].path == workspace / "digest" print("✓ test_build_watch_rules_basic passed") def test_build_watch_rules_multiple_suffixes(): """Multiple suffixes are forwarded to each rule.""" app_config = MagicMock() app_config.daily_dir = "daily" app_config.resource_dir = "resource" workspace = Path("/fake/workspace") rules = build_watch_rules( app_config, workspace, watch_dirs=["daily_dir", "resource_dir"], watch_suffixes=["md", "jsonl"], ) assert len(rules) == 2 assert rules[0].suffixes == ["md", "jsonl"] assert rules[1].suffixes == ["md", "jsonl"] print("✓ test_build_watch_rules_multiple_suffixes passed") def test_build_watch_rules_fallback_literal(): """Unknown field names are used as literal directory names.""" app_config = MagicMock(spec=[]) # no attributes workspace = Path("/fake/workspace") rules = build_watch_rules(app_config, workspace, watch_dirs=["custom_dir"], watch_suffixes=["txt"]) assert rules[0].path == workspace / "custom_dir" print("✓ test_build_watch_rules_fallback_literal passed") def test_match_file_suffix(): """match_file accepts files matching suffix under rule path.""" rules = [WatchRule(path=Path("/workspace/daily"), suffixes=["md"])] assert match_file("/workspace/daily/2026-01-01.md", rules) assert match_file("/workspace/daily/sub/note.md", rules) assert not match_file("/workspace/daily/file.txt", rules) assert not match_file("/workspace/other/file.md", rules) print("✓ test_match_file_suffix passed") def test_match_file_no_suffix_filter(): """Empty suffixes list means all files match.""" rules = [WatchRule(path=Path("/workspace/resource"), suffixes=[])] assert match_file("/workspace/resource/anything.xyz", rules) assert match_file("/workspace/resource/sub/deep.pdf", rules) assert not match_file("/workspace/other/file.md", rules) print("✓ test_match_file_no_suffix_filter passed") def test_collect_existing_filters(): """collect_existing applies suffix rules correctly.""" with tempfile.TemporaryDirectory() as tmpdir: workspace = Path(tmpdir) daily = workspace / "daily" resource = workspace / "resource" write_file(daily / "note.md") write_file(daily / "ignore.txt") write_file(resource / "data.json") write_file(resource / "binary.png") rules = [ WatchRule(path=daily, suffixes=["md"]), WatchRule(path=resource, suffixes=["json"]), ] result = collect_existing(rules, recursive=True) paths = set(result.keys()) assert str((daily / "note.md").absolute()) in paths assert str((daily / "ignore.txt").absolute()) not in paths assert str((resource / "data.json").absolute()) in paths assert str((resource / "binary.png").absolute()) not in paths print("✓ test_collect_existing_filters passed") # --------------------------------------------------------------------------- # InitChangesStep # --------------------------------------------------------------------------- def test_clear_and_scan_flow_includes_jsonl_when_configured(): """The legacy clear-and-scan primitives accept configured JSONL inputs.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() write_file(cwd / "daily" / "note.md", "alpha") write_file(cwd / "resource" / "events.jsonl", '{"a": 1}\n') write_file(cwd / "resource" / "ignore.txt", "skip") fs = LocalFileStore(name="test_store", embedding_store="") await fs.start() try: clear_step = ClearStoreStep(file_store=fs, app_context=_make_app_context(cwd)) scan_step = InitChangesStep(store="file_store", file_store=fs, app_context=_make_app_context(cwd)) ctx = RuntimeContext(watch_dirs=["daily_dir", "resource_dir"], watch_suffixes=["md", "jsonl"]) await clear_step(ctx) resp = await scan_step(ctx) paths = {Path(item["path"]).name for item in ctx["changes"]} assert resp.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0} assert paths == {"note.md", "events.jsonl"} finally: await fs.close() print("✓ test_clear_and_scan_defaults_include_jsonl passed") asyncio.run(run()) async def _make_scan_step(workspace_path: Path, watch_dirs=None, watch_suffixes=None, recursive=True): fs = LocalFileStore(name="test_store", embedding_store="") chunker = DefaultFileChunker() await fs.start() await chunker.start() app_ctx = _make_app_context(workspace_path) step = InitChangesStep( store="file_store", recursive=recursive, file_store=fs, file_chunker=chunker, app_context=app_ctx, ) context = RuntimeContext( watch_dirs=watch_dirs or ["daily_dir", "digest_dir"], watch_suffixes=watch_suffixes or ["md"], ) return step, context, fs, chunker async def _teardown(fs: LocalFileStore, chunker: DefaultFileChunker) -> None: await chunker.close() await fs.close() def test_scan_changes_initial_all_added(): """First run on a fresh store emits 'added' for every existing file.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() write_file(cwd / "daily" / "a.md", "alpha") write_file(cwd / "daily" / "b.md", "beta") (cwd / "digest").mkdir(parents=True, exist_ok=True) step, ctx, fs, chunker = await _make_scan_step(cwd) try: resp = await step(ctx) counts = resp.metadata["counts"] assert counts == {"added": 2, "modified": 0, "deleted": 0} assert len(ctx["changes"]) == 2 finally: await _teardown(fs, chunker) print("✓ test_scan_changes_initial_all_added passed") asyncio.run(run()) def test_scan_changes_no_changes(): """Second run over an unchanged store reports zero counts.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() a = write_file(cwd / "daily" / "a.md", "alpha") (cwd / "digest").mkdir(parents=True, exist_ok=True) step, ctx, fs, chunker = await _make_scan_step(cwd) try: node, chunks = await chunker.chunk(a) await fs.upsert([(node, chunks)]) resp = await step(ctx) assert resp.metadata["counts"] == {"added": 0, "modified": 0, "deleted": 0} assert ctx["changes"] == [] finally: await _teardown(fs, chunker) print("✓ test_scan_changes_no_changes passed") asyncio.run(run()) def test_scan_changes_detect_modify_delete(): """Second pass distinguishes added/modified/deleted.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() a = write_file(cwd / "daily" / "a.md", "alpha") b = write_file(cwd / "daily" / "b.md", "beta") (cwd / "digest").mkdir(parents=True, exist_ok=True) step, ctx, fs, chunker = await _make_scan_step(cwd) try: for p in (a, b): node, chunks = await chunker.chunk(p) await fs.upsert([(node, chunks)]) a.write_text("alpha-v2", encoding="utf-8") os.utime(a, (9_999_999_999, 9_999_999_999)) b.unlink() write_file(cwd / "daily" / "c.md", "gamma") resp = await step(ctx) counts = resp.metadata["counts"] assert counts == {"added": 1, "modified": 1, "deleted": 1} finally: await _teardown(fs, chunker) print("✓ test_scan_changes_detect_modify_delete passed") asyncio.run(run()) def test_scan_changes_missing_dir_skipped(): """Non-existent watch_dirs entries are dropped silently.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() (cwd / "daily").mkdir() # digest dir missing step, ctx, fs, chunker = await _make_scan_step(cwd) try: resp = await step(ctx) assert resp.metadata["counts"] == {"added": 0, "modified": 0, "deleted": 0} finally: await _teardown(fs, chunker) print("✓ test_scan_changes_missing_dir_skipped passed") asyncio.run(run()) def test_scan_changes_resource_dir(): """Scanning resource_dir with multiple suffixes works.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() resource = cwd / "resource" write_file(resource / "data.json", "{}") write_file(resource / "note.md", "# Note") write_file(resource / "image.png", "binary") step, ctx, fs, chunker = await _make_scan_step( cwd, watch_dirs=["resource_dir"], watch_suffixes=["md", "json"], ) try: resp = await step(ctx) assert resp.metadata["counts"]["added"] == 2 finally: await _teardown(fs, chunker) print("✓ test_scan_changes_resource_dir passed") asyncio.run(run()) def test_init_changes_named_file_catalog_monitor(): """monitor_type/monitor_name selects the requested file_catalog component.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() write_file(cwd / "resource" / "2026-01-01" / "a.md", "alpha") catalog = LocalFileCatalog(name="resource") await catalog.start() try: app_ctx = _make_app_context(cwd) app_ctx.components = {ComponentEnum.FILE_CATALOG: {"resource": catalog}} step = InitChangesStep(monitor_type="file_catalog", monitor_name="resource", app_context=app_ctx) ctx = RuntimeContext(watch_dirs=["resource_dir"], watch_suffixes=["md"]) resp = await step(ctx) assert resp.metadata["counts"] == {"added": 1, "modified": 0, "deleted": 0} assert ctx["changes"][0]["change"] == "added" finally: await catalog.close() print("✓ test_init_changes_named_file_catalog_monitor passed") asyncio.run(run()) def test_bucket_changes_coalesces_by_final_file_state(): """A delete+add replacement batch for an existing file becomes one modified event.""" with tempfile.TemporaryDirectory() as tmpdir: p = write_file(Path(tmpdir) / "daily" / "a.md", "alpha") buckets = bucket_changes( [ {"change": "deleted", "path": str(p)}, {"change": "added", "path": str(p)}, ], ) assert buckets[Change.modified] == [str(p)] assert buckets[Change.added] == [] assert buckets[Change.deleted] == [] print("✓ test_bucket_changes_coalesces_by_final_file_state passed") def test_update_catalog_relative_path_uses_workspace(): """update_catalog_step resolves workspace-relative change paths against workspace_path.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() write_file(cwd / "daily" / "a.md", "alpha") catalog = LocalFileCatalog(name="test_catalog") await catalog.start() try: step = UpdateCatalogStep(file_catalog=catalog, app_context=_make_app_context(cwd)) ctx = RuntimeContext(changes=[{"change": "added", "path": "daily/a.md"}]) resp = await step(ctx) assert resp.success is True nodes = await catalog.get_nodes() assert [n.path for n in nodes] == ["daily/a.md"] finally: await catalog.close() print("✓ test_update_catalog_relative_path_uses_workspace passed") asyncio.run(run()) def test_update_catalog_upserts_in_batches_of_at_most_100_files(): """Large change sets are committed incrementally instead of retained as one list.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index:03d}.md", "x") for index in range(205)] catalog = LocalFileCatalog(name="test_catalog") await catalog.start() batch_sizes = [] original_upsert = catalog.upsert async def record_upsert(nodes): batch_sizes.append(len(nodes)) await original_upsert(nodes) try: step = UpdateCatalogStep(file_catalog=catalog, persist=False, app_context=_make_app_context(cwd)) changes = [{"change": "added", "path": str(path)} for path in paths] available = MagicMock(available=8 * 1024 * 1024 * 1024) with ( patch.object(catalog, "upsert", side_effect=record_upsert), patch("reme.steps.index.update_changes.psutil.virtual_memory", return_value=available), ): response = await step(RuntimeContext(changes=changes)) assert response.success is True assert batch_sizes == [100, 100, 5] assert len(await catalog.get_nodes()) == 205 finally: await catalog.close() asyncio.run(run()) def test_update_catalog_deletes_in_batches_of_at_most_100_paths(): """Large delete sets use the same bounded failure domain as upserts.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() deleted = [cwd / "daily" / f"{index:03d}.md" for index in range(205)] catalog = LocalFileCatalog(name="test_catalog") await catalog.start() batch_sizes = [] original_delete = catalog.delete async def record_delete(paths): batch_sizes.append(len(paths)) await original_delete(paths) try: step = UpdateCatalogStep(file_catalog=catalog, persist=False, app_context=_make_app_context(cwd)) changes = [{"change": "deleted", "path": str(path)} for path in deleted] with patch.object(catalog, "delete", side_effect=record_delete): response = await step(RuntimeContext(changes=changes)) assert response.success is True assert batch_sizes == [100, 100, 5] assert len(response.answer) == 205 finally: await catalog.close() asyncio.run(run()) def test_update_index_memory_budget_can_reduce_batches_to_one_file(): """A file larger than the target batch budget is still processed alone.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index}.md", f"# Note {index}\nbody\n") for index in range(3)] fs = LocalFileStore(name="default", embedding_store="") chunker = DefaultFileChunker(supported_extensions=["md"]) await fs.start() await chunker.start() batch_sizes = [] original_upsert = fs.upsert async def record_upsert(items): batch_sizes.append(len(items)) await original_upsert(items) 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) changes = [{"change": "added", "path": str(path)} for path in paths] available = MagicMock(available=1_000) with ( patch.object(fs, "upsert", side_effect=record_upsert), patch("reme.steps.index.update_changes.psutil.virtual_memory", return_value=available), ): response = await step(RuntimeContext(changes=changes)) assert response.success is True assert batch_sizes == [1, 1, 1] assert len(await fs.get_nodes()) == 3 finally: await chunker.close() await fs.close() asyncio.run(run()) def test_update_catalog_memory_target_limits_cumulative_batch_size(): """The memory target applies to the accumulated batch, not only individual files.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index}.md", "x") for index in range(3)] catalog = LocalFileCatalog(name="test_catalog") await catalog.start() batch_sizes = [] original_upsert = catalog.upsert async def record_upsert(nodes): batch_sizes.append(len(nodes)) await original_upsert(nodes) try: step = UpdateCatalogStep( file_catalog=catalog, persist=False, batch_memory_target_bytes=64 * 1024, app_context=_make_app_context(cwd), ) changes = [{"change": "added", "path": str(path)} for path in paths] available = MagicMock(available=8 * 1024 * 1024 * 1024) with ( patch.object(catalog, "upsert", side_effect=record_upsert), patch("reme.steps.index.update_changes.psutil.virtual_memory", return_value=available), ): response = await step(RuntimeContext(changes=changes)) assert response.success is True assert batch_sizes == [2, 1] finally: await catalog.close() asyncio.run(run()) @pytest.mark.parametrize( ("kwargs", "message"), [ ({"batch_available_memory_ratio": float("nan")}, "batch_available_memory_ratio"), ({"batch_memory_target_bytes": 0}, "batch_memory_target_bytes"), ({"batch_memory_expansion_factor": float("inf")}, "batch_memory_expansion_factor"), ({"file_memory_overhead_bytes": -1}, "file_memory_overhead_bytes"), ({"chunk_memory_overhead_bytes": -1}, "chunk_memory_overhead_bytes"), ({"estimated_chunk_bytes": 0}, "estimated_chunk_bytes"), ({"float16_bytes": 0}, "float16_bytes"), ], ) def test_update_step_rejects_invalid_batch_memory_settings(kwargs, message): """Invalid batch-memory settings fail during step construction.""" with pytest.raises(ValueError, match=message): UpdateCatalogStep(**kwargs) def test_update_step_accepts_configured_batch_memory_settings(): """Step configuration can override all batching and memory-estimation defaults.""" step = UpdateCatalogStep( batch_max_files=25, batch_available_memory_ratio=0.25, batch_memory_target_bytes=1024 * 1024, batch_memory_expansion_factor=4.5, file_memory_overhead_bytes=1024, chunk_memory_overhead_bytes=512, estimated_chunk_bytes=5000, float16_bytes=4, ) assert step.batch_max_files == 25 assert step.batch_available_memory_ratio == 0.25 assert step.batch_memory_target_bytes == 1024 * 1024 assert step.batch_memory_expansion_factor == 4.5 assert step.file_memory_overhead_bytes == 1024 assert step.chunk_memory_overhead_bytes == 512 assert step.estimated_chunk_bytes == 5000 assert step.float16_bytes == 4 @pytest.mark.parametrize("estimate_method", ["estimate_source_memory", "estimate_item_memory"]) def test_update_catalog_memory_estimate_failure_processes_each_file_alone(estimate_method): """Advisory estimate failures isolate files without skipping their updates.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index}.md", "x") for index in range(2)] catalog = LocalFileCatalog(name="test_catalog") await catalog.start() batch_sizes = [] original_upsert = catalog.upsert async def record_upsert(nodes): batch_sizes.append(len(nodes)) await original_upsert(nodes) try: step = UpdateCatalogStep(file_catalog=catalog, persist=False, app_context=_make_app_context(cwd)) changes = [{"change": "added", "path": str(path)} for path in paths] with ( patch.object(step, estimate_method, side_effect=RuntimeError("estimate failed")), patch.object(catalog, "upsert", side_effect=record_upsert), ): response = await step(RuntimeContext(changes=changes)) assert response.success is True assert batch_sizes == [1, 1] assert len(await catalog.get_nodes()) == 2 finally: await catalog.close() asyncio.run(run()) def test_update_catalog_releases_flushed_item_before_building_next_file(): """A one-file batch does not retain its payload while the next payload is built.""" class TrackedItem: """Weak-referenceable payload used to observe the local item lifetime.""" class TrackingCatalogStep(UpdateCatalogStep): """Catalog step that records whether the prior payload was released.""" def __init__(self, **kwargs): super().__init__(**kwargs) self.previous_item = None self.release_observations = [] async def build_item(self, path): """Build a payload and inspect the preceding payload reference.""" del path if self.previous_item is not None: self.release_observations.append(self.previous_item() is None) item = TrackedItem() self.previous_item = weakref.ref(item) return item async def upsert_items(self, items): """Discard the batch so only the apply loop could retain its payload.""" del items async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index}.md", "x") for index in range(2)] step = TrackingCatalogStep( persist=False, batch_max_files=1, app_context=_make_app_context(cwd), ) changes = [{"change": "added", "path": str(path)} for path in paths] response = await step(RuntimeContext(changes=changes)) assert response.success is True assert step.release_observations == [True] asyncio.run(run()) def test_update_catalog_continues_after_one_batch_fails(): """An upsert failure is isolated to its bounded batch and later files continue.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() paths = [write_file(cwd / "daily" / f"{index}.md", "x") for index in range(3)] catalog = LocalFileCatalog(name="test_catalog") await catalog.start() original_upsert = catalog.upsert call_count = 0 async def fail_first_batch(nodes): nonlocal call_count call_count += 1 if call_count == 1: raise RuntimeError("batch failed") await original_upsert(nodes) try: step = UpdateCatalogStep( file_catalog=catalog, persist=False, batch_max_files=2, app_context=_make_app_context(cwd), ) changes = [{"change": "added", "path": str(path)} for path in paths] available = MagicMock(available=8 * 1024 * 1024 * 1024) with ( patch.object(catalog, "upsert", side_effect=fail_first_batch), patch("reme.steps.index.update_changes.psutil.virtual_memory", return_value=available), ): response = await step(RuntimeContext(changes=changes)) assert response.success is False assert [result["success"] for result in response.answer] == [False, False, True] assert [node.path for node in await catalog.get_nodes()] == ["daily/2.md"] finally: await catalog.close() asyncio.run(run()) def test_update_catalog_yields_to_event_loop_while_building_batch(): """Synchronous file inspection does not monopolize the application event loop.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() source = write_file(cwd / "daily" / "a.md", "alpha") catalog = LocalFileCatalog(name="test_catalog") await catalog.start() yielded = False observed = [] original_upsert = catalog.upsert def mark_yielded(): nonlocal yielded yielded = True async def observe_upsert(nodes): observed.append(yielded) await original_upsert(nodes) try: step = UpdateCatalogStep(file_catalog=catalog, persist=False, app_context=_make_app_context(cwd)) asyncio.get_running_loop().call_soon(mark_yielded) with patch.object(catalog, "upsert", side_effect=observe_upsert): response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}])) assert response.success is True assert observed == [True] finally: await catalog.close() asyncio.run(run()) class _CountingEmbeddingStore: dimensions = 2 max_batch_size = 10 def __init__(self): self.calls = 0 async def get_node_embeddings(self, nodes, **_kwargs): """Record one embedding call and populate deterministic vectors.""" self.calls += 1 for node in nodes: node.embedding = np.array([1.0, 0.0], dtype=np.float16) return nodes def test_update_index_modified_file_reuses_unchanged_embedding(): """The step relies on replace-aware upsert instead of deleting reusable chunks first.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() source = write_file(cwd / "daily" / "a.md", "alpha") fs = LocalFileStore(name="default", embedding_store="") chunker = DefaultFileChunker(supported_extensions=["md"]) await fs.start() await chunker.start() embedding_store = _CountingEmbeddingStore() fs.embedding_store = embedding_store try: app_ctx = _make_app_context(cwd) app_ctx.components = {ComponentEnum.FILE_CHUNKER: {"default": chunker}} await fs.upsert([await chunker.chunk(source)]) assert embedding_store.calls == 1 step = UpdateIndexStep(file_store=fs, persist=False, app_context=app_ctx) with patch.object(fs, "delete", wraps=fs.delete) as delete_mock: response = await step( RuntimeContext(changes=[{"change": "modified", "path": str(source)}]), ) assert response.success is True assert embedding_store.calls == 1 delete_mock.assert_not_awaited() finally: await chunker.close() await fs.close() 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(persist=False, app_context=_make_app_context(Path.cwd())) with ( patch.object(Path, "is_file", return_value=True), patch.object(Path, "stat", side_effect=FileNotFoundError("file disappeared")), ): response = await step(RuntimeContext(changes=[{"change": "modified", "path": "daily/a.md"}])) assert response.success is False assert response.answer == [ { "change": "modified", "path": "daily/a.md", "success": False, "error": "file disappeared", }, ] asyncio.run(run()) def test_update_index_reports_memory_estimation_failure_without_aborting(): """A missing chunker is isolated to the affected file and returned as a normal failure.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() source = write_file(cwd / "daily" / "a.unknown", "alpha") app_ctx = _make_app_context(cwd) app_ctx.components = {ComponentEnum.FILE_CHUNKER: {}} step = UpdateIndexStep(persist=False, app_context=app_ctx) response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}])) assert response.success is False assert response.answer == [ { "change": "added", "path": str(source), "success": False, "error": ( f"No file chunker supports {source} (suffix='unknown') and no default chunker is configured" ), }, ] 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.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() daily_a = write_file(cwd / "daily" / "a.md", "alpha\n[[digest/report.md]]\n") write_file(cwd / "digest" / "report.md", "# Report\nbeta\n") write_file(cwd / "daily" / "ignore.txt", "skip") 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_STORE: {"default": fs}, ComponentEnum.FILE_CHUNKER: {"default": chunker}, } ctx = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"]) init_step = InitChangesStep( monitor_type="file_store", monitor_name="default", dispatch_steps=["update_index_step"], app_context=app_ctx, ) first = await init_step(ctx) assert first.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0} nodes = {n.path: n for n in await fs.get_nodes()} assert set(nodes) == {"daily/a.md", "digest/report.md"} assert all(nodes[p].chunk_ids for p in nodes) daily_a.write_text("alpha v2\n[[digest/report.md]]\n", encoding="utf-8") os.utime(daily_a, (9_999_999_999, 9_999_999_999)) (cwd / "digest" / "report.md").unlink() write_file(cwd / "daily" / "c.md", "gamma\n") second = await init_step(ctx) assert second.metadata["counts"] == {"added": 1, "modified": 1, "deleted": 1} assert {(c["change"], Path(c["path"]).name) for c in ctx["changes"]} == { ("modified", "a.md"), ("deleted", "report.md"), ("added", "c.md"), } nodes = {n.path: n for n in await fs.get_nodes()} assert set(nodes) == {"daily/a.md", "daily/c.md"} assert all(nodes[p].chunk_ids for p in nodes) assert all(chunk.path in nodes for chunk in fs.file_chunks.values()) finally: await chunker.close() await fs.close() print("✓ test_index_update_loop_init_dispatch_updates_store_across_batches passed") asyncio.run(run()) def test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs(): """digest_watch_loop style config updates the digest catalog without touching resource/default catalogs.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() daily = write_file(cwd / "daily" / "2026-01-01.md", "day one") digest = write_file(cwd / "digest" / "week.md", "weekly") write_file(cwd / "resource" / "asset.md", "not watched by digest loop") digest_catalog = LocalFileCatalog(name="digest") resource_catalog = LocalFileCatalog(name="resource") await digest_catalog.start() await resource_catalog.start() try: app_ctx = _make_app_context(cwd) app_ctx.components = { ComponentEnum.FILE_CATALOG: { "digest": digest_catalog, "resource": resource_catalog, }, } ctx = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"]) init_step = InitChangesStep( monitor_type="file_catalog", monitor_name="digest", dispatch_steps=[ {"backend": "update_catalog_step", "file_catalog": "digest"}, {"backend": "log_changes_step"}, ], app_context=app_ctx, ) first = await init_step(ctx) assert first.metadata["counts"] == {"added": 2, "modified": 0, "deleted": 0} assert {n.path for n in await digest_catalog.get_nodes()} == { "daily/2026-01-01.md", "digest/week.md", } assert await resource_catalog.get_nodes() == [] daily.write_text("day two", encoding="utf-8") os.utime(daily, (9_999_999_999, 9_999_999_999)) digest.unlink() write_file(cwd / "daily" / "2026-01-02.md", "next day") second = await init_step(ctx) assert second.metadata["counts"] == {"added": 1, "modified": 1, "deleted": 1} assert {(c["change"], Path(c["path"]).name) for c in ctx["changes"]} == { ("modified", "2026-01-01.md"), ("deleted", "week.md"), ("added", "2026-01-02.md"), } assert {n.path for n in await digest_catalog.get_nodes()} == { "daily/2026-01-01.md", "daily/2026-01-02.md", } assert await resource_catalog.get_nodes() == [] finally: await resource_catalog.close() await digest_catalog.close() print("✓ test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs passed") asyncio.run(run()) # --------------------------------------------------------------------------- # WatchChangesStep # --------------------------------------------------------------------------- def test_watch_changes_default_low_power_timing(): """Default watcher timing favors lower resource use.""" step = WatchChangesStep() assert step.debounce == DEFAULT_WATCH_DEBOUNCE_MS assert step.step == DEFAULT_WATCH_STEP_MS assert step.poll_delay_ms == DEFAULT_LOW_POWER_POLL_MS custom = WatchChangesStep(debounce=1000, step=250, poll_delay_ms=3000) assert custom.debounce == 1000 assert custom.step == 250 assert custom.poll_delay_ms == 3000 print("✓ test_watch_changes_default_low_power_timing passed") def test_watch_changes_requires_stop_event(): """Missing stop_event in context raises a clear error.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() (cwd / "daily").mkdir() app_ctx = _make_app_context(cwd) step = WatchChangesStep(app_context=app_ctx) step.context = RuntimeContext(watch_dirs=["daily_dir"], watch_suffixes=["md"]) try: await step.execute() except RuntimeError as e: assert "stop_event" in str(e) else: raise AssertionError("expected RuntimeError") print("✓ test_watch_changes_requires_stop_event passed") asyncio.run(run()) def test_watch_changes_raises_no_valid_paths(): """With no valid watch_paths, the step raises.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): cwd = Path.cwd() app_ctx = _make_app_context(cwd) step = WatchChangesStep(app_context=app_ctx) stop = asyncio.Event() step.context = RuntimeContext(stop_event=stop, watch_dirs=["daily_dir"], watch_suffixes=["md"]) try: await step.execute() except RuntimeError as e: assert "No valid watch paths" in str(e) else: raise AssertionError("expected RuntimeError") print("✓ test_watch_changes_raises_no_valid_paths passed") asyncio.run(run()) def test_watch_changes_filter_matches_rules(): """The internal filter uses watch rules from context.""" with tempfile.TemporaryDirectory() as tmpdir: workspace = Path(tmpdir) (workspace / "daily").mkdir() (workspace / "digest").mkdir() (workspace / "resource").mkdir() app_ctx = _make_app_context(workspace) step = WatchChangesStep(app_context=app_ctx) step.context = RuntimeContext(watch_dirs=["daily_dir", "digest_dir"], watch_suffixes=["md"]) step._rules = step._get_watch_rules() assert step._filter(Change.added, str(workspace / "daily/foo.md")) assert step._filter(Change.added, str(workspace / "digest/bar.md")) assert not step._filter(Change.added, str(workspace / "daily/foo.txt")) assert not step._filter(Change.added, str(workspace / "resource/file.md")) print("✓ test_watch_changes_filter_matches_rules passed") def test_watch_changes_dispatch_steps_list(): """dispatch_steps config is stored by BaseStep.""" step = WatchChangesStep(dispatch_steps=["update_catalog_step", "auto_resource_step"]) assert step.dispatch_step_specs == ["update_catalog_step", "auto_resource_step"] print("✓ test_watch_changes_dispatch_steps_list passed") def test_auto_resource_batch_deleted_changes(): """AutoResourceStep accepts a batch of change dicts from dispatch_steps.""" 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: filename = "file.md" note_stem = _compute_note_stem(filename) note_path = cwd / "daily" / "2026-01-01" / f"{note_stem}.md" write_file( note_path, "---\nname: test\nsource_resource: '[[resource/2026-01-01/file.md]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs) ctx = RuntimeContext( changes=[ {"change": "deleted", "path": str(cwd / "resource" / "2026-01-01" / filename)}, ], ) resp = await step(ctx) assert resp.success is True assert resp.answer == "Deleted resource note: daily/2026-01-01/file.md" assert resp.metadata["processed"] == 1 assert resp.metadata["results"][0]["path"] == "resource/2026-01-01/file.md" assert not note_path.exists() finally: await fs.close() print("✓ test_auto_resource_batch_deleted_changes passed") 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.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): workspace = Path.cwd() app_ctx = _make_app_context(workspace) source = write_file(workspace / "resource" / "report.txt", "hello") today = datetime.datetime.now().strftime("%Y-%m-%d") captured = {} step = AutoResourceStep(app_context=app_ctx) async def fake_upsert(file_path, date_str, note_stem, created): captured.update( {"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created}, ) step.context.response.success = True step.context.response.answer = "ok" step._handle_upsert = fake_upsert resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}])) assert resp.success is True assert source.read_text(encoding="utf-8") == "hello" assert not (workspace / "resource" / today / "report.txt").exists() assert captured == { "file_path": "resource/report.txt", "date_str": today, "note_stem": "report", "created": True, } print("✓ test_auto_resource_accepts_loose_root_resource passed") asyncio.run(run()) def test_auto_resource_loose_root_resource_keeps_existing_dated_resource(): """Loose-resource compatibility does not overwrite dated resource files.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): workspace = Path.cwd() app_ctx = _make_app_context(workspace) today = datetime.datetime.now().strftime("%Y-%m-%d") write_file(workspace / "resource" / today / "report.txt", "existing") source = write_file(workspace / "resource" / "report.txt", "new") captured = {} step = AutoResourceStep(app_context=app_ctx) async def fake_upsert(file_path, date_str, note_stem, created): captured.update( {"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created}, ) step.context.response.success = True step.context.response.answer = "ok" step._handle_upsert = fake_upsert resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}])) assert resp.success is True assert (workspace / "resource" / today / "report.txt").read_text(encoding="utf-8") == "existing" assert source.read_text(encoding="utf-8") == "new" assert captured == { "file_path": "resource/report.txt", "date_str": today, "note_stem": "report", "created": True, } print("✓ test_auto_resource_loose_root_resource_keeps_existing_dated_resource passed") asyncio.run(run()) def test_auto_resource_modified_missing_note_uses_create_tools(): """A modified resource with no daily note is treated as a create.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "hello") wrapper.on_reply = lambda *_: write_file( cwd / "daily" / "2026-01-01" / "report.md", "---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) assert resp.success is True assert resp.metadata["results"][0]["metadata"]["created"] is True assert resp.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/resource-summary.md" assert wrapper.kwargs["job_tools"] == ["write"] assert "The target file does not exist" in wrapper.inputs assert "Target note path: daily/2026-01-01/report.md" in wrapper.inputs assert not (cwd / "daily" / "2026-01-01" / "report.md").exists() assert (cwd / "daily" / "2026-01-01" / "resource-summary.md").exists() finally: await fs.close() print("✓ test_auto_resource_modified_missing_note_uses_create_tools passed") asyncio.run(run()) def test_auto_resource_sanitizes_invalid_generated_name(): """Invalid LLM-suggested names are sanitized before renaming.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "hello") wrapper.on_reply = lambda *_: write_file( cwd / "daily" / "2026-01-01" / "report.md", "---\nname: bad/name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]), ) assert resp.success is True assert resp.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/bad-name.md" assert not (cwd / "daily" / "2026-01-01" / "report.md").exists() assert "name: bad-name" in (cwd / "daily" / "2026-01-01" / "bad-name.md").read_text(encoding="utf-8") finally: await fs.close() print("✓ test_auto_resource_sanitizes_invalid_generated_name passed") asyncio.run(run()) def test_auto_resource_uniquifies_conflicting_generated_name(): """Conflicting LLM-suggested names get a stable source-derived suffix.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "hello") write_file( cwd / "daily" / "2026-01-01" / "resource-summary.md", "---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/other.txt]]'\n---\nother\n", ) wrapper.on_reply = lambda *_: write_file( cwd / "daily" / "2026-01-01" / "report.md", "---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]), ) path = resp.metadata["results"][0]["metadata"]["path"] assert resp.success is True assert path.startswith("daily/2026-01-01/resource-summary--") assert path.endswith(".md") assert (cwd / path).exists() assert "name: resource-summary--" in (cwd / path).read_text(encoding="utf-8") assert (cwd / "daily" / "2026-01-01" / "resource-summary.md").exists() assert not (cwd / "daily" / "2026-01-01" / "report.md").exists() finally: await fs.close() print("✓ test_auto_resource_uniquifies_conflicting_generated_name passed") asyncio.run(run()) def test_auto_resource_update_finds_renamed_note_by_source_resource(): """A modified resource updates the renamed note linked by source_resource.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "hello v2") write_file( cwd / "daily" / "2026-01-01" / "generated-name.md", "---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nold body\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) assert resp.success is True assert resp.metadata["results"][0]["metadata"]["created"] is False assert resp.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/generated-name.md" assert wrapper.kwargs["job_tools"] == ["read", "edit", "frontmatter_update", "write"] assert "Target note path: daily/2026-01-01/generated-name.md" in wrapper.inputs finally: await fs.close() print("✓ test_auto_resource_update_finds_renamed_note_by_source_resource passed") asyncio.run(run()) def test_auto_resource_update_keeps_existing_renamed_path(): """Updates do not rename an already-renamed note from a fresh LLM suggestion.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "hello v2") write_file( cwd / "daily" / "2026-01-01" / "generated-name.md", "---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nold body\n", ) def rewrite_frontmatter(*_args): write_file( cwd / "daily" / "2026-01-01" / "generated-name.md", "---\nname: new-suggestion\nsource_resource: " "'[[resource/2026-01-01/report.txt]]'\n---\nnew body\n", ) wrapper.on_reply = rewrite_frontmatter step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) path = cwd / "daily" / "2026-01-01" / "generated-name.md" assert resp.success is True assert resp.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/generated-name.md" assert path.exists() text = path.read_text(encoding="utf-8") assert "name: generated-name" in text assert not (cwd / "daily" / "2026-01-01" / "new-suggestion.md").exists() finally: await fs.close() print("✓ test_auto_resource_update_keeps_existing_renamed_path passed") asyncio.run(run()) def test_auto_resource_reports_unmodified_when_agent_skips_existing_note(): """A modified event that leaves the note bytes unchanged reports modified=False.""" 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: write_file(cwd / "resource" / "2026-01-01" / "report.txt", "same") write_file( cwd / "daily" / "2026-01-01" / "report.md", "---\nname: report\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]), ) result_meta = resp.metadata["results"][0]["metadata"] assert resp.success is True assert result_meta["modified"] is False assert resp.metadata["modified"] is False finally: await fs.close() print("✓ test_auto_resource_reports_unmodified_when_agent_skips_existing_note passed") asyncio.run(run()) def test_auto_resource_deletes_loose_root_resource_note_for_today(): """Deleting a loose root resource deletes today's same-stem note.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): workspace = Path.cwd() app_ctx = _make_app_context(workspace) fs = LocalFileStore(name="test_store", embedding_store="") await fs.start() _install_file_jobs(app_ctx, fs) try: today = datetime.datetime.now().strftime("%Y-%m-%d") note_path = write_file(workspace / "daily" / today / "report.md", "---\nname: report\n---\nbody\n") step = AutoResourceStep(app_context=app_ctx, file_store=fs) resp = await step(RuntimeContext(changes=[{"change": "deleted", "path": "resource/report.txt"}])) assert resp.success is True assert resp.metadata["results"][0]["path"] == "resource/report.txt" assert resp.metadata["results"][0]["metadata"]["modified"] is True assert resp.metadata["modified"] is True assert not note_path.exists() finally: await fs.close() print("✓ test_auto_resource_deletes_loose_root_resource_note_for_today passed") asyncio.run(run()) def test_auto_resource_deletes_renamed_note_by_source_resource(): """Deleting a resource removes the note even after frontmatter-name rename.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir): workspace = Path.cwd() app_ctx = _make_app_context(workspace) fs = LocalFileStore(name="test_store", embedding_store="") await fs.start() _install_file_jobs(app_ctx, fs) try: note_path = write_file( workspace / "daily" / "2026-01-01" / "generated-name.md", "---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n", ) step = AutoResourceStep(app_context=app_ctx, file_store=fs) resp = await step( RuntimeContext(changes=[{"change": "deleted", "path": "resource/2026-01-01/report.txt"}]), ) assert resp.success is True assert resp.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/generated-name.md" assert not note_path.exists() finally: await fs.close() print("✓ test_auto_resource_deletes_renamed_note_by_source_resource passed") asyncio.run(run()) def test_auto_memory_reports_modified_for_create_and_false_for_skip(): """AutoMemoryStep reports whether a daily note actually changed.""" 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: today = datetime.datetime.now().strftime("%Y-%m-%d") wrapper.on_reply = lambda *_: write_file( cwd / "daily" / today / "memory.md", "---\nname: memory\nsession_id: s1\n" "source_conversation: '[[session/dialog/s1.jsonl]]'\n---\nbody\n", ) step = AutoMemoryStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext( messages=[{"name": "user", "role": "user", "content": "remember project detail"}], session_id="s1", ), ) resp = resp or step.context.response assert resp.success is True assert resp.metadata["created"] is True assert resp.metadata["modified"] is True wrapper.on_reply = None resp = await step(RuntimeContext(messages=[], session_id="s2")) resp = resp or step.context.response assert resp.success is True assert resp.metadata["modified"] is False assert resp.metadata["n_messages"] == 0 finally: await fs.close() print("✓ test_auto_memory_reports_modified_for_create_and_false_for_skip passed") asyncio.run(run()) def test_auto_memory_uses_message_day_for_historical_create(): """AutoMemoryStep creates historical daily notes from message timestamps.""" 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: historical_day = "2023-01-19" def write_historical_note(inputs, _kwargs): assert f"Today: {historical_day}" in inputs assert f"date={historical_day}" in inputs assert "injected_job_kwargs" not in _kwargs write_file( cwd / "daily" / historical_day / "memory.md", "---\nname: memory\nsession_id: s1\n" "source_conversation: '[[session/dialog/s1.jsonl]]'\n---\nbody\n", ) wrapper.on_reply = write_historical_note step = AutoMemoryStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext( messages=[ { "name": "user", "role": "user", "content": "remember historical project detail", "created_at": "2023-01-19T08:00:00", }, ], session_id="s1", ), ) resp = resp or step.context.response assert resp.success is True assert resp.metadata["date"] == historical_day assert resp.metadata["path"] == "daily/2023-01-19/memory.md" assert (cwd / "daily" / historical_day / "memory.md").is_file() finally: await fs.close() print("✓ test_auto_memory_uses_message_day_for_historical_create passed") asyncio.run(run()) def test_auto_memory_rejects_invalid_explicit_date_before_saving_session(): """Invalid explicit dates fail before writing session history.""" 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: step = AutoMemoryStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper) resp = await step( RuntimeContext( messages=[ { "name": "user", "role": "user", "content": "remember historical project detail", "created_at": "2023-01-19T08:00:00", }, ], session_id="s1", date="2023-01-19T08:00:00", ), ) resp = resp or step.context.response assert resp.success is False assert resp.answer == "Error: date must be YYYY-MM-DD" assert resp.metadata == { "date": "2023-01-19T08:00:00", "modified": False, "n_messages": 1, } assert not (cwd / "session" / "dialog" / "s1.jsonl").exists() assert not (cwd / "daily").exists() assert wrapper.inputs == "" finally: await fs.close() print("✓ test_auto_memory_rejects_invalid_explicit_date_before_saving_session passed") asyncio.run(run()) def test_auto_resource_result_hook_is_optional_and_isolated(): """AutoResourceStep optionally emits host result hooks without coupling.""" async def run(): with tempfile.TemporaryDirectory() as tmpdir: app_ctx = _make_app_context(Path(tmpdir)) step = AutoResourceStep(app_context=app_ctx) step.context = RuntimeContext() changes = [{"change": "added", "path": "resource/2026-01-01/file.md"}] results = [{"success": True, "path": "resource/2026-01-01/file.md"}] app_ctx.metadata = {} await step._emit_result_hook(changes=changes, results=results) calls = [] async def hook(**kwargs): calls.append(kwargs) app_ctx.metadata = {"qwenpaw_memory_result_hook": hook} step.context.response.metadata["modified"] = False await step._emit_result_hook(changes=changes, results=results) assert not calls step.context.response.metadata["modified"] = True await step._emit_result_hook(changes=changes, results=results) assert len(calls) == 1 assert calls[0]["job_name"] == "auto_resource" assert calls[0]["response"] is step.context.response assert calls[0]["kwargs"] == {"changes": changes} assert calls[0]["metadata"] == {"results": results} def failing_hook(**_kwargs): raise RuntimeError("boom") app_ctx.metadata = {"qwenpaw_memory_result_hook": failing_hook} await step._emit_result_hook(changes=changes, results=results) print("✓ test_auto_resource_result_hook_is_optional_and_isolated passed") asyncio.run(run()) # LogChangesStep # --------------------------------------------------------------------------- def test_log_changes_step(): """LogChangesStep logs and reports count.""" async def run(): step = LogChangesStep() changes = [ {"change": "added", "path": "/workspace/daily/note.md"}, {"change": "deleted", "path": "/workspace/daily/old.md"}, ] ctx = RuntimeContext(changes=changes) resp = await step(ctx) assert resp.success is True assert resp.metadata["count"] == 2 print("✓ test_log_changes_step passed") asyncio.run(run()) if __name__ == "__main__": print("\n=== Background Steps Tests ===") # _watch_rules test_build_watch_rules_basic() test_build_watch_rules_multiple_suffixes() test_build_watch_rules_fallback_literal() test_match_file_suffix() test_match_file_no_suffix_filter() test_collect_existing_filters() # InitChangesStep test_clear_and_scan_flow_includes_jsonl_when_configured() test_scan_changes_initial_all_added() test_scan_changes_no_changes() test_scan_changes_detect_modify_delete() test_scan_changes_missing_dir_skipped() test_scan_changes_resource_dir() test_index_update_loop_init_dispatch_updates_store_across_batches() test_digest_watch_loop_init_dispatch_updates_named_catalog_and_logs() # WatchChangesStep test_watch_changes_default_low_power_timing() test_watch_changes_requires_stop_event() test_watch_changes_raises_no_valid_paths() test_watch_changes_filter_matches_rules() test_watch_changes_dispatch_steps_list() test_auto_resource_batch_deleted_changes() test_auto_resource_accepts_loose_root_resource() test_auto_resource_loose_root_resource_keeps_existing_dated_resource() test_auto_resource_modified_missing_note_uses_create_tools() test_auto_resource_sanitizes_invalid_generated_name() test_auto_resource_uniquifies_conflicting_generated_name() test_auto_resource_update_finds_renamed_note_by_source_resource() test_auto_resource_update_keeps_existing_renamed_path() test_auto_memory_uses_message_day_for_historical_create() test_auto_memory_rejects_invalid_explicit_date_before_saving_session() test_auto_resource_deletes_loose_root_resource_note_for_today() test_auto_resource_deletes_renamed_note_by_source_resource() test_auto_resource_result_hook_is_optional_and_isolated() # LogChangesStep test_log_changes_step() print("\n所有测试通过!")