From 37c704e8753a0d7646a28a50ff68e9e568770b0b Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Wed, 13 May 2026 13:57:56 -0400 Subject: [PATCH] feat(versions): auto-pin scan promotes stable versions _promote_stable_versions_for_note is the pure-function core: walks versions chronologically and pins any with a >= AUTO_PIN_STABILITY_DAYS (2-day) gap to the next version (or to now, for the latest). Auto- generated label describes the stability window. _scan_one_note loads versions for one note, runs the promotion, commits mutations to the attached rows, then calls prune_auto_pins to cap the auto bucket. scan_user_for_auto_pins fans out across the user's notes; scan_all_users_for_auto_pins is the top-level entrypoint for the cron. Per-note and per-user errors are caught and logged. --- .../services/version_pinning.py | 124 ++++++++++++++ tests/test_version_pinning_scan.py | 161 ++++++++++++++++++ 2 files changed, 285 insertions(+) create mode 100644 tests/test_version_pinning_scan.py diff --git a/src/fabledassistant/services/version_pinning.py b/src/fabledassistant/services/version_pinning.py index 780b795..159c3c7 100644 --- a/src/fabledassistant/services/version_pinning.py +++ b/src/fabledassistant/services/version_pinning.py @@ -90,6 +90,130 @@ async def unpin_version( return version +def _format_auto_pin_label( + start: datetime.datetime, end: datetime.datetime | None, +) -> str: + """Auto-generated label describing the stability window. + + end=None → version is the latest with no successor; render as + "stable since {start_iso}". Otherwise render as + "stable {start_iso} → {end_iso}". + """ + s = start.date().isoformat() + if end is None: + return f"stable since {s}" + return f"stable {s} → {end.date().isoformat()}" + + +def _promote_stable_versions_for_note(versions_chrono: list) -> list: + """Mutate the input list: set pin_kind='auto' + auto-label on any + unpinned version whose gap to its successor (or to now, for the + latest) is >= AUTO_PIN_STABILITY_DAYS. + + Returns the list of versions that were newly pinned (caller uses + this for logging / counts). + """ + now = datetime.datetime.now(timezone.utc) + newly_pinned: list = [] + for i, v in enumerate(versions_chrono): + if v.pin_kind is not None: + continue + if i + 1 < len(versions_chrono): + next_ts = versions_chrono[i + 1].created_at + is_latest = False + else: + next_ts = now + is_latest = True + v_ts = v.created_at + if v_ts.tzinfo is None: + v_ts = v_ts.replace(tzinfo=timezone.utc) + if next_ts.tzinfo is None: + next_ts = next_ts.replace(tzinfo=timezone.utc) + gap_days = (next_ts - v_ts).total_seconds() / 86400 + if gap_days >= AUTO_PIN_STABILITY_DAYS: + v.pin_kind = "auto" + v.pin_label = _format_auto_pin_label( + v_ts, None if is_latest else next_ts, + ) + newly_pinned.append(v) + return newly_pinned + + +async def _list_user_note_ids_with_versions(user_id: int) -> list[int]: + """Return note_ids belonging to user_id that have at least one row in + note_versions. Notes with no version history are ignored.""" + async with async_session() as session: + rows = ( + await session.execute( + text( + """ + SELECT DISTINCT note_id FROM note_versions + WHERE user_id = :user_id + """ + ).bindparams(user_id=user_id) + ) + ).all() + return [r[0] for r in rows] + + +async def _scan_one_note(user_id: int, note_id: int) -> int: + """Run the promote+prune flow for one note. Returns count of newly- + pinned versions.""" + async with async_session() as session: + versions = ( + await session.execute( + select(NoteVersion) + .where( + NoteVersion.note_id == note_id, + NoteVersion.user_id == user_id, + ) + .order_by(NoteVersion.created_at.asc()) + ) + ).scalars().all() + if not versions: + return 0 + newly_pinned = _promote_stable_versions_for_note(list(versions)) + if newly_pinned: + await session.commit() + if newly_pinned: + await prune_auto_pins(user_id=user_id, note_id=note_id) + return len(newly_pinned) + + +async def scan_user_for_auto_pins(user_id: int) -> int: + """Run the scan across every versioned note for one user. Returns + total newly-pinned-this-pass.""" + total = 0 + note_ids = await _list_user_note_ids_with_versions(user_id) + for note_id in note_ids: + try: + total += await _scan_one_note(user_id, note_id) + except Exception: + logger.exception( + "auto-pin scan failed for user=%d note=%d", user_id, note_id, + ) + return total + + +async def scan_all_users_for_auto_pins() -> dict[int, int]: + """Top-level scan entrypoint. Iterates over all users and runs the + per-user scan. Returns {user_id: newly_pinned_count}. Per-user errors + are caught and logged so one user's failure doesn't stop the scan.""" + from fabledassistant.models import User + + async with async_session() as session: + users = (await session.execute(select(User.id))).scalars().all() + + out: dict[int, int] = {} + for uid in users: + try: + out[uid] = await scan_user_for_auto_pins(uid) + except Exception: + logger.exception("auto-pin scan failed for user=%d", uid) + out[uid] = 0 + return out + + async def prune_auto_pins(user_id: int, note_id: int) -> None: """FIFO-prune the auto-pinned bucket for one note past MAX_AUTO_PINS. diff --git a/tests/test_version_pinning_scan.py b/tests/test_version_pinning_scan.py new file mode 100644 index 0000000..772ec87 --- /dev/null +++ b/tests/test_version_pinning_scan.py @@ -0,0 +1,161 @@ +"""Tests for the auto-pin scan algorithm. + +Tests the per-note promotion logic directly via a pure helper so the +algorithm can be driven without standing up a real DB. +""" +from datetime import datetime, timedelta, timezone +from types import SimpleNamespace + + +def _v(version_id: int, days_ago: int, pin_kind=None): + """Build a SimpleNamespace mimicking a NoteVersion row.""" + return SimpleNamespace( + id=version_id, + created_at=datetime.now(timezone.utc) - timedelta(days=days_ago), + pin_kind=pin_kind, + pin_label=None, + ) + + +def test_promote_stable_versions_pins_versions_with_2_day_gap(): + """Versions followed by another version >= 2 days later get promoted.""" + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + # 3 versions: v1 (10 days ago), v2 (5 days ago), v3 (1 day ago) + # Gaps: v1→v2 = 5d, v2→v3 = 4d. Both ≥ 2d → both pinned. + # v3 (latest) has gap to now = 1d → NOT pinned (< 2 days). + versions = [_v(1, 10), _v(2, 5), _v(3, 1)] + pinned = _promote_stable_versions_for_note(versions) + assert {v.id for v in pinned} == {1, 2} + assert versions[0].pin_kind == "auto" + assert versions[1].pin_kind == "auto" + assert versions[2].pin_kind is None + + +def test_promote_stable_versions_pins_latest_if_old_enough(): + """If the latest version is >= 2 days old with no successor, pin it.""" + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + versions = [_v(1, 10), _v(2, 3)] # latest is 3 days old + pinned = _promote_stable_versions_for_note(versions) + assert {v.id for v in pinned} == {1, 2} + + +def test_promote_stable_versions_skips_already_pinned(): + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + versions = [_v(1, 10, pin_kind="manual"), _v(2, 5)] + pinned = _promote_stable_versions_for_note(versions) + # v1 already manual — not re-pinned. v2 has 5-day gap to v1, then + # latest-with-no-successor gap of 5d to now → pinned auto. + assert {v.id for v in pinned} == {2} + assert versions[0].pin_kind == "manual" + assert versions[1].pin_kind == "auto" + + +def test_promote_stable_versions_skips_active_editing(): + """Versions a few hours apart with no >2-day gaps → no pins.""" + base = datetime.now(timezone.utc) + v1 = SimpleNamespace( + id=1, created_at=base - timedelta(hours=10), + pin_kind=None, pin_label=None, + ) + v2 = SimpleNamespace( + id=2, created_at=base - timedelta(hours=2), + pin_kind=None, pin_label=None, + ) + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + pinned = _promote_stable_versions_for_note([v1, v2]) + assert pinned == [] + + +def test_promote_stable_versions_writes_descriptive_label(): + """The auto-generated label references stability.""" + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + versions = [_v(1, 10), _v(2, 5), _v(3, 1)] + _promote_stable_versions_for_note(versions) + assert versions[0].pin_label is not None + assert "stable" in versions[0].pin_label.lower() + + +def test_promote_stable_versions_naive_datetime_is_treated_as_utc(): + """created_at may come in as a naive datetime from some code paths; + the algorithm coerces it to UTC rather than crashing on subtraction.""" + naive_old = datetime.utcnow() - timedelta(days=10) + naive_recent = datetime.utcnow() - timedelta(days=1) + versions = [ + SimpleNamespace(id=1, created_at=naive_old, pin_kind=None, pin_label=None), + SimpleNamespace(id=2, created_at=naive_recent, pin_kind=None, pin_label=None), + ] + from fabledassistant.services.version_pinning import ( + _promote_stable_versions_for_note, + ) + pinned = _promote_stable_versions_for_note(versions) + # v1 → v2 gap is 9 days → v1 pinned. v2 → now is 1 day → not pinned. + assert {v.id for v in pinned} == {1} + + +async def test_scan_user_for_auto_pins_iterates_all_notes(monkeypatch): + """scan_user_for_auto_pins iterates every note for the user and calls + the per-note flow on each, summing returned counts.""" + from fabledassistant.services import version_pinning + + note_ids = [10, 20, 30] + seen: list[int] = [] + + async def fake_list_note_ids(uid): + return note_ids + + async def fake_one_note(uid, nid): + seen.append(nid) + # Pretend each note pinned one version. + return 1 + + monkeypatch.setattr( + version_pinning, + "_list_user_note_ids_with_versions", + fake_list_note_ids, + ) + monkeypatch.setattr( + version_pinning, "_scan_one_note", fake_one_note, + ) + + total = await version_pinning.scan_user_for_auto_pins(user_id=1) + + assert seen == note_ids + assert total == 3 + + +async def test_scan_user_for_auto_pins_swallows_per_note_errors(monkeypatch): + """One bad note doesn't stop the scan; the error is logged and the + others continue.""" + from fabledassistant.services import version_pinning + + async def fake_list_note_ids(uid): + return [10, 20, 30] + + async def fake_one_note(uid, nid): + if nid == 20: + raise RuntimeError("simulated DB error") + return 1 + + monkeypatch.setattr( + version_pinning, + "_list_user_note_ids_with_versions", + fake_list_note_ids, + ) + monkeypatch.setattr( + version_pinning, "_scan_one_note", fake_one_note, + ) + + total = await version_pinning.scan_user_for_auto_pins(user_id=1) + + # 10 and 30 each contributed 1; 20 raised and contributed 0. + assert total == 2