a6491df5a6
* refactor(memory): lazy-init embedder env reads * feat(memory): reindex log to .rag/reindex.log * refactor(memory): search handles None embedder * test(memory): cover None embedder in search and index * docs(handoff): add handoff and ADR for lazy-init embedder * docs(handoff): set PR number * docs(handoff): rename to pr-100 prefix for pipeline detection * refactor(memory): split run_search to satisfy xenon rank A * fix(memory): cast search result fields to satisfy mypy * style(memory): ruff format search.py --------- Co-authored-by: opencode-agent <agent@opencode.local>
390 lines
14 KiB
Python
390 lines
14 KiB
Python
import json
|
|
import os
|
|
from argparse import Namespace
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
os.environ.setdefault("OPENAI_BASE_URL", "http://test/v1")
|
|
|
|
from src.memory.index import _atomic_write, _content_hash, _extract_text, run_index
|
|
|
|
|
|
class TestExtractText:
|
|
def test_with_frontmatter(self) -> None:
|
|
md = "---\ntitle: test\n---\n\nbody content"
|
|
assert _extract_text(md) == "body content"
|
|
|
|
def test_no_frontmatter(self) -> None:
|
|
md = "just content"
|
|
assert _extract_text(md) == "just content"
|
|
|
|
def test_empty(self) -> None:
|
|
assert _extract_text("") == ""
|
|
|
|
|
|
class TestRunIndex:
|
|
def test_index_creates_json(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
|
|
(memory_dir / "test.md").write_text("---\ntitle: test\n---\n\nhello world")
|
|
|
|
def fake_embed_texts(texts):
|
|
return [[0.1, 0.2, 0.3]] * len(texts)
|
|
|
|
with patch("src.memory.index.embed_texts", fake_embed_texts):
|
|
args = Namespace(memory_dir=str(memory_dir), output=str(index_dir))
|
|
run_index(args)
|
|
|
|
assert (index_dir / "index.json").exists()
|
|
index = json.loads((index_dir / "index.json").read_text())
|
|
assert len(index["files"]) == 1
|
|
assert index["files"][0]["source"] == "test.md"
|
|
assert index["files"][0]["embedding"] == [0.1, 0.2, 0.3]
|
|
assert index["files"][0]["chunk_idx"] == 0
|
|
assert index["files"][0]["offset"] == 0
|
|
|
|
def test_no_md_files(self, tmp_path: Path, capsys) -> None:
|
|
memory_dir = tmp_path / "empty"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
|
|
args = Namespace(memory_dir=str(memory_dir), output=str(index_dir))
|
|
run_index(args)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "No .md files found" in captured.out
|
|
|
|
def test_chunking_long_file(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
|
|
long_body = "a" * 1500
|
|
(memory_dir / "long.md").write_text(long_body)
|
|
|
|
def fake_embed_texts(texts):
|
|
return [[0.1, 0.2, 0.3]] * len(texts)
|
|
|
|
with patch("src.memory.index.embed_texts", fake_embed_texts):
|
|
args = Namespace(memory_dir=str(memory_dir), output=str(index_dir))
|
|
run_index(args)
|
|
|
|
index = json.loads((index_dir / "index.json").read_text())
|
|
assert len(index["files"]) >= 3
|
|
chunk_idxs = [f["chunk_idx"] for f in index["files"]]
|
|
assert chunk_idxs == sorted(chunk_idxs)
|
|
assert chunk_idxs[0] == 0
|
|
assert index["files"][0]["offset"] == 0
|
|
offsets = [f["offset"] for f in index["files"]]
|
|
assert offsets == sorted(offsets)
|
|
|
|
def test_chunking_size_env_override(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
|
|
monkeypatch.setenv("MEMORY_CHUNK_SIZE", "256")
|
|
monkeypatch.setenv("MEMORY_CHUNK_OVERLAP", "0")
|
|
(memory_dir / "f.md").write_text("a" * 600)
|
|
|
|
def fake_embed_texts(texts):
|
|
return [[0.1]] * len(texts)
|
|
|
|
with patch("src.memory.index.embed_texts", fake_embed_texts):
|
|
args = Namespace(memory_dir=str(memory_dir), output=str(index_dir))
|
|
run_index(args)
|
|
|
|
index = json.loads((index_dir / "index.json").read_text())
|
|
assert len(index["files"]) == 3
|
|
for i, f in enumerate(index["files"]):
|
|
assert f["chunk_idx"] == i
|
|
assert f["offset"] == i * 256
|
|
|
|
def test_chunking_skip_rag_dir(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = memory_dir / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
|
|
(memory_dir / "real.md").write_text("real content")
|
|
(index_dir).mkdir()
|
|
(index_dir / "index.json").write_text('{"files": []}')
|
|
(index_dir / "ignore.md").write_text("should be ignored")
|
|
|
|
def fake_embed_texts(texts):
|
|
return [[0.1]] * len(texts)
|
|
|
|
output_dir = tmp_path / "out"
|
|
with patch("src.memory.index.embed_texts", fake_embed_texts):
|
|
args = Namespace(memory_dir=str(memory_dir), output=str(output_dir))
|
|
run_index(args)
|
|
|
|
index = json.loads((output_dir / "index.json").read_text())
|
|
sources = [f["source"] for f in index["files"]]
|
|
assert "real.md" in sources
|
|
assert all(".rag" not in s for s in sources)
|
|
|
|
|
|
class TestIncrementalIndex:
|
|
def _make_args(self, memory_dir: Path, output_dir: Path) -> Namespace:
|
|
return Namespace(memory_dir=str(memory_dir), output=str(output_dir))
|
|
|
|
def _fake_embed(self):
|
|
def fake_embed_texts(texts):
|
|
return [[0.1, 0.2] for _ in texts]
|
|
|
|
return fake_embed_texts
|
|
|
|
def test_incremental_add(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A content")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
(memory_dir / "b.md").write_text("file B content")
|
|
calls: list[list[str]] = []
|
|
|
|
def tracking_embed(texts):
|
|
calls.append(list(texts))
|
|
return [[0.1, 0.2] for _ in texts]
|
|
|
|
with patch("src.memory.index.embed_texts", tracking_embed):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
assert len(calls) == 1
|
|
embedded_text = calls[0][0]
|
|
assert "file B content" in embedded_text
|
|
assert "file A content" not in embedded_text
|
|
|
|
index = json.loads((index_dir / "index.json").read_text())
|
|
sources = {f["source"] for f in index["files"]}
|
|
assert {"a.md", "b.md"} <= sources
|
|
|
|
def test_incremental_edit(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("original content")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
(memory_dir / "a.md").write_text("edited content here")
|
|
calls: list[list[str]] = []
|
|
|
|
def tracking_embed(texts):
|
|
calls.append(list(texts))
|
|
return [[0.1, 0.2] for _ in texts]
|
|
|
|
with patch("src.memory.index.embed_texts", tracking_embed):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
assert len(calls) == 1
|
|
assert "edited content here" in calls[0][0]
|
|
|
|
def test_incremental_delete(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
(memory_dir / "b.md").write_text("file B")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
(memory_dir / "b.md").unlink()
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
index = json.loads((index_dir / "index.json").read_text())
|
|
sources = [f["source"] for f in index["files"]]
|
|
assert "a.md" in sources
|
|
assert "b.md" not in sources
|
|
|
|
meta = json.loads((index_dir / "meta.json").read_text())
|
|
assert "b.md" not in meta["files"]
|
|
assert "a.md" in meta["files"]
|
|
|
|
def test_no_changes_noop(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("stable content")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
index_mtime_before = (index_dir / "index.json").stat().st_mtime_ns
|
|
|
|
embed_mock = Mock(return_value=[[0.1, 0.2]])
|
|
with patch("src.memory.index.embed_texts", embed_mock):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
embed_mock.assert_not_called()
|
|
index_mtime_after = (index_dir / "index.json").stat().st_mtime_ns
|
|
assert index_mtime_after == index_mtime_before
|
|
|
|
def test_meta_json_persistence(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
(memory_dir / "b.md").write_text("file B")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
meta_path = index_dir / "meta.json"
|
|
assert meta_path.exists()
|
|
meta = json.loads(meta_path.read_text())
|
|
assert "a.md" in meta["files"]
|
|
assert "b.md" in meta["files"]
|
|
assert "version" in meta
|
|
assert meta["files"]["a.md"]["sha256"]
|
|
assert meta["files"]["a.md"]["chunks"] == 1
|
|
|
|
(memory_dir / "c.md").write_text("file C")
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
meta2 = json.loads(meta_path.read_text())
|
|
assert "a.md" in meta2["files"]
|
|
assert "b.md" in meta2["files"]
|
|
assert "c.md" in meta2["files"]
|
|
|
|
def test_sha256_content_hash(self, tmp_path: Path) -> None:
|
|
assert _content_hash("hello") == _content_hash("hello")
|
|
assert _content_hash("hello") != _content_hash("world")
|
|
assert len(_content_hash("hello")) == 64
|
|
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("---\ntitle: T\n---\n\nbody content")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
meta_before = json.loads((index_dir / "meta.json").read_text())
|
|
sha_before = meta_before["files"]["a.md"]["sha256"]
|
|
|
|
os.utime(memory_dir / "a.md", None)
|
|
|
|
embed_mock = Mock(return_value=[[0.1, 0.2]])
|
|
with patch("src.memory.index.embed_texts", embed_mock):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
embed_mock.assert_not_called()
|
|
meta_after = json.loads((index_dir / "meta.json").read_text())
|
|
assert meta_after["files"]["a.md"]["sha256"] == sha_before
|
|
|
|
def test_atomic_writes(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
original_index = (index_dir / "index.json").read_text()
|
|
|
|
(memory_dir / "a.md").write_text("file A edited")
|
|
|
|
with (
|
|
patch("src.memory.index.embed_texts", self._fake_embed()),
|
|
patch("src.memory.index.os.replace", side_effect=OSError("simulated crash")),
|
|
pytest.raises(OSError),
|
|
):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
assert (index_dir / "index.json").read_text() == original_index
|
|
|
|
def test_index_versioning(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
meta = json.loads((index_dir / "meta.json").read_text())
|
|
meta["version"] = "old-model:512:64"
|
|
(index_dir / "meta.json").write_text(json.dumps(meta))
|
|
|
|
calls: list[int] = []
|
|
|
|
def tracking_embed(texts):
|
|
calls.append(len(texts))
|
|
return [[0.1, 0.2] for _ in texts]
|
|
|
|
with patch("src.memory.index.embed_texts", tracking_embed):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
assert len(calls) == 1
|
|
assert calls[0] >= 1
|
|
|
|
meta2 = json.loads((index_dir / "meta.json").read_text())
|
|
assert meta2["version"] != "old-model:512:64"
|
|
|
|
def test_index_writes_reindex_log(self, tmp_path: Path) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
(memory_dir / "b.md").write_text("file B")
|
|
|
|
with patch("src.memory.index.embed_texts", self._fake_embed()):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
log_path = index_dir / "reindex.log"
|
|
assert log_path.exists()
|
|
log_content = log_path.read_text(encoding="utf-8")
|
|
assert "start:" in log_content
|
|
assert "done:" in log_content
|
|
assert "total=2" in log_content
|
|
|
|
def test_index_embedder_none_logs_and_exits_zero(self, tmp_path: Path, capsys) -> None:
|
|
memory_dir = tmp_path / "memory"
|
|
index_dir = tmp_path / ".rag"
|
|
memory_dir.mkdir(parents=True)
|
|
(memory_dir / "a.md").write_text("file A")
|
|
|
|
def none_embed(texts):
|
|
return None
|
|
|
|
with patch("src.memory.index.embed_texts", none_embed):
|
|
run_index(self._make_args(memory_dir, index_dir))
|
|
|
|
log_path = index_dir / "reindex.log"
|
|
log_content = log_path.read_text(encoding="utf-8")
|
|
assert "failed:" in log_content
|
|
assert "embeddings unavailable" in log_content
|
|
|
|
captured = capsys.readouterr()
|
|
assert "embeddings unavailable" in captured.out
|
|
assert not (index_dir / "index.json").exists()
|
|
|
|
|
|
class TestAtomicWrite:
|
|
def test_atomic_write_replaces(self, tmp_path: Path) -> None:
|
|
target = tmp_path / "file.json"
|
|
target.write_text("old")
|
|
_atomic_write(target, "new content")
|
|
assert target.read_text() == "new content"
|
|
assert not (tmp_path / "file.json.tmp").exists()
|
|
|
|
def test_atomic_write_creates(self, tmp_path: Path) -> None:
|
|
target = tmp_path / "new.json"
|
|
_atomic_write(target, "fresh")
|
|
assert target.read_text() == "fresh"
|