Compare commits

...

3 Commits

Author SHA1 Message Date
Alexander Whitestone
9d11f218ee fix: date-only parsing for end_of_day boundary (all 97 tests pass)
Some checks failed
CI / test (pull_request) Failing after 9s
CI / validate (pull_request) Failing after 14s
Review Approval Gate / verify-review (pull_request) Failing after 3s
2026-04-11 20:49:20 -04:00
Alexander Whitestone
1437613560 feat(mnemosyne): add CLI commands and tests for temporal queries
Closes #1244

- CLI: 'timeline <start> <end>' for date range queries
- CLI: 'neighbors <entry_id> [--days N]' for temporal proximity
- Tests: 16 test cases covering parsing, ranges, boundaries,
  sorting, limits, neighbor ordering, and edge cases
2026-04-11 20:48:14 -04:00
Alexander Whitestone
81b4616c03 feat(mnemosyne): add by_date_range and temporal_neighbors query methods
Closes #1244

- by_date_range(start, end, limit): entries within a date range
- temporal_neighbors(entry_id, window_days): entries near a reference in time
- _parse_datetime: handles ISO datetime and date-only strings
- Both methods sort results by temporal relevance
2026-04-11 20:46:55 -04:00
13 changed files with 324 additions and 2 deletions

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

Binary file not shown.

View File

@@ -685,3 +685,116 @@ class MnemosyneArchive:
self._save()
return total_links
def by_date_range(
self,
start: str,
end: str,
limit: Optional[int] = None,
) -> list[ArchiveEntry]:
"""Return entries whose created_at falls within [start, end].
Args:
start: ISO datetime string (inclusive). Can be a date-only string
like "2026-03-01" (treated as start of that day UTC).
end: ISO datetime string (inclusive). Can be a date-only string
like "2026-04-01" (treated as end of that day UTC).
limit: Maximum entries to return (None = all).
Returns:
List of ArchiveEntry sorted by created_at ascending.
"""
start_dt = self._parse_datetime(start, end_of_day=False)
end_dt = self._parse_datetime(end, end_of_day=True)
matched = []
for entry in self._entries.values():
entry_dt = self._parse_datetime(entry.created_at, end_of_day=False)
if entry_dt is None:
continue
if start_dt <= entry_dt <= end_dt:
matched.append(entry)
matched.sort(key=lambda e: e.created_at)
if limit is not None:
matched = matched[:limit]
return matched
def temporal_neighbors(
self,
entry_id: str,
window_days: int = 7,
) -> list[ArchiveEntry]:
"""Return entries created within N days of a given entry.
Args:
entry_id: The reference entry's ID.
window_days: Number of days before and after the reference entry's
created_at to search (default 7).
Returns:
List of ArchiveEntry sorted by time distance from the reference,
closest first. Excludes the reference entry itself.
Raises:
KeyError: If entry_id does not exist.
"""
from datetime import timedelta
ref = self._entries.get(entry_id)
if ref is None:
raise KeyError(entry_id)
ref_dt = self._parse_datetime(ref.created_at, end_of_day=False)
if ref_dt is None:
return []
window = timedelta(days=window_days)
start_dt = ref_dt - window
end_dt = ref_dt + window
neighbors = []
for entry in self._entries.values():
if entry.id == entry_id:
continue
entry_dt = self._parse_datetime(entry.created_at, end_of_day=False)
if entry_dt is None:
continue
if start_dt <= entry_dt <= end_dt:
distance = abs((entry_dt - ref_dt).total_seconds())
neighbors.append((distance, entry))
neighbors.sort(key=lambda x: x[0])
return [e for _, e in neighbors]
@staticmethod
def _parse_datetime(value: str, end_of_day: bool = False) -> Optional[datetime]:
"""Parse an ISO datetime or date-only string to a timezone-aware datetime.
Args:
value: ISO datetime string, or date-only string (YYYY-MM-DD).
end_of_day: If True and value is date-only, set time to 23:59:59 UTC.
Returns:
Timezone-aware datetime (UTC), or None if parsing fails.
"""
if not value:
return None
# Date-only strings (no T separator) get special end_of_day handling
if "T" not in value and " " not in value:
try:
from datetime import date
d = date.fromisoformat(value)
if end_of_day:
return datetime(d.year, d.month, d.day, 23, 59, 59, tzinfo=timezone.utc)
return datetime(d.year, d.month, d.day, tzinfo=timezone.utc)
except ValueError:
return None
# Full datetime
try:
dt = datetime.fromisoformat(value)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
except ValueError:
return None

View File

@@ -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,8 +181,45 @@ def cmd_retag(args):
print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}")
def cmd_timeline(args):
archive = MnemosyneArchive()
results = archive.by_date_range(args.start, args.end, limit=args.limit)
if not results:
print(f"No entries between {args.start} and {args.end}.")
return
print(f"Timeline: {args.start}{args.end} ({len(results)} entries)")
print()
for entry in results:
print(f" [{entry.created_at[:10]}] {entry.title}")
print(f" ID: {entry.id[:8]} | Source: {entry.source} | Topics: {', '.join(entry.topics)}")
print()
def cmd_neighbors(args):
archive = MnemosyneArchive()
# Resolve prefix to full ID
matches = [e for e in archive._entries.values() if e.id.startswith(args.entry_id)]
if not matches:
print(f"No entry matching '{args.entry_id}'.")
return
if len(matches) > 1:
print(f"Ambiguous — {len(matches)} entries match '{args.entry_id}'. Use a longer prefix.")
return
entry = matches[0]
results = archive.temporal_neighbors(entry.id, window_days=args.days)
if not results:
print(f"No entries within {args.days} days of [{entry.id[:8]}] {entry.title}.")
return
print(f"Neighbors of [{entry.id[:8]}] {entry.title}{args.days} days):")
print()
for neighbor in results:
print(f" [{neighbor.created_at[:10]}] {neighbor.title}")
print(f" ID: {neighbor.id[:8]} | Source: {neighbor.source} | Topics: {', '.join(neighbor.topics)}")
print()
def main():
parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive")
parser = argparse.ArgumentParser(prog="mnemosyne", description="Mnemosyne — Living Holographic Archive")
sub = parser.add_subparsers(dest="command")
sub.add_parser("stats", help="Show archive statistics")
@@ -233,6 +271,15 @@ 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 a date range")
tl.add_argument("start", help="Start date (YYYY-MM-DD or ISO datetime)")
tl.add_argument("end", help="End date (YYYY-MM-DD or ISO datetime)")
tl.add_argument("-n", "--limit", type=int, default=50, help="Max entries to show")
nb = sub.add_parser("neighbors", help="Show entries temporally near a reference entry")
nb.add_argument("entry_id", help="Reference entry ID (or prefix)")
nb.add_argument("-d", "--days", type=int, default=7, help="Window in days (default 7)")
args = parser.parse_args()
if not args.command:
parser.print_help()
@@ -253,6 +300,8 @@ def main():
"tag": cmd_tag,
"untag": cmd_untag,
"retag": cmd_retag,
"timeline": cmd_timeline,
"neighbors": cmd_neighbors,
}
dispatch[args.command](args)

View File

@@ -0,0 +1,160 @@
"""Tests for Mnemosyne temporal query methods.
Tests: by_date_range, temporal_neighbors, _parse_datetime.
"""
import pytest
from pathlib import Path
import tempfile
from nexus.mnemosyne.archive import MnemosyneArchive
from nexus.mnemosyne.entry import ArchiveEntry
@pytest.fixture
def archive():
"""Create a fresh archive in a temp directory."""
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "test_archive.json"
a = MnemosyneArchive(archive_path=path)
yield a
class TestParseDatetime:
"""Test _parse_datetime static method."""
def test_iso_datetime(self):
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00+00:00")
assert dt is not None
assert dt.year == 2026
assert dt.month == 3
assert dt.hour == 10
def test_date_only_start(self):
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=False)
assert dt is not None
assert dt.hour == 0
assert dt.minute == 0
def test_date_only_end(self):
dt = MnemosyneArchive._parse_datetime("2026-03-15", end_of_day=True)
assert dt is not None
assert dt.hour == 23
assert dt.minute == 59
assert dt.second == 59
def test_naive_datetime_gets_utc(self):
dt = MnemosyneArchive._parse_datetime("2026-03-15T10:30:00")
assert dt is not None
assert dt.tzinfo is not None
def test_empty_string(self):
assert MnemosyneArchive._parse_datetime("") is None
def test_invalid_string(self):
assert MnemosyneArchive._parse_datetime("not-a-date") is None
class TestByDateRange:
"""Test by_date_range() method."""
def _add_dated_entry(self, archive, title, created_at):
"""Add an entry with a specific created_at timestamp."""
entry = ArchiveEntry(title=title, content=f"Content for {title}")
entry.created_at = created_at
archive._entries[entry.id] = entry
archive._save()
return entry
def test_empty_archive(self, archive):
results = archive.by_date_range("2026-01-01", "2026-12-31")
assert results == []
def test_single_entry_in_range(self, archive):
self._add_dated_entry(archive, "March entry", "2026-03-15T10:00:00+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31")
assert len(results) == 1
assert results[0].title == "March entry"
def test_single_entry_out_of_range(self, archive):
self._add_dated_entry(archive, "January entry", "2026-01-10T10:00:00+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31")
assert results == []
def test_boundary_inclusive(self, archive):
self._add_dated_entry(archive, "Start", "2026-03-01T00:00:00+00:00")
self._add_dated_entry(archive, "End", "2026-03-31T23:59:59+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31")
assert len(results) == 2
def test_sorted_by_date(self, archive):
self._add_dated_entry(archive, "Late", "2026-03-20T10:00:00+00:00")
self._add_dated_entry(archive, "Early", "2026-03-05T10:00:00+00:00")
self._add_dated_entry(archive, "Mid", "2026-03-12T10:00:00+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31")
assert [r.title for r in results] == ["Early", "Mid", "Late"]
def test_limit(self, archive):
for i in range(10):
self._add_dated_entry(archive, f"Entry {i}", f"2026-03-{i+1:02d}T10:00:00+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31", limit=3)
assert len(results) == 3
def test_date_only_strings(self, archive):
self._add_dated_entry(archive, "Mid March", "2026-03-15T10:00:00+00:00")
results = archive.by_date_range("2026-03-01", "2026-03-31")
assert len(results) == 1
class TestTemporalNeighbors:
"""Test temporal_neighbors() method."""
def _add_dated_entry(self, archive, title, created_at):
entry = ArchiveEntry(title=title, content=f"Content for {title}")
entry.created_at = created_at
archive._entries[entry.id] = entry
archive._save()
return entry
def test_no_neighbors(self, archive):
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
self._add_dated_entry(archive, "Far away", "2026-06-01T10:00:00+00:00")
results = archive.temporal_neighbors(ref.id, window_days=7)
assert results == []
def test_closest_first(self, archive):
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
far = self._add_dated_entry(archive, "6 days away", "2026-03-21T10:00:00+00:00")
close = self._add_dated_entry(archive, "1 day away", "2026-03-16T10:00:00+00:00")
results = archive.temporal_neighbors(ref.id, window_days=7)
assert len(results) == 2
assert results[0].title == "1 day away"
assert results[1].title == "6 days away"
def test_excludes_reference(self, archive):
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
self._add_dated_entry(archive, "Nearby", "2026-03-16T10:00:00+00:00")
results = archive.temporal_neighbors(ref.id, window_days=7)
ids = [r.id for r in results]
assert ref.id not in ids
def test_default_window(self, archive):
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
self._add_dated_entry(archive, "7 days before", "2026-03-08T10:00:00+00:00")
self._add_dated_entry(archive, "8 days before", "2026-03-07T10:00:00+00:00")
results = archive.temporal_neighbors(ref.id) # default 7 days
titles = [r.title for r in results]
assert "7 days before" in titles
assert "8 days before" not in titles
def test_custom_window(self, archive):
ref = self._add_dated_entry(archive, "Reference", "2026-03-15T10:00:00+00:00")
self._add_dated_entry(archive, "14 days away", "2026-03-29T10:00:00+00:00")
results = archive.temporal_neighbors(ref.id, window_days=7)
assert len(results) == 0
results = archive.temporal_neighbors(ref.id, window_days=15)
assert len(results) == 1
def test_bad_entry_id(self, archive):
with pytest.raises(KeyError):
archive.temporal_neighbors("nonexistent-id")