Compare commits
4 Commits
feat/mnemo
...
fix/mnemos
| Author | SHA1 | Date | |
|---|---|---|---|
| 3361a0e259 | |||
| 8fb0a50b91 | |||
| 99e4baf54b | |||
| b0e24af7fe |
@@ -67,7 +67,7 @@ modules:
|
||||
cli:
|
||||
status: shipped
|
||||
files: [cli.py]
|
||||
description: CLI interface — stats, search, ingest, link, topics, remove, export, clusters, hubs, bridges, rebuild, tag/untag/retag, timeline, neighbors, consolidate
|
||||
description: CLI interface — stats, search, ingest, link, topics, remove, export, clusters, hubs, bridges, rebuild, tag/untag/retag, timeline, neighbors, consolidate, path, touch, decay, vitality, fading, vibrant
|
||||
|
||||
tests:
|
||||
status: shipped
|
||||
|
||||
@@ -4,7 +4,9 @@ Provides: mnemosyne ingest, mnemosyne search, mnemosyne link, mnemosyne stats,
|
||||
mnemosyne topics, mnemosyne remove, mnemosyne export,
|
||||
mnemosyne clusters, mnemosyne hubs, mnemosyne bridges, mnemosyne rebuild,
|
||||
mnemosyne tag, mnemosyne untag, mnemosyne retag,
|
||||
mnemosyne timeline, mnemosyne neighbors
|
||||
mnemosyne timeline, mnemosyne neighbors, mnemosyne path,
|
||||
mnemosyne touch, mnemosyne decay, mnemosyne vitality,
|
||||
mnemosyne fading, mnemosyne vibrant
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -208,7 +210,7 @@ def cmd_timeline(args):
|
||||
|
||||
|
||||
def cmd_path(args):
|
||||
archive = _load(args.archive)
|
||||
archive = MnemosyneArchive(archive_path=args.archive) if args.archive else MnemosyneArchive()
|
||||
path = archive.shortest_path(args.start, args.end)
|
||||
if path is None:
|
||||
print(f"No path found between {args.start} and {args.end}")
|
||||
@@ -254,6 +256,65 @@ def cmd_neighbors(args):
|
||||
print()
|
||||
|
||||
|
||||
def cmd_touch(args):
|
||||
archive = MnemosyneArchive()
|
||||
try:
|
||||
entry = archive.touch(args.entry_id)
|
||||
except KeyError:
|
||||
print(f"Entry not found: {args.entry_id}")
|
||||
sys.exit(1)
|
||||
v = archive.get_vitality(entry.id)
|
||||
print(f"[{entry.id[:8]}] {entry.title}")
|
||||
print(f" Vitality: {v['vitality']:.4f} (boosted)")
|
||||
|
||||
|
||||
def cmd_decay(args):
|
||||
archive = MnemosyneArchive()
|
||||
result = archive.apply_decay()
|
||||
print(f"Applied decay to {result['total_entries']} entries")
|
||||
print(f" Decayed: {result['decayed_count']}")
|
||||
print(f" Avg vitality: {result['avg_vitality']:.4f}")
|
||||
print(f" Fading (<0.3): {result['fading_count']}")
|
||||
print(f" Vibrant (>0.7): {result['vibrant_count']}")
|
||||
|
||||
|
||||
def cmd_vitality(args):
|
||||
archive = MnemosyneArchive()
|
||||
try:
|
||||
v = archive.get_vitality(args.entry_id)
|
||||
except KeyError:
|
||||
print(f"Entry not found: {args.entry_id}")
|
||||
sys.exit(1)
|
||||
print(f"[{v['entry_id'][:8]}] {v['title']}")
|
||||
print(f" Vitality: {v['vitality']:.4f}")
|
||||
print(f" Last accessed: {v['last_accessed'] or 'never'}")
|
||||
print(f" Age: {v['age_days']} days")
|
||||
|
||||
|
||||
def cmd_fading(args):
|
||||
archive = MnemosyneArchive()
|
||||
results = archive.fading(limit=args.limit)
|
||||
if not results:
|
||||
print("Archive is empty.")
|
||||
return
|
||||
for v in results:
|
||||
print(f"[{v['entry_id'][:8]}] {v['title']}")
|
||||
print(f" Vitality: {v['vitality']:.4f} | Age: {v['age_days']}d | Last: {v['last_accessed'] or 'never'}")
|
||||
print()
|
||||
|
||||
|
||||
def cmd_vibrant(args):
|
||||
archive = MnemosyneArchive()
|
||||
results = archive.vibrant(limit=args.limit)
|
||||
if not results:
|
||||
print("Archive is empty.")
|
||||
return
|
||||
for v in results:
|
||||
print(f"[{v['entry_id'][:8]}] {v['title']}")
|
||||
print(f" Vitality: {v['vitality']:.4f} | Age: {v['age_days']}d | Last: {v['last_accessed'] or 'never'}")
|
||||
print()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive")
|
||||
sub = parser.add_subparsers(dest="command")
|
||||
@@ -325,6 +386,21 @@ def main():
|
||||
co.add_argument("--dry-run", action="store_true", help="Show what would be merged without applying")
|
||||
co.add_argument("--threshold", type=float, default=0.9, help="Similarity threshold (default: 0.9)")
|
||||
|
||||
|
||||
tc = sub.add_parser("touch", help="Boost an entry's vitality by accessing it")
|
||||
tc.add_argument("entry_id", help="Entry ID to touch")
|
||||
|
||||
dc = sub.add_parser("decay", help="Apply time-based decay to all entries")
|
||||
|
||||
vy = sub.add_parser("vitality", help="Show an entry's vitality status")
|
||||
vy.add_argument("entry_id", help="Entry ID to check")
|
||||
|
||||
fg = sub.add_parser("fading", help="Show most neglected entries (lowest vitality)")
|
||||
fg.add_argument("-n", "--limit", type=int, default=10, help="Max entries to show")
|
||||
|
||||
vb = sub.add_parser("vibrant", help="Show most alive entries (highest vitality)")
|
||||
vb.add_argument("-n", "--limit", type=int, default=10, help="Max entries to show")
|
||||
|
||||
args = parser.parse_args()
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
@@ -348,6 +424,12 @@ def main():
|
||||
"timeline": cmd_timeline,
|
||||
"neighbors": cmd_neighbors,
|
||||
"consolidate": cmd_consolidate,
|
||||
"path": cmd_path,
|
||||
"touch": cmd_touch,
|
||||
"decay": cmd_decay,
|
||||
"vitality": cmd_vitality,
|
||||
"fading": cmd_fading,
|
||||
"vibrant": cmd_vibrant,
|
||||
}
|
||||
dispatch[args.command](args)
|
||||
|
||||
|
||||
138
nexus/mnemosyne/tests/test_cli_commands.py
Normal file
138
nexus/mnemosyne/tests/test_cli_commands.py
Normal file
@@ -0,0 +1,138 @@
|
||||
"""Tests for Mnemosyne CLI commands — path, touch, decay, vitality, fading, vibrant."""
|
||||
|
||||
import json
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
import sys
|
||||
import io
|
||||
|
||||
import pytest
|
||||
|
||||
from nexus.mnemosyne.archive import MnemosyneArchive
|
||||
from nexus.mnemosyne.entry import ArchiveEntry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def archive(tmp_path):
|
||||
path = tmp_path / "test_archive.json"
|
||||
return MnemosyneArchive(archive_path=path)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def linked_archive(tmp_path):
|
||||
"""Archive with entries linked to each other for path testing."""
|
||||
path = tmp_path / "test_archive.json"
|
||||
arch = MnemosyneArchive(archive_path=path, auto_embed=False)
|
||||
e1 = arch.add(ArchiveEntry(title="Alpha", content="first entry about python", topics=["code"]))
|
||||
e2 = arch.add(ArchiveEntry(title="Beta", content="second entry about python coding", topics=["code"]))
|
||||
e3 = arch.add(ArchiveEntry(title="Gamma", content="third entry about cooking recipes", topics=["food"]))
|
||||
return arch, e1, e2, e3
|
||||
|
||||
|
||||
class TestPathCommand:
|
||||
def test_shortest_path_exists(self, linked_archive):
|
||||
arch, e1, e2, e3 = linked_archive
|
||||
path = arch.shortest_path(e1.id, e2.id)
|
||||
assert path is not None
|
||||
assert path[0] == e1.id
|
||||
assert path[-1] == e2.id
|
||||
|
||||
def test_shortest_path_no_connection(self, linked_archive):
|
||||
arch, e1, e2, e3 = linked_archive
|
||||
# e3 (cooking) likely not linked to e1 (python coding)
|
||||
path = arch.shortest_path(e1.id, e3.id)
|
||||
# Path may or may not exist depending on linking threshold
|
||||
# Either None or a list is valid
|
||||
|
||||
def test_shortest_path_same_entry(self, linked_archive):
|
||||
arch, e1, _, _ = linked_archive
|
||||
path = arch.shortest_path(e1.id, e1.id)
|
||||
assert path == [e1.id]
|
||||
|
||||
def test_shortest_path_missing_entry(self, linked_archive):
|
||||
arch, e1, _, _ = linked_archive
|
||||
path = arch.shortest_path(e1.id, "nonexistent-id")
|
||||
assert path is None
|
||||
|
||||
|
||||
class TestTouchCommand:
|
||||
def test_touch_boosts_vitality(self, archive):
|
||||
entry = archive.add(ArchiveEntry(title="Test", content="Content"))
|
||||
# Simulate time passing by setting old last_accessed
|
||||
old_time = "2020-01-01T00:00:00+00:00"
|
||||
entry.last_accessed = old_time
|
||||
entry.vitality = 0.5
|
||||
archive._save()
|
||||
|
||||
touched = archive.touch(entry.id)
|
||||
assert touched.vitality > 0.5
|
||||
assert touched.last_accessed != old_time
|
||||
|
||||
def test_touch_missing_entry(self, archive):
|
||||
with pytest.raises(KeyError):
|
||||
archive.touch("nonexistent-id")
|
||||
|
||||
|
||||
class TestDecayCommand:
|
||||
def test_apply_decay_returns_stats(self, archive):
|
||||
archive.add(ArchiveEntry(title="Test", content="Content"))
|
||||
result = archive.apply_decay()
|
||||
assert result["total_entries"] == 1
|
||||
assert "avg_vitality" in result
|
||||
assert "fading_count" in result
|
||||
assert "vibrant_count" in result
|
||||
|
||||
def test_decay_on_empty_archive(self, archive):
|
||||
result = archive.apply_decay()
|
||||
assert result["total_entries"] == 0
|
||||
assert result["avg_vitality"] == 0.0
|
||||
|
||||
|
||||
class TestVitalityCommand:
|
||||
def test_get_vitality(self, archive):
|
||||
entry = archive.add(ArchiveEntry(title="Test", content="Content"))
|
||||
v = archive.get_vitality(entry.id)
|
||||
assert v["entry_id"] == entry.id
|
||||
assert v["title"] == "Test"
|
||||
assert 0.0 <= v["vitality"] <= 1.0
|
||||
assert v["age_days"] >= 0
|
||||
|
||||
def test_get_vitality_missing(self, archive):
|
||||
with pytest.raises(KeyError):
|
||||
archive.get_vitality("nonexistent-id")
|
||||
|
||||
|
||||
class TestFadingVibrant:
|
||||
def test_fading_returns_sorted_ascending(self, archive):
|
||||
# Add entries with different vitalities
|
||||
e1 = archive.add(ArchiveEntry(title="Vibrant", content="High energy"))
|
||||
e2 = archive.add(ArchiveEntry(title="Fading", content="Low energy"))
|
||||
e2.vitality = 0.1
|
||||
e2.last_accessed = "2020-01-01T00:00:00+00:00"
|
||||
archive._save()
|
||||
|
||||
results = archive.fading(limit=10)
|
||||
assert len(results) == 2
|
||||
assert results[0]["vitality"] <= results[1]["vitality"]
|
||||
|
||||
def test_vibrant_returns_sorted_descending(self, archive):
|
||||
e1 = archive.add(ArchiveEntry(title="Fresh", content="New"))
|
||||
e2 = archive.add(ArchiveEntry(title="Old", content="Ancient"))
|
||||
e2.vitality = 0.1
|
||||
e2.last_accessed = "2020-01-01T00:00:00+00:00"
|
||||
archive._save()
|
||||
|
||||
results = archive.vibrant(limit=10)
|
||||
assert len(results) == 2
|
||||
assert results[0]["vitality"] >= results[1]["vitality"]
|
||||
|
||||
def test_fading_limit(self, archive):
|
||||
for i in range(15):
|
||||
archive.add(ArchiveEntry(title=f"Entry {i}", content=f"Content {i}"))
|
||||
results = archive.fading(limit=5)
|
||||
assert len(results) == 5
|
||||
|
||||
def test_vibrant_empty(self, archive):
|
||||
results = archive.vibrant()
|
||||
assert results == []
|
||||
Reference in New Issue
Block a user