Anto01
775960c8c8
Pipeline observability: - Retrieval harness runs nightly (Step E in batch-extract.sh) - Pipeline summary persisted to project state after each run (pipeline_last_run, pipeline_summary, retrieval_harness_result) - Dashboard enhanced: interaction total + by_client, pipeline health (last_run, hours_since, harness results, triage stats), dynamic project list from registry Phase 10 — reinforcement-based auto-promotion: - auto_promote_reinforced(): candidates with reference_count >= 3 and confidence >= 0.7 auto-graduate to active - expire_stale_candidates(): candidates unreinforced for 14+ days auto-rejected to prevent unbounded queue growth - Both wired into nightly cron (Step B2) - Batch script: scripts/auto_promote_reinforced.py (--dry-run support) Knowledge seeding: - scripts/seed_project_state.py: 26 curated Trusted Project State entries across p04-gigabit, p05-interferometer, p06-polisher, atomizer-v2, abb-space, atocore (decisions, requirements, facts, contacts, milestones) Tests: 299 → 303 (4 new Phase 10 tests) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""Tests for Memory Core."""
|
|
|
|
import os
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
import atocore.config as _config
|
|
from atocore.models.database import init_db
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def isolated_db():
|
|
"""Give each test a completely isolated database."""
|
|
tmpdir = tempfile.mkdtemp()
|
|
os.environ["ATOCORE_DATA_DIR"] = tmpdir
|
|
|
|
# Replace the global settings so all modules see the new data_dir
|
|
_config.settings = _config.Settings()
|
|
|
|
# Also reset any module-level references to the old settings
|
|
import atocore.models.database
|
|
# database.py now uses _config.settings dynamically, so no patch needed
|
|
|
|
init_db()
|
|
yield tmpdir
|
|
|
|
|
|
def test_create_memory(isolated_db):
|
|
from atocore.memory.service import create_memory
|
|
mem = create_memory("identity", "User is a mechanical engineer specializing in optics")
|
|
assert mem.memory_type == "identity"
|
|
assert mem.status == "active"
|
|
assert mem.confidence == 1.0
|
|
|
|
|
|
def test_create_memory_invalid_type(isolated_db):
|
|
from atocore.memory.service import create_memory
|
|
with pytest.raises(ValueError, match="Invalid memory type"):
|
|
create_memory("invalid_type", "some content")
|
|
|
|
|
|
def test_create_memory_dedup(isolated_db):
|
|
from atocore.memory.service import create_memory
|
|
m1 = create_memory("identity", "User is an engineer")
|
|
m2 = create_memory("identity", "User is an engineer")
|
|
assert m1.id == m2.id
|
|
|
|
|
|
def test_create_memory_dedup_is_project_scoped(isolated_db):
|
|
from atocore.memory.service import create_memory
|
|
m1 = create_memory("project", "Uses SQLite for local state", project="atocore")
|
|
m2 = create_memory("project", "Uses SQLite for local state", project="openclaw")
|
|
assert m1.id != m2.id
|
|
|
|
|
|
def test_project_is_persisted_and_filterable(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories
|
|
create_memory("project", "Uses SQLite for local state", project="atocore")
|
|
create_memory("project", "Uses Postgres in production", project="openclaw")
|
|
|
|
atocore_memories = get_memories(memory_type="project", project="atocore")
|
|
assert len(atocore_memories) == 1
|
|
assert atocore_memories[0].project == "atocore"
|
|
|
|
|
|
def test_get_memories_all(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories
|
|
create_memory("identity", "User is an engineer")
|
|
create_memory("preference", "Prefers Python with type hints")
|
|
create_memory("knowledge", "Zerodur has near-zero thermal expansion")
|
|
|
|
mems = get_memories()
|
|
assert len(mems) == 3
|
|
|
|
|
|
def test_get_memories_by_type(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories
|
|
create_memory("identity", "User is an engineer")
|
|
create_memory("preference", "Prefers concise code")
|
|
create_memory("preference", "Uses FastAPI for APIs")
|
|
|
|
mems = get_memories(memory_type="preference")
|
|
assert len(mems) == 2
|
|
|
|
|
|
def test_get_memories_active_only(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories, invalidate_memory
|
|
m = create_memory("knowledge", "Fact about optics")
|
|
invalidate_memory(m.id)
|
|
|
|
assert len(get_memories(active_only=True)) == 0
|
|
assert len(get_memories(active_only=False)) == 1
|
|
|
|
|
|
def test_get_memories_min_confidence(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories
|
|
create_memory("knowledge", "High confidence fact", confidence=0.9)
|
|
create_memory("knowledge", "Low confidence fact", confidence=0.3)
|
|
|
|
high = get_memories(min_confidence=0.5)
|
|
assert len(high) == 1
|
|
assert high[0].confidence == 0.9
|
|
|
|
|
|
def test_update_memory(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories, update_memory
|
|
mem = create_memory("knowledge", "Initial fact")
|
|
update_memory(mem.id, content="Updated fact", confidence=0.8)
|
|
|
|
mems = get_memories()
|
|
assert len(mems) == 1
|
|
assert mems[0].content == "Updated fact"
|
|
assert mems[0].confidence == 0.8
|
|
|
|
|
|
def test_update_memory_rejects_duplicate_active_memory(isolated_db):
|
|
from atocore.memory.service import create_memory, update_memory
|
|
import pytest
|
|
|
|
first = create_memory("knowledge", "Canonical fact", project="atocore")
|
|
second = create_memory("knowledge", "Different fact", project="atocore")
|
|
|
|
with pytest.raises(ValueError, match="duplicate active memory"):
|
|
update_memory(second.id, content="Canonical fact")
|
|
|
|
|
|
def test_create_memory_validates_confidence(isolated_db):
|
|
from atocore.memory.service import create_memory
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError, match="Confidence must be between 0.0 and 1.0"):
|
|
create_memory("knowledge", "Out of range", confidence=1.5)
|
|
|
|
|
|
def test_invalidate_memory(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories, invalidate_memory
|
|
mem = create_memory("knowledge", "Wrong fact")
|
|
invalidate_memory(mem.id)
|
|
assert len(get_memories(active_only=True)) == 0
|
|
|
|
|
|
def test_supersede_memory(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories, supersede_memory
|
|
mem = create_memory("knowledge", "Old fact")
|
|
supersede_memory(mem.id)
|
|
|
|
mems = get_memories(active_only=False)
|
|
assert len(mems) == 1
|
|
assert mems[0].status == "superseded"
|
|
|
|
|
|
def test_memories_for_context(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories_for_context
|
|
create_memory("identity", "User is a senior mechanical engineer")
|
|
create_memory("preference", "Prefers Python with type hints")
|
|
|
|
text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=500)
|
|
assert "--- AtoCore Memory ---" in text
|
|
assert "[identity]" in text
|
|
assert "[preference]" in text
|
|
assert chars > 0
|
|
|
|
|
|
def test_memories_for_context_reserves_room_for_each_type(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories_for_context
|
|
create_memory("identity", "Identity entry that is intentionally long so it could consume the whole budget on its own")
|
|
create_memory("preference", "Preference entry that should still appear")
|
|
|
|
text, _ = get_memories_for_context(memory_types=["identity", "preference"], budget=120)
|
|
assert "[preference]" in text
|
|
|
|
|
|
def test_memories_for_context_respects_actual_serialized_budget(isolated_db):
|
|
from atocore.memory.service import create_memory, get_memories_for_context
|
|
create_memory("identity", "Identity text that should fit the wrapper-aware memory budget calculation")
|
|
create_memory("preference", "Preference text that should also fit")
|
|
|
|
text, chars = get_memories_for_context(memory_types=["identity", "preference"], budget=140)
|
|
assert chars == len(text)
|
|
assert chars <= 140
|
|
|
|
|
|
def test_memories_for_context_empty(isolated_db):
|
|
from atocore.memory.service import get_memories_for_context
|
|
text, chars = get_memories_for_context()
|
|
assert text == ""
|
|
assert chars == 0
|
|
|
|
|
|
# --- Phase 10: auto-promotion + candidate expiry ---
|
|
|
|
|
|
def _get_memory_by_id(memory_id):
|
|
"""Helper: fetch a single memory by ID."""
|
|
from atocore.models.database import get_connection
|
|
with get_connection() as conn:
|
|
row = conn.execute("SELECT * FROM memories WHERE id = ?", (memory_id,)).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def test_auto_promote_reinforced_basic(isolated_db):
|
|
from atocore.memory.service import (
|
|
auto_promote_reinforced,
|
|
create_memory,
|
|
reinforce_memory,
|
|
)
|
|
|
|
mem_obj = create_memory("knowledge", "Zerodur has near-zero CTE", status="candidate", confidence=0.7)
|
|
mid = mem_obj.id
|
|
# reinforce_memory only touches active memories, so we need to
|
|
# promote first to reinforce, then demote back to candidate —
|
|
# OR just bump reference_count + last_referenced_at directly
|
|
from atocore.models.database import get_connection
|
|
from datetime import datetime, timezone
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE memories SET reference_count = 3, last_referenced_at = ? WHERE id = ?",
|
|
(now, mid),
|
|
)
|
|
|
|
promoted = auto_promote_reinforced(min_reference_count=3, min_confidence=0.7)
|
|
assert mid in promoted
|
|
mem = _get_memory_by_id(mid)
|
|
assert mem["status"] == "active"
|
|
|
|
|
|
def test_auto_promote_reinforced_ignores_low_refs(isolated_db):
|
|
from atocore.memory.service import auto_promote_reinforced, create_memory
|
|
from atocore.models.database import get_connection
|
|
from datetime import datetime, timezone
|
|
|
|
mem_obj = create_memory("knowledge", "Some knowledge", status="candidate", confidence=0.7)
|
|
mid = mem_obj.id
|
|
now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE memories SET reference_count = 1, last_referenced_at = ? WHERE id = ?",
|
|
(now, mid),
|
|
)
|
|
|
|
promoted = auto_promote_reinforced(min_reference_count=3, min_confidence=0.7)
|
|
assert mid not in promoted
|
|
mem = _get_memory_by_id(mid)
|
|
assert mem["status"] == "candidate"
|
|
|
|
|
|
def test_expire_stale_candidates(isolated_db):
|
|
from atocore.memory.service import create_memory, expire_stale_candidates
|
|
from atocore.models.database import get_connection
|
|
|
|
mem_obj = create_memory("knowledge", "Old unreferenced fact", status="candidate")
|
|
mid = mem_obj.id
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE memories SET created_at = datetime('now', '-30 days') WHERE id = ?",
|
|
(mid,),
|
|
)
|
|
|
|
expired = expire_stale_candidates(max_age_days=14)
|
|
assert mid in expired
|
|
mem = _get_memory_by_id(mid)
|
|
assert mem["status"] == "invalid"
|
|
|
|
|
|
def test_expire_stale_candidates_keeps_reinforced(isolated_db):
|
|
from atocore.memory.service import create_memory, expire_stale_candidates
|
|
from atocore.models.database import get_connection
|
|
|
|
mem_obj = create_memory("knowledge", "Referenced fact", status="candidate")
|
|
mid = mem_obj.id
|
|
with get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE memories SET reference_count = 1, "
|
|
"created_at = datetime('now', '-30 days') WHERE id = ?",
|
|
(mid,),
|
|
)
|
|
|
|
expired = expire_stale_candidates(max_age_days=14)
|
|
assert mid not in expired
|
|
mem = _get_memory_by_id(mid)
|
|
assert mem["status"] == "candidate"
|