my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

#1	"""Regression tests for [C20]: hermes_plugin caches `_triple_store` against
#2	the first session's DB and never invalidates it. After a session change that
#3	moves `_memory_instance` to a different DB (bank switch, custom MNEMOSYNE_DATA_DIR,
#4	etc.), `_get_triples()` keeps returning the OLD store — so triple writes go
#5	to the original DB while memory writes go to the new one.
#6
#7	Bug: hermes_plugin/__init__.py:42-68. `_get_memory()` rebinds `_memory_instance`
#8	on session change, but `_triple_store` is never reset alongside it.
#9
#10	Tests:
#11	1. After _get_memory(b) rebinds memory, _get_triples() returns a store
#12	aligned with b's db_path, not a's.
#13	2. Triples written via the public `mnemosyne_triple_add` tool after a
#14	session switch land in the new DB, not the old one.
#15	"""
#16
#17	import json
#18	from pathlib import Path
#19
#20	import pytest
#21
#22	import hermes_plugin
#23	from hermes_plugin import tools
#24	from mnemosyne.core.memory import Mnemosyne
#25	from mnemosyne.core.triples import TripleStore
#26
#27
#28	def _route_mnemosyne(monkeypatch, session_to_db):
#29	"""Patch hermes_plugin.Mnemosyne so each session_id resolves to a fixed db_path.
#30
#31	Production code can move db_path between sessions via bank switches or
#32	runtime data-dir changes; this helper simulates that without depending on
#33	env-var resolution timing.
#34	"""
#35	real_mnemosyne = hermes_plugin.Mnemosyne
#36
#37	def fake_mnemosyne(session_id, **kwargs):
#38	kwargs.pop("db_path", None)
#39	if session_id in session_to_db:
#40	return real_mnemosyne(session_id=session_id,
#41	db_path=session_to_db[session_id],
#42	**kwargs)
#43	return real_mnemosyne(session_id=session_id, **kwargs)
#44
#45	monkeypatch.setattr(hermes_plugin, "Mnemosyne", fake_mnemosyne)
#46
#47
#48	class TestTripleStoreCacheInvalidation:
#49
#50	def test_get_triples_follows_active_memory_after_session_switch(
#51	self, tmp_path, monkeypatch
#52	):
#53	"""_get_triples() must return a store aligned with the active
#54	Mnemosyne instance's db_path, not the first one captured.
#55
#56	Sets HERMES_SESSION_ID env to mirror each explicit session, the
#57	way Hermes does in production, so any internal _get_memory() call
#58	that reads env stays consistent with the caller's intent.
#59	"""
#60	db_a = tmp_path / "a.db"
#61	db_b = tmp_path / "b.db"
#62	_route_mnemosyne(monkeypatch, {"session_a": db_a, "session_b": db_b})
#63
#64	# First session
#65	monkeypatch.setenv("HERMES_SESSION_ID", "session_a")
#66	mem_a = hermes_plugin._get_memory("session_a")
#67	assert Path(mem_a.db_path) == db_a
#68	triples_first = hermes_plugin._get_triples()
#69	assert Path(triples_first.db_path) == db_a
#70
#71	# Second session — different db
#72	monkeypatch.setenv("HERMES_SESSION_ID", "session_b")
#73	mem_b = hermes_plugin._get_memory("session_b")
#74	assert Path(mem_b.db_path) == db_b
#75
#76	# Critical: triples must follow the new memory, not stay cached at db_a
#77	triples_second = hermes_plugin._get_triples()
#78	assert Path(triples_second.db_path) == db_b, (
#79	f"Triple store cached at {triples_first.db_path} after session "
#80	f"switch; expected to follow new memory at {db_b}"
#81	)
#82
#83	def test_triple_writes_route_to_new_db_after_session_switch(
#84	self, tmp_path, monkeypatch
#85	):
#86	"""End-to-end: mnemosyne_triple_add after a session switch must
#87	write to the new session's DB, not silently to the old one."""
#88	db_a = tmp_path / "a.db"
#89	db_b = tmp_path / "b.db"
#90	_route_mnemosyne(monkeypatch, {"session_a": db_a, "session_b": db_b})
#91
#92	# Session a — write triple_a
#93	monkeypatch.setenv("HERMES_SESSION_ID", "session_a")
#94	hermes_plugin._get_memory("session_a")
#95	result_a = json.loads(tools.mnemosyne_triple_add({
#96	"subject": "alice",
#97	"predicate": "knows",
#98	"object": "bob",
#99	"source": "test",
#100	}))
#101	assert result_a.get("status") == "added", f"unexpected: {result_a}"
#102
#103	# Session b — write triple_b
#104	monkeypatch.setenv("HERMES_SESSION_ID", "session_b")
#105	hermes_plugin._get_memory("session_b")
#106	result_b = json.loads(tools.mnemosyne_triple_add({
#107	"subject": "carol",
#108	"predicate": "owns",
#109	"object": "project_b",
#110	"source": "test",
#111	}))
#112	assert result_b.get("status") == "added", f"unexpected: {result_b}"
#113
#114	# Read each DB directly — bypassing the plugin cache entirely.
#115	triples_in_a = TripleStore(db_path=db_a).query(subject="alice")
#116	triples_in_b = TripleStore(db_path=db_b).query(subject="carol")
#117
#118	assert len(triples_in_a) == 1, (
#119	f"alice/knows/bob should live in db_a but found {len(triples_in_a)} matches"
#120	)
#121	assert len(triples_in_b) == 1, (
#122	f"carol/owns/project_b should live in db_b but found {len(triples_in_b)} matches"
#123	)
#124
#125	# Cross-check: data must NOT have leaked across DBs.
#126	leaked_to_a = TripleStore(db_path=db_a).query(subject="carol")
#127	leaked_to_b = TripleStore(db_path=db_b).query(subject="alice")
#128	assert len(leaked_to_a) == 0, (
#129	f"carol triple leaked into db_a (the bug — cached store)"
#130	)
#131	assert len(leaked_to_b) == 0, (
#132	f"alice triple leaked into db_b"
#133	)
#134
#135	def test_get_triples_honors_env_change_without_explicit_memory_call(
#136	self, tmp_path, monkeypatch
#137	):
#138	"""If HERMES_SESSION_ID env changes but no explicit _get_memory(session_id)
#139	is made before the next triple call, _get_triples() should still route
#140	to the new session's DB. Locks in env-honoring behavior; pre-review
#141	revisions of this fix regressed this scenario.
#142	"""
#143	db_a = tmp_path / "a.db"
#144	db_b = tmp_path / "b.db"
#145	_route_mnemosyne(monkeypatch, {"session_a": db_a, "session_b": db_b})
#146
#147	# Bind memory to session_a
#148	monkeypatch.setenv("HERMES_SESSION_ID", "session_a")
#149	hermes_plugin._get_memory("session_a")
#150	triples_a = hermes_plugin._get_triples()
#151	assert Path(triples_a.db_path) == db_a
#152
#153	# Env changes to session_b — no explicit _get_memory call.
#154	monkeypatch.setenv("HERMES_SESSION_ID", "session_b")
#155
#156	# Next _get_triples() call should follow env, not stay on session_a.
#157	triples_b = hermes_plugin._get_triples()
#158	assert Path(triples_b.db_path) == db_b, (
#159	f"_get_triples() did not honor env change: still at "
#160	f"{triples_b.db_path} after env switch to session_b"
#161	)
#162

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public