diff --git a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/middleware/knowledge_store_persistence/commit_turn.py b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/middleware/knowledge_store_persistence/commit_turn.py index 7ec49418d..e7fe33d16 100644 --- a/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/middleware/knowledge_store_persistence/commit_turn.py +++ b/surfsense_backend/app/agents/chat/multi_agent_chat/main_agent/middleware/knowledge_store_persistence/commit_turn.py @@ -27,7 +27,12 @@ from app.observability import metrics logger = logging.getLogger(__name__) -_OPERATION_BY_KIND = {"added": "write_file", "modified": "edit_file", "removed": "rm"} +_OPERATION_BY_KIND = { + "added": "write_file", + "modified": "edit_file", + "removed": "rm", + "renamed": "move_file", +} async def commit_turn_working_copy( diff --git a/surfsense_backend/app/knowledge_store/engines/base.py b/surfsense_backend/app/knowledge_store/engines/base.py index 4abed36e9..35643525a 100644 --- a/surfsense_backend/app/knowledge_store/engines/base.py +++ b/surfsense_backend/app/knowledge_store/engines/base.py @@ -9,7 +9,7 @@ from datetime import datetime from pathlib import Path from typing import Literal -ChangeKind = Literal["added", "modified", "removed"] +ChangeKind = Literal["added", "modified", "removed", "renamed"] @dataclass(frozen=True) @@ -33,6 +33,8 @@ class Change: kind: ChangeKind #: Content address after the change (``None`` when removed). content_id: str | None + #: Where a renamed path came from (``None`` for every other kind). + previous_path: str | None = None @dataclass(frozen=True) @@ -90,8 +92,14 @@ class VersionedContentEngine(ABC): """Revisions newest-first, optionally scoped to a single path.""" @abstractmethod - def list_changes(self, revision: str) -> list[Change]: - """Paths added, modified, or removed by ``revision`` (vs its parent).""" + def list_changes(self, revision: str, *, since: str | None = None) -> list[Change]: + """What ``revision`` changed, against its parent or against ``since``. + + ``since`` compares two snapshots directly, so a path touched repeatedly + in between appears once, with its net effect. A path that moved is one + ``renamed`` change carrying both paths, not a removal plus an addition, + so callers can keep whatever they hold against the old path. + """ @abstractmethod def list_paths(self, revision: str) -> list[TrackedPath]: diff --git a/surfsense_backend/app/knowledge_store/engines/git.py b/surfsense_backend/app/knowledge_store/engines/git.py index 08316b472..17b9a3a47 100644 --- a/surfsense_backend/app/knowledge_store/engines/git.py +++ b/surfsense_backend/app/knowledge_store/engines/git.py @@ -14,7 +14,14 @@ from datetime import UTC, datetime from pathlib import Path from dulwich import porcelain -from dulwich.diff_tree import CHANGE_ADD, CHANGE_DELETE, CHANGE_MODIFY, tree_changes +from dulwich.diff_tree import ( + CHANGE_ADD, + CHANGE_DELETE, + CHANGE_MODIFY, + CHANGE_RENAME, + RenameDetector, + tree_changes, +) from dulwich.object_store import iter_tree_contents, tree_lookup_path from dulwich.objects import Blob from dulwich.repo import Repo @@ -32,6 +39,7 @@ _CHANGE_KINDS = { CHANGE_ADD: "added", CHANGE_MODIFY: "modified", CHANGE_DELETE: "removed", + CHANGE_RENAME: "renamed", } # Serializes working-copy creation against parallel tool calls in one process. @@ -126,13 +134,20 @@ class GitContentEngine(VersionedContentEngine): finally: repo.close() - def list_changes(self, revision: str) -> list[Change]: + def list_changes(self, revision: str, *, since: str | None = None) -> list[Change]: repo = Repo(str(self._path)) try: commit = repo[revision.encode()] - parent_tree = repo[commit.parents[0]].tree if commit.parents else None + base_tree = self._base_tree(repo, commit, since) + # Renames come from git's own detection, not a guess of ours: identical + # content is matched by hash, so a plain move is always found. Only + # similarity matching (a move that also edits) is bounded — dulwich stops + # at 200 candidates, past which such a move reads as a removal and an add. + detector = RenameDetector(repo.object_store) changes = [] - for change in tree_changes(repo.object_store, parent_tree, commit.tree): + for change in tree_changes( + repo.object_store, base_tree, commit.tree, rename_detector=detector + ): kind = _CHANGE_KINDS.get(change.type) if kind is None: continue @@ -142,6 +157,9 @@ class GitContentEngine(VersionedContentEngine): path=entry.path.decode(), kind=kind, content_id=None if kind == "removed" else entry.sha.decode(), + previous_path=( + change.old.path.decode() if kind == "renamed" else None + ), ) ) return changes @@ -254,6 +272,13 @@ class GitContentEngine(VersionedContentEngine): def compute_content_id(data: bytes) -> str: return Blob.from_string(data).id.decode() + @staticmethod + def _base_tree(repo: Repo, commit, since: str | None): + """What to diff against: ``since``'s tree, else the commit's first parent.""" + if since is not None: + return repo[since.encode()].tree + return repo[commit.parents[0]].tree if commit.parents else None + @staticmethod def _working_copy_base(copy_path: Path) -> str | None: """Revision an existing copy was opened at (``None`` for a bare directory).""" diff --git a/surfsense_backend/app/knowledge_store/index/converge.py b/surfsense_backend/app/knowledge_store/index/converge.py index 4f2788586..07e74469b 100644 --- a/surfsense_backend/app/knowledge_store/index/converge.py +++ b/surfsense_backend/app/knowledge_store/index/converge.py @@ -76,6 +76,8 @@ class _Plan: upserts: list[str] removals: list[str] + #: Paths that moved, as ``(from, to)``; the row follows instead of being remade. + renames: list[tuple[str, str]] = field(default_factory=list) #: Every path in the tree, when the run is a full rebuild; ``None`` otherwise. tree: set[str] | None = field(default=None) @@ -132,6 +134,11 @@ async def _plan(store: KnowledgeStore, head: str, since: str | None) -> _Plan: return _Plan( upserts=[c.path for c in changes if c.kind != "removed"], removals=[c.path for c in changes if c.kind == "removed"], + renames=[ + (c.previous_path, c.path) + for c in changes + if c.kind == "renamed" and c.previous_path + ], ) tracked = [entry.path for entry in await store.list_paths(head)] return _Plan(upserts=tracked, removals=[], tree=set(tracked)) @@ -140,11 +147,16 @@ async def _plan(store: KnowledgeStore, head: str, since: str | None) -> _Plan: async def _changes_since( store: KnowledgeStore, head: str, since: str ) -> list[Change] | None: - """Net change set from ``since`` (exclusive) to ``head``, newest write wins. + """Net change set from ``since`` (exclusive) to ``head``. + + One diff of the two snapshots rather than a fold of each revision between + them, which matters because a queued task can be several commits behind by + the time it runs: a path written twice in that window appears once, a path + written then deleted appears not at all, and a move is a single ``renamed`` + change that keeps both of its paths. ``None`` when ``since`` is not in the history any more, which asks the caller - for a full rebuild rather than a guess. Folding every revision in between - matters because a queued task can be two commits behind by the time it runs. + for a full rebuild rather than a guess. ``ponytail:`` walks the whole revision list to locate ``since``; upgrade path is a bounded walk once histories get long enough to notice. @@ -152,12 +164,7 @@ async def _changes_since( ids = [revision.id for revision in await store.list_revisions()] if since not in ids: return None - newer = ids[: ids.index(since)] - merged: dict[str, Change] = {} - for revision_id in reversed(newer): - for change in await store.list_changes(revision_id): - merged[change.path] = change - return list(merged.values()) + return await store.list_changes(head, since=since) async def _converge( @@ -171,6 +178,14 @@ async def _converge( owned = await _load_owned(session, workspace.id) author_id = await _revision_author_id(store, head, workspace) + for from_path, to_path in plan.renames: + _follow_rename( + owned, + workspace.id, + to_virtual_path(from_path), + to_virtual_path(to_path), + ) + for store_path in plan.upserts: virtual_path = to_virtual_path(store_path) content = await _read_indexable(store, head, store_path) @@ -317,6 +332,39 @@ async def _index_one( return True +def _follow_rename( + owned: dict[str, Document], + workspace_id: int, + from_virtual: str, + to_virtual: str, +) -> None: + """Point the row living at ``from_virtual`` at the path it moved to. + + A move has to leave the row's id alone: ``document_versions`` and an upload's + stored original both cascade from it, and citations saved in earlier answers + name it. Re-keying is the whole trick — the upsert of the new path then + resolves to this row and updates it in place, rather than inserting one row + and deleting the other. + """ + document = owned.pop(from_virtual, None) + if document is None: + # Nothing marked at the old path: an unindexed file, or a recorder that + # already moved the marker. Either way the upsert resolves it by itself. + return + owned[to_virtual] = document + from_hash = generate_unique_identifier_hash( + DocumentType.NOTE, from_virtual, workspace_id + ) + if document.unique_identifier_hash == from_hash: + # Carry _resolve's fallback key along with the marker, or a later file at + # the old path resolves to this row. Only when the key is the path's own: + # an upload identifies by filename, and rewriting that would let a + # re-upload of the same file insert a second row. + document.unique_identifier_hash = generate_unique_identifier_hash( + DocumentType.NOTE, to_virtual, workspace_id + ) + + async def _resolve( session: AsyncSession, workspace_id: int, @@ -364,11 +412,11 @@ async def _delete( return 0 marker = (document.document_metadata or {}).get(PATH_MARKER) if marker and marker != virtual_path: - # The row moved, it did not go away. A rename arrives as a removal of the - # old path plus an upsert of the new one, and the upsert has already - # claimed this row; deleting on the removal would drop what the same run - # just wrote. Reachable because a retitle moves the marker and leaves - # unique_identifier_hash — which _resolve falls back to — on the old path. + # The row moved, it did not go away: the upsert has already claimed it, so + # deleting here would drop what this same run just wrote. Reached when git + # cannot see the move — a rewrite in flight leaves nothing to match, so it + # arrives as a removal and an addition — while the recorder has moved the + # marker and left unique_identifier_hash, _resolve's fallback, behind. return 0 owned.pop(virtual_path, None) await session.delete(document) diff --git a/surfsense_backend/app/knowledge_store/store.py b/surfsense_backend/app/knowledge_store/store.py index 7d03b3a00..beabe5179 100644 --- a/surfsense_backend/app/knowledge_store/store.py +++ b/surfsense_backend/app/knowledge_store/store.py @@ -65,9 +65,11 @@ class KnowledgeStore: self._engine.list_revisions, path=path, limit=limit ) - async def list_changes(self, revision: str) -> list[Change]: - """Paths added, modified, or removed by ``revision`` (vs its parent).""" - return await asyncio.to_thread(self._engine.list_changes, revision) + async def list_changes( + self, revision: str, *, since: str | None = None + ) -> list[Change]: + """What ``revision`` changed, against its parent or against ``since``.""" + return await asyncio.to_thread(self._engine.list_changes, revision, since=since) async def list_paths(self, revision: str) -> list[TrackedPath]: """Every path stored at ``revision``, with its content address.""" diff --git a/surfsense_backend/tests/integration/knowledge_store/index/test_converge.py b/surfsense_backend/tests/integration/knowledge_store/index/test_converge.py index 7ac0e7357..e39f68490 100644 --- a/surfsense_backend/tests/integration/knowledge_store/index/test_converge.py +++ b/surfsense_backend/tests/integration/knowledge_store/index/test_converge.py @@ -16,7 +16,7 @@ import pytest from sqlalchemy import func, select from app.config import config as app_config -from app.db import Chunk, Document, DocumentStatus, DocumentType +from app.db import Chunk, Document, DocumentStatus, DocumentType, DocumentVersion from app.indexing_pipeline.connector_document import ConnectorDocument from app.indexing_pipeline.indexing_pipeline_service import IndexingPipelineService from app.knowledge_store import KnowledgeStore @@ -26,6 +26,10 @@ from app.utils.document_converters import generate_unique_identifier_hash pytestmark = pytest.mark.integration +# Content for the move tests. Git recognises a moved file by its content, so what +# matters is that the same bytes land at the new path. +MOVABLE = "# Content\n\na body git can match at its new path\n" + @pytest.fixture def knowledge_root(tmp_path, monkeypatch): @@ -66,6 +70,15 @@ async def chunk_ids(session, document_id) -> list[int]: return list(result.scalars()) +async def versions(session, document_id) -> list[int]: + result = await session.execute( + select(DocumentVersion.version_number) + .where(DocumentVersion.document_id == document_id) + .order_by(DocumentVersion.version_number) + ) + return list(result.scalars()) + + async def chunk_shapes(session, document_id) -> list[tuple]: """Text and line span of every chunk, in document order.""" result = await session.execute( @@ -194,29 +207,66 @@ async def test_removed_path_deletes_the_document_and_its_chunks( assert await chunk_ids(db_session, doomed_id) == [] -async def test_a_rename_leaves_exactly_one_document( +async def test_a_move_keeps_the_document_and_its_history( store, db_session, db_workspace, patched_embed_texts ): - """Git reports a rename as add+delete, so both halves must be applied.""" - await commit(store, {"documents/old.xml": "# Content"}) + """The row has to outlive a move, not just be replaced by an equivalent one. + Its version history cascades from the id, so a new row means an agent moving + a file silently destroys every saved version of it — and dangles the + ``document_id`` in citations already written into past answers.""" + await commit(store, {"documents/old.xml": MOVABLE}) await index_changes(db_session, db_workspace.id) - - await commit( - store, {"documents/new.xml": "# Content"}, removes=["documents/old.xml"] + document_id = (await titles(db_session, db_workspace.id))["old"].id + db_session.add( + DocumentVersion( + document_id=document_id, + version_number=1, + source_markdown=MOVABLE, + content_hash=f"hash-{uuid.uuid4().hex}", + title="old", + ) ) + await db_session.commit() + + await commit(store, {"documents/new.xml": MOVABLE}, removes=["documents/old.xml"]) await index_changes(db_session, db_workspace.id) - assert set(await titles(db_session, db_workspace.id)) == {"new"} + rows = await titles(db_session, db_workspace.id) + assert set(rows) == {"new"} + assert rows["new"].id == document_id + assert await versions(db_session, document_id) == [1] -async def test_a_rename_the_recorder_already_marked_keeps_its_document( +async def test_a_new_file_at_a_moved_from_path_gets_its_own_row( store, db_session, db_workspace, patched_embed_texts ): - """An editor retitle moves the row's marker before the index ever runs, and - leaves unique_identifier_hash on the old path. The removal half then resolves - by that hash to the row the upsert half just updated — dropping a document - whose file is still in the tree, invisible until the next full rebuild.""" - await commit(store, {"documents/old.xml": "# Content"}) + """The moved row's fallback identity has to travel with it. Left behind on the + old path, it makes the next document written there resolve to the moved row — + one row claiming two paths, and the new file never getting a row of its own.""" + await commit(store, {"documents/old.xml": MOVABLE}) + await index_changes(db_session, db_workspace.id) + moved_id = (await titles(db_session, db_workspace.id))["old"].id + + await commit(store, {"documents/new.xml": MOVABLE}, removes=["documents/old.xml"]) + await index_changes(db_session, db_workspace.id) + await commit(store, {"documents/old.xml": "# Reused path\n\nnew note here\n"}) + await index_changes(db_session, db_workspace.id) + + rows = await titles(db_session, db_workspace.id) + assert set(rows) == {"new", "old"} + assert rows["new"].id == moved_id + assert rows["old"].id != moved_id + + +async def test_a_move_that_rewrites_the_file_keeps_the_marked_document( + store, db_session, db_workspace, patched_embed_texts +): + """With nothing in common git sees no move, so the halves arrive separately — + and an editor retitle has already moved the marker while leaving + unique_identifier_hash behind. The removal then resolves by that hash to the + row the upsert just updated, dropping a document whose file is still in the + tree, invisible until the next full rebuild.""" + await commit(store, {"documents/old.xml": MOVABLE}) await index_changes(db_session, db_workspace.id) row = (await titles(db_session, db_workspace.id))["old"] document_id = row.id @@ -224,7 +274,9 @@ async def test_a_rename_the_recorder_already_marked_keeps_its_document( await db_session.commit() await commit( - store, {"documents/new.xml": "# Content"}, removes=["documents/old.xml"] + store, + {"documents/new.xml": "# Other\n\nnothing whatever in common\n"}, + removes=["documents/old.xml"], ) await index_changes(db_session, db_workspace.id) diff --git a/surfsense_backend/tests/integration/knowledge_store/test_commit_turn.py b/surfsense_backend/tests/integration/knowledge_store/test_commit_turn.py index 8a8d1a702..2d7e8916f 100644 --- a/surfsense_backend/tests/integration/knowledge_store/test_commit_turn.py +++ b/surfsense_backend/tests/integration/knowledge_store/test_commit_turn.py @@ -249,20 +249,17 @@ async def test_a_deleted_file_is_recorded_as_a_removal( assert await store.list_paths(revision) == [] -async def test_a_moved_file_is_recorded_as_a_removal_and_an_addition( - knowledge_root, workspace_id, llm -): - """The store has no rename verb: a move is the two changes it decomposes into.""" +async def test_a_moved_file_is_recorded_as_one_move(knowledge_root, workspace_id, llm): + """A move is committed as a removal plus a write, but read back as a rename: + git recognises the content, which is what lets the index move the document's + row instead of replacing it. The receipt reports the move it was.""" store = await _committed_turn(workspace_id, llm, {"documents/old.md": b"hello"}) copy = await _next_turn_copy(store) (copy.path / "documents/old.md").rename(copy.path / "documents/new.md") delta = await _commit(workspace_id, llm) - assert _operations(delta) == { - "documents/old.md": "rm", - "documents/new.md": "write_file", - } + assert _operations(delta) == {"documents/new.md": "move_file"} revision = (await store.list_revisions())[0].id assert [e.path for e in await store.list_paths(revision)] == ["documents/new.md"] diff --git a/surfsense_backend/tests/unit/knowledge_store/engines/test_git.py b/surfsense_backend/tests/unit/knowledge_store/engines/test_git.py index fcaea3806..26afc0347 100644 --- a/surfsense_backend/tests/unit/knowledge_store/engines/test_git.py +++ b/surfsense_backend/tests/unit/knowledge_store/engines/test_git.py @@ -137,6 +137,76 @@ class TestHistoryQueries: assert changed["a.xml"].content_id == engine.compute_content_id(b"a2") assert changed["b.xml"].content_id is None + def test_list_changes_reports_a_move_as_one_rename(self, engine): + """A move must not read as an unrelated removal and addition: the index + would replace the document's row, cascading away its version history.""" + body = b"# Note\n\nenough body text for git to recognise it again\n" + engine.record( + writes={"old.xml": body}, removes=[], message="seed", author=AUTHOR + ) + moved = engine.record( + writes={"new.xml": body}, + removes=["old.xml"], + message="move", + author=AUTHOR, + ) + + (change,) = engine.list_changes(moved) + assert (change.path, change.kind, change.previous_path) == ( + "new.xml", + "renamed", + "old.xml", + ) + assert change.content_id == engine.compute_content_id(body) + + def test_a_move_that_rewrites_the_file_is_not_a_rename(self, engine): + """The ceiling of matching by similarity: with nothing left in common there + is no move to see, so the index falls back to replacing the row.""" + engine.record( + writes={"old.xml": b"# Note\n\nthe original body\n"}, + removes=[], + message="seed", + author=AUTHOR, + ) + rewritten = engine.record( + writes={"new.xml": b"# Other\n\nnothing whatever in common\n"}, + removes=["old.xml"], + message="rewrite", + author=AUTHOR, + ) + + assert {c.path: c.kind for c in engine.list_changes(rewritten)} == { + "old.xml": "removed", + "new.xml": "added", + } + + def test_list_changes_since_reports_a_window_net(self, engine): + """What a queued index run needs: several revisions behind, it asks once + for the net effect rather than replaying each revision and folding.""" + body = b"# Note\n\nenough body text for git to recognise it again\n" + base = engine.record( + writes={"old.xml": body}, removes=[], message="seed", author=AUTHOR + ) + engine.record( + writes={"new.xml": body, "gone.xml": b"transient"}, + removes=["old.xml"], + message="move", + author=AUTHOR, + ) + head = engine.record( + writes={"new.xml": body + b"edited later\n"}, + removes=["gone.xml"], + message="edit", + author=AUTHOR, + ) + + window = {c.path: c for c in engine.list_changes(head, since=base)} + # The move survives being edited afterwards, and a file that came and went + # inside the window is not reported at all. + assert window["new.xml"].kind == "renamed" + assert window["new.xml"].previous_path == "old.xml" + assert "gone.xml" not in window + def test_list_paths_reflects_the_given_revision(self, engine): first = engine.record( writes={"a.xml": b"a1", "sub/b.xml": b"b1"},