feat(versions): pin_version and unpin_version services

pin_version sets pin_kind='manual' and pin_label on the target row.
Accepts already-pinned rows (promotes auto→manual, updates label).
Labels are capped at PIN_LABEL_MAX_LEN=500 chars; longer values raise
ValueError before any DB access.

unpin_version clears both fields, downgrading the row to rolling. Does
NOT delete — if the row is past the rolling FIFO depth, the next
autosave's prune will drop it.
This commit is contained in:
2026-05-13 13:56:21 -04:00
parent b65d736869
commit 925a53e0f7
2 changed files with 218 additions and 0 deletions
@@ -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