diff --git a/server/api/memories.py b/server/api/memories.py index cf93b5b..50ff203 100644 --- a/server/api/memories.py +++ b/server/api/memories.py @@ -446,6 +446,17 @@ async def search_memories( session: AsyncSession = Depends(get_session), tenant_id: str | None = Depends(get_tenant_id), ): + """Search a subject's memories by text or semantic similarity. + + The `search_mode` field in the response reports which path actually ran, + so callers can tell whether semantic search really executed: + - `semantic`: embedding (or hybrid) search ran. + - `text`: plain text search — either `semantic` was not requested, or it + was requested without a `q` query (in which case it is ignored). + - `text_fallback`: `semantic` was requested with a `q`, but could not run + (no embedding provider configured, or the provider errored), so text + search ran instead. Check the server logs for the underlying cause. + """ with span( "search_memories", { @@ -457,6 +468,10 @@ async def search_memories( "rerank": rerank, }, ): + # `search_mode` records which path actually ran (issue #281): text is + # the default; the semantic branch upgrades it to "semantic" on success + # or "text_fallback" if semantic was requested but could not run. + search_mode = "text" # Try semantic / hybrid search if requested and query text is provided if semantic and query: provider = get_embedding_provider() @@ -501,7 +516,8 @@ async def search_memories( else: rows = rows[:limit] return SearchMemoriesResponse( - memories=[_to_response(row) for row in rows] + memories=[_to_response(row) for row in rows], + search_mode="semantic", ) results = await repo.search_memories_by_embedding( session, @@ -512,17 +528,26 @@ async def search_memories( limit=limit, ) return SearchMemoriesResponse( - memories=[_to_response(row) for row, _dist in results] + memories=[_to_response(row) for row, _dist in results], + search_mode="semantic", ) except Exception: logger.warning("semantic_search_failed_falling_back", exc_info=True) - # Fall through to text search + # Semantic was requested but errored — fall through to text + # search and tell the caller it was a fallback. + search_mode = "text_fallback" + else: + # Semantic requested with a query, but no embedding provider is + # configured — text search runs instead of semantic. + search_mode = "text_fallback" # Default: exact/text search rows = await repo.search_memories( session, subject_id, tenant_id=tenant_id, kind=kind, query=query, limit=limit ) - return SearchMemoriesResponse(memories=[_to_response(r) for r in rows]) + return SearchMemoriesResponse( + memories=[_to_response(r) for r in rows], search_mode=search_mode + ) def _to_response(row) -> MemoryResponse: diff --git a/server/api/subjects.py b/server/api/subjects.py index ba91d67..528852a 100644 --- a/server/api/subjects.py +++ b/server/api/subjects.py @@ -56,15 +56,21 @@ async def delete_subject( # boost from ghost memories. await repo.delete_entities_by_subject(session, subject_id, tenant_id=tenant_id) await session.commit() - await webhooks.fire( - "subject.deleted", - { - "subject_id": subject_id, - "episodes_deleted": ep_count, - "memories_deleted": mem_count, - }, - tenant_id=tenant_id, - ) + # Only fire the webhook when something was actually deleted (issue #282): + # deleting a missing subject (or the same subject twice) is a no-op, so a + # subject.deleted event with zero counts would be a spurious deletion + # signal to consumers (cache invalidation, audit, compliance records). + # The HTTP response stays 200 with honest zero counts either way. + if ep_count + mem_count > 0: + await webhooks.fire( + "subject.deleted", + { + "subject_id": subject_id, + "episodes_deleted": ep_count, + "memories_deleted": mem_count, + }, + tenant_id=tenant_id, + ) return DeleteSubjectResponse( subject_id=subject_id, episodes_deleted=ep_count, diff --git a/server/schemas/responses.py b/server/schemas/responses.py index e646a41..8966531 100644 --- a/server/schemas/responses.py +++ b/server/schemas/responses.py @@ -4,7 +4,7 @@ import uuid from datetime import datetime -from typing import Any +from typing import Any, Literal from pydantic import BaseModel, Field @@ -67,6 +67,17 @@ class CompileMemoriesResponse(BaseModel): class SearchMemoriesResponse(BaseModel): memories: list[MemoryResponse] + search_mode: Literal["semantic", "text", "text_fallback"] = Field( + "text", + description=( + "Which search path actually ran: 'semantic' (embedding/hybrid " + "search executed), 'text' (plain text search — semantic was not " + "requested, or requested without a q), or 'text_fallback' " + "(semantic was requested with a q but could not run — no embedding " + "provider configured, or the provider errored — so text search ran " + "instead)." + ), + ) class SessionInfo(BaseModel): diff --git a/tests/test_memories_search_mode.py b/tests/test_memories_search_mode.py new file mode 100644 index 0000000..a9d6c33 --- /dev/null +++ b/tests/test_memories_search_mode.py @@ -0,0 +1,90 @@ +"""GET /v1/memories/search reports which path actually ran via `search_mode` +(issue #281), so callers can tell semantic search from a silent text fallback. +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, patch + +from httpx import AsyncClient + + +async def test_search_mode_text_when_semantic_not_requested(client: AsyncClient): + with patch( + "server.api.memories.repo.search_memories", new=AsyncMock(return_value=[]) + ): + resp = await client.get( + "/v1/memories/search", params={"subject_id": "u", "q": "hello"} + ) + assert resp.status_code == 200 + assert resp.json()["search_mode"] == "text" + + +async def test_search_mode_text_when_semantic_without_query(client: AsyncClient): + # semantic=true but no q → the `if semantic and query` guard is false → text. + with patch( + "server.api.memories.repo.search_memories", new=AsyncMock(return_value=[]) + ): + resp = await client.get( + "/v1/memories/search", + params={"subject_id": "u", "semantic": "true"}, + ) + assert resp.status_code == 200 + assert resp.json()["search_mode"] == "text" + + +async def test_search_mode_text_fallback_when_no_provider(client: AsyncClient): + # semantic + q requested, but no embedding provider configured → text_fallback. + with ( + patch("server.api.memories.get_embedding_provider", return_value=None), + patch( + "server.api.memories.repo.search_memories", new=AsyncMock(return_value=[]) + ), + ): + resp = await client.get( + "/v1/memories/search", + params={"subject_id": "u", "q": "hello", "semantic": "true"}, + ) + assert resp.status_code == 200 + assert resp.json()["search_mode"] == "text_fallback" + + +async def test_search_mode_text_fallback_when_provider_errors(client: AsyncClient): + # provider present, but the embedding lookup raises → caught → text_fallback. + with ( + patch("server.api.memories.get_embedding_provider", return_value=object()), + patch( + "server.services.embeddings.query_cache.cached_embed_query", + new=AsyncMock(side_effect=RuntimeError("provider down")), + ), + patch( + "server.api.memories.repo.search_memories", new=AsyncMock(return_value=[]) + ), + ): + resp = await client.get( + "/v1/memories/search", + params={"subject_id": "u", "q": "hello", "semantic": "true"}, + ) + assert resp.status_code == 200 + assert resp.json()["search_mode"] == "text_fallback" + + +async def test_search_mode_semantic_when_embedding_search_runs(client: AsyncClient): + # provider + embedding + (hybrid) search all succeed → semantic. + with ( + patch("server.api.memories.get_embedding_provider", return_value=object()), + patch( + "server.services.embeddings.query_cache.cached_embed_query", + new=AsyncMock(return_value=[0.0] * 8), + ), + patch( + "server.api.memories.repo.search_memories_hybrid", + new=AsyncMock(return_value=[]), + ), + ): + resp = await client.get( + "/v1/memories/search", + params={"subject_id": "u", "q": "hello", "semantic": "true"}, + ) + assert resp.status_code == 200 + assert resp.json()["search_mode"] == "semantic" diff --git a/tests/test_subject_delete_webhook.py b/tests/test_subject_delete_webhook.py new file mode 100644 index 0000000..4e3a648 --- /dev/null +++ b/tests/test_subject_delete_webhook.py @@ -0,0 +1,83 @@ +"""DELETE /v1/subjects fires subject.deleted only when something was actually +deleted (issue #282). + +The delete is idempotent — deleting a missing subject (or the same subject +twice) still returns 200 with honest zero counts — but it must NOT emit a +subject.deleted webhook for a no-op, which would be a spurious deletion signal +to consumers (cache invalidation, audit logging, compliance records). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +from server.api import subjects as subjects_api + + +class _FakeSession: + """Minimal async session — delete_subject only calls .commit() on it; the + repo delete helpers are mocked.""" + + async def commit(self): + return None + + +def _stub_repo_deletes(monkeypatch, *, episodes: int, memories: int) -> None: + monkeypatch.setattr( + subjects_api.repo, "delete_episodes_by_subject", AsyncMock(return_value=episodes) + ) + monkeypatch.setattr( + subjects_api.repo, "delete_memories_by_subject", AsyncMock(return_value=memories) + ) + for name in ( + "delete_resolutions_by_subject", + "delete_health_cache_by_subject", + "delete_entities_by_subject", + ): + monkeypatch.setattr(subjects_api.repo, name, AsyncMock(return_value=0)) + + +async def test_no_webhook_when_nothing_deleted(monkeypatch): + _stub_repo_deletes(monkeypatch, episodes=0, memories=0) + fire = AsyncMock() + monkeypatch.setattr(subjects_api.webhooks, "fire", fire) + + resp = await subjects_api.delete_subject( + "ghost-subject", session=_FakeSession(), tenant_id=None + ) + + # Idempotent: still 200 with honest zero counts... + assert resp.episodes_deleted == 0 + assert resp.memories_deleted == 0 + # ...but no spurious subject.deleted event. + fire.assert_not_called() + + +async def test_webhook_fires_when_something_deleted(monkeypatch): + _stub_repo_deletes(monkeypatch, episodes=3, memories=2) + fire = AsyncMock() + monkeypatch.setattr(subjects_api.webhooks, "fire", fire) + + resp = await subjects_api.delete_subject( + "real-subject", session=_FakeSession(), tenant_id=None + ) + + assert resp.episodes_deleted == 3 + assert resp.memories_deleted == 2 + fire.assert_called_once() + event_name = fire.call_args.args[0] + payload = fire.call_args.args[1] + assert event_name == "subject.deleted" + assert payload["episodes_deleted"] == 3 + assert payload["memories_deleted"] == 2 + + +async def test_webhook_fires_when_only_memories_deleted(monkeypatch): + # ep_count=0 but mem_count>0 is still a real deletion → fire. + _stub_repo_deletes(monkeypatch, episodes=0, memories=1) + fire = AsyncMock() + monkeypatch.setattr(subjects_api.webhooks, "fire", fire) + + await subjects_api.delete_subject("m-only", session=_FakeSession(), tenant_id=None) + + fire.assert_called_once() diff --git a/tests/test_webhooks.py b/tests/test_webhooks.py index 65cca4c..6f31413 100644 --- a/tests/test_webhooks.py +++ b/tests/test_webhooks.py @@ -327,3 +327,105 @@ async def test_get_event_status_without_tenant_filter_returns_event(): ): result = await webhooks.get_event_status(row.id) assert result is not None and result["id"] == str(row.id) + + +# ── purge_events (issue #279) ──────────────────────────────────────────────── +# +# purge_events() is an operator-facing DESTRUCTIVE delete. These tests lock in +# its two safety guards (no-filter, terminal-status) and its filter scoping so a +# future refactor that drops a guard turns CI red. + + +class _CaptureSession: + """Async-context session that records the statement passed to execute() and + reports a fixed rowcount, so tests can assert what the DELETE targets + without a live database.""" + + def __init__(self, rowcount: int = 0): + self.executed = None + self.committed = False + self._rowcount = rowcount + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return False + + async def execute(self, statement): + self.executed = statement + return SimpleNamespace(rowcount=self._rowcount) + + async def commit(self): + self.committed = True + + +def _purge_sql(session) -> str: + """The captured DELETE rendered with literal values inlined, for asserting + on the WHERE clause.""" + from sqlalchemy.dialects import postgresql + + return str( + session.executed.compile( + dialect=postgresql.dialect(), + compile_kwargs={"literal_binds": True}, + ) + ) + + +async def test_purge_events_requires_at_least_one_filter(): + """No-filter guard: purge_events() with no filters raises ValueError instead + of wiping every delivered + dead-letter event in one call.""" + with pytest.raises(ValueError, match="at least one filter"): + await webhooks.purge_events() + + +async def test_purge_events_rejects_non_terminal_status(): + """Terminal-status guard: 'pending' is rejected — those events may still be + in flight in the delivery loop.""" + with pytest.raises(ValueError, match="must be one of"): + await webhooks.purge_events(status="pending") + + +async def test_purge_events_status_filter_targets_only_that_status(): + """purge_events(status='delivered') deletes only delivered rows (dead_letter + is left untouched) and returns the deleted row count.""" + session = _CaptureSession(rowcount=3) + with patch( + "server.services.webhooks.get_session_factory", return_value=lambda: session + ): + count = await webhooks.purge_events(status="delivered") + + assert count == 3 + assert session.committed + sql = _purge_sql(session) + assert "'delivered'" in sql + assert "'dead_letter'" not in sql # only the requested status is purged + + +async def test_purge_events_without_status_targets_all_terminal_only(): + """With only an event_type filter, both terminal statuses are in scope and + 'pending' never is.""" + session = _CaptureSession(rowcount=2) + with patch( + "server.services.webhooks.get_session_factory", return_value=lambda: session + ): + count = await webhooks.purge_events(event_type="episode.created") + + assert count == 2 + sql = _purge_sql(session) + assert "'delivered'" in sql and "'dead_letter'" in sql + assert "'pending'" not in sql + assert "'episode.created'" in sql # event_type filter applied + + +async def test_purge_events_tenant_filter_applied(): + """tenant_id filter scopes the delete to a single tenant.""" + session = _CaptureSession(rowcount=1) + with patch( + "server.services.webhooks.get_session_factory", return_value=lambda: session + ): + count = await webhooks.purge_events(tenant_id="tenant-a") + + assert count == 1 + assert "'tenant-a'" in _purge_sql(session)