diff --git a/nexus/mnemosyne/archive.py b/nexus/mnemosyne/archive.py index a84a676..6a13b20 100644 --- a/nexus/mnemosyne/archive.py +++ b/nexus/mnemosyne/archive.py @@ -510,6 +510,77 @@ class MnemosyneArchive: bridges.sort(key=lambda b: b["components_after_removal"], reverse=True) return bridges + def add_tags(self, entry_id: str, tags: list[str]) -> ArchiveEntry: + """Add new tags to an existing entry (deduplicates, case-preserving). + + Args: + entry_id: ID of the entry to update. + tags: Tags to add. Already-present tags (case-insensitive) are skipped. + + Returns: + The updated ArchiveEntry. + + Raises: + KeyError: If entry_id does not exist. + """ + entry = self._entries.get(entry_id) + if entry is None: + raise KeyError(entry_id) + existing_lower = {t.lower() for t in entry.topics} + for tag in tags: + if tag.lower() not in existing_lower: + entry.topics.append(tag) + existing_lower.add(tag.lower()) + self._save() + return entry + + def remove_tags(self, entry_id: str, tags: list[str]) -> ArchiveEntry: + """Remove specific tags from an existing entry (case-insensitive match). + + Args: + entry_id: ID of the entry to update. + tags: Tags to remove. Tags not present are silently ignored. + + Returns: + The updated ArchiveEntry. + + Raises: + KeyError: If entry_id does not exist. + """ + entry = self._entries.get(entry_id) + if entry is None: + raise KeyError(entry_id) + remove_lower = {t.lower() for t in tags} + entry.topics = [t for t in entry.topics if t.lower() not in remove_lower] + self._save() + return entry + + def retag(self, entry_id: str, tags: list[str]) -> ArchiveEntry: + """Replace all tags on an existing entry (deduplicates new list). + + Args: + entry_id: ID of the entry to update. + tags: New tag list. Duplicates (case-insensitive) are collapsed. + + Returns: + The updated ArchiveEntry. + + Raises: + KeyError: If entry_id does not exist. + """ + entry = self._entries.get(entry_id) + if entry is None: + raise KeyError(entry_id) + seen: set[str] = set() + deduped: list[str] = [] + for tag in tags: + if tag.lower() not in seen: + seen.add(tag.lower()) + deduped.append(tag) + entry.topics = deduped + self._save() + return entry + def rebuild_links(self, threshold: Optional[float] = None) -> int: """Recompute all links from scratch. diff --git a/nexus/mnemosyne/cli.py b/nexus/mnemosyne/cli.py index 6bc692b..d712c9d 100644 --- a/nexus/mnemosyne/cli.py +++ b/nexus/mnemosyne/cli.py @@ -2,7 +2,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 clusters, mnemosyne hubs, mnemosyne bridges, mnemosyne rebuild, + mnemosyne tag, mnemosyne untag, mnemosyne retag """ from __future__ import annotations @@ -143,6 +144,42 @@ def cmd_rebuild(args): print(f"Rebuilt links: {total} connections across {archive.count} entries") +def cmd_tag(args): + archive = MnemosyneArchive() + tags = [t.strip() for t in args.tags.split(",") if t.strip()] + try: + entry = archive.add_tags(args.entry_id, tags) + except KeyError: + print(f"Entry not found: {args.entry_id}") + sys.exit(1) + print(f"[{entry.id[:8]}] {entry.title}") + print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}") + + +def cmd_untag(args): + archive = MnemosyneArchive() + tags = [t.strip() for t in args.tags.split(",") if t.strip()] + try: + entry = archive.remove_tags(args.entry_id, tags) + except KeyError: + print(f"Entry not found: {args.entry_id}") + sys.exit(1) + print(f"[{entry.id[:8]}] {entry.title}") + print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}") + + +def cmd_retag(args): + archive = MnemosyneArchive() + tags = [t.strip() for t in args.tags.split(",") if t.strip()] + try: + entry = archive.retag(args.entry_id, tags) + except KeyError: + print(f"Entry not found: {args.entry_id}") + sys.exit(1) + print(f"[{entry.id[:8]}] {entry.title}") + print(f" Topics: {', '.join(entry.topics) if entry.topics else '(none)'}") + + def main(): parser = argparse.ArgumentParser(prog="mnemosyne", description="The Living Holographic Archive") sub = parser.add_subparsers(dest="command") @@ -184,6 +221,18 @@ def main(): rb = sub.add_parser("rebuild", help="Recompute all links from scratch") rb.add_argument("-t", "--threshold", type=float, default=None, help="Similarity threshold override") + tg = sub.add_parser("tag", help="Add tags to an existing entry") + tg.add_argument("entry_id", help="Entry ID") + tg.add_argument("tags", help="Comma-separated tags to add") + + ut = sub.add_parser("untag", help="Remove tags from an existing entry") + ut.add_argument("entry_id", help="Entry ID") + ut.add_argument("tags", help="Comma-separated tags to remove") + + rt = sub.add_parser("retag", help="Replace all tags on an existing entry") + rt.add_argument("entry_id", help="Entry ID") + rt.add_argument("tags", help="Comma-separated new tag list") + args = parser.parse_args() if not args.command: parser.print_help() @@ -201,6 +250,9 @@ def main(): "hubs": cmd_hubs, "bridges": cmd_bridges, "rebuild": cmd_rebuild, + "tag": cmd_tag, + "untag": cmd_untag, + "retag": cmd_retag, } dispatch[args.command](args) diff --git a/nexus/mnemosyne/tests/test_archive.py b/nexus/mnemosyne/tests/test_archive.py index e269c1d..4a4ab61 100644 --- a/nexus/mnemosyne/tests/test_archive.py +++ b/nexus/mnemosyne/tests/test_archive.py @@ -343,3 +343,151 @@ def test_archive_topic_counts(): assert counts["automation"] == 2 # sorted by count desc — both tied but must be present assert set(counts.keys()) == {"python", "automation"} + + +# --- Tag management tests --- + +def test_add_tags_basic(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha"]) + archive.add_tags(e.id, ["beta", "gamma"]) + fresh = archive.get(e.id) + assert "beta" in fresh.topics + assert "gamma" in fresh.topics + assert "alpha" in fresh.topics + + +def test_add_tags_deduplication(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha"]) + archive.add_tags(e.id, ["alpha", "ALPHA", "beta"]) + fresh = archive.get(e.id) + lower_topics = [t.lower() for t in fresh.topics] + assert lower_topics.count("alpha") == 1 + assert "beta" in lower_topics + + +def test_add_tags_missing_entry(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + try: + archive.add_tags("nonexistent-id", ["tag"]) + assert False, "Expected KeyError" + except KeyError: + pass + + +def test_add_tags_empty_list(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha"]) + archive.add_tags(e.id, []) + fresh = archive.get(e.id) + assert fresh.topics == ["alpha"] + + +def test_remove_tags_basic(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha", "beta", "gamma"]) + archive.remove_tags(e.id, ["beta"]) + fresh = archive.get(e.id) + assert "beta" not in fresh.topics + assert "alpha" in fresh.topics + assert "gamma" in fresh.topics + + +def test_remove_tags_case_insensitive(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["Python", "rust"]) + archive.remove_tags(e.id, ["PYTHON"]) + fresh = archive.get(e.id) + assert "Python" not in fresh.topics + assert "rust" in fresh.topics + + +def test_remove_tags_missing_tag_silent(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha"]) + archive.remove_tags(e.id, ["nope"]) # should not raise + fresh = archive.get(e.id) + assert fresh.topics == ["alpha"] + + +def test_remove_tags_missing_entry(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + try: + archive.remove_tags("nonexistent-id", ["tag"]) + assert False, "Expected KeyError" + except KeyError: + pass + + +def test_retag_basic(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["old1", "old2"]) + archive.retag(e.id, ["new1", "new2"]) + fresh = archive.get(e.id) + assert fresh.topics == ["new1", "new2"] + + +def test_retag_deduplication(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["x"]) + archive.retag(e.id, ["go", "GO", "rust"]) + fresh = archive.get(e.id) + lower_topics = [t.lower() for t in fresh.topics] + assert lower_topics.count("go") == 1 + assert "rust" in lower_topics + + +def test_retag_empty_list(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + e = ingest_event(archive, title="T", content="c", topics=["alpha"]) + archive.retag(e.id, []) + fresh = archive.get(e.id) + assert fresh.topics == [] + + +def test_retag_missing_entry(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + archive = MnemosyneArchive(archive_path=path) + try: + archive.retag("nonexistent-id", ["tag"]) + assert False, "Expected KeyError" + except KeyError: + pass + + +def test_tag_persistence_across_reload(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "test_archive.json" + a1 = MnemosyneArchive(archive_path=path) + e = ingest_event(a1, title="T", content="c", topics=["alpha"]) + a1.add_tags(e.id, ["beta"]) + a1.remove_tags(e.id, ["alpha"]) + + a2 = MnemosyneArchive(archive_path=path) + fresh = a2.get(e.id) + assert "beta" in fresh.topics + assert "alpha" not in fresh.topics