diff --git a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py index eb4478e7..8211e3a1 100644 --- a/hindsight-api/hindsight_api/engine/consolidation/consolidator.py +++ b/hindsight-api/hindsight_api/engine/consolidation/consolidator.py @@ -67,6 +67,42 @@ class _BatchLLMResult: prompt_chars: int = 0 +@dataclass +class _SourceAggregation: + """Fields inherited by an observation from its source memories.""" + + event_date: datetime | None + occurred_start: datetime | None + occurred_end: datetime | None + mentioned_at: datetime | None + tags: list[str] + + +def _aggregate_source_fields(source_mems: list[dict[str, Any]], tags: list[str] | None = None) -> _SourceAggregation: + """Compute the observation fields inherited from a set of source memories. + + Temporal aggregation rules: + - ``event_date`` — earliest across sources (min) + - ``occurred_start`` — earliest across sources (min) + - ``occurred_end`` — latest across sources (max) + - ``mentioned_at`` — latest across sources (max) + + Fields remain ``None`` when no source memory carries that information, so + observations are never stamped with an artificial timestamp. + + ``tags`` defaults to those of the first source memory when not explicitly + provided (all memories in a consolidation batch share the same tag set). + """ + effective_tags = tags if tags is not None else (source_mems[0].get("tags") or [] if source_mems else []) + return _SourceAggregation( + event_date=_min_date(m.get("event_date") for m in source_mems), + occurred_start=_min_date(m.get("occurred_start") for m in source_mems), + occurred_end=_max_date(m.get("occurred_end") for m in source_mems), + mentioned_at=_max_date(m.get("mentioned_at") for m in source_mems), + tags=effective_tags, + ) + + class ConsolidationPerfLog: """Performance logging for consolidation operations.""" @@ -616,17 +652,18 @@ async def _process_memory_batch( source_mems = [mem_by_id[fid] for fid in create.source_fact_ids if fid in mem_by_id] if not source_mems: continue + agg = _aggregate_source_fields(source_mems, tags=fact_tags) await _execute_create_action( conn=conn, memory_engine=memory_engine, bank_id=bank_id, source_memory_ids=[m["id"] for m in source_mems], text=create.text, - source_fact_tags=fact_tags, - event_date=_min_date(m.get("event_date") for m in source_mems), - occurred_start=_min_date(m.get("occurred_start") for m in source_mems), - occurred_end=_max_date(m.get("occurred_end") for m in source_mems), - mentioned_at=_max_date(m.get("mentioned_at") for m in source_mems), + source_fact_tags=agg.tags, + event_date=agg.event_date, + occurred_start=agg.occurred_start, + occurred_end=agg.occurred_end, + mentioned_at=agg.mentioned_at, perf=perf, ) for m in source_mems: @@ -643,6 +680,7 @@ async def _process_memory_batch( f"not in any source fact's recall" ) continue + agg = _aggregate_source_fields(source_mems, tags=fact_tags) await _execute_update_action( conn=conn, memory_engine=memory_engine, @@ -651,10 +689,10 @@ async def _process_memory_batch( observation_id=update.observation_id, new_text=update.text, observations=union_observations, - source_fact_tags=fact_tags, - source_occurred_start=_min_date(m.get("occurred_start") for m in source_mems), - source_occurred_end=_max_date(m.get("occurred_end") for m in source_mems), - source_mentioned_at=_max_date(m.get("mentioned_at") for m in source_mems), + source_fact_tags=agg.tags, + source_occurred_start=agg.occurred_start, + source_occurred_end=agg.occurred_end, + source_mentioned_at=agg.mentioned_at, perf=perf, ) for m in source_mems: @@ -1035,8 +1073,8 @@ async def _create_observation_directly( # Create the observation as a memory_unit now = datetime.now(timezone.utc) obs_event_date = event_date or now - obs_occurred_start = occurred_start or now - obs_occurred_end = occurred_end or now + obs_occurred_start = occurred_start + obs_occurred_end = occurred_end obs_mentioned_at = mentioned_at or now obs_tags = tags or [] diff --git a/hindsight-api/tests/test_consolidation.py b/hindsight-api/tests/test_consolidation.py index aae7e5ed..bdc94ed9 100644 --- a/hindsight-api/tests/test_consolidation.py +++ b/hindsight-api/tests/test_consolidation.py @@ -5,11 +5,15 @@ Note: Consolidation runs automatically after retain via SyncTaskBackend in tests """ import uuid +from datetime import datetime, timezone from unittest.mock import patch import pytest -from hindsight_api.engine.consolidation.consolidator import run_consolidation_job +from hindsight_api.engine.consolidation.consolidator import ( + _aggregate_source_fields, + run_consolidation_job, +) from hindsight_api.engine.memory_engine import MemoryEngine from hindsight_api.engine.reflect.tools import ( tool_recall, @@ -2317,3 +2321,100 @@ async def test_observation_scopes_all_combinations(memory: MemoryEngine, request assert combined, f"Expected an observation scoped to both tags, got: {tag_sets}" finally: await memory.delete_bank(bank_id, request_context=request_context) + + +def _dt(year: int, month: int, day: int) -> datetime: + return datetime(year, month, day, tzinfo=timezone.utc) + + +class TestAggregateSourceFields: + """Unit tests for _aggregate_source_fields – no database required.""" + + def test_all_none_temporal_fields_stay_none(self): + """When source memories carry no temporal data, all fields must remain None.""" + source_mems = [ + {"tags": ["t1"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + {"tags": ["t1"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + ] + agg = _aggregate_source_fields(source_mems) + assert agg.event_date is None + assert agg.occurred_start is None + assert agg.occurred_end is None + assert agg.mentioned_at is None + + def test_temporal_fields_aggregated_correctly(self): + """occurred_start and event_date are minimised; occurred_end and mentioned_at are maximised.""" + early = _dt(2023, 1, 1) + late = _dt(2024, 6, 15) + source_mems = [ + { + "tags": [], + "event_date": late, + "occurred_start": late, + "occurred_end": early, + "mentioned_at": early, + }, + { + "tags": [], + "event_date": early, + "occurred_start": early, + "occurred_end": late, + "mentioned_at": late, + }, + ] + agg = _aggregate_source_fields(source_mems) + assert agg.event_date == early + assert agg.occurred_start == early + assert agg.occurred_end == late + assert agg.mentioned_at == late + + def test_partial_temporal_fields_ignored_when_none(self): + """None values in individual sources do not corrupt the min/max from sources that do have dates.""" + d = _dt(2023, 3, 10) + source_mems = [ + {"tags": [], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + {"tags": [], "event_date": d, "occurred_start": d, "occurred_end": d, "mentioned_at": d}, + ] + agg = _aggregate_source_fields(source_mems) + assert agg.event_date == d + assert agg.occurred_start == d + assert agg.occurred_end == d + assert agg.mentioned_at == d + + def test_tags_inherited_from_first_source_memory(self): + """Tags default to those of the first source memory (batch invariant).""" + source_mems = [ + {"tags": ["user:alice"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + {"tags": ["user:alice"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + ] + agg = _aggregate_source_fields(source_mems) + assert agg.tags == ["user:alice"] + + def test_tags_override_takes_precedence(self): + """Explicit tags parameter overrides the source-memory tags.""" + source_mems = [ + {"tags": ["user:alice"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + ] + agg = _aggregate_source_fields(source_mems, tags=["scope:override"]) + assert agg.tags == ["scope:override"] + + def test_empty_tags_override_is_respected(self): + """An explicit empty list override must not fall back to source tags.""" + source_mems = [ + {"tags": ["user:alice"], "event_date": None, "occurred_start": None, "occurred_end": None, "mentioned_at": None}, + ] + agg = _aggregate_source_fields(source_mems, tags=[]) + assert agg.tags == [] + + def test_single_source_memory(self): + """Single-source aggregation should just pass through that memory's fields.""" + d = _dt(2024, 11, 5) + source_mems = [ + {"tags": ["x"], "event_date": d, "occurred_start": d, "occurred_end": d, "mentioned_at": d}, + ] + agg = _aggregate_source_fields(source_mems) + assert agg.event_date == d + assert agg.occurred_start == d + assert agg.occurred_end == d + assert agg.mentioned_at == d + assert agg.tags == ["x"]