Timmy_Foundation/hermes-agent
15
0
Fork 0
Code Issues 282 Pull Requests 87 Actions 1 Packages Projects Releases Wiki Activity
Files
75e31bee270b21b5dd7aada7e5f8a0358f58c41f
hermes-agent/tests/test_lossless_context.py
Alexander Whitestone 75e31bee27
All checks were successful
Lint / lint (pull_request) Successful in 9s
Details
feat(atlas): lossless context + memory subsystem from hermes-lcm and gbrain 
Implements the ATLAS (Adaptive Turn-Lineage Archival System) lossless
memory subsystem as described in issue #985. All 183 fixture-backed
tests pass.

agent/atlas/ — modular ATLAS package:
- turns.py: RawTurnStore, immutable append-only turn records with stable
  lineage IDs (session_id:seq:06d format). Turns are never deleted.
- dag.py: SummaryDAGStore, compaction that creates summary nodes storing
  source_turn_ids — every compaction is traceable back to raw turns.
- stores.py: WorldKnowledgeStore, DurableMemoryStore, SessionStateStore
  with strict routing (no mixed bucket).
- extractor.py: TypedLinkExtractor, deterministic regex-based extraction
  of 7 typed relation types on every write (no LLM calls):
  DEFINES, MODIFIES, REFERENCES, DEPENDS_ON, CONTRADICTS, PREFERS, LOCATES
- recall.py: RecallEngine with 3 explicit recall ops: search/describe/expand
- db.py: AtlasDB SQLite connection/schema bootstrap

agent/atlas_memory.py — AtlasMemory facade (higher-level API)
agent/lossless_context.py — JSONL-based lossless context subsystem
plugins/memory/atlas/ — MemoryProvider plugin wiring into Hermes lifecycle
tools/lossless_recall_tool.py — lossless_recall tool (search/describe/expand)
tools/memory_tool.py — extended with world_knowledge store target

tests/ — 183 fixture-backed tests covering all acceptance criteria:
- Every turn persisted with stable lineage IDs ✅
- Compaction builds retrievable summary DAG nodes with source references ✅
- 3 explicit recall operations (search/describe/expand) ✅
- Writes route to explicit stores (no mixed bucket) ✅
- 7 typed relation types with fixture-backed tests ✅
- test_recover_fact_from_compacted_context proves fact recovery without
  re-injecting the full original transcript ✅

Fixes #985
2026-04-22 10:15:21 -04:00

716 lines
29 KiB
Python
Raw Blame History

"""Tests for the lossless context + memory subsystem.
Covers:
- TurnRecord: lineage IDs, immutability, SHA-256
- SessionTurnStore: append-only persistence, load, search
- SummaryDAG: DAG node persistence, source references, expand
- LinkExtractor: all 5+ relation types, fixture-backed
- StoreRouter: three-store routing
- RecallEngine: search / describe / expand (the key recovery proof)
- Recovery test: fact survives compaction and is recoverable via expand()
"""
import json
import time
import pytest
from pathlib import Path
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from agent.lossless_context import (
TurnRecord,
SessionTurnStore,
SummaryDAG,
SummaryNode,
RelationLink,
RelationLinkStore,
RelationType,
StoreTier,
LinkExtractor,
StoreRouter,
RecallEngine,
ingest_turn,
compact_turns_to_dag,
create_lossless_context,
)
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_hermes_home(tmp_path):
"""Return a temporary directory that acts as HERMES_HOME."""
return tmp_path
@pytest.fixture
def session_id():
return "test-session-abc123"
@pytest.fixture
def turn_store(session_id, tmp_hermes_home):
return SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
@pytest.fixture
def summary_dag(session_id, tmp_hermes_home):
return SummaryDAG(session_id, hermes_home=tmp_hermes_home)
@pytest.fixture
def link_store(session_id, tmp_hermes_home):
return RelationLinkStore(tmp_hermes_home / "sessions" / session_id / "links.jsonl")
@pytest.fixture
def store_router(tmp_hermes_home):
return StoreRouter(hermes_home=tmp_hermes_home)
@pytest.fixture
def recall_engine(turn_store, summary_dag, link_store, store_router):
return RecallEngine(turn_store, summary_dag, link_store, store_router)
@pytest.fixture
def link_extractor():
return LinkExtractor()
# ---------------------------------------------------------------------------
# TurnRecord tests
# ---------------------------------------------------------------------------
class TestTurnRecord:
def test_lineage_id_format(self):
rec = TurnRecord.create(session_id="sess1", seq=7, role="user", content="Hello")
assert rec.lineage_id == "sess1:7"
def test_content_sha256_computed(self):
rec = TurnRecord.create(session_id="s", seq=0, role="user", content="test content")
import hashlib
expected = hashlib.sha256(b"test content").hexdigest()
assert rec.content_sha256 == expected
def test_roundtrip_dict(self):
rec = TurnRecord.create(session_id="s", seq=3, role="assistant", content="reply")
restored = TurnRecord.from_dict(rec.to_dict())
assert restored.lineage_id == rec.lineage_id
assert restored.content == rec.content
assert restored.role == rec.role
assert restored.seq == rec.seq
def test_stable_lineage_id_across_roundtrip(self):
"""lineage_id must survive serialization unchanged."""
rec = TurnRecord.create(session_id="stable", seq=42, role="tool", content="output")
d = rec.to_dict()
restored = TurnRecord.from_dict(d)
assert restored.lineage_id == "stable:42"
def test_tool_call_fields(self):
rec = TurnRecord.create(
session_id="s", seq=0, role="tool", content="result",
tool_name="read_file", tool_call_id="call-001"
)
assert rec.tool_name == "read_file"
assert rec.tool_call_id == "call-001"
d = rec.to_dict()
restored = TurnRecord.from_dict(d)
assert restored.tool_name == "read_file"
# ---------------------------------------------------------------------------
# SessionTurnStore tests
# ---------------------------------------------------------------------------
class TestSessionTurnStore:
def test_append_returns_record_with_lineage(self, turn_store, session_id):
rec = turn_store.append(role="user", content="Hi there")
assert rec.lineage_id == f"{session_id}:0"
assert rec.role == "user"
assert rec.content == "Hi there"
def test_sequence_increments(self, turn_store, session_id):
r1 = turn_store.append(role="user", content="first")
r2 = turn_store.append(role="assistant", content="second")
r3 = turn_store.append(role="tool", content="result")
assert r1.seq == 0
assert r2.seq == 1
assert r3.seq == 2
def test_persistence_across_reload(self, session_id, tmp_hermes_home):
store1 = SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
store1.append(role="user", content="persisted message")
store1.append(role="assistant", content="persisted reply")
# Reload from disk
store2 = SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
records = store2.load_all()
assert len(records) == 2
assert records[0].content == "persisted message"
assert records[1].content == "persisted reply"
def test_load_preserves_order(self, turn_store):
for i in range(5):
turn_store.append(role="user", content=f"message {i}")
records = turn_store.load_all()
assert [r.seq for r in records] == [0, 1, 2, 3, 4]
def test_search_case_insensitive(self, turn_store):
turn_store.append(role="user", content="I prefer Python for backend work")
turn_store.append(role="assistant", content="Understood")
turn_store.append(role="user", content="Unrelated message")
results = turn_store.search("PYTHON")
assert len(results) == 1
assert "Python" in results[0].content
def test_get_by_id(self, turn_store, session_id):
turn_store.append(role="user", content="first")
r = turn_store.append(role="user", content="target")
turn_store.append(role="user", content="after")
found = turn_store.get_by_id(r.lineage_id)
assert found is not None
assert found.content == "target"
def test_get_by_id_missing(self, turn_store):
assert turn_store.get_by_id("nonexistent:999") is None
def test_turns_are_never_deleted(self, session_id, tmp_hermes_home):
"""Appending more turns never removes old ones (lossless)."""
store = SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
for i in range(10):
store.append(role="user", content=f"turn {i}")
records = store.load_all()
assert len(records) == 10
def test_seq_resumes_after_reload(self, session_id, tmp_hermes_home):
"""New store instance continues from the last sequence number."""
store1 = SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
store1.append(role="user", content="a")
store1.append(role="user", content="b")
store2 = SessionTurnStore(session_id, hermes_home=tmp_hermes_home)
r = store2.append(role="user", content="c")
assert r.seq == 2
# ---------------------------------------------------------------------------
# SummaryDAG tests
# ---------------------------------------------------------------------------
class TestSummaryDAG:
def test_add_node_returns_node_with_id(self, summary_dag, session_id):
node = summary_dag.add_node(
summary_text="Worked on issue #123",
source_turn_ids=["sess:0", "sess:1", "sess:2"],
)
assert node.node_id == f"{session_id}:summary:0"
assert node.session_id == session_id
assert node.source_turn_ids == ["sess:0", "sess:1", "sess:2"]
def test_sequential_node_ids(self, summary_dag, session_id):
n1 = summary_dag.add_node("first summary", ["s:0"])
n2 = summary_dag.add_node("second summary", ["s:1"])
assert n1.node_id.endswith(":summary:0")
assert n2.node_id.endswith(":summary:1")
def test_persistence_across_reload(self, session_id, tmp_hermes_home):
dag1 = SummaryDAG(session_id, hermes_home=tmp_hermes_home)
dag1.add_node("first", ["t:0"])
dag1.add_node("second", ["t:1"])
dag2 = SummaryDAG(session_id, hermes_home=tmp_hermes_home)
nodes = dag2.load_all()
assert len(nodes) == 2
assert nodes[0].summary_text == "first"
def test_parent_node_id(self, summary_dag):
n1 = summary_dag.add_node("initial", ["t:0"])
n2 = summary_dag.add_node("update", ["t:1"], parent_node_id=n1.node_id)
assert n2.parent_node_id == n1.node_id
def test_get_by_id(self, summary_dag):
n = summary_dag.add_node("target summary", ["t:5"])
found = summary_dag.get_by_id(n.node_id)
assert found is not None
assert found.summary_text == "target summary"
def test_get_latest(self, summary_dag):
summary_dag.add_node("older", ["t:0"])
latest = summary_dag.add_node("newest", ["t:1"])
assert summary_dag.get_latest().node_id == latest.node_id
def test_search(self, summary_dag):
summary_dag.add_node("Fixed a bug in auth module", ["t:0"])
summary_dag.add_node("Refactored the database layer", ["t:1"])
results = summary_dag.search("auth")
assert len(results) == 1
assert "auth" in results[0].summary_text
# ---------------------------------------------------------------------------
# LinkExtractor tests — fixture-backed, deterministic
# ---------------------------------------------------------------------------
class TestLinkExtractor:
"""Fixture-backed tests for all 5+ relation types."""
FIXTURES = [
# (input_text, expected_relation, partial_subject, partial_object)
# PREFERS
("I prefer Python over JavaScript for backend work.",
RelationType.PREFERS, "user", "Python"),
("My preferred editor is Neovim.",
RelationType.PREFERS, "user", "Neovim"),
# CORRECTS
("Actually, the port is 8081 not 8080.",
RelationType.CORRECTS, "user", "the port is 8081"),
("I meant the config file is at /etc/app/config.yaml.",
RelationType.CORRECTS, "user", "the config"),
# USES
("The project uses PostgreSQL for data storage.",
RelationType.USES, "project", "PostgreSQL"),
("Using Docker for containerization.",
RelationType.USES, "project", "Docker"),
# LOCATED_AT
("The config is at /etc/hermes/config.yaml.",
RelationType.LOCATED_AT, "config", "/etc/hermes/config.yaml"),
# DEPENDS_ON
("hermes-agent requires openai>=2.0.",
RelationType.DEPENDS_ON, "hermes-agent", "openai"),
# CONFIGURES
("debug is set to true.",
RelationType.CONFIGURES, "debug", "true"),
("Set log_level to DEBUG.",
RelationType.CONFIGURES, "log_level", "DEBUG"),
]
@pytest.mark.parametrize("text,expected_rel,partial_subj,partial_obj", FIXTURES)
def test_extraction(self, link_extractor, text, expected_rel, partial_subj, partial_obj):
links = link_extractor.extract(text, source_turn_id="test:0")
matching = [
lk for lk in links
if lk.relation == expected_rel.value
and (partial_subj.lower() in lk.subject.lower() or
partial_subj.lower() in lk.object_.lower())
]
assert len(matching) >= 1, (
f"Expected at least one {expected_rel.value} link "
f"with subject/object containing '{partial_subj}' or '{partial_obj}'. "
f"Got links: {[(lk.relation, lk.subject, lk.object_) for lk in links]}"
)
def test_source_turn_id_preserved(self, link_extractor):
links = link_extractor.extract("I prefer Go over Python.", source_turn_id="session:7")
assert all(lk.source_turn_id == "session:7" for lk in links)
def test_empty_text_returns_empty(self, link_extractor):
assert link_extractor.extract("", source_turn_id="s:0") == []
def test_short_text_returns_empty(self, link_extractor):
assert link_extractor.extract("Hi", source_turn_id="s:0") == []
def test_no_duplicates_for_same_pattern(self, link_extractor):
links = link_extractor.extract(
"I prefer Python. I prefer Python.", source_turn_id="s:0"
)
# Should deduplicate
prefers = [lk for lk in links if lk.relation == RelationType.PREFERS.value]
assert len(prefers) <= 2 # May match twice on different patterns but not the same exact key
def test_confidence_in_range(self, link_extractor):
links = link_extractor.extract("The project uses Redis.", source_turn_id="s:0")
for lk in links:
assert 0.0 <= lk.confidence <= 1.0
def test_store_tier_assigned(self, link_extractor):
links = link_extractor.extract("I prefer Python.", source_turn_id="s:0")
for lk in links:
assert lk.store_tier in (
StoreTier.WORLD_KNOWLEDGE.value,
StoreTier.DURABLE_MEMORY.value,
StoreTier.SESSION_STATE.value,
)
def test_five_distinct_relation_types_covered(self, link_extractor):
"""A single input set covers all 5+ relation types."""
texts = [
("I prefer TypeScript over JavaScript.", RelationType.PREFERS),
("Actually the database is PostgreSQL.", RelationType.CORRECTS),
("The project uses FastAPI.", RelationType.USES),
("The config is at /app/settings.py.", RelationType.LOCATED_AT),
("myapp requires redis>=4.0.", RelationType.DEPENDS_ON),
("cache_timeout is set to 3600.", RelationType.CONFIGURES),
]
found_types = set()
for text, expected_rel in texts:
links = link_extractor.extract(text, source_turn_id="s:0")
for lk in links:
found_types.add(lk.relation)
assert RelationType.PREFERS.value in found_types
assert RelationType.CORRECTS.value in found_types
assert RelationType.USES.value in found_types
assert RelationType.DEPENDS_ON.value in found_types or RelationType.CONFIGURES.value in found_types
# At least 5 distinct types found
assert len(found_types) >= 5, f"Expected ≥5 relation types, found: {found_types}"
# ---------------------------------------------------------------------------
# StoreRouter tests
# ---------------------------------------------------------------------------
class TestStoreRouter:
def test_write_and_load_durable(self, store_router):
store_router.write(
tier=StoreTier.DURABLE_MEMORY,
category="user_pref",
content="user prefers dark mode",
source_id="s:0",
)
facts = store_router.load(StoreTier.DURABLE_MEMORY)
assert len(facts) == 1
assert "dark mode" in facts[0]["content"]
def test_write_and_load_world_knowledge(self, store_router):
store_router.write(
tier=StoreTier.WORLD_KNOWLEDGE,
category="USES",
content="project USES PostgreSQL",
source_id="s:1",
)
facts = store_router.load(StoreTier.WORLD_KNOWLEDGE)
assert len(facts) >= 1
assert any("PostgreSQL" in f["content"] for f in facts)
def test_session_state_tier_is_noop(self, store_router):
# Writing to SESSION_STATE should not create any file
store_router.write(tier=StoreTier.SESSION_STATE, category="turn", content="stuff")
# No file should be created for session_state (it's handled by SessionTurnStore)
session_path = store_router._stores.get(StoreTier.SESSION_STATE)
assert session_path is None
def test_search_across_tiers(self, store_router):
store_router.write(StoreTier.DURABLE_MEMORY, "pref", "user prefers vim editor")
store_router.write(StoreTier.WORLD_KNOWLEDGE, "USES", "project uses React")
# Search both
vim_results = store_router.search("vim")
assert len(vim_results) == 1
assert "vim" in vim_results[0]["content"]
react_results = store_router.search("React")
assert len(react_results) == 1
def test_search_tier_filter(self, store_router):
store_router.write(StoreTier.DURABLE_MEMORY, "pref", "prefers dark mode")
store_router.write(StoreTier.WORLD_KNOWLEDGE, "fact", "dark web reference")
# Search only DURABLE_MEMORY
results = store_router.search("dark", tier=StoreTier.DURABLE_MEMORY)
assert all(r["tier"] == StoreTier.DURABLE_MEMORY for r in results)
# ---------------------------------------------------------------------------
# RecallEngine tests
# ---------------------------------------------------------------------------
class TestRecallEngine:
def test_search_finds_turn_content(self, recall_engine, turn_store):
turn_store.append(role="user", content="I prefer Go for systems programming")
result = recall_engine.search("Go for systems")
assert result["total_results"] > 0
assert len(result["turns"]) >= 1
assert "Go for systems" in result["turns"][0]["content_preview"]
def test_search_finds_summary_content(self, recall_engine, summary_dag):
summary_dag.add_node(
"User prefers TypeScript. Fixed auth bug.", source_turn_ids=["s:0"]
)
result = recall_engine.search("TypeScript")
assert len(result["summaries"]) >= 1
def test_search_returns_structured_result(self, recall_engine, turn_store):
turn_store.append(role="user", content="test query")
result = recall_engine.search("test query")
assert "query" in result
assert "turns" in result
assert "summaries" in result
assert "links" in result
assert "total_results" in result
def test_describe_turn(self, recall_engine, turn_store, session_id):
r = turn_store.append(role="user", content="The config is at /home/user/app.yaml")
result = recall_engine.describe(r.lineage_id)
assert result["type"] == "turn"
assert result["lineage_id"] == r.lineage_id
assert result["content"] == "The config is at /home/user/app.yaml"
assert result["role"] == "user"
def test_describe_summary_node(self, recall_engine, summary_dag, session_id):
node = summary_dag.add_node("summary content", source_turn_ids=["x:0", "x:1"])
result = recall_engine.describe(node.node_id)
assert result["type"] == "summary_node"
assert result["node_id"] == node.node_id
assert result["source_turn_ids"] == ["x:0", "x:1"]
def test_describe_missing_returns_error(self, recall_engine):
result = recall_engine.describe("nonexistent:999")
assert "error" in result
def test_expand_shows_source_turns(self, recall_engine, turn_store, summary_dag, session_id):
"""expand() must retrieve the original turns that produced a summary."""
# Ingest some turns
r0 = turn_store.append(role="user", content="Fact: the DB is Postgres")
r1 = turn_store.append(role="assistant", content="Noted — using PostgreSQL")
r2 = turn_store.append(role="user", content="Also use Redis for caching")
# Create a summary DAG node referencing those turns
node = summary_dag.add_node(
summary_text="Established DB stack: Postgres + Redis cache",
source_turn_ids=[r0.lineage_id, r1.lineage_id, r2.lineage_id],
)
# Expand the summary — should retrieve original turn content
result = recall_engine.expand(node.node_id)
assert "error" not in result
assert result["node_id"] == node.node_id
assert result["source_count"] == 3
assert result["found_count"] == 3
# Verify we can read the original content
contents = {t["content"] for t in result["source_turns"]}
assert "Fact: the DB is Postgres" in contents
assert "Noted — using PostgreSQL" in contents
assert "Also use Redis for caching" in contents
def test_expand_missing_node(self, recall_engine):
result = recall_engine.expand("nonexistent:summary:0")
assert "error" in result
def test_expand_follows_parent_chain(self, recall_engine, turn_store, summary_dag):
"""expand() returns the parent chain for DAG traversal."""
r = turn_store.append(role="user", content="initial context")
n1 = summary_dag.add_node("first compaction", source_turn_ids=[r.lineage_id])
r2 = turn_store.append(role="user", content="more context")
n2 = summary_dag.add_node(
"second compaction (includes first)",
source_turn_ids=[r2.lineage_id],
parent_node_id=n1.node_id,
)
result = recall_engine.expand(n2.node_id)
assert result["parent_chain"]
assert result["parent_chain"][0]["node_id"] == n1.node_id
# ---------------------------------------------------------------------------
# Integration test: fact recovery from compacted context
# ---------------------------------------------------------------------------
class TestFactRecoveryFromCompactedContext:
"""Prove that the agent can recover a fact from compacted context
without re-injecting the full original transcript.
Scenario:
1. Ingest multiple turns (including a key fact)
2. Create a summary DAG node (simulating context compaction)
3. The original turns are "gone" from the live context window
4. Use RecallEngine.expand() to recover the original fact
"""
def test_recover_fact_via_expand(self, session_id, tmp_hermes_home):
"""Core acceptance criterion: recover a fact from compacted context."""
turn_store, summary_dag, link_extractor, link_store, store_router, recall_engine = \
create_lossless_context(session_id, hermes_home=tmp_hermes_home)
# Step 1: Ingest turns that contain a key fact
key_fact = "The production database password is stored in /etc/secrets/db.env"
r0 = ingest_turn("What's the database setup?", "user", turn_store, link_extractor, link_store, store_router)
r1 = ingest_turn(key_fact, "assistant", turn_store, link_extractor, link_store, store_router)
r2 = ingest_turn("Thanks, understood.", "user", turn_store, link_extractor, link_store, store_router)
# Step 2: Simulate context compaction — create a summary node
summary_text = "User asked about database setup. Assistant explained that credentials are stored in /etc/secrets/db.env."
node = compact_turns_to_dag(
summary_text=summary_text,
source_turn_ids=[r0.lineage_id, r1.lineage_id, r2.lineage_id],
summary_dag=summary_dag,
)
# Step 3: Simulate that live context is now ONLY the summary
# (original turns r0, r1, r2 are "gone" from the active window)
# The agent only sees the summary node ID from the compacted message.
# Step 4: Use expand() to recover the original content
expanded = recall_engine.expand(node.node_id)
assert "error" not in expanded, f"expand() failed: {expanded}"
assert expanded["found_count"] == 3, f"Expected 3 source turns, got: {expanded}"
# The key fact must be in the expanded source turns
all_content = " ".join(t["content"] for t in expanded["source_turns"])
assert key_fact in all_content, (
f"Key fact not found in expanded turns. Content: {all_content}"
)
def test_search_recovers_fact_from_compacted_summary(self, session_id, tmp_hermes_home):
"""search() can find a fact that's only in a summary node."""
turn_store, summary_dag, link_extractor, link_store, store_router, recall_engine = \
create_lossless_context(session_id, hermes_home=tmp_hermes_home)
# Ingest minimal turns then compact them
r = ingest_turn("The API key is ABCD-1234.", "assistant", turn_store, link_extractor, link_store, store_router)
compact_turns_to_dag(
summary_text="User verified API key ABCD-1234 for production.",
source_turn_ids=[r.lineage_id],
summary_dag=summary_dag,
)
# Search for the API key value — should find it in summaries
result = recall_engine.search("ABCD-1234")
assert result["total_results"] > 0
# Found in either turns or summaries
found_in_turns = any("ABCD-1234" in t["content_preview"] for t in result["turns"])
found_in_summaries = any("ABCD-1234" in s["summary_preview"] for s in result["summaries"])
assert found_in_turns or found_in_summaries, (
f"Fact not found. turns={result['turns']}, summaries={result['summaries']}"
)
def test_typed_links_survive_compaction(self, session_id, tmp_hermes_home):
"""Typed links extracted at write time survive even if turns are compacted."""
turn_store, summary_dag, link_extractor, link_store, store_router, recall_engine = \
create_lossless_context(session_id, hermes_home=tmp_hermes_home)
# Ingest a turn with a clear preference
r = ingest_turn(
"I prefer PostgreSQL over MySQL for this project.",
"user",
turn_store,
link_extractor,
link_store,
store_router,
)
# Compact the turn
compact_turns_to_dag(
summary_text="Database preference established.",
source_turn_ids=[r.lineage_id],
summary_dag=summary_dag,
)
# The original typed link should still be searchable via RecallEngine
result = recall_engine.search("PostgreSQL")
# Should find the preference in typed links or turns
assert result["total_results"] > 0
# Verify typed links were extracted
link_results = result["links"]
prefers_links = [lk for lk in link_results if lk["relation"] == RelationType.PREFERS.value]
assert len(prefers_links) >= 1, f"Expected PREFERS link, got: {link_results}"
# ---------------------------------------------------------------------------
# ingest_turn integration test
# ---------------------------------------------------------------------------
class TestIngestTurn:
def test_ingest_appends_and_extracts(self, session_id, tmp_hermes_home):
turn_store, _, link_extractor, link_store, store_router, _ = \
create_lossless_context(session_id, hermes_home=tmp_hermes_home)
record = ingest_turn(
"The project uses FastAPI for the REST endpoints.",
"user",
turn_store, link_extractor, link_store, store_router,
)
assert record.role == "user"
assert "FastAPI" in record.content
# Check that a USES link was extracted and stored
links = link_store.load_all()
uses_links = [lk for lk in links if lk.relation == RelationType.USES.value]
assert len(uses_links) >= 1
def test_tool_turns_not_link_extracted(self, session_id, tmp_hermes_home):
turn_store, _, link_extractor, link_store, store_router, _ = \
create_lossless_context(session_id, hermes_home=tmp_hermes_home)
ingest_turn(
"The project uses FastAPI", # Tool results are noisy — don't extract
"tool",
turn_store, link_extractor, link_store, store_router,
tool_name="read_file", tool_call_id="call-abc",
)
links = link_store.load_all()
assert len(links) == 0
# ---------------------------------------------------------------------------
# Memory tool world_knowledge target test
# ---------------------------------------------------------------------------
class TestMemoryStoreWorldKnowledge:
"""Verify that the world_knowledge store target works in MemoryStore."""
def test_world_knowledge_add_and_load(self, tmp_hermes_home, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_hermes_home))
from hermes_constants import get_hermes_home
# Patch get_memory_dir to use tmp_hermes_home
import tools.memory_tool as mt
monkeypatch.setattr(mt, "get_memory_dir", lambda: tmp_hermes_home / "memories")
store = mt.MemoryStore()
store.load_from_disk()
# Add to world_knowledge
result = store.add("world_knowledge", "Python was created by Guido van Rossum")
assert result["success"] is True
# Verify entry exists
entries = store._entries_for("world_knowledge")
assert any("Guido" in e for e in entries)
def test_world_knowledge_char_limit_separate(self, tmp_hermes_home, monkeypatch):
import tools.memory_tool as mt
monkeypatch.setattr(mt, "get_memory_dir", lambda: tmp_hermes_home / "memories")
store = mt.MemoryStore(world_knowledge_char_limit=100)
store.load_from_disk()
# Add something long that exceeds the limit
long_entry = "A" * 150
result = store.add("world_knowledge", long_entry)
assert result["success"] is False
assert "limit" in result["error"].lower() or "exceed" in result["error"].lower()
def test_invalid_target_rejected(self, tmp_hermes_home, monkeypatch):
import tools.memory_tool as mt
monkeypatch.setattr(mt, "get_memory_dir", lambda: tmp_hermes_home / "memories")
store = mt.MemoryStore()
store.load_from_disk()
result = mt.memory_tool("add", target="unknown_store", content="test", store=store)
data = json.loads(result)
assert data.get("success") is False
assert "unknown_store" in data.get("error", "")
Reference in New Issue View Git Blame Copy Permalink