Compare commits

...

4 Commits

Author SHA1 Message Date
b08df4f79d docs: add discover to CLI docstring
Some checks failed
CI / test (pull_request) Failing after 7s
CI / validate (pull_request) Failing after 11s
Review Approval Gate / verify-review (pull_request) Failing after 3s
2026-04-12 10:29:04 +00:00
7b3f8f09df test: add tests for discover() functionality
8 tests covering:
- Basic discovery returns entries
- Count limits
- Topic filtering (case-insensitive)
- Empty archive handling
- Vitality boost via touch
2026-04-12 10:28:39 +00:00
af297927a2 feat: add CLI commands for discover
- mnemosyne discover [-n COUNT] [-t TOPIC] [--vibrant]
- Serendipitous entry exploration weighted by vitality decay
2026-04-12 10:28:05 +00:00
46eeca9ad1 feat: add discover() for serendipitous entry exploration
Probabilistic entry selection weighted by vitality decay.
Fading entries surface more often, vibrant ones less.
Touches selected entries to boost their vitality.
Optional topic filter.
2026-04-12 10:27:33 +00:00
3 changed files with 195 additions and 1 deletions

View File

@@ -1308,3 +1308,88 @@ class MnemosyneArchive:
self._save()
return total_links
# ─── Discovery ──────────────────────────────────────────────
def discover(
self,
count: int = 5,
prefer_fading: bool = True,
topic: Optional[str] = None,
) -> list[dict]:
"""Serendipitous entry discovery — surface forgotten knowledge.
Selects entries probabilistically, weighting toward fading (low vitality)
entries when prefer_fading=True, or toward vibrant entries when False.
Optionally filter by topic.
Touches selected entries to boost their vitality, preventing the same
entries from being repeatedly surfaced.
Args:
count: Number of entries to discover.
prefer_fading: If True, weight toward neglected entries. If False,
weight toward vibrant entries.
topic: Optional topic filter — only discover entries with this tag.
Returns:
List of dicts with keys: entry_id, title, content_preview, topics,
vitality, age_days, last_accessed
"""
import random
candidates = list(self._entries.values())
# Filter by topic if specified
if topic:
topic_lower = topic.lower()
candidates = [
e for e in candidates
if topic_lower in [t.lower() for t in e.topics]
]
if not candidates:
return []
# Compute vitality for each candidate
scored = []
for entry in candidates:
v = self._compute_vitality(entry)
scored.append((entry, v))
# Build selection weights
if prefer_fading:
# Lower vitality = higher weight. Invert and normalize.
weights = [max(0.01, 1.0 - v) for _, v in scored]
else:
# Higher vitality = higher weight
weights = [max(0.01, v) for _, v in scored]
# Sample without replacement
k = min(count, len(scored))
selected_indices = random.choices(range(len(scored)), weights=weights, k=k)
# Deduplicate while preserving order
seen = set()
unique_indices = []
for idx in selected_indices:
if idx not in seen:
seen.add(idx)
unique_indices.append(idx)
results = []
for idx in unique_indices:
entry, v = scored[idx]
# Touch to boost vitality
self.touch(entry.id)
created = self._parse_dt(entry.created_at)
age_days = (datetime.now(timezone.utc) - created).days
results.append({
"entry_id": entry.id,
"title": entry.title,
"content_preview": entry.content[:200] + "..." if len(entry.content) > 200 else entry.content,
"topics": entry.topics,
"vitality": round(v, 4),
"age_days": age_days,
"last_accessed": entry.last_accessed,
})
return results

View File

@@ -8,6 +8,7 @@ Provides: mnemosyne ingest, mnemosyne search, mnemosyne link, mnemosyne stats,
mnemosyne touch, mnemosyne decay, mnemosyne vitality,
mnemosyne fading, mnemosyne vibrant,
mnemosyne snapshot create|list|restore|diff
mnemosyne discover [-n COUNT] [-t TOPIC] [--vibrant]
"""
from __future__ import annotations
@@ -366,6 +367,23 @@ def cmd_snapshot(args):
sys.exit(1)
def cmd_discover(args):
archive = MnemosyneArchive()
results = archive.discover(
count=args.count,
prefer_fading=not args.vibrant,
topic=args.topic if args.topic else None,
)
if not results:
print("No entries found." + (" (topic filter too narrow?)" if args.topic else ""))
return
for r in results:
print(f"[{r['entry_id'][:8]}] {r['title']}")
print(f" Topics: {', '.join(r['topics'])} | Vitality: {r['vitality']} | Age: {r['age_days']}d")
print(f" {r['content_preview']}")
print()
def cmd_vibrant(args):
archive = MnemosyneArchive()
results = archive.vibrant(limit=args.limit)
@@ -464,7 +482,12 @@ def main():
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")
sn = sub.add_parser("snapshot", help="Point-in-time backup and restore")
dc = sub.add_parser("discover", help="Serendipitous entry discovery")
dc.add_argument("-n", "--count", type=int, default=5, help="Number of entries to discover")
dc.add_argument("-t", "--topic", default=None, help="Filter by topic")
dc.add_argument("--vibrant", action="store_true", help="Prefer vibrant (alive) entries over fading ones")
sn = sub.add_parser("snapshot", help="Point-in-time backup and restore")
sn_sub = sn.add_subparsers(dest="snapshot_cmd")
sn_create = sn_sub.add_parser("create", help="Create a new snapshot")
sn_create.add_argument("--label", default="", help="Human-readable label for the snapshot")
@@ -507,6 +530,7 @@ def main():
"fading": cmd_fading,
"vibrant": cmd_vibrant,
"snapshot": cmd_snapshot,
"discover": cmd_discover,
}
dispatch[args.command](args)

View File

@@ -0,0 +1,85 @@
"""Tests for Mnemosyne discover functionality."""
import tempfile
from pathlib import Path
from nexus.mnemosyne.archive import MnemosyneArchive
from nexus.mnemosyne.ingest import ingest_event
def _make_archive_with_entries():
"""Helper: create an archive with test entries."""
path = Path(tempfile.mkdtemp()) / "test_archive.json"
archive = MnemosyneArchive(archive_path=path, auto_embed=False)
ingest_event(archive, title="Python automation", content="Building tools in Python", topics=["python", "automation"])
ingest_event(archive, title="Cooking pasta", content="How to make carbonara", topics=["cooking"])
ingest_event(archive, title="Bitcoin basics", content="Understanding Bitcoin and blockchain", topics=["bitcoin", "crypto"])
ingest_event(archive, title="AI agents", content="Building autonomous AI agents", topics=["ai", "agents"])
ingest_event(archive, title="Meditation guide", content="Mindfulness and meditation techniques", topics=["wellness"])
return archive
def test_discover_returns_entries():
archive = _make_archive_with_entries()
results = archive.discover(count=3)
assert len(results) == 3
for r in results:
assert "entry_id" in r
assert "title" in r
assert "content_preview" in r
assert "topics" in r
assert "vitality" in r
assert "age_days" in r
def test_discover_respects_count():
archive = _make_archive_with_entries()
results = archive.discover(count=2)
assert len(results) == 2
def test_discover_count_exceeds_entries():
archive = _make_archive_with_entries()
results = archive.discover(count=100)
assert len(results) == archive.count
def test_discover_topic_filter():
archive = _make_archive_with_entries()
results = archive.discover(count=10, topic="python")
assert len(results) == 1
assert results[0]["title"] == "Python automation"
def test_discover_topic_case_insensitive():
archive = _make_archive_with_entries()
results = archive.discover(count=10, topic="Python")
assert len(results) == 1
def test_discover_empty_topic_returns_nothing():
archive = _make_archive_with_entries()
results = archive.discover(count=10, topic="nonexistent")
assert len(results) == 0
def test_discover_boosts_vitality():
archive = _make_archive_with_entries()
# Get initial vitality
before = archive.fading(limit=5)
# Discover (which touches entries)
archive.discover(count=3)
# The touched entries should have higher vitality now
after = archive.fading(limit=5)
# At least some entries should have changed vitality
before_vitals = {e["entry_id"]: e["vitality"] for e in before}
after_vitals = {e["entry_id"]: e["vitality"] for e in after}
changed = sum(1 for eid in before_vitals if eid in after_vitals and abs(before_vitals[eid] - after_vitals[eid]) > 0.001)
assert changed >= 1, "Discover should touch and boost vitality of selected entries"
def test_discover_empty_archive():
path = Path(tempfile.mkdtemp()) / "empty.json"
archive = MnemosyneArchive(archive_path=path, auto_embed=False)
results = archive.discover(count=5)
assert len(results) == 0