Compare commits

...
Author SHA1 Message Date
Nicolò Boschi 8f6006e72d refactor: use dedicated response model for clear_memory_observations, remove code example from docs 2026-02-24 12:16:47 +01:00
Nicolò Boschi 518562baae feat: add clear_memory_observations endpoint and regenerate clients
- Add DELETE /banks/{id}/memories/{memory_id}/observations endpoint
- Add observations lifecycle/invalidation section to docs
- Regenerate OpenAPI spec and all clients (Python, TypeScript, Go, Rust)
2026-02-24 10:40:16 +01:00
Nicolò Boschi df8e2dd790 fix: handle observations regeneration when memories get deleted 2026-02-24 10:37:27 +01:00
18 changed files with 1626 additions and 50 deletions
+42
View File
@@ -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,
@@ -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:
@@ -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
@@ -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)
+59
View File
@@ -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:
+126
View File
@@ -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
@@ -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)
}
@@ -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
@@ -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
@@ -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,
@@ -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
@@ -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
@@ -17,6 +17,9 @@ import type {
ClearBankMemoriesData,
ClearBankMemoriesErrors,
ClearBankMemoriesResponses,
ClearMemoryObservationsData,
ClearMemoryObservationsErrors,
ClearMemoryObservationsResponses,
ClearObservationsData,
ClearObservationsErrors,
ClearObservationsResponses,
@@ -829,6 +832,23 @@ export const clearObservations = <ThrowOnError extends boolean = false>(
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 = <ThrowOnError extends boolean = false>(
options: Options<ClearMemoryObservationsData, ThrowOnError>,
) =>
(options.client ?? client).delete<
ClearMemoryObservationsResponses,
ClearMemoryObservationsErrors,
ThrowOnError
>({
url: "/v1/default/banks/{bank_id}/memories/{memory_id}/observations",
...options,
});
/**
* Reset bank configuration
*
@@ -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?: {
@@ -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: [],
@@ -608,12 +608,25 @@ export function DataView({ factType }: DataViewProps) {
<Table className="table-fixed">
<TableHeader>
<TableRow className="bg-muted/50">
<TableHead className="w-[45%]">
<TableHead
className={factType === "observation" ? "w-[40%]" : "w-[45%]"}
>
{factType === "observation" ? "Observation" : "Memory"}
</TableHead>
<TableHead className="w-[20%]">Entities</TableHead>
<TableHead className="w-[17%]">Occurred</TableHead>
<TableHead className="w-[18%]">Mentioned</TableHead>
{factType === "observation" && (
<TableHead className="w-[10%]">Sources</TableHead>
)}
<TableHead
className={factType === "observation" ? "w-[15%]" : "w-[17%]"}
>
Occurred
</TableHead>
<TableHead
className={factType === "observation" ? "w-[15%]" : "w-[18%]"}
>
Mentioned
</TableHead>
</TableRow>
</TableHeader>
<TableBody>
@@ -673,6 +686,11 @@ export function DataView({ factType }: DataViewProps) {
<span className="text-xs text-muted-foreground">-</span>
)}
</TableCell>
{factType === "observation" && (
<TableCell className="text-xs py-2 text-foreground">
{row.proof_count ?? 1}
</TableCell>
)}
<TableCell className="text-xs py-2 text-foreground">
{occurredDisplay || (
<span className="text-muted-foreground">-</span>
@@ -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).
+85
View File
@@ -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": {