Mdmem Agent
8fdec33037
- Remove references to PLAN.md, SECURITY_ANALYSIS.md, deploy/ paths - Delete test_deploy.py (tests deleted K8s manifests) - Clean up eval/ references in comments and docstrings
105 lines
3.6 KiB
Python
105 lines
3.6 KiB
Python
from pathlib import Path
|
|
|
|
from mdmem.logging import AuditLog
|
|
|
|
|
|
def test_recent_merges_persisted_history_across_restart_without_duplicates(tmp_path):
|
|
"""`AuditLog.recent` must return the merged tail of persisted history plus
|
|
in-memory entries across a restart, newest-first, without duplicating the
|
|
entries that show up in both (logging.py:77-92)."""
|
|
path = tmp_path / "audit.jsonl"
|
|
first = AuditLog(path)
|
|
for i in range(5):
|
|
first.record(tool="memory_search", seq=i)
|
|
|
|
# A fresh AuditLog over the same file simulates a process restart: its
|
|
# in-memory list starts empty even though history is already on disk.
|
|
second = AuditLog(path)
|
|
second.record(tool="memory_write", seq=5)
|
|
|
|
got = second.recent(100)
|
|
seqs = [entry["seq"] for entry in got]
|
|
assert seqs == [5, 4, 3, 2, 1, 0]
|
|
assert len(seqs) == len(set(seqs))
|
|
|
|
|
|
def test_recent_respects_limit_across_the_persisted_in_memory_boundary(tmp_path):
|
|
path = tmp_path / "audit.jsonl"
|
|
first = AuditLog(path)
|
|
for i in range(5):
|
|
first.record(tool="memory_search", seq=i)
|
|
|
|
second = AuditLog(path)
|
|
second.record(tool="memory_write", seq=5)
|
|
second.record(tool="memory_write", seq=6)
|
|
|
|
got = second.recent(3)
|
|
assert [entry["seq"] for entry in got] == [6, 5, 4]
|
|
|
|
|
|
def test_recent_before_any_new_call_still_sees_persisted_history(tmp_path):
|
|
"""Even with zero in-memory entries, a fresh instance over an existing
|
|
file must not report an empty history."""
|
|
path = tmp_path / "audit.jsonl"
|
|
first = AuditLog(path)
|
|
for i in range(3):
|
|
first.record(tool="memory_search", seq=i)
|
|
|
|
second = AuditLog(path)
|
|
got = second.recent(100)
|
|
assert [entry["seq"] for entry in got] == [2, 1, 0]
|
|
|
|
|
|
def test_recent_without_a_persisted_path_uses_memory_only(tmp_path):
|
|
log = AuditLog(None)
|
|
log.record(tool="memory_search", seq=0)
|
|
log.record(tool="memory_write", seq=1)
|
|
assert [entry["seq"] for entry in log.recent(100)] == [1, 0]
|
|
|
|
|
|
def test_recent_does_not_read_the_whole_audit_file(tmp_path, monkeypatch):
|
|
"""/metrics calls recent() on every Prometheus scrape and the audit file is
|
|
append-only, so a months-old trail must not become a multi-megabyte read
|
|
every 15 seconds."""
|
|
path = tmp_path / "audit.jsonl"
|
|
audit = AuditLog(path)
|
|
for i in range(20_000):
|
|
audit.record(tool="memory_search", ok=True, seq=i)
|
|
assert path.stat().st_size > 1_000_000
|
|
|
|
fresh = AuditLog(path) # restart: nothing in memory
|
|
read_bytes = 0
|
|
real_read = Path.open
|
|
|
|
def counting_open(self, *a, **kw):
|
|
fh = real_read(self, *a, **kw)
|
|
inner_read = fh.read
|
|
|
|
def tracked(n=-1):
|
|
data = inner_read(n)
|
|
nonlocal read_bytes
|
|
read_bytes += len(data)
|
|
return data
|
|
|
|
fh.read = tracked # type: ignore[method-assign]
|
|
return fh
|
|
|
|
monkeypatch.setattr(Path, "open", counting_open)
|
|
rows = fresh.recent(100)
|
|
assert [r["seq"] for r in rows] == list(range(19_999, 19_899, -1))
|
|
assert read_bytes < path.stat().st_size // 4, f"read {read_bytes} of {path.stat().st_size} bytes"
|
|
|
|
|
|
def test_recent_widens_its_window_for_unusually_long_entries(tmp_path):
|
|
"""The tail window is sized by an estimate, not a guarantee: long entries must
|
|
still yield the full requested count, not a short page."""
|
|
path = tmp_path / "audit.jsonl"
|
|
audit = AuditLog(path)
|
|
for i in range(300):
|
|
audit.record(tool="memory_search", ok=True, seq=i, filler="x" * 4000)
|
|
|
|
fresh = AuditLog(path)
|
|
rows = fresh.recent(100)
|
|
assert len(rows) == 100
|
|
assert [r["seq"] for r in rows] == list(range(299, 199, -1))
|