diff --git a/src/fabledassistant/services/version_pinning.py b/src/fabledassistant/services/version_pinning.py new file mode 100644 index 0000000..6b32803 --- /dev/null +++ b/src/fabledassistant/services/version_pinning.py @@ -0,0 +1,90 @@ +"""Pin/unpin and auto-pin operations on NoteVersion. + +The autosave-rolling system in `services/note_versions.py` continues to +own the existing rolling-cap behavior; this module adds the manual-pin +API and the daily auto-pin scan. + +Tiers (pin_kind values): + None → rolling autosave; capped at MAX_VERSIONS, FIFO. + "auto" → system-declared via the stability scan; capped at MAX_AUTO_PINS, FIFO. + "manual" → user-declared; unlimited, never pruned. + +Design: docs/superpowers/specs/2026-05-13-note-version-pinning-design.md +""" +from __future__ import annotations + +import datetime +import logging +from datetime import timezone + +from sqlalchemy import select, text + +from fabledassistant.models import async_session +from fabledassistant.models.note_version import NoteVersion + +logger = logging.getLogger(__name__) + +MAX_AUTO_PINS = 25 +AUTO_PIN_STABILITY_DAYS = 2 +PIN_LABEL_MAX_LEN = 500 + + +async def pin_version( + user_id: int, note_id: int, version_id: int, *, label: str | None, +) -> NoteVersion | None: + """Mark a version as manually pinned. Returns the updated row or None + if not found (or wrong user/note scope). + + Acceptable on already-pinned rows (manual or auto) — promotes to + manual and updates the label. Labels are capped at PIN_LABEL_MAX_LEN + chars; longer values raise ValueError. + """ + if label is not None and len(label) > PIN_LABEL_MAX_LEN: + raise ValueError( + f"pin_label too long ({len(label)} > {PIN_LABEL_MAX_LEN} chars)" + ) + async with async_session() as session: + version = ( + await session.execute( + select(NoteVersion).where( + NoteVersion.id == version_id, + NoteVersion.note_id == note_id, + NoteVersion.user_id == user_id, + ) + ) + ).scalars().first() + if version is None: + return None + version.pin_kind = "manual" + version.pin_label = label + await session.commit() + await session.refresh(version) + return version + + +async def unpin_version( + user_id: int, note_id: int, version_id: int, +) -> NoteVersion | None: + """Clear pin_kind and pin_label, downgrading the row to rolling. + + Does NOT delete the row. If the row is older than the rolling cap + depth, the next create_version call will prune it via the rolling + FIFO. Returns the updated row or None if not found. + """ + async with async_session() as session: + version = ( + await session.execute( + select(NoteVersion).where( + NoteVersion.id == version_id, + NoteVersion.note_id == note_id, + NoteVersion.user_id == user_id, + ) + ) + ).scalars().first() + if version is None: + return None + version.pin_kind = None + version.pin_label = None + await session.commit() + await session.refresh(version) + return version diff --git a/tests/test_version_pinning_pin_unpin.py b/tests/test_version_pinning_pin_unpin.py new file mode 100644 index 0000000..d018640 --- /dev/null +++ b/tests/test_version_pinning_pin_unpin.py @@ -0,0 +1,128 @@ +"""Tests for manual pin / unpin on note versions.""" +from unittest.mock import AsyncMock, MagicMock, patch + + +def _mock_session_for_version(mock_version): + mock_session = AsyncMock() + result = MagicMock() + result.scalars.return_value.first.return_value = mock_version + mock_session.execute = AsyncMock(return_value=result) + mock_session.commit = AsyncMock() + mock_session.refresh = AsyncMock() + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=False) + return mock_session + + +async def test_pin_version_sets_manual_kind_and_label(): + mock_version = MagicMock() + mock_version.pin_kind = None + mock_version.pin_label = None + + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=_mock_session_for_version(mock_version), + ): + from fabledassistant.services.version_pinning import pin_version + result = await pin_version( + user_id=1, note_id=42, version_id=7, label="the runbook circa Q2", + ) + + assert result is mock_version + assert mock_version.pin_kind == "manual" + assert mock_version.pin_label == "the runbook circa Q2" + + +async def test_pin_version_accepts_null_label(): + """label is optional — pinning with no label is allowed.""" + mock_version = MagicMock() + mock_version.pin_kind = None + mock_version.pin_label = None + + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=_mock_session_for_version(mock_version), + ): + from fabledassistant.services.version_pinning import pin_version + await pin_version(user_id=1, note_id=42, version_id=7, label=None) + + assert mock_version.pin_kind == "manual" + assert mock_version.pin_label is None + + +async def test_pin_version_promotes_auto_to_manual_with_label_update(): + """Pinning an already-auto-pinned version promotes it and updates label.""" + mock_version = MagicMock() + mock_version.pin_kind = "auto" + mock_version.pin_label = "stable 2026-04-01 → 2026-04-05" + + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=_mock_session_for_version(mock_version), + ): + from fabledassistant.services.version_pinning import pin_version + await pin_version( + user_id=1, note_id=42, version_id=7, label="post-network-refresh", + ) + + assert mock_version.pin_kind == "manual" + assert mock_version.pin_label == "post-network-refresh" + + +async def test_pin_version_rejects_overlong_label(): + """Labels are capped at 500 chars to keep the UI sane.""" + mock_version = MagicMock() + mock_version.pin_kind = None + + # The cap is checked before any DB access, so the session shouldn't + # even be entered. We still patch it to avoid accidental DB lookups + # if the implementation order changes. + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=_mock_session_for_version(mock_version), + ): + from fabledassistant.services.version_pinning import pin_version + try: + await pin_version( + user_id=1, note_id=42, version_id=7, label="x" * 501, + ) + assert False, "expected ValueError" + except ValueError as e: + assert "500" in str(e) or "long" in str(e).lower() + + +async def test_pin_version_returns_none_when_not_found(): + mock_session = AsyncMock() + result = MagicMock() + result.scalars.return_value.first.return_value = None + mock_session.execute = AsyncMock(return_value=result) + mock_session.commit = AsyncMock() + mock_session.refresh = AsyncMock() + mock_session.__aenter__ = AsyncMock(return_value=mock_session) + mock_session.__aexit__ = AsyncMock(return_value=False) + + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=mock_session, + ): + from fabledassistant.services.version_pinning import pin_version + out = await pin_version(user_id=1, note_id=42, version_id=99, label=None) + + assert out is None + + +async def test_unpin_version_clears_kind_and_label(): + mock_version = MagicMock() + mock_version.pin_kind = "manual" + mock_version.pin_label = "previous label" + + with patch( + "fabledassistant.services.version_pinning.async_session", + return_value=_mock_session_for_version(mock_version), + ): + from fabledassistant.services.version_pinning import unpin_version + result = await unpin_version(user_id=1, note_id=42, version_id=7) + + assert result is mock_version + assert mock_version.pin_kind is None + assert mock_version.pin_label is None