diff --git a/hindsight-api/hindsight_api/api/http.py b/hindsight-api/hindsight_api/api/http.py index a21c9cc5..c93c8f8a 100644 --- a/hindsight-api/hindsight_api/api/http.py +++ b/hindsight-api/hindsight_api/api/http.py @@ -1235,6 +1235,14 @@ class DeleteResponse(BaseModel): deleted_count: int | None = None +class ClearMemoryObservationsResponse(BaseModel): + """Response model for clearing observations for a specific memory.""" + + model_config = ConfigDict(json_schema_extra={"example": {"deleted_count": 3}}) + + deleted_count: int + + class BankStatsResponse(BaseModel): """Response model for bank statistics endpoint.""" @@ -3549,6 +3557,40 @@ def _register_routes(app: FastAPI): logger.error(f"Error in DELETE /v1/default/banks/{bank_id}/observations: {error_detail}") raise HTTPException(status_code=500, detail=str(e)) + @app.delete( + "/v1/default/banks/{bank_id}/memories/{memory_id}/observations", + response_model=ClearMemoryObservationsResponse, + summary="Clear observations for a memory", + description="Delete all observations derived from a specific memory and reset it for re-consolidation. " + "The memory itself is not deleted. A consolidation job is triggered automatically so the memory " + "will produce fresh observations on the next consolidation run.", + operation_id="clear_memory_observations", + tags=["Memory"], + ) + async def api_clear_memory_observations( + bank_id: str, + memory_id: str, + request_context: RequestContext = Depends(get_request_context), + ): + """Clear all observations derived from a specific memory.""" + try: + result = await app.state.memory.clear_observations_for_memory( + bank_id=bank_id, + memory_id=memory_id, + request_context=request_context, + ) + return ClearMemoryObservationsResponse(deleted_count=result["deleted_count"]) + except (AuthenticationError, HTTPException): + raise + except Exception as e: + import traceback + + error_detail = f"{str(e)}\n\nTraceback:\n{traceback.format_exc()}" + logger.error( + f"Error in DELETE /v1/default/banks/{bank_id}/memories/{memory_id}/observations: {error_detail}" + ) + raise HTTPException(status_code=500, detail=str(e)) + @app.get( "/v1/default/banks/{bank_id}/config", response_model=BankConfigResponse, diff --git a/hindsight-api/hindsight_api/engine/consolidation/prompts.py b/hindsight-api/hindsight_api/engine/consolidation/prompts.py index 5b4142b5..13f1b873 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/prompts.py +++ b/hindsight-api/hindsight_api/engine/consolidation/prompts.py @@ -39,7 +39,8 @@ Rules: - Keep specifics: names, numbers, locations. Never abstract into general principles. - NEVER merge observations about different people or unrelated topics. - REDUNDANT: same info worded differently → update existing. -- CONTRADICTION/UPDATE: capture both states with temporal markers ("used to X, now Y").""" +- CONTRADICTION/UPDATE: capture both states with temporal markers ("used to X, now Y"). +- RESOLVE REFERENCES: When a new fact provides a concrete value that resolves a vague placeholder in an existing observation (e.g., a location that corresponds to "home country", "hometown", "birthplace", "native language", "her ex", "that city"), UPDATE the existing observation to embed the resolved value explicitly. Example: new fact mentions grandma in Sweden + existing observation says "moved from her home country" → update to state "home country is Sweden".""" def build_consolidation_prompt(observations_mission: str | None = None) -> str: diff --git a/hindsight-api/hindsight_api/engine/memory_engine.py b/hindsight-api/hindsight_api/engine/memory_engine.py index 3fbb2a1a..fc0486fe 100644 --- a/hindsight-api/hindsight_api/engine/memory_engine.py +++ b/hindsight-api/hindsight_api/engine/memory_engine.py @@ -3169,14 +3169,22 @@ class MemoryEngine(MemoryEngineInterface): """ await self._authenticate_tenant(request_context) pool = await self._get_pool() + invalidated_obs = 0 async with acquire_with_retry(pool) as conn: async with conn.transaction(): - # Get memory unit IDs before deletion (for mental model invalidation) + # Get memory unit IDs before deletion (for observation cleanup) unit_rows = await conn.fetch( - f"SELECT id FROM {fq_table('memory_units')} WHERE document_id = $1", document_id + f"SELECT id FROM {fq_table('memory_units')} WHERE document_id = $1 AND fact_type IN ('experience', 'world')", + document_id, ) unit_ids = [str(row["id"]) for row in unit_rows] - units_count = len(unit_ids) + units_count = await conn.fetchval( + f"SELECT COUNT(*) FROM {fq_table('memory_units')} WHERE document_id = $1", document_id + ) + + # Invalidate observations referencing these memories before deletion + if unit_ids: + invalidated_obs = await self._delete_stale_observations_for_memories(conn, bank_id, unit_ids) # Delete document (cascades to memory_units and all their links) deleted = await conn.fetchval( @@ -3185,11 +3193,15 @@ class MemoryEngine(MemoryEngineInterface): bank_id, ) - # Invalidate deleted fact IDs from mental models - if deleted and unit_ids: - await self._invalidate_facts_from_mental_models(conn, bank_id, unit_ids) + result = { + "document_deleted": 1 if deleted else 0, + "memory_units_deleted": units_count if deleted else 0, + } - return {"document_deleted": 1 if deleted else 0, "memory_units_deleted": units_count if deleted else 0} + if invalidated_obs > 0: + await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context) + + return result async def delete_memory_unit( self, @@ -3205,6 +3217,9 @@ class MemoryEngine(MemoryEngineInterface): - All links to this unit (memory_links where to_unit_id = unit_id) - All entity associations (unit_entities where unit_id = unit_id) + Observations referencing this memory are deleted and their other source + memories are reset for re-consolidation. + Args: unit_id: UUID of the memory unit to delete request_context: Request context for authentication. @@ -3214,21 +3229,30 @@ class MemoryEngine(MemoryEngineInterface): """ await self._authenticate_tenant(request_context) pool = await self._get_pool() + invalidated_obs = 0 + bank_id_for_consolidation: str | None = None async with acquire_with_retry(pool) as conn: async with conn.transaction(): - # Get bank_id before deletion (for mental model invalidation) - bank_id = await conn.fetchval(f"SELECT bank_id FROM {fq_table('memory_units')} WHERE id = $1", unit_id) + # Get bank_id and fact_type before deletion + row = await conn.fetchrow( + f"SELECT bank_id, fact_type FROM {fq_table('memory_units')} WHERE id = $1", + unit_id, + ) + bank_id = row["bank_id"] if row else None + fact_type = row["fact_type"] if row else None + + # Invalidate observations before deletion (only for source memory types) + if bank_id and fact_type in ("experience", "world"): + invalidated_obs = await self._delete_stale_observations_for_memories(conn, bank_id, [unit_id]) + if invalidated_obs > 0: + bank_id_for_consolidation = bank_id # Delete the memory unit (cascades to links and associations) deleted = await conn.fetchval( f"DELETE FROM {fq_table('memory_units')} WHERE id = $1 RETURNING id", unit_id ) - # Invalidate deleted fact ID from mental models - if deleted and bank_id: - await self._invalidate_facts_from_mental_models(conn, bank_id, [str(deleted)]) - - return { + result = { "success": deleted is not None, "unit_id": str(deleted) if deleted else None, "message": "Memory unit and all its links deleted successfully" @@ -3236,6 +3260,11 @@ class MemoryEngine(MemoryEngineInterface): else "Memory unit not found", } + if bank_id_for_consolidation: + await self.submit_async_consolidation(bank_id=bank_id_for_consolidation, request_context=request_context) + + return result + async def delete_bank( self, bank_id: str, @@ -3264,12 +3293,27 @@ class MemoryEngine(MemoryEngineInterface): """ await self._authenticate_tenant(request_context) pool = await self._get_pool() + invalidated_obs = 0 + result: dict[str, int] = {} async with acquire_with_retry(pool) as conn: # Ensure connection is not in read-only mode (can happen with connection poolers) await conn.execute("SET SESSION CHARACTERISTICS AS TRANSACTION READ WRITE") async with conn.transaction(): try: if fact_type: + # For source memory types, clean up observations before deletion + if fact_type in ("experience", "world"): + unit_id_rows = await conn.fetch( + f"SELECT id FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = $2", + bank_id, + fact_type, + ) + unit_ids = [str(row["id"]) for row in unit_id_rows] + if unit_ids: + invalidated_obs = await self._delete_stale_observations_for_memories( + conn, bank_id, unit_ids + ) + # Delete only memories of a specific fact type units_count = await conn.fetchval( f"SELECT COUNT(*) FROM {fq_table('memory_units')} WHERE bank_id = $1 AND fact_type = $2", @@ -3284,9 +3328,9 @@ class MemoryEngine(MemoryEngineInterface): # Note: We don't delete entities when fact_type is specified, # as they may be referenced by other memory units - return {"memory_units_deleted": units_count, "entities_deleted": 0} + result = {"memory_units_deleted": units_count, "entities_deleted": 0} else: - # Delete all data for the bank + # Delete all data for the bank — observations are included, no invalidation needed units_count = await conn.fetchval( f"SELECT COUNT(*) FROM {fq_table('memory_units')} WHERE bank_id = $1", bank_id ) @@ -3309,7 +3353,7 @@ class MemoryEngine(MemoryEngineInterface): # Delete the bank profile itself await conn.execute(f"DELETE FROM {fq_table('banks')} WHERE bank_id = $1", bank_id) - return { + result = { "memory_units_deleted": units_count, "entities_deleted": entities_count, "documents_deleted": documents_count, @@ -3319,6 +3363,11 @@ class MemoryEngine(MemoryEngineInterface): except Exception as e: raise Exception(f"Failed to delete agent data: {str(e)}") + if invalidated_obs > 0: + await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context) + + return result + async def clear_observations( self, bank_id: str, @@ -3351,6 +3400,12 @@ class MemoryEngine(MemoryEngineInterface): bank_id, ) + # Reset consolidated_at on source memories so they get re-consolidated + await conn.execute( + f"UPDATE {fq_table('memory_units')} SET consolidated_at = NULL WHERE bank_id = $1 AND fact_type IN ('experience', 'world')", + bank_id, + ) + # Reset consolidation timestamp await conn.execute( f"UPDATE {fq_table('banks')} SET last_consolidated_at = NULL WHERE bank_id = $1", @@ -3359,6 +3414,59 @@ class MemoryEngine(MemoryEngineInterface): return {"deleted_count": count or 0} + async def clear_observations_for_memory( + self, + bank_id: str, + memory_id: str, + *, + request_context: "RequestContext", + ) -> dict[str, int]: + """ + Clear all observations derived from a specific memory and mark source memories + (including the given memory itself) for re-consolidation. + + Unlike deleting the memory, the memory itself is preserved. This is useful + when you want to force re-consolidation of a specific memory's observations + without losing the underlying fact. + + Args: + bank_id: Bank ID + memory_id: ID of the memory whose observations should be cleared + request_context: Request context for authentication. + + Returns: + Dictionary with count of deleted observations + """ + await self._authenticate_tenant(request_context) + pool = await self._get_pool() + deleted_count = 0 + + async with acquire_with_retry(pool) as conn: + async with conn.transaction(): + import uuid as uuid_module + + deleted_count = await self._delete_stale_observations_for_memories(conn, bank_id, [memory_id]) + + # Also reset this memory's own consolidated_at so it gets re-consolidated + # (the memory was a source for the deleted observations, so it needs new ones) + if deleted_count > 0: + await conn.execute( + f""" + UPDATE {fq_table("memory_units")} + SET consolidated_at = NULL + WHERE id = $1 + AND bank_id = $2 + AND fact_type IN ('experience', 'world') + """, + uuid_module.UUID(memory_id), + bank_id, + ) + + if deleted_count > 0: + await self.submit_async_consolidation(bank_id=bank_id, request_context=request_context) + + return {"deleted_count": deleted_count} + async def run_consolidation( self, bank_id: str, @@ -3747,7 +3855,7 @@ class MemoryEngine(MemoryEngineInterface): units = await conn.fetch( f""" - SELECT id, text, event_date, context, fact_type, mentioned_at, occurred_start, occurred_end, chunk_id + SELECT id, text, event_date, context, fact_type, mentioned_at, occurred_start, occurred_end, chunk_id, proof_count FROM {fq_table("memory_units")} {where_clause} ORDER BY mentioned_at DESC NULLS LAST, created_at DESC @@ -3799,6 +3907,7 @@ class MemoryEngine(MemoryEngineInterface): "occurred_end": row["occurred_end"].isoformat() if row["occurred_end"] else None, "entities": ", ".join(entities) if entities else "", "chunk_id": row["chunk_id"] if row["chunk_id"] else None, + "proof_count": row["proof_count"] if row["proof_count"] is not None else 1, } ) @@ -4974,61 +5083,88 @@ class MemoryEngine(MemoryEngineInterface): ) return count or 0 - async def _invalidate_facts_from_mental_models( + async def _delete_stale_observations_for_memories( self, conn, bank_id: str, fact_ids: list[str], ) -> int: """ - Remove fact IDs from observation source_memory_ids when memories are deleted. + Handle cleanup of observations when source memories are deleted. - Observations are stored in memory_units with fact_type='observation' - and have a source_memory_ids column (UUID[]) tracking their source memories. + For each observation referencing any of the deleted fact IDs: + 1. Delete the observation (its text is stale without those source memories) + 2. Reset consolidated_at=NULL on the remaining source memories so they get re-consolidated + + Must be called within an active transaction, before the source memories are deleted. Args: - conn: Database connection + conn: Database connection (must be in an active transaction) bank_id: Bank identifier - fact_ids: List of fact IDs to remove from observations + fact_ids: List of fact IDs (as strings) that are being deleted Returns: - Number of observations updated + Number of observations deleted """ if not fact_ids: return 0 - # Convert string IDs to UUIDs for the array comparison import uuid as uuid_module fact_uuids = [uuid_module.UUID(fid) for fid in fact_ids] - # Update observations (memory_units with fact_type='observation') - # by removing the deleted fact IDs from source_memory_ids - # Use array subtraction: source_memory_ids - deleted_ids - result = await conn.execute( + # Find all observations referencing any of the deleted facts + affected_obs = await conn.fetch( f""" - UPDATE {fq_table("memory_units")} - SET source_memory_ids = ( - SELECT COALESCE(array_agg(elem), ARRAY[]::uuid[]) - FROM unnest(source_memory_ids) AS elem - WHERE elem != ALL($2::uuid[]) - ), - updated_at = NOW() + SELECT id, source_memory_ids + FROM {fq_table("memory_units")} WHERE bank_id = $1 - AND fact_type = 'observation' - AND source_memory_ids && $2::uuid[] + AND fact_type = 'observation' + AND source_memory_ids && $2::uuid[] """, bank_id, fact_uuids, ) - # Parse the result to get number of updated rows - updated_count = int(result.split()[-1]) if result and "UPDATE" in result else 0 - if updated_count > 0: - logger.info( - f"[OBSERVATIONS] Invalidated {len(fact_ids)} fact IDs from {updated_count} observations in bank {bank_id}" + if not affected_obs: + return 0 + + # Collect observation IDs to delete and remaining source memory IDs to reset + deleted_set = {str(uid) for uid in fact_uuids} + obs_ids = [obs["id"] for obs in affected_obs] + seen_remaining: set[str] = set() + remaining_source_ids: list[uuid_module.UUID] = [] + + for obs in affected_obs: + for src_id in obs["source_memory_ids"] or []: + src_str = str(src_id) + if src_str not in deleted_set and src_str not in seen_remaining: + remaining_source_ids.append(src_id) + seen_remaining.add(src_str) + + # Delete the stale observations + await conn.execute( + f"DELETE FROM {fq_table('memory_units')} WHERE id = ANY($1::uuid[])", + obs_ids, + ) + + # Reset consolidated_at on remaining source memories so they get re-consolidated + if remaining_source_ids: + await conn.execute( + f""" + UPDATE {fq_table("memory_units")} + SET consolidated_at = NULL + WHERE id = ANY($1::uuid[]) + AND fact_type IN ('experience', 'world') + """, + remaining_source_ids, ) - return updated_count + + logger.info( + f"[OBSERVATIONS] Deleted {len(obs_ids)} observations, reset {len(remaining_source_ids)} " + f"source memories for re-consolidation in bank {bank_id}" + ) + return len(obs_ids) # ========================================================================= # MENTAL MODELS (CONSOLIDATED) - Read-only access to auto-consolidated mental models diff --git a/hindsight-api/tests/test_observation_invalidation.py b/hindsight-api/tests/test_observation_invalidation.py new file mode 100644 index 00000000..97511ddb --- /dev/null +++ b/hindsight-api/tests/test_observation_invalidation.py @@ -0,0 +1,457 @@ +""" +Tests for observation invalidation when source memories are deleted. + +These tests verify that: +1. Observations are deleted (not just updated) when their source memories are removed +2. Remaining source memories are reset for re-consolidation (consolidated_at=NULL) +3. The clear_observations_for_memory method correctly clears observations and + resets the target memory itself for re-consolidation +4. delete_bank(fact_type=...) also cleans up affected observations +""" +import uuid +from unittest.mock import AsyncMock, patch + +import pytest + +from hindsight_api import RequestContext +from hindsight_api.engine.memory_engine import MemoryEngine + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +async def _insert_memory(conn, bank_id: str, text: str, fact_type: str = "experience") -> uuid.UUID: + """Insert a memory unit directly, bypassing LLM retain pipeline.""" + mem_id = uuid.uuid4() + await conn.execute( + """ + INSERT INTO memory_units (id, bank_id, text, fact_type, event_date, created_at, updated_at, consolidated_at) + VALUES ($1, $2, $3, $4, NOW(), NOW(), NOW(), NOW()) + """, + mem_id, + bank_id, + text, + fact_type, + ) + return mem_id + + +async def _insert_observation( + conn, bank_id: str, text: str, source_memory_ids: list[uuid.UUID] +) -> uuid.UUID: + """Insert an observation unit directly.""" + obs_id = uuid.uuid4() + await conn.execute( + """ + INSERT INTO memory_units ( + id, bank_id, text, fact_type, event_date, source_memory_ids, proof_count, created_at, updated_at + ) VALUES ($1, $2, $3, 'observation', NOW(), $4, $5, NOW(), NOW()) + """, + obs_id, + bank_id, + text, + source_memory_ids, + len(source_memory_ids), + ) + return obs_id + + +async def _get_observation_ids(conn, bank_id: str) -> list[str]: + rows = await conn.fetch( + "SELECT id FROM memory_units WHERE bank_id = $1 AND fact_type = 'observation'", + bank_id, + ) + return [str(r["id"]) for r in rows] + + +async def _get_consolidated_at(conn, memory_id: uuid.UUID): + return await conn.fetchval( + "SELECT consolidated_at FROM memory_units WHERE id = $1", + memory_id, + ) + + +async def _ensure_bank(memory: MemoryEngine, bank_id: str, request_context: RequestContext): + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + + +# --------------------------------------------------------------------------- +# Tests: delete_memory_unit +# --------------------------------------------------------------------------- + +class TestDeleteMemoryUnitObservationCleanup: + + @pytest.mark.asyncio + async def test_deleting_source_memory_removes_observation( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Deleting a source memory removes observations derived from it.""" + bank_id = f"test-invalidate-del-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice goes hiking every weekend.") + obs_id = await _insert_observation(conn, bank_id, "Alice enjoys hiking regularly.", [m1, m2]) + + await memory.delete_memory_unit(str(m1), request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids, "Observation should have been deleted" + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_deleting_source_memory_resets_remaining_source_consolidated_at( + self, memory: MemoryEngine, request_context: RequestContext + ): + """After deleting a source memory, remaining source memories are reset for re-consolidation.""" + bank_id = f"test-invalidate-reset-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice goes hiking every weekend.") + await _insert_observation(conn, bank_id, "Alice enjoys hiking regularly.", [m1, m2]) + + # Verify m2 starts with consolidated_at set + assert await _get_consolidated_at(conn, m2) is not None + + # Patch out consolidation so it doesn't re-set consolidated_at before we can check it + with patch.object(memory, "submit_async_consolidation", new=AsyncMock()): + await memory.delete_memory_unit(str(m1), request_context=request_context) + + async with pool.acquire() as conn: + # m2 should have consolidated_at reset to NULL + consolidated_at = await _get_consolidated_at(conn, m2) + assert consolidated_at is None, "Remaining source memory should be reset for re-consolidation" + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_deleting_non_source_memory_leaves_observations_intact( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Deleting a memory that is not a source of any observation leaves observations unchanged.""" + bank_id = f"test-invalidate-noop-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice goes hiking every weekend.") + unrelated = await _insert_memory(conn, bank_id, "Bob likes cycling.") + obs_id = await _insert_observation(conn, bank_id, "Alice enjoys hiking regularly.", [m1, m2]) + + await memory.delete_memory_unit(str(unrelated), request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) in obs_ids, "Observation should remain untouched" + # m1 and m2 should still be consolidated + assert await _get_consolidated_at(conn, m1) is not None + assert await _get_consolidated_at(conn, m2) is not None + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_deleting_sole_source_memory_removes_observation_no_remaining_reset( + self, memory: MemoryEngine, request_context: RequestContext + ): + """When an observation has only one source and it's deleted, observation is removed with no remaining memories to reset.""" + bank_id = f"test-invalidate-sole-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + obs_id = await _insert_observation(conn, bank_id, "Alice enjoys hiking.", [m1]) + + await memory.delete_memory_unit(str(m1), request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids, "Observation should have been deleted" + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_deleting_observation_type_memory_does_not_trigger_invalidation( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Deleting a memory with fact_type='observation' directly does not trigger invalidation logic.""" + bank_id = f"test-invalidate-obstype-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + obs_id = await _insert_observation(conn, bank_id, "Alice enjoys hiking.", [m1]) + + # Delete the observation directly (not the source memory) + await memory.delete_memory_unit(str(obs_id), request_context=request_context) + + async with pool.acquire() as conn: + # Source memory should still be consolidated (not reset) + assert await _get_consolidated_at(conn, m1) is not None + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids + + await memory.delete_bank(bank_id, request_context=request_context) + + +# --------------------------------------------------------------------------- +# Tests: delete_document +# --------------------------------------------------------------------------- + +class TestDeleteDocumentObservationCleanup: + + @pytest.mark.asyncio + async def test_deleting_document_removes_observations( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Deleting a document removes observations derived from its memory units.""" + bank_id = f"test-invalidate-doc-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + + # Create a document and attach memories to it + async with pool.acquire() as conn: + doc_id = str(uuid.uuid4()) # documents.id is TEXT + await conn.execute( + """ + INSERT INTO documents (id, bank_id, original_text, content_hash, created_at, updated_at) + VALUES ($1, $2, 'some doc', 'hash123', NOW(), NOW()) + """, + doc_id, + bank_id, + ) + m1 = uuid.uuid4() + m2 = uuid.uuid4() + for mem_id, text in [(m1, "Alice loves hiking."), (m2, "Alice goes hiking every weekend.")]: + await conn.execute( + """ + INSERT INTO memory_units (id, bank_id, text, fact_type, event_date, document_id, created_at, updated_at, consolidated_at) + VALUES ($1, $2, $3, 'experience', NOW(), $4, NOW(), NOW(), NOW()) + """, + mem_id, + bank_id, + text, + doc_id, + ) + + # Standalone memory (not in document) + m3 = await _insert_memory(conn, bank_id, "Alice is an avid outdoor person.") + + # Observation referencing both doc memories and the standalone memory + obs_id = await _insert_observation( + conn, bank_id, "Alice enjoys outdoor activities.", [m1, m2, m3] + ) + + # Patch out consolidation so it doesn't re-set consolidated_at before we can check it + with patch.object(memory, "submit_async_consolidation", new=AsyncMock()): + await memory.delete_document(str(doc_id), bank_id, request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids, "Observation should have been deleted" + + # m3 (remaining source) should be reset for re-consolidation + consolidated_at = await _get_consolidated_at(conn, m3) + assert consolidated_at is None, "Remaining source memory should be reset" + + await memory.delete_bank(bank_id, request_context=request_context) + + +# --------------------------------------------------------------------------- +# Tests: delete_bank with fact_type filter +# --------------------------------------------------------------------------- + +class TestDeleteBankByTypeObservationCleanup: + + @pytest.mark.asyncio + async def test_clearing_experience_memories_removes_affected_observations( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Clearing all experience memories removes observations sourced from them.""" + bank_id = f"test-invalidate-banktype-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + exp1 = await _insert_memory(conn, bank_id, "Alice went hiking last week.", "experience") + world1 = await _insert_memory(conn, bank_id, "Alice is a hiker.", "world") + obs_id = await _insert_observation( + conn, bank_id, "Alice is a regular hiker.", [exp1, world1] + ) + + # Patch out consolidation so it doesn't re-set consolidated_at before we can check it + with patch.object(memory, "submit_async_consolidation", new=AsyncMock()): + await memory.delete_bank(bank_id, fact_type="experience", request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids, "Observation should have been deleted" + + # world1 (remaining source) should be reset for re-consolidation + consolidated_at = await _get_consolidated_at(conn, world1) + assert consolidated_at is None, "World memory should be reset for re-consolidation" + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_clearing_unrelated_type_leaves_observations_intact( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Clearing memories of a type that is not a source of any observation leaves observations untouched.""" + bank_id = f"test-invalidate-banktype-noop-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + world1 = await _insert_memory(conn, bank_id, "Alice is a hiker.", "world") + obs_id = await _insert_observation(conn, bank_id, "Alice is a regular hiker.", [world1]) + + # Deleting 'experience' type should not affect observations sourced only from 'world' + await memory.delete_bank(bank_id, fact_type="experience", request_context=request_context) + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) in obs_ids, "Observation should remain untouched" + + await memory.delete_bank(bank_id, request_context=request_context) + + +# --------------------------------------------------------------------------- +# Tests: clear_observations_for_memory +# --------------------------------------------------------------------------- + +class TestClearObservationsForMemory: + + @pytest.mark.asyncio + async def test_clears_observations_and_resets_all_source_memories( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Clearing observations for a memory deletes them and resets all related source memories.""" + bank_id = f"test-clear-obs-mem-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice hikes every weekend.") + obs_id = await _insert_observation(conn, bank_id, "Alice is an avid hiker.", [m1, m2]) + + # Patch out consolidation so it doesn't re-set consolidated_at before we can check it + with patch.object(memory, "submit_async_consolidation", new=AsyncMock()): + result = await memory.clear_observations_for_memory( + bank_id, str(m1), request_context=request_context + ) + + assert result["deleted_count"] == 1 + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs_id) not in obs_ids, "Observation should be deleted" + + # Both m1 (target) and m2 (remaining source) should be reset + assert await _get_consolidated_at(conn, m1) is None, "Target memory should be reset" + assert await _get_consolidated_at(conn, m2) is None, "Remaining source should be reset" + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_no_observations_returns_zero( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Returns 0 when the memory has no associated observations.""" + bank_id = f"test-clear-obs-noop-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + + result = await memory.clear_observations_for_memory( + bank_id, str(m1), request_context=request_context + ) + + assert result["deleted_count"] == 0 + + async with pool.acquire() as conn: + # Memory should still be consolidated (no observations were cleared) + assert await _get_consolidated_at(conn, m1) is not None + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_only_clears_observations_referencing_target_memory( + self, memory: MemoryEngine, request_context: RequestContext + ): + """Clearing observations for m1 does not affect observations that only reference m2.""" + bank_id = f"test-clear-obs-selective-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice hikes every weekend.") + m3 = await _insert_memory(conn, bank_id, "Alice climbed a mountain.") + + obs1_id = await _insert_observation(conn, bank_id, "Alice is an avid hiker.", [m1, m2]) + obs2_id = await _insert_observation(conn, bank_id, "Alice is a mountaineer.", [m3]) + + result = await memory.clear_observations_for_memory( + bank_id, str(m1), request_context=request_context + ) + + assert result["deleted_count"] == 1 + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs1_id) not in obs_ids, "obs1 (references m1) should be deleted" + assert str(obs2_id) in obs_ids, "obs2 (does not reference m1) should remain" + + # m3 should still be consolidated + assert await _get_consolidated_at(conn, m3) is not None + + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_multiple_observations_for_same_memory_all_cleared( + self, memory: MemoryEngine, request_context: RequestContext + ): + """All observations referencing the target memory are cleared in one call.""" + bank_id = f"test-clear-obs-multi-{uuid.uuid4().hex[:8]}" + await _ensure_bank(memory, bank_id, request_context) + + pool = await memory._get_pool() + async with pool.acquire() as conn: + m1 = await _insert_memory(conn, bank_id, "Alice loves hiking.") + m2 = await _insert_memory(conn, bank_id, "Alice hikes every weekend.") + + obs1_id = await _insert_observation(conn, bank_id, "Alice hikes often.", [m1]) + obs2_id = await _insert_observation(conn, bank_id, "Alice is outdoorsy.", [m1, m2]) + + # Patch out consolidation so it doesn't re-set consolidated_at before we can check it + with patch.object(memory, "submit_async_consolidation", new=AsyncMock()): + result = await memory.clear_observations_for_memory( + bank_id, str(m1), request_context=request_context + ) + + assert result["deleted_count"] == 2 + + async with pool.acquire() as conn: + obs_ids = await _get_observation_ids(conn, bank_id) + assert str(obs1_id) not in obs_ids + assert str(obs2_id) not in obs_ids + + # m1 and m2 should both be reset + assert await _get_consolidated_at(conn, m1) is None + assert await _get_consolidated_at(conn, m2) is None + + await memory.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-clients/go/api/openapi.yaml b/hindsight-clients/go/api/openapi.yaml index 8ca7a402..365c7ead 100644 --- a/hindsight-clients/go/api/openapi.yaml +++ b/hindsight-clients/go/api/openapi.yaml @@ -1852,6 +1852,54 @@ paths: summary: Clear all observations tags: - Banks + /v1/default/banks/{bank_id}/memories/{memory_id}/observations: + delete: + description: Delete all observations derived from a specific memory and reset + it for re-consolidation. The memory itself is not deleted. A consolidation + job is triggered automatically so the memory will produce fresh observations + on the next consolidation run. + operationId: clear_memory_observations + parameters: + - explode: false + in: path + name: bank_id + required: true + schema: + title: Bank Id + type: string + style: simple + - explode: false + in: path + name: memory_id + required: true + schema: + title: Memory Id + type: string + style: simple + - explode: false + in: header + name: authorization + required: false + schema: + nullable: true + type: string + style: simple + responses: + "200": + content: + application/json: + schema: + $ref: '#/components/schemas/ClearMemoryObservationsResponse' + description: Successful Response + "422": + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' + description: Validation Error + summary: Clear observations for a memory + tags: + - Memory /v1/default/banks/{bank_id}/config: delete: description: Reset bank configuration to defaults by removing all bank-specific @@ -2597,6 +2645,17 @@ components: - created_at - document_id title: ChunkResponse + ClearMemoryObservationsResponse: + description: Response model for clearing observations for a specific memory. + example: + deleted_count: 3 + properties: + deleted_count: + title: Deleted Count + type: integer + required: + - deleted_count + title: ClearMemoryObservationsResponse ConsolidationResponse: description: Response model for consolidation trigger endpoint. example: diff --git a/hindsight-clients/go/api_memory.go b/hindsight-clients/go/api_memory.go index 950604f3..4cc49d7e 100644 --- a/hindsight-clients/go/api_memory.go +++ b/hindsight-clients/go/api_memory.go @@ -155,6 +155,132 @@ func (a *MemoryAPIService) ClearBankMemoriesExecute(r ApiClearBankMemoriesReques return localVarReturnValue, localVarHTTPResponse, nil } +type ApiClearMemoryObservationsRequest struct { + ctx context.Context + ApiService *MemoryAPIService + bankId string + memoryId string + authorization *string +} + +func (r ApiClearMemoryObservationsRequest) Authorization(authorization string) ApiClearMemoryObservationsRequest { + r.authorization = &authorization + return r +} + +func (r ApiClearMemoryObservationsRequest) Execute() (*ClearMemoryObservationsResponse, *http.Response, error) { + return r.ApiService.ClearMemoryObservationsExecute(r) +} + +/* +ClearMemoryObservations Clear observations for a memory + +Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. + + @param ctx context.Context - for authentication, logging, cancellation, deadlines, tracing, etc. Passed from http.Request or context.Background(). + @param bankId + @param memoryId + @return ApiClearMemoryObservationsRequest +*/ +func (a *MemoryAPIService) ClearMemoryObservations(ctx context.Context, bankId string, memoryId string) ApiClearMemoryObservationsRequest { + return ApiClearMemoryObservationsRequest{ + ApiService: a, + ctx: ctx, + bankId: bankId, + memoryId: memoryId, + } +} + +// Execute executes the request +// @return ClearMemoryObservationsResponse +func (a *MemoryAPIService) ClearMemoryObservationsExecute(r ApiClearMemoryObservationsRequest) (*ClearMemoryObservationsResponse, *http.Response, error) { + var ( + localVarHTTPMethod = http.MethodDelete + localVarPostBody interface{} + formFiles []formFile + localVarReturnValue *ClearMemoryObservationsResponse + ) + + localBasePath, err := a.client.cfg.ServerURLWithContext(r.ctx, "MemoryAPIService.ClearMemoryObservations") + if err != nil { + return localVarReturnValue, nil, &GenericOpenAPIError{error: err.Error()} + } + + localVarPath := localBasePath + "/v1/default/banks/{bank_id}/memories/{memory_id}/observations" + localVarPath = strings.Replace(localVarPath, "{"+"bank_id"+"}", url.PathEscape(parameterValueToString(r.bankId, "bankId")), -1) + localVarPath = strings.Replace(localVarPath, "{"+"memory_id"+"}", url.PathEscape(parameterValueToString(r.memoryId, "memoryId")), -1) + + localVarHeaderParams := make(map[string]string) + localVarQueryParams := url.Values{} + localVarFormParams := url.Values{} + + // to determine the Content-Type header + localVarHTTPContentTypes := []string{} + + // set Content-Type header + localVarHTTPContentType := selectHeaderContentType(localVarHTTPContentTypes) + if localVarHTTPContentType != "" { + localVarHeaderParams["Content-Type"] = localVarHTTPContentType + } + + // to determine the Accept header + localVarHTTPHeaderAccepts := []string{"application/json"} + + // set Accept header + localVarHTTPHeaderAccept := selectHeaderAccept(localVarHTTPHeaderAccepts) + if localVarHTTPHeaderAccept != "" { + localVarHeaderParams["Accept"] = localVarHTTPHeaderAccept + } + if r.authorization != nil { + parameterAddToHeaderOrQuery(localVarHeaderParams, "authorization", r.authorization, "simple", "") + } + req, err := a.client.prepareRequest(r.ctx, localVarPath, localVarHTTPMethod, localVarPostBody, localVarHeaderParams, localVarQueryParams, localVarFormParams, formFiles) + if err != nil { + return localVarReturnValue, nil, err + } + + localVarHTTPResponse, err := a.client.callAPI(req) + if err != nil || localVarHTTPResponse == nil { + return localVarReturnValue, localVarHTTPResponse, err + } + + localVarBody, err := io.ReadAll(localVarHTTPResponse.Body) + localVarHTTPResponse.Body.Close() + localVarHTTPResponse.Body = io.NopCloser(bytes.NewBuffer(localVarBody)) + if err != nil { + return localVarReturnValue, localVarHTTPResponse, err + } + + if localVarHTTPResponse.StatusCode >= 300 { + newErr := &GenericOpenAPIError{ + body: localVarBody, + error: localVarHTTPResponse.Status, + } + if localVarHTTPResponse.StatusCode == 422 { + var v HTTPValidationError + err = a.client.decode(&v, localVarBody, localVarHTTPResponse.Header.Get("Content-Type")) + if err != nil { + newErr.error = err.Error() + return localVarReturnValue, localVarHTTPResponse, newErr + } + newErr.error = formatErrorMessage(localVarHTTPResponse.Status, &v) + newErr.model = v + } + return localVarReturnValue, localVarHTTPResponse, newErr + } + + err = a.client.decode(&localVarReturnValue, localVarBody, localVarHTTPResponse.Header.Get("Content-Type")) + if err != nil { + newErr := &GenericOpenAPIError{ + body: localVarBody, + error: err.Error(), + } + return localVarReturnValue, localVarHTTPResponse, newErr + } + + return localVarReturnValue, localVarHTTPResponse, nil +} + type ApiGetGraphRequest struct { ctx context.Context ApiService *MemoryAPIService diff --git a/hindsight-clients/go/model_clear_memory_observations_response.go b/hindsight-clients/go/model_clear_memory_observations_response.go new file mode 100644 index 00000000..a9eb8f4a --- /dev/null +++ b/hindsight-clients/go/model_clear_memory_observations_response.go @@ -0,0 +1,158 @@ +/* +Hindsight HTTP API + +HTTP API for Hindsight + +API version: 0.4.13 +*/ + +// Code generated by OpenAPI Generator (https://openapi-generator.tech); DO NOT EDIT. + +package hindsight + +import ( + "encoding/json" + "bytes" + "fmt" +) + +// checks if the ClearMemoryObservationsResponse type satisfies the MappedNullable interface at compile time +var _ MappedNullable = &ClearMemoryObservationsResponse{} + +// ClearMemoryObservationsResponse Response model for clearing observations for a specific memory. +type ClearMemoryObservationsResponse struct { + DeletedCount int32 `json:"deleted_count"` +} + +type _ClearMemoryObservationsResponse ClearMemoryObservationsResponse + +// NewClearMemoryObservationsResponse instantiates a new ClearMemoryObservationsResponse object +// This constructor will assign default values to properties that have it defined, +// and makes sure properties required by API are set, but the set of arguments +// will change when the set of required properties is changed +func NewClearMemoryObservationsResponse(deletedCount int32) *ClearMemoryObservationsResponse { + this := ClearMemoryObservationsResponse{} + this.DeletedCount = deletedCount + return &this +} + +// NewClearMemoryObservationsResponseWithDefaults instantiates a new ClearMemoryObservationsResponse object +// This constructor will only assign default values to properties that have it defined, +// but it doesn't guarantee that properties required by API are set +func NewClearMemoryObservationsResponseWithDefaults() *ClearMemoryObservationsResponse { + this := ClearMemoryObservationsResponse{} + return &this +} + +// GetDeletedCount returns the DeletedCount field value +func (o *ClearMemoryObservationsResponse) GetDeletedCount() int32 { + if o == nil { + var ret int32 + return ret + } + + return o.DeletedCount +} + +// GetDeletedCountOk returns a tuple with the DeletedCount field value +// and a boolean to check if the value has been set. +func (o *ClearMemoryObservationsResponse) GetDeletedCountOk() (*int32, bool) { + if o == nil { + return nil, false + } + return &o.DeletedCount, true +} + +// SetDeletedCount sets field value +func (o *ClearMemoryObservationsResponse) SetDeletedCount(v int32) { + o.DeletedCount = v +} + +func (o ClearMemoryObservationsResponse) MarshalJSON() ([]byte, error) { + toSerialize,err := o.ToMap() + if err != nil { + return []byte{}, err + } + return json.Marshal(toSerialize) +} + +func (o ClearMemoryObservationsResponse) ToMap() (map[string]interface{}, error) { + toSerialize := map[string]interface{}{} + toSerialize["deleted_count"] = o.DeletedCount + return toSerialize, nil +} + +func (o *ClearMemoryObservationsResponse) UnmarshalJSON(data []byte) (err error) { + // This validates that all required properties are included in the JSON object + // by unmarshalling the object into a generic map with string keys and checking + // that every required field exists as a key in the generic map. + requiredProperties := []string{ + "deleted_count", + } + + allProperties := make(map[string]interface{}) + + err = json.Unmarshal(data, &allProperties) + + if err != nil { + return err; + } + + for _, requiredProperty := range(requiredProperties) { + if _, exists := allProperties[requiredProperty]; !exists { + return fmt.Errorf("no value given for required property %v", requiredProperty) + } + } + + varClearMemoryObservationsResponse := _ClearMemoryObservationsResponse{} + + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + err = decoder.Decode(&varClearMemoryObservationsResponse) + + if err != nil { + return err + } + + *o = ClearMemoryObservationsResponse(varClearMemoryObservationsResponse) + + return err +} + +type NullableClearMemoryObservationsResponse struct { + value *ClearMemoryObservationsResponse + isSet bool +} + +func (v NullableClearMemoryObservationsResponse) Get() *ClearMemoryObservationsResponse { + return v.value +} + +func (v *NullableClearMemoryObservationsResponse) Set(val *ClearMemoryObservationsResponse) { + v.value = val + v.isSet = true +} + +func (v NullableClearMemoryObservationsResponse) IsSet() bool { + return v.isSet +} + +func (v *NullableClearMemoryObservationsResponse) Unset() { + v.value = nil + v.isSet = false +} + +func NewNullableClearMemoryObservationsResponse(val *ClearMemoryObservationsResponse) *NullableClearMemoryObservationsResponse { + return &NullableClearMemoryObservationsResponse{value: val, isSet: true} +} + +func (v NullableClearMemoryObservationsResponse) MarshalJSON() ([]byte, error) { + return json.Marshal(v.value) +} + +func (v *NullableClearMemoryObservationsResponse) UnmarshalJSON(src []byte) error { + v.isSet = true + return json.Unmarshal(src, &v.value) +} + + diff --git a/hindsight-clients/python/.openapi-generator/FILES b/hindsight-clients/python/.openapi-generator/FILES index 286e294b..2ca8bee1 100644 --- a/hindsight-clients/python/.openapi-generator/FILES +++ b/hindsight-clients/python/.openapi-generator/FILES @@ -29,6 +29,7 @@ hindsight_client_api/models/child_operation_status.py hindsight_client_api/models/chunk_data.py hindsight_client_api/models/chunk_include_options.py hindsight_client_api/models/chunk_response.py +hindsight_client_api/models/clear_memory_observations_response.py hindsight_client_api/models/consolidation_response.py hindsight_client_api/models/create_bank_request.py hindsight_client_api/models/create_directive_request.py diff --git a/hindsight-clients/python/hindsight_client_api/__init__.py b/hindsight-clients/python/hindsight_client_api/__init__.py index f3e88aeb..2ae21a44 100644 --- a/hindsight-clients/python/hindsight_client_api/__init__.py +++ b/hindsight-clients/python/hindsight_client_api/__init__.py @@ -54,6 +54,7 @@ from hindsight_client_api.models.child_operation_status import ChildOperationSta from hindsight_client_api.models.chunk_data import ChunkData from hindsight_client_api.models.chunk_include_options import ChunkIncludeOptions from hindsight_client_api.models.chunk_response import ChunkResponse +from hindsight_client_api.models.clear_memory_observations_response import ClearMemoryObservationsResponse from hindsight_client_api.models.consolidation_response import ConsolidationResponse from hindsight_client_api.models.create_bank_request import CreateBankRequest from hindsight_client_api.models.create_directive_request import CreateDirectiveRequest diff --git a/hindsight-clients/python/hindsight_client_api/api/memory_api.py b/hindsight-clients/python/hindsight_client_api/api/memory_api.py index 5c28319f..21814b28 100644 --- a/hindsight-clients/python/hindsight_client_api/api/memory_api.py +++ b/hindsight-clients/python/hindsight_client_api/api/memory_api.py @@ -19,6 +19,7 @@ from typing_extensions import Annotated from pydantic import Field, StrictInt, StrictStr from typing import Any, Optional from typing_extensions import Annotated +from hindsight_client_api.models.clear_memory_observations_response import ClearMemoryObservationsResponse from hindsight_client_api.models.delete_response import DeleteResponse from hindsight_client_api.models.graph_data_response import GraphDataResponse from hindsight_client_api.models.list_memory_units_response import ListMemoryUnitsResponse @@ -343,6 +344,299 @@ class MemoryApi: + @validate_call + async def clear_memory_observations( + self, + bank_id: StrictStr, + memory_id: StrictStr, + authorization: Optional[StrictStr] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> ClearMemoryObservationsResponse: + """Clear observations for a memory + + Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. + + :param bank_id: (required) + :type bank_id: str + :param memory_id: (required) + :type memory_id: str + :param authorization: + :type authorization: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._clear_memory_observations_serialize( + bank_id=bank_id, + memory_id=memory_id, + authorization=authorization, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "ClearMemoryObservationsResponse", + '422': "HTTPValidationError", + } + response_data = await self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + await response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ).data + + + @validate_call + async def clear_memory_observations_with_http_info( + self, + bank_id: StrictStr, + memory_id: StrictStr, + authorization: Optional[StrictStr] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> ApiResponse[ClearMemoryObservationsResponse]: + """Clear observations for a memory + + Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. + + :param bank_id: (required) + :type bank_id: str + :param memory_id: (required) + :type memory_id: str + :param authorization: + :type authorization: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._clear_memory_observations_serialize( + bank_id=bank_id, + memory_id=memory_id, + authorization=authorization, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "ClearMemoryObservationsResponse", + '422': "HTTPValidationError", + } + response_data = await self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + await response_data.read() + return self.api_client.response_deserialize( + response_data=response_data, + response_types_map=_response_types_map, + ) + + + @validate_call + async def clear_memory_observations_without_preload_content( + self, + bank_id: StrictStr, + memory_id: StrictStr, + authorization: Optional[StrictStr] = None, + _request_timeout: Union[ + None, + Annotated[StrictFloat, Field(gt=0)], + Tuple[ + Annotated[StrictFloat, Field(gt=0)], + Annotated[StrictFloat, Field(gt=0)] + ] + ] = None, + _request_auth: Optional[Dict[StrictStr, Any]] = None, + _content_type: Optional[StrictStr] = None, + _headers: Optional[Dict[StrictStr, Any]] = None, + _host_index: Annotated[StrictInt, Field(ge=0, le=0)] = 0, + ) -> RESTResponseType: + """Clear observations for a memory + + Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. + + :param bank_id: (required) + :type bank_id: str + :param memory_id: (required) + :type memory_id: str + :param authorization: + :type authorization: str + :param _request_timeout: timeout setting for this request. If one + number provided, it will be total request + timeout. It can also be a pair (tuple) of + (connection, read) timeouts. + :type _request_timeout: int, tuple(int, int), optional + :param _request_auth: set to override the auth_settings for an a single + request; this effectively ignores the + authentication in the spec for a single request. + :type _request_auth: dict, optional + :param _content_type: force content-type for the request. + :type _content_type: str, Optional + :param _headers: set to override the headers for a single + request; this effectively ignores the headers + in the spec for a single request. + :type _headers: dict, optional + :param _host_index: set to override the host_index for a single + request; this effectively ignores the host_index + in the spec for a single request. + :type _host_index: int, optional + :return: Returns the result object. + """ # noqa: E501 + + _param = self._clear_memory_observations_serialize( + bank_id=bank_id, + memory_id=memory_id, + authorization=authorization, + _request_auth=_request_auth, + _content_type=_content_type, + _headers=_headers, + _host_index=_host_index + ) + + _response_types_map: Dict[str, Optional[str]] = { + '200': "ClearMemoryObservationsResponse", + '422': "HTTPValidationError", + } + response_data = await self.api_client.call_api( + *_param, + _request_timeout=_request_timeout + ) + return response_data.response + + + def _clear_memory_observations_serialize( + self, + bank_id, + memory_id, + authorization, + _request_auth, + _content_type, + _headers, + _host_index, + ) -> RequestSerialized: + + _host = None + + _collection_formats: Dict[str, str] = { + } + + _path_params: Dict[str, str] = {} + _query_params: List[Tuple[str, str]] = [] + _header_params: Dict[str, Optional[str]] = _headers or {} + _form_params: List[Tuple[str, str]] = [] + _files: Dict[ + str, Union[str, bytes, List[str], List[bytes], List[Tuple[str, bytes]]] + ] = {} + _body_params: Optional[bytes] = None + + # process the path parameters + if bank_id is not None: + _path_params['bank_id'] = bank_id + if memory_id is not None: + _path_params['memory_id'] = memory_id + # process the query parameters + # process the header parameters + if authorization is not None: + _header_params['authorization'] = authorization + # process the form parameters + # process the body parameter + + + # set the HTTP header `Accept` + if 'Accept' not in _header_params: + _header_params['Accept'] = self.api_client.select_header_accept( + [ + 'application/json' + ] + ) + + + # authentication setting + _auth_settings: List[str] = [ + ] + + return self.api_client.param_serialize( + method='DELETE', + resource_path='/v1/default/banks/{bank_id}/memories/{memory_id}/observations', + path_params=_path_params, + query_params=_query_params, + header_params=_header_params, + body=_body_params, + post_params=_form_params, + files=_files, + auth_settings=_auth_settings, + collection_formats=_collection_formats, + _host=_host, + _request_auth=_request_auth + ) + + + + @validate_call async def get_graph( self, diff --git a/hindsight-clients/python/hindsight_client_api/models/__init__.py b/hindsight-clients/python/hindsight_client_api/models/__init__.py index f916bc19..a713633b 100644 --- a/hindsight-clients/python/hindsight_client_api/models/__init__.py +++ b/hindsight-clients/python/hindsight_client_api/models/__init__.py @@ -29,6 +29,7 @@ from hindsight_client_api.models.child_operation_status import ChildOperationSta from hindsight_client_api.models.chunk_data import ChunkData from hindsight_client_api.models.chunk_include_options import ChunkIncludeOptions from hindsight_client_api.models.chunk_response import ChunkResponse +from hindsight_client_api.models.clear_memory_observations_response import ClearMemoryObservationsResponse from hindsight_client_api.models.consolidation_response import ConsolidationResponse from hindsight_client_api.models.create_bank_request import CreateBankRequest from hindsight_client_api.models.create_directive_request import CreateDirectiveRequest diff --git a/hindsight-clients/python/hindsight_client_api/models/clear_memory_observations_response.py b/hindsight-clients/python/hindsight_client_api/models/clear_memory_observations_response.py new file mode 100644 index 00000000..1624fed6 --- /dev/null +++ b/hindsight-clients/python/hindsight_client_api/models/clear_memory_observations_response.py @@ -0,0 +1,87 @@ +# coding: utf-8 + +""" + Hindsight HTTP API + + HTTP API for Hindsight + + The version of the OpenAPI document: 0.4.13 + Generated by OpenAPI Generator (https://openapi-generator.tech) + + Do not edit the class manually. +""" # noqa: E501 + + +from __future__ import annotations +import pprint +import re # noqa: F401 +import json + +from pydantic import BaseModel, ConfigDict, StrictInt +from typing import Any, ClassVar, Dict, List +from typing import Optional, Set +from typing_extensions import Self + +class ClearMemoryObservationsResponse(BaseModel): + """ + Response model for clearing observations for a specific memory. + """ # noqa: E501 + deleted_count: StrictInt + __properties: ClassVar[List[str]] = ["deleted_count"] + + model_config = ConfigDict( + populate_by_name=True, + validate_assignment=True, + protected_namespaces=(), + ) + + + def to_str(self) -> str: + """Returns the string representation of the model using alias""" + return pprint.pformat(self.model_dump(by_alias=True)) + + def to_json(self) -> str: + """Returns the JSON representation of the model using alias""" + # TODO: pydantic v2: use .model_dump_json(by_alias=True, exclude_unset=True) instead + return json.dumps(self.to_dict()) + + @classmethod + def from_json(cls, json_str: str) -> Optional[Self]: + """Create an instance of ClearMemoryObservationsResponse from a JSON string""" + return cls.from_dict(json.loads(json_str)) + + def to_dict(self) -> Dict[str, Any]: + """Return the dictionary representation of the model using alias. + + This has the following differences from calling pydantic's + `self.model_dump(by_alias=True)`: + + * `None` is only added to the output dict for nullable fields that + were set at model initialization. Other fields with value `None` + are ignored. + """ + excluded_fields: Set[str] = set([ + ]) + + _dict = self.model_dump( + by_alias=True, + exclude=excluded_fields, + exclude_none=True, + ) + return _dict + + @classmethod + def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional[Self]: + """Create an instance of ClearMemoryObservationsResponse from a dict""" + if obj is None: + return None + + if not isinstance(obj, dict): + return cls.model_validate(obj) + + _obj = cls.model_validate({ + "deleted_count": obj.get("deleted_count") + }) + return _obj + + diff --git a/hindsight-clients/typescript/generated/sdk.gen.ts b/hindsight-clients/typescript/generated/sdk.gen.ts index 36a518d3..60b995a4 100644 --- a/hindsight-clients/typescript/generated/sdk.gen.ts +++ b/hindsight-clients/typescript/generated/sdk.gen.ts @@ -17,6 +17,9 @@ import type { ClearBankMemoriesData, ClearBankMemoriesErrors, ClearBankMemoriesResponses, + ClearMemoryObservationsData, + ClearMemoryObservationsErrors, + ClearMemoryObservationsResponses, ClearObservationsData, ClearObservationsErrors, ClearObservationsResponses, @@ -829,6 +832,23 @@ export const clearObservations = ( ThrowOnError >({ url: "/v1/default/banks/{bank_id}/observations", ...options }); +/** + * Clear observations for a memory + * + * Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run. + */ +export const clearMemoryObservations = ( + options: Options, +) => + (options.client ?? client).delete< + ClearMemoryObservationsResponses, + ClearMemoryObservationsErrors, + ThrowOnError + >({ + url: "/v1/default/banks/{bank_id}/memories/{memory_id}/observations", + ...options, + }); + /** * Reset bank configuration * diff --git a/hindsight-clients/typescript/generated/types.gen.ts b/hindsight-clients/typescript/generated/types.gen.ts index 37bb3570..2125fb17 100644 --- a/hindsight-clients/typescript/generated/types.gen.ts +++ b/hindsight-clients/typescript/generated/types.gen.ts @@ -396,6 +396,18 @@ export type ChunkResponse = { created_at: string; }; +/** + * ClearMemoryObservationsResponse + * + * Response model for clearing observations for a specific memory. + */ +export type ClearMemoryObservationsResponse = { + /** + * Deleted Count + */ + deleted_count: number; +}; + /** * ConsolidationResponse * @@ -3655,6 +3667,48 @@ export type ClearObservationsResponses = { export type ClearObservationsResponse = ClearObservationsResponses[keyof ClearObservationsResponses]; +export type ClearMemoryObservationsData = { + body?: never; + headers?: { + /** + * Authorization + */ + authorization?: string | null; + }; + path: { + /** + * Bank Id + */ + bank_id: string; + /** + * Memory Id + */ + memory_id: string; + }; + query?: never; + url: "/v1/default/banks/{bank_id}/memories/{memory_id}/observations"; +}; + +export type ClearMemoryObservationsErrors = { + /** + * Validation Error + */ + 422: HttpValidationError; +}; + +export type ClearMemoryObservationsError = + ClearMemoryObservationsErrors[keyof ClearMemoryObservationsErrors]; + +export type ClearMemoryObservationsResponses = { + /** + * Successful Response + */ + 200: ClearMemoryObservationsResponse; +}; + +export type ClearMemoryObservationsResponse2 = + ClearMemoryObservationsResponses[keyof ClearMemoryObservationsResponses]; + export type ResetBankConfigData = { body?: never; headers?: { diff --git a/hindsight-control-plane/src/app/api/banks/[bankId]/observations/route.ts b/hindsight-control-plane/src/app/api/banks/[bankId]/observations/route.ts index 225d54b2..86bc80a4 100644 --- a/hindsight-control-plane/src/app/api/banks/[bankId]/observations/route.ts +++ b/hindsight-control-plane/src/app/api/banks/[bankId]/observations/route.ts @@ -29,7 +29,7 @@ export async function GET(request: Request, { params }: { params: Promise<{ bank id: item.id, bank_id: bankId, text: item.text, - proof_count: 1, + proof_count: (item as any).proof_count ?? 1, history: [], tags: item.tags || [], source_memory_ids: [], diff --git a/hindsight-control-plane/src/components/data-view.tsx b/hindsight-control-plane/src/components/data-view.tsx index 159b39b1..9629ade5 100644 --- a/hindsight-control-plane/src/components/data-view.tsx +++ b/hindsight-control-plane/src/components/data-view.tsx @@ -608,12 +608,25 @@ export function DataView({ factType }: DataViewProps) { - + {factType === "observation" ? "Observation" : "Memory"} Entities - Occurred - Mentioned + {factType === "observation" && ( + Sources + )} + + Occurred + + + Mentioned + @@ -673,6 +686,11 @@ export function DataView({ factType }: DataViewProps) { - )} + {factType === "observation" && ( + + {row.proof_count ?? 1} + + )} {occurredDisplay || ( - diff --git a/hindsight-docs/docs/developer/observations.mdx b/hindsight-docs/docs/developer/observations.mdx index 799211b9..b59de72e 100644 --- a/hindsight-docs/docs/developer/observations.mdx +++ b/hindsight-docs/docs/developer/observations.mdx @@ -156,6 +156,42 @@ Set `observations_mission` via the [bank config API](/developer/api/memory-banks --- +## Observation Lifecycle & Invalidation + +### When Memories Are Deleted + +Observations are derived from source memories. When source memories are removed, Hindsight automatically keeps observations consistent: + +| Action | Effect on observations | +|--------|----------------------| +| Delete a document | All observations derived from the document's memories are deleted | +| Delete individual memories (by type) | Observations sourced from those memories are deleted | +| Delete an entire bank | All observations are deleted along with everything else | + +After deletion, the **remaining source memories** that fed the affected observations have their consolidation state reset, so they will be re-consolidated on the next consolidation run and produce fresh observations. + +### Clearing Observations for a Specific Memory + +You can clear all observations derived from a single memory without deleting the memory itself. This is useful when you want to force re-synthesis of a memory's contribution to consolidated knowledge. + +Use the `DELETE /v1/default/banks/{bank_id}/memories/{memory_id}/observations` endpoint. This will: +1. Delete all observations that list the memory as a source +2. Reset `consolidated_at` on the memory itself and any other source memories that contributed to those observations +3. Trigger a consolidation job so fresh observations are produced automatically + +### Resetting All Observations + +To wipe all consolidated knowledge and start over: + +```python +# Clear all observations for a bank +client.clear_observations(bank_id="my-bank") +``` + +This resets the consolidation state for all source memories in the bank, so the next consolidation run will re-derive all observations from scratch. + +--- + ## Configuration Observation consolidation runs automatically. You can monitor consolidation via the [Operations API](./api/operations). diff --git a/hindsight-docs/static/openapi.json b/hindsight-docs/static/openapi.json index ad83de4c..c0ab6f3f 100644 --- a/hindsight-docs/static/openapi.json +++ b/hindsight-docs/static/openapi.json @@ -2751,6 +2751,74 @@ } } }, + "/v1/default/banks/{bank_id}/memories/{memory_id}/observations": { + "delete": { + "tags": [ + "Memory" + ], + "summary": "Clear observations for a memory", + "description": "Delete all observations derived from a specific memory and reset it for re-consolidation. The memory itself is not deleted. A consolidation job is triggered automatically so the memory will produce fresh observations on the next consolidation run.", + "operationId": "clear_memory_observations", + "parameters": [ + { + "name": "bank_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Bank Id" + } + }, + { + "name": "memory_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "title": "Memory Id" + } + }, + { + "name": "authorization", + "in": "header", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Authorization" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ClearMemoryObservationsResponse" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + } + } + }, "/v1/default/banks/{bank_id}/config": { "get": { "tags": [ @@ -3842,6 +3910,23 @@ "document_id": "session_1" } }, + "ClearMemoryObservationsResponse": { + "properties": { + "deleted_count": { + "type": "integer", + "title": "Deleted Count" + } + }, + "type": "object", + "required": [ + "deleted_count" + ], + "title": "ClearMemoryObservationsResponse", + "description": "Response model for clearing observations for a specific memory.", + "example": { + "deleted_count": 3 + } + }, "ConsolidationResponse": { "properties": { "operation_id": {