diff --git a/src/crud/__init__.py b/src/crud/__init__.py index ff8f87ac..58bcd9f0 100644 --- a/src/crud/__init__.py +++ b/src/crud/__init__.py @@ -9,6 +9,7 @@ from .document import ( create_observations, delete_document, delete_document_by_id, + delete_documents, fetch_documents_by_ids, get_all_documents, get_child_observations, @@ -95,6 +96,7 @@ __all__ = [ "query_external_vector_document_ids", "delete_document", "delete_document_by_id", + "delete_documents", # Message "create_messages", "get_messages", diff --git a/src/crud/document.py b/src/crud/document.py index ba652eea..97a57e36 100644 --- a/src/crud/document.py +++ b/src/crud/document.py @@ -661,6 +661,48 @@ async def delete_document( await db.commit() +async def delete_documents( + db: AsyncSession, + workspace_name: str, + document_ids: Sequence[str], + *, + observer: str, + observed: str, + session_name: str | None = None, +) -> list[tuple[str, str]]: + """ + Soft-delete multiple documents in a single UPDATE ... RETURNING statement. + + Returns (id, level) tuples for rows that actually got deleted — i.e. rows + that matched the workspace/observer/observed filter and were not already + soft-deleted. IDs that didn't match are silently skipped; callers can diff + the returned ids against the input to detect misses. + """ + if not document_ids: + return [] + + conditions = [ + models.Document.id.in_(document_ids), + models.Document.workspace_name == workspace_name, + models.Document.observer == observer, + models.Document.observed == observed, + models.Document.deleted_at.is_(None), + ] + if session_name is not None: + conditions.append(models.Document.session_name == session_name) + + stmt = ( + update(models.Document) + .where(*conditions) + .values(deleted_at=func.now()) + .returning(models.Document.id, models.Document.level) + ) + result = await db.execute(stmt) + rows = result.all() + await db.commit() + return [(row.id, row.level) for row in rows] + + async def delete_document_by_id( db: AsyncSession, workspace_name: str, diff --git a/src/telemetry/events/agent.py b/src/telemetry/events/agent.py index 77abe2c5..fe7ff72b 100644 --- a/src/telemetry/events/agent.py +++ b/src/telemetry/events/agent.py @@ -121,7 +121,7 @@ class AgentToolConclusionsDeletedEvent(BaseEvent): """ _event_type: ClassVar[str] = "agent.tool.conclusions.deleted" - _schema_version: ClassVar[int] = 1 + _schema_version: ClassVar[int] = 2 _category: ClassVar[str] = "agent" # Run identification @@ -141,6 +141,10 @@ class AgentToolConclusionsDeletedEvent(BaseEvent): # What was deleted conclusion_count: int = Field(..., description="Number of conclusions deleted") + levels: list[str] = Field( + default_factory=list, + description="Level of each deleted conclusion (e.g., ['explicit', 'deductive', 'deductive'])", + ) def get_resource_id(self) -> str: """Resource ID includes run_id and iteration for uniqueness.""" diff --git a/src/utils/agent_tools.py b/src/utils/agent_tools.py index 36168f34..d3bd93ab 100644 --- a/src/utils/agent_tools.py +++ b/src/utils/agent_tools.py @@ -1812,22 +1812,24 @@ async def _handle_delete_observations( if not observation_ids: return "ERROR: observation_ids list is empty" - deleted_count = 0 async with ctx.db_lock, tracked_db("tool.delete_observations") as db: - for obs_id in observation_ids: - try: - await crud.delete_document( - db, - workspace_name=ctx.workspace_name, - document_id=obs_id, - observer=ctx.observer, - observed=ctx.observed, - ) - deleted_count += 1 - except Exception as e: - logger.warning("Failed to delete observation %s: %s", obs_id, e) + deleted = await crud.delete_documents( + db, + workspace_name=ctx.workspace_name, + document_ids=observation_ids, + observer=ctx.observer, + observed=ctx.observed, + ) - # Emit telemetry event if context is available + deleted_ids = {doc_id for doc_id, _ in deleted} + for obs_id in observation_ids: + if obs_id not in deleted_ids: + logger.warning( + "Failed to delete observation %s (not found, already deleted, or wrong scope)", + obs_id, + ) + + deleted_count = len(deleted) if deleted_count > 0 and ctx.run_id and ctx.agent_type and ctx.parent_category: emit( AgentToolConclusionsDeletedEvent( @@ -1839,6 +1841,7 @@ async def _handle_delete_observations( observer=ctx.observer, observed=ctx.observed, conclusion_count=deleted_count, + levels=[level for _, level in deleted], ) ) diff --git a/tests/integration/test_telemetry.py b/tests/integration/test_telemetry.py index 96b45c53..9a627885 100644 --- a/tests/integration/test_telemetry.py +++ b/tests/integration/test_telemetry.py @@ -246,6 +246,7 @@ def create_conclusions_deleted_event( observer="assistant", observed="user_peer", conclusion_count=3, + levels=["explicit", "deductive", "explicit"], ) @@ -651,6 +652,7 @@ class TestAllEventTypes: received = mock_transport.received_events[0] assert received["type"] == "agent.tool.conclusions.deleted" assert received["data"]["conclusion_count"] == 3 + assert received["data"]["levels"] == ["explicit", "deductive", "explicit"] @pytest.mark.asyncio async def test_peer_card_updated_event( diff --git a/tests/telemetry/conftest.py b/tests/telemetry/conftest.py index bf296ef3..a52194e2 100644 --- a/tests/telemetry/conftest.py +++ b/tests/telemetry/conftest.py @@ -184,6 +184,7 @@ def sample_conclusions_deleted_event( observer="assistant", observed="user_peer", conclusion_count=3, + levels=["explicit", "deductive", "explicit"], ) diff --git a/tests/telemetry/test_events.py b/tests/telemetry/test_events.py index 94f34aaa..c60c3923 100644 --- a/tests/telemetry/test_events.py +++ b/tests/telemetry/test_events.py @@ -422,7 +422,7 @@ class TestAgentToolConclusionsDeletedEvent: def test_schema_version(self): """schema_version() returns correct value.""" - assert AgentToolConclusionsDeletedEvent.schema_version() == 1 + assert AgentToolConclusionsDeletedEvent.schema_version() == 2 def test_category(self): """category() returns correct value.""" diff --git a/tests/utils/test_agent_tools.py b/tests/utils/test_agent_tools.py index bb0ff900..4114cfe7 100644 --- a/tests/utils/test_agent_tools.py +++ b/tests/utils/test_agent_tools.py @@ -140,6 +140,9 @@ def make_tool_context(tool_test_data: Any) -> Callable[..., ToolContext]: include_observation_ids: bool = False, history_token_limit: int = 8192, session_name: str | None = None, + run_id: str | None = None, + agent_type: str | None = None, + parent_category: str | None = None, ) -> ToolContext: return ToolContext( workspace_name=workspace.name, @@ -150,6 +153,9 @@ def make_tool_context(tool_test_data: Any) -> Callable[..., ToolContext]: include_observation_ids=include_observation_ids, history_token_limit=history_token_limit, db_lock=shared_lock, + run_id=run_id, + agent_type=agent_type, + parent_category=parent_category, ) return _make_context @@ -412,6 +418,82 @@ class TestDeleteObservations: # Should report 0 deleted (graceful handling) assert "Deleted 0 observations" in result + async def test_delete_batch_emits_levels_for_successful_only( + self, + db_session: AsyncSession, + tool_test_data: Any, + make_tool_context: Callable[..., ToolContext], + monkeypatch: pytest.MonkeyPatch, + ): + """Batch delete with mixed levels emits levels only for rows actually deleted.""" + workspace, peer1, peer2, session, _messages, documents = tool_test_data + + # Add two extra documents with non-explicit levels so the batch spans levels. + deductive_doc = models.Document( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + content="Works in tech", + embedding=[0.42] * 1536, + session_name=session.name, + level="deductive", + metadata={}, + ) + inductive_doc = models.Document( + workspace_name=workspace.name, + observer=peer1.name, + observed=peer2.name, + content="Tends to be an early riser", + embedding=[0.43] * 1536, + session_name=session.name, + level="inductive", + metadata={}, + ) + db_session.add_all([deductive_doc, inductive_doc]) + await db_session.flush() + await db_session.refresh(deductive_doc) + await db_session.refresh(inductive_doc) + await db_session.commit() + + # Capture emitted telemetry events. + from src.telemetry.events import AgentToolConclusionsDeletedEvent + from src.telemetry.events.base import BaseEvent + from src.utils import agent_tools as agent_tools_module + + captured: list[BaseEvent] = [] + + def _capture(event: BaseEvent) -> None: + captured.append(event) + + monkeypatch.setattr(agent_tools_module, "emit", _capture) + + ctx = make_tool_context( + include_observation_ids=True, + run_id="test_run", + agent_type="deduction", + parent_category="dream", + ) + + explicit_doc_id = documents[0].id + ids_to_delete = [ + explicit_doc_id, + deductive_doc.id, + inductive_doc.id, + "nonexistent_id_12345", + ] + + result = await _handle_delete_observations( + ctx, {"observation_ids": ids_to_delete} + ) + + assert "Deleted 3 observations" in result + assert len(captured) == 1 + event = captured[0] + assert isinstance(event, AgentToolConclusionsDeletedEvent) + assert event.conclusion_count == 3 + # RETURNING order is not guaranteed; compare as multiset. + assert sorted(event.levels) == sorted(["explicit", "deductive", "inductive"]) + @pytest.mark.asyncio class TestGetRecentObservations: