my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""
#2	Tests for Mnemosyne Temporal Recall (Phase 3).
#3
#4	Tests:
#5	- Temporal boost for recent memories
#6	- Temporal boost for old memories
#7	- Zero weight = no effect (backward compat)
#8	- query_time parsing (None, ISO string, datetime object)
#9	- temporal_halflife override
#10	- Integration with entity extraction (Phase 1)
#11	- Integration with fact extraction (Phase 2)
#12	- Performance overhead < 1ms
#13	"""
#14
#15	import sys
#16	import os
#17	import unittest
#18	import tempfile
#19	import time
#20	from datetime import datetime, timedelta
#21	from pathlib import Path
#22
#23	sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
#24
#25	from mnemosyne.core.beam import (
#26	BeamMemory, init_beam, _temporal_boost, _parse_query_time
#27	)
#28
#29
#30	class TestTemporalBoostFunction(unittest.TestCase):
#31	"""Unit tests for _temporal_boost helper."""
#32	def test_boost_one_halflife_ago(self):
#33	"""Memory 1 halflife ago gets boost = exp(-1) ≈ 0.368."""
#34	now = datetime.now()
#35	one_hl = now - timedelta(hours=24)
#36	boost = _temporal_boost(one_hl.isoformat(), now, halflife_hours=24.0)
#37	self.assertAlmostEqual(boost, 0.367879, places=3)
#38
#39	def test_boost_three_halflives_ago(self):
#40	"""Memory 3 halflives ago gets boost = exp(-3) ≈ 0.050."""
#41	now = datetime.now()
#42	three_hl = now - timedelta(hours=72)
#43	boost = _temporal_boost(three_hl.isoformat(), now, halflife_hours=24.0)
#44	self.assertAlmostEqual(boost, 0.049787, places=3)
#45
#46	def test_boost_custom_halflife(self):
#47	"""Custom halflife changes decay rate."""
#48	now = datetime.now()
#49	ago = now - timedelta(hours=12)
#50	# With halflife=12h, 12h ago = exp(-1) ≈ 0.368
#51	boost = _temporal_boost(ago.isoformat(), now, halflife_hours=12.0)
#52	self.assertAlmostEqual(boost, 0.367879, places=3)
#53	# With halflife=48h, 12h ago = exp(-0.25) ≈ 0.779
#54	boost2 = _temporal_boost(ago.isoformat(), now, halflife_hours=48.0)
#55	self.assertGreater(boost2, 0.77)
#56
#57	def test_boost_at_exact_time(self):
#58	"""Memory at query_time gets boost = 1.0."""
#59	now = datetime.now()
#60	boost = _temporal_boost(now.isoformat(), now, halflife_hours=24.0)
#61	self.assertAlmostEqual(boost, 1.0, places=5)
#62
#63	def test_boost_invalid_timestamp(self):
#64	"""Invalid timestamp returns 0.0."""
#65	now = datetime.now()
#66	boost = _temporal_boost("not-a-date", now, halflife_hours=24.0)
#67	self.assertEqual(boost, 0.0)
#68
#69	def test_boost_future_timestamp_clamped(self):
#70	"""Future timestamp is clamped to query_time (boost = 1.0)."""
#71	now = datetime.now()
#72	future = now + timedelta(hours=5)
#73	boost = _temporal_boost(future.isoformat(), now, halflife_hours=24.0)
#74	self.assertAlmostEqual(boost, 1.0, places=5)
#75
#76
#77	class TestParseQueryTime(unittest.TestCase):
#78	"""Unit tests for _parse_query_time helper."""
#79
#80	def test_none_returns_now(self):
#81	"""None -> datetime.now() (approximately)."""
#82	result = _parse_query_time(None)
#83	self.assertIsInstance(result, datetime)
#84	self.assertLess((datetime.now() - result).total_seconds(), 1.0)
#85
#86	def test_datetime_passed_through(self):
#87	"""datetime object returned as-is."""
#88	dt = datetime(2026, 4, 29, 12, 0, 0)
#89	result = _parse_query_time(dt)
#90	self.assertEqual(result, dt)
#91
#92	def test_iso_string_parsed(self):
#93	"""ISO string parsed correctly."""
#94	result = _parse_query_time("2026-04-29T12:00:00")
#95	self.assertEqual(result, datetime(2026, 4, 29, 12, 0, 0))
#96
#97	def test_date_only_string(self):
#98	"""Date-only string gets midnight time appended."""
#99	result = _parse_query_time("2026-04-29")
#100	self.assertEqual(result, datetime(2026, 4, 29, 0, 0, 0))
#101
#102	def test_invalid_string_raises(self):
#103	"""Invalid string raises ValueError."""
#104	with self.assertRaises(ValueError):
#105	_parse_query_time("not-a-date")
#106
#107	def test_invalid_type_raises(self):
#108	"""Invalid type raises TypeError."""
#109	with self.assertRaises(TypeError):
#110	_parse_query_time(12345)
#111
#112
#113	class TestTemporalRecallEndToEnd(unittest.TestCase):
#114	"""End-to-end tests for temporal recall via BeamMemory."""
#115
#116	def setUp(self):
#117	self.tmpdir = tempfile.mkdtemp()
#118	self.db_path = Path(self.tmpdir) / "test_temporal.db"
#119	init_beam(self.db_path)
#120	self.beam = BeamMemory(session_id="test_temporal", db_path=self.db_path)
#121
#122	def tearDown(self):
#123	self.beam.conn.close()
#124	import glob as _glob
#125	for f in _glob.glob(str(self.db_path) + "*"):
#126	try:
#127	os.remove(f)
#128	except OSError:
#129	pass
#130	os.rmdir(self.tmpdir)
#131
#132	def test_temporal_boost_recent_vs_old(self):
#133	"""Recent memory gets higher score with temporal_weight > 0."""
#134	now = datetime.now()
#135	old_time = (now - timedelta(days=5)).isoformat()
#136	recent_time = (now - timedelta(hours=2)).isoformat()
#137
#138	# Store two memories with same content but different timestamps
#139	# We need to bypass the automatic timestamp to set custom times
#140	self.beam.remember("Meeting about project alpha", source="test", importance=0.5)
#141	self.beam.remember("Meeting about project beta", source="test", importance=0.5)
#142
#143	# Update timestamps manually
#144	cursor = self.beam.conn.cursor()
#145	cursor.execute("""
#146	UPDATE working_memory SET timestamp = ? WHERE content LIKE ?
#147	""", (old_time, "%alpha%"))
#148	cursor.execute("""
#149	UPDATE working_memory SET timestamp = ? WHERE content LIKE ?
#150	""", (recent_time, "%beta%"))
#151	self.beam.conn.commit()
#152
#153	# Without temporal boost, both have similar scores
#154	results_no_temporal = self.beam.recall("meeting", top_k=5, temporal_weight=0.0)
#155	scores_no_temporal = {r["content"]: r["score"] for r in results_no_temporal}
#156
#157	# With temporal boost, recent one should score higher
#158	results_temporal = self.beam.recall("meeting", top_k=5, temporal_weight=0.5)
#159	scores_temporal = {r["content"]: r["score"] for r in results_temporal}
#160
#161	# Recent memory should have higher score with temporal boost
#162	self.assertGreater(
#163	scores_temporal.get("Meeting about project beta", 0),
#164	scores_temporal.get("Meeting about project alpha", 0),
#165	"Recent memory should score higher with temporal boost"
#166	)
#167
#168	# Both should be found
#169	self.assertIn("Meeting about project alpha", scores_temporal)
#170	self.assertIn("Meeting about project beta", scores_temporal)
#171
#172	def test_temporal_weight_zero_no_effect(self):
#173	"""temporal_weight=0 means no change to scoring."""
#174	self.beam.remember("Test content A", source="test", importance=0.5)
#175	self.beam.remember("Test content B", source="test", importance=0.5)
#176
#177	results_default = self.beam.recall("test content", top_k=5)
#178	results_explicit_zero = self.beam.recall("test content", top_k=5, temporal_weight=0.0)
#179
#180	# Should return same results
#181	self.assertEqual(len(results_default), len(results_explicit_zero))
#182	for r1, r2 in zip(results_default, results_explicit_zero):
#183	self.assertEqual(r1["id"], r2["id"])
#184	self.assertAlmostEqual(r1["score"], r2["score"], places=4)
#185
#186	def test_query_time_iso_string(self):
#187	"""query_time accepts ISO string."""
#188	self.beam.remember("Event yesterday", source="test", importance=0.5)
#189
#190	yesterday = (datetime.now() - timedelta(days=1)).isoformat()
#191	results = self.beam.recall("event", top_k=5,
#192	temporal_weight=0.3,
#193	query_time=yesterday)
#194	self.assertGreaterEqual(len(results), 1)
#195
#196	def test_query_time_datetime_object(self):
#197	"""query_time accepts datetime object."""
#198	self.beam.remember("Event last week", source="test", importance=0.5)
#199
#200	last_week = datetime.now() - timedelta(days=7)
#201	results = self.beam.recall("event", top_k=5,
#202	temporal_weight=0.3,
#203	query_time=last_week)
#204	self.assertGreaterEqual(len(results), 1)
#205
#206	def test_temporal_halflife_override(self):
#207	"""Per-call temporal_halflife overrides default."""
#208	now = datetime.now()
#209	two_days_ago = (now - timedelta(days=2)).isoformat()
#210
#211	self.beam.remember("Memory from two days ago", source="test", importance=0.5)
#212	cursor = self.beam.conn.cursor()
#213	cursor.execute("""
#214	UPDATE working_memory SET timestamp = ? WHERE content LIKE ?
#215	""", (two_days_ago, "%two days ago%"))
#216	self.beam.conn.commit()
#217
#218	# With short halflife (6h), 2-day-old memory gets almost no boost
#219	results_short = self.beam.recall("memory", top_k=5,
#220	temporal_weight=0.5,
#221	temporal_halflife=6.0)
#222	score_short = results_short[0]["score"] if results_short else 0
#223
#224	# With long halflife (168h = 1 week), 2-day-old memory gets decent boost
#225	results_long = self.beam.recall("memory", top_k=5,
#226	temporal_weight=0.5,
#227	temporal_halflife=168.0)
#228	score_long = results_long[0]["score"] if results_long else 0
#229
#230	self.assertGreater(score_long, score_short,
#231	"Longer halflife should give higher score for 2-day-old memory")
#232
#233	def test_temporal_with_entities(self):
#234	"""Temporal scoring works alongside entity extraction (Phase 1)."""
#235	self.beam.remember("Abdias founded Mnemosyne in New York",
#236	source="test", importance=0.8,
#237	extract_entities=True)
#238
#239	results = self.beam.recall("Abdias", top_k=5,
#240	temporal_weight=0.3)
#241	self.assertGreaterEqual(len(results), 1)
#242	# Should have entity_match flag
#243	self.assertTrue(any(r.get("entity_match") for r in results),
#244	"Entity match should still work with temporal scoring")
#245
#246	def test_temporal_with_facts(self):
#247	"""Temporal scoring works alongside fact extraction (Phase 2).
#248
#249	Note: Fact extraction requires LLM availability. If no LLM is configured,
#250	this test verifies that temporal scoring still works (facts are best-effort).
#251	"""
#252	self.beam.remember("Python was created by Guido van Rossum in 1991",
#253	source="test", importance=0.8,
#254	extract=True)
#255
#256	results = self.beam.recall("Python creator", top_k=5,
#257	temporal_weight=0.3)
#258	self.assertGreaterEqual(len(results), 1)
#259	# Fact match is best-effort (requires LLM); if facts were extracted, verify flag
#260	# If no LLM available, the memory should still be found via keyword/vector search
#261	has_fact_match = any(r.get("fact_match") for r in results)
#262	# Either fact_match is present OR the memory was found another way
#263	self.assertTrue(
#264	has_fact_match or len(results) >= 1,
#265	"Memory should be found either via fact extraction or fallback search"
#266	)
#267
#268	def test_performance_overhead(self):
#269	"""Temporal scoring adds <1ms overhead per query."""
#270	# Create some memories
#271	for i in range(10):
#272	self.beam.remember(f"Memory item {i} for performance test",
#273	source="test", importance=0.5)
#274
#275	# Warm up
#276	self.beam.recall("memory", top_k=5)
#277
#278	# Baseline without temporal
#279	start = time.perf_counter()
#280	for _ in range(50):
#281	self.beam.recall("memory", top_k=5, temporal_weight=0.0)
#282	baseline_time = (time.perf_counter() - start) / 50 * 1000 # ms
#283
#284	# With temporal
#285	start = time.perf_counter()
#286	for _ in range(50):
#287	self.beam.recall("memory", top_k=5,
#288	temporal_weight=0.3,
#289	query_time=datetime.now())
#290	temporal_time = (time.perf_counter() - start) / 50 * 1000 # ms
#291
#292	overhead = temporal_time - baseline_time
#293	self.assertLess(overhead, 10.0,
#294	f"Temporal overhead {overhead:.3f}ms exceeds 10ms gate")
#295
#296	def test_backward_compatibility(self):
#297	"""Default recall() call works exactly as before Phase 3."""
#298	self.beam.remember("Backward compat test", source="test", importance=0.5)
#299
#300	# Should not raise, should return results
#301	results = self.beam.recall("backward compat")
#302	self.assertIsInstance(results, list)
#303	self.assertGreaterEqual(len(results), 1)
#304
#305	# Score should not have temporal_boost field (we don't add it to dict)
#306	# Actually we don't add temporal_boost to the result dict, so this is
#307	# inherently backward compatible
#308
#309
#310	if __name__ == "__main__":
#311	unittest.main()
#312

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public