[claude] Add temporal query methods: by_date_range and temporal_neighbors (#1244) (#1246)

nexus/mnemosyne/archive.py

@@ -7,7 +7,7 @@ and provides query interfaces for retrieving connected knowledge.
 from __future__ import annotations
 
 import json
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Optional
 
@@ -651,6 +651,68 @@ class MnemosyneArchive:
         self._save()
         return entry
 
+    @staticmethod
+    def _parse_dt(dt_str: str) -> datetime:
+        """Parse an ISO datetime string. Assumes UTC if no timezone is specified."""
+        dt = datetime.fromisoformat(dt_str)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt
+
+    def by_date_range(self, start: str, end: str) -> list[ArchiveEntry]:
+        """Return entries whose ``created_at`` falls within [start, end] (inclusive).
+
+        Args:
+            start: ISO datetime string for the range start (e.g. "2024-01-01" or
+                   "2024-01-01T00:00:00Z"). Timezone-naive strings are treated as UTC.
+            end: ISO datetime string for the range end. Timezone-naive strings are
+                 treated as UTC.
+
+        Returns:
+            List of ArchiveEntry sorted by ``created_at`` ascending.
+        """
+        start_dt = self._parse_dt(start)
+        end_dt = self._parse_dt(end)
+        results = []
+        for entry in self._entries.values():
+            entry_dt = self._parse_dt(entry.created_at)
+            if start_dt <= entry_dt <= end_dt:
+                results.append(entry)
+        results.sort(key=lambda e: e.created_at)
+        return results
+
+    def temporal_neighbors(self, entry_id: str, window_days: int = 7) -> list[ArchiveEntry]:
+        """Return entries created within ``window_days`` of a given entry.
+
+        The reference entry itself is excluded from results.
+
+        Args:
+            entry_id: ID of the anchor entry.
+            window_days: Number of days around the anchor's ``created_at`` to search.
+
+        Returns:
+            List of ArchiveEntry sorted by ``created_at`` ascending.
+
+        Raises:
+            KeyError: If ``entry_id`` does not exist in the archive.
+        """
+        anchor = self._entries.get(entry_id)
+        if anchor is None:
+            raise KeyError(entry_id)
+        anchor_dt = self._parse_dt(anchor.created_at)
+        delta = timedelta(days=window_days)
+        window_start = anchor_dt - delta
+        window_end = anchor_dt + delta
+        results = []
+        for entry in self._entries.values():
+            if entry.id == entry_id:
+                continue
+            entry_dt = self._parse_dt(entry.created_at)
+            if window_start <= entry_dt <= window_end:
+                results.append(entry)
+        results.sort(key=lambda e: e.created_at)
+        return results
+
     def rebuild_links(self, threshold: Optional[float] = None) -> int:
         """Recompute all links from scratch.
 

nexus/mnemosyne/cli.py

@@ -3,7 +3,8 @@
 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 tag, mnemosyne untag, mnemosyne retag,
+          mnemosyne timeline, mnemosyne neighbors
 """
 
 from __future__ import annotations
@@ -180,6 +181,38 @@ def cmd_retag(args):
     print(f"  Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
 
 
+def cmd_timeline(args):
+    archive = MnemosyneArchive()
+    try:
+        results = archive.by_date_range(args.start, args.end)
+    except ValueError as e:
+        print(f"Invalid date format: {e}")
+        sys.exit(1)
+    if not results:
+        print("No entries found in that date range.")
+        return
+    for entry in results:
+        print(f"[{entry.id[:8]}] {entry.created_at[:10]}  {entry.title}")
+        print(f"  Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
+        print()
+
+
+def cmd_neighbors(args):
+    archive = MnemosyneArchive()
+    try:
+        results = archive.temporal_neighbors(args.entry_id, window_days=args.days)
+    except KeyError:
+        print(f"Entry not found: {args.entry_id}")
+        sys.exit(1)
+    if not results:
+        print("No temporal neighbors found.")
+        return
+    for entry in results:
+        print(f"[{entry.id[:8]}] {entry.created_at[:10]}  {entry.title}")
+        print(f"  Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
+        print()
+
+
 def main():
     parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive")
     sub = parser.add_subparsers(dest="command")
@@ -233,6 +266,14 @@ def main():
     rt.add_argument("entry_id", help="Entry ID")
     rt.add_argument("tags", help="Comma-separated new tag list")
 
+    tl = sub.add_parser("timeline", help="Show entries within an ISO date range")
+    tl.add_argument("start", help="Start datetime (ISO format, e.g. 2024-01-01 or 2024-01-01T00:00:00Z)")
+    tl.add_argument("end", help="End datetime (ISO format)")
+
+    nb = sub.add_parser("neighbors", help="Show entries temporally near a given entry")
+    nb.add_argument("entry_id", help="Anchor entry ID")
+    nb.add_argument("--days", type=int, default=7, help="Window in days (default: 7)")
+
     args = parser.parse_args()
     if not args.command:
         parser.print_help()
@@ -253,6 +294,8 @@ def main():
         "tag": cmd_tag,
         "untag": cmd_untag,
         "retag": cmd_retag,
+        "timeline": cmd_timeline,
+        "neighbors": cmd_neighbors,
     }
     dispatch[args.command](args)
 

nexus/mnemosyne/tests/test_archive.py

@@ -2,6 +2,7 @@
 
 import json
 import tempfile
+from datetime import datetime, timezone, timedelta
 from pathlib import Path
 
 from nexus.mnemosyne.entry import ArchiveEntry
@@ -666,3 +667,189 @@ def test_update_entry_no_change_no_crash():
         e = ingest_event(archive, title="T", content="c")
         result = archive.update_entry(e.id)
         assert result.title == "T"
+
+
+# --- by_date_range tests ---
+
+def _make_entry_at(archive: MnemosyneArchive, title: str, dt: datetime) -> ArchiveEntry:
+    """Helper: ingest an entry and backdate its created_at."""
+    e = ingest_event(archive, title=title, content=title)
+    e.created_at = dt.isoformat()
+    archive._save()
+    return e
+
+
+def test_by_date_range_empty_archive():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        results = archive.by_date_range("2024-01-01", "2024-12-31")
+        assert results == []
+
+
+def test_by_date_range_returns_matching_entries():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        jan = datetime(2024, 1, 15, tzinfo=timezone.utc)
+        mar = datetime(2024, 3, 10, tzinfo=timezone.utc)
+        jun = datetime(2024, 6, 1, tzinfo=timezone.utc)
+        e1 = _make_entry_at(archive, "Jan entry", jan)
+        e2 = _make_entry_at(archive, "Mar entry", mar)
+        e3 = _make_entry_at(archive, "Jun entry", jun)
+
+        results = archive.by_date_range("2024-01-01", "2024-04-01")
+        ids = {e.id for e in results}
+        assert e1.id in ids
+        assert e2.id in ids
+        assert e3.id not in ids
+
+
+def test_by_date_range_boundary_inclusive():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        exact = datetime(2024, 3, 1, tzinfo=timezone.utc)
+        e = _make_entry_at(archive, "Exact boundary", exact)
+
+        results = archive.by_date_range("2024-03-01T00:00:00+00:00", "2024-03-01T00:00:00+00:00")
+        assert len(results) == 1
+        assert results[0].id == e.id
+
+
+def test_by_date_range_no_results():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        jan = datetime(2024, 1, 15, tzinfo=timezone.utc)
+        _make_entry_at(archive, "Jan entry", jan)
+
+        results = archive.by_date_range("2023-01-01", "2023-12-31")
+        assert results == []
+
+
+def test_by_date_range_timezone_naive_treated_as_utc():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        dt = datetime(2024, 6, 15, tzinfo=timezone.utc)
+        e = _make_entry_at(archive, "Summer", dt)
+
+        # Timezone-naive start/end should still match
+        results = archive.by_date_range("2024-06-01", "2024-07-01")
+        assert any(r.id == e.id for r in results)
+
+
+def test_by_date_range_sorted_ascending():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        dates = [
+            datetime(2024, 3, 5, tzinfo=timezone.utc),
+            datetime(2024, 1, 10, tzinfo=timezone.utc),
+            datetime(2024, 2, 20, tzinfo=timezone.utc),
+        ]
+        for i, dt in enumerate(dates):
+            _make_entry_at(archive, f"Entry {i}", dt)
+
+        results = archive.by_date_range("2024-01-01", "2024-12-31")
+        assert len(results) == 3
+        assert results[0].created_at < results[1].created_at < results[2].created_at
+
+
+def test_by_date_range_single_entry_archive():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        dt = datetime(2024, 5, 1, tzinfo=timezone.utc)
+        e = _make_entry_at(archive, "Only", dt)
+
+        assert archive.by_date_range("2024-01-01", "2024-12-31") == [e]
+        assert archive.by_date_range("2025-01-01", "2025-12-31") == []
+
+
+# --- temporal_neighbors tests ---
+
+def test_temporal_neighbors_empty_archive():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        e = ingest_event(archive, title="Lone", content="c")
+        results = archive.temporal_neighbors(e.id, window_days=7)
+        assert results == []
+
+
+def test_temporal_neighbors_missing_entry_raises():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        try:
+            archive.temporal_neighbors("nonexistent-id")
+            assert False, "Expected KeyError"
+        except KeyError:
+            pass
+
+
+def test_temporal_neighbors_returns_within_window():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        anchor_dt = datetime(2024, 4, 10, tzinfo=timezone.utc)
+        near_dt = datetime(2024, 4, 14, tzinfo=timezone.utc)   # +4 days — within 7
+        far_dt = datetime(2024, 4, 20, tzinfo=timezone.utc)    # +10 days — outside 7
+
+        anchor = _make_entry_at(archive, "Anchor", anchor_dt)
+        near = _make_entry_at(archive, "Near", near_dt)
+        far = _make_entry_at(archive, "Far", far_dt)
+
+        results = archive.temporal_neighbors(anchor.id, window_days=7)
+        ids = {e.id for e in results}
+        assert near.id in ids
+        assert far.id not in ids
+        assert anchor.id not in ids
+
+
+def test_temporal_neighbors_excludes_anchor():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        dt = datetime(2024, 4, 10, tzinfo=timezone.utc)
+        anchor = _make_entry_at(archive, "Anchor", dt)
+        same = _make_entry_at(archive, "Same day", dt)
+
+        results = archive.temporal_neighbors(anchor.id, window_days=0)
+        ids = {e.id for e in results}
+        assert anchor.id not in ids
+        assert same.id in ids
+
+
+def test_temporal_neighbors_custom_window():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        anchor_dt = datetime(2024, 4, 10, tzinfo=timezone.utc)
+        within_3 = datetime(2024, 4, 12, tzinfo=timezone.utc)   # +2 days
+        outside_3 = datetime(2024, 4, 15, tzinfo=timezone.utc)  # +5 days
+
+        anchor = _make_entry_at(archive, "Anchor", anchor_dt)
+        e_near = _make_entry_at(archive, "Near", within_3)
+        e_far = _make_entry_at(archive, "Far", outside_3)
+
+        results = archive.temporal_neighbors(anchor.id, window_days=3)
+        ids = {e.id for e in results}
+        assert e_near.id in ids
+        assert e_far.id not in ids
+
+
+def test_temporal_neighbors_sorted_ascending():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        anchor_dt = datetime(2024, 6, 15, tzinfo=timezone.utc)
+        anchor = _make_entry_at(archive, "Anchor", anchor_dt)
+        for offset in [5, 1, 3]:
+            _make_entry_at(archive, f"Offset {offset}", anchor_dt + timedelta(days=offset))
+
+        results = archive.temporal_neighbors(anchor.id, window_days=7)
+        assert len(results) == 3
+        assert results[0].created_at < results[1].created_at < results[2].created_at
+
+
+def test_temporal_neighbors_boundary_inclusive():
+    with tempfile.TemporaryDirectory() as tmp:
+        archive = MnemosyneArchive(archive_path=Path(tmp) / "a.json")
+        anchor_dt = datetime(2024, 6, 15, tzinfo=timezone.utc)
+        boundary_dt = anchor_dt + timedelta(days=7)  # exactly at window edge
+
+        anchor = _make_entry_at(archive, "Anchor", anchor_dt)
+        boundary = _make_entry_at(archive, "Boundary", boundary_dt)
+
+        results = archive.temporal_neighbors(anchor.id, window_days=7)
+        assert any(r.id == boundary.id for r in results)