FabledCurator/tests/test_discord_grouping.py

"""Milestone 388 E2: FC authors a post out of a Discord drop.

The one failure that would make this feature worse than nothing is
OVER-GROUPING — merging pieces that merely look alike into a post claiming
they belong together. Similarity alone does that: any two pieces of the same
character by the same artist sit close in SigLIP space. So the tests that
matter most here are the ones that prove the predicate REFUSES, and per rule
167 each is written so it would fail if the axis it guards were dropped.
"""
import math
from datetime import UTC, datetime, timedelta

import pytest
from sqlalchemy import select

from backend.app.models import Artist, ImageProvenance, ImageRecord, MLSettings, Post, Source
from backend.app.services.discord_grouping import (
    DROP_GROUPER,
    assign_to_group,
    build_groups,
    cosine_distance,
    group_source,
    join_open_groups,
    should_resurface,
    sweep,
)
from backend.app.services.post_feed_service import PostFeedService

pytestmark = pytest.mark.integration

DIM = 1152


def _vec(angle: float) -> list[float]:
    """A unit vector at `angle` radians, in the first two dimensions.

    Constructed this way so a test can STATE the distance it wants rather than
    hope: cosine distance between two of these is exactly `1 - cos(a - b)`. The
    first draft perturbed one component of an all-ones vector, which moved the
    vector by ~1e-6 — every distance assertion would have passed no matter what
    the predicate did (rule 167: a guard has to be able to fail).
    """
    v = [0.0] * DIM
    v[0] = math.cos(angle)
    v[1] = math.sin(angle)
    return v


# --- the predicate, in isolation ------------------------------------------


def test_identical_vectors_have_zero_distance():
    assert cosine_distance(_vec(0.4), _vec(0.4)) == pytest.approx(0.0, abs=1e-9)


def test_the_helper_produces_the_distance_it_claims():
    """The test-support vector itself, pinned — every threshold assertion below
    is only meaningful if `_vec` really moves by `1 - cos(delta)`."""
    assert cosine_distance(_vec(0.0), _vec(0.5)) == pytest.approx(1 - math.cos(0.5))


def test_a_zero_vector_never_pulls_anything_in():
    """No direction means no meaningful distance. Returning 0 ("identical")
    would let an all-zero embedding — a failed embed that stored something
    rather than nothing — vacuum every drop into one post."""
    assert cosine_distance([0.0] * DIM, _vec(0.0)) == 1.0


def test_a_gap_longer_than_the_window_ends_the_drop():
    """The time axis, alone. Both images are IDENTICAL, so if this grouped it
    would prove the window is not consulted — which is exactly the
    over-grouping failure (a month of one character collapsing into one post)."""
    now = datetime.now(UTC)
    rows = [
        (1, now, _vec(0.0)),
        (2, now + timedelta(days=30), _vec(0.0)),
    ]
    groups = build_groups(rows, max_distance=0.5, window=timedelta(minutes=60))
    assert [g.member_ids for g in groups] == [[1], [2]]


def test_unrelated_images_in_the_same_minute_do_not_group():
    """The similarity axis, alone. Same second, so the window cannot be what
    separates them."""
    now = datetime.now(UTC)
    # A quarter turn apart: distance 1.0, the far end of the scale.
    rows = [(1, now, _vec(0.0)), (2, now, _vec(math.pi / 2))]
    groups = build_groups(rows, max_distance=0.10, window=timedelta(minutes=60))
    assert [g.member_ids for g in groups] == [[1], [2]]


def test_near_variants_dropped_together_become_one_group():
    now = datetime.now(UTC)
    # 0.10 apart in angle = 0.005 in cosine distance, comfortably inside 0.10.
    rows = [
        (1, now, _vec(0.0)),
        (2, now + timedelta(minutes=2), _vec(0.10)),
        (3, now + timedelta(minutes=5), _vec(0.20)),
    ]
    groups = build_groups(rows, max_distance=0.10, window=timedelta(minutes=60))
    assert [g.member_ids for g in groups] == [[1, 2, 3]]


def test_the_window_is_measured_between_consecutive_messages():
    """An artist trickling variants out over an evening is ONE drop. Anchoring
    the window on the first message would cut this in half at minute 60."""
    now = datetime.now(UTC)
    rows = [
        (i, now + timedelta(minutes=50 * i), _vec(0.02 * i))
        for i in range(5)
    ]
    groups = build_groups(rows, max_distance=0.10, window=timedelta(minutes=60))
    assert len(groups) == 1
    assert len(groups[0].member_ids) == 5


def test_distance_is_measured_to_the_seed_so_a_group_cannot_drift():
    """Twenty small steps must not walk a group from one piece to another.

    Each hop here is within threshold of its PREDECESSOR; only the total
    departure from the seed exceeds it. Chaining to the previous member would
    swallow all of them.
    """
    now = datetime.now(UTC)
    # Each 0.30rad step is 0.0447 from its predecessor — inside the 0.05 cut.
    # Two steps out is already 0.1747 from the seed, three times the cut.
    rows = [
        (i, now + timedelta(minutes=i), _vec(0.30 * i))
        for i in range(8)
    ]
    assert cosine_distance(_vec(0.0), _vec(0.30)) < 0.05, "hop must be inside the cut"
    assert cosine_distance(_vec(0.0), _vec(0.60)) > 0.05, "seed distance must exceed it"
    groups = build_groups(rows, max_distance=0.05, window=timedelta(minutes=60))
    assert len(groups) > 1, "chained distance let the group drift"


# --- end to end, against the database -------------------------------------


async def _seed(db, *, name: str, platform: str = "discord"):
    artist = Artist(name=name, slug=name.lower().replace(" ", "-"))
    db.add(artist)
    await db.flush()
    source = Source(
        artist_id=artist.id, platform=platform,
        url=f"https://discord.com/channels/1/{name}", enabled=True,
    )
    db.add(source)
    await db.flush()
    return artist, source


async def _message(db, source, artist, *, ext: str, at, vec, text=None):
    post = Post(
        source_id=source.id, artist_id=artist.id, external_post_id=ext,
        post_date=at, description=text,
    )
    db.add(post)
    await db.flush()
    img = ImageRecord(
        path=f"/images/{source.id}-{ext}.jpg",
        sha256=f"{source.id:04d}{ext:0>60}"[:64],
        size_bytes=10, mime="image/jpeg", width=10, height=10,
        origin="downloaded", primary_post_id=post.id, artist_id=artist.id,
        siglip_embedding=vec,
    )
    db.add(img)
    await db.flush()
    return post, img


@pytest.mark.asyncio
async def test_a_drop_becomes_one_synthetic_post_carrying_every_image(db):
    artist, source = await _seed(db, name="drop-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    a, img_a = await _message(db, source, artist, ext="m1", at=long_ago, vec=_vec(0.0), text="blonde")
    b, img_b = await _message(
        db, source, artist, ext="m2", at=long_ago + timedelta(minutes=3),
        vec=_vec(0.10), text="and redhead",
    )
    await db.commit()

    created = await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()
    assert created == 1

    post = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    # The honesty rule: it says FC made it, and what from.
    assert post.synthesized_by == DROP_GROUPER
    assert post.synthesis_details["message_count"] == 2
    assert sorted(post.synthesis_details["member_post_ids"]) == sorted([a.id, b.id])
    # The thresholds AS THEY WERE, so the decision stays explicable after a tune.
    assert post.synthesis_details["max_distance"] == 0.10
    assert post.synthesis_details["window_minutes"] == 60
    # The messages' text, accumulated in arrival order, IS the body.
    assert post.description == "blonde\n\nand redhead"
    # No invented title — the one place this could put words in a creator's mouth.
    assert post.post_title is None

    linked = set((await db.execute(
        select(ImageProvenance.image_record_id).where(ImageProvenance.post_id == post.id)
    )).scalars().all())
    assert linked == {img_a.id, img_b.id}


@pytest.mark.asyncio
async def test_members_are_absorbed_not_destroyed_and_leave_the_feed(db):
    artist, source = await _seed(db, name="absorb-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    a, _ = await _message(db, source, artist, ext="n1", at=long_ago, vec=_vec(0.0))
    b, _ = await _message(
        db, source, artist, ext="n2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    synthetic = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    db.expunge_all()

    # Still there — they are the images' true origin and the audit trail.
    for member_id in (a.id, b.id):
        member = await db.get(Post, member_id)
        assert member is not None
        assert member.absorbed_by_post_id == synthetic.id

    feed_ids = [
        i["id"] for i in
        (await PostFeedService(db).scroll(cursor=None, artist_id=artist.id, limit=50))["items"]
    ]
    assert synthetic.id in feed_ids
    assert a.id not in feed_ids and b.id not in feed_ids

    # ...but still reachable by id. That is how a grouping gets inspected.
    assert (await PostFeedService(db).get_post(a.id))["absorbed_by_post_id"] == synthetic.id


@pytest.mark.asyncio
async def test_deleting_the_synthetic_post_returns_its_members_to_the_feed(db):
    """The reversal path is one DELETE — no repair step, no orphan."""
    artist, source = await _seed(db, name="undo-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    a, _ = await _message(db, source, artist, ext="u1", at=long_ago, vec=_vec(0.0))
    b, _ = await _message(
        db, source, artist, ext="u2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    synthetic = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    await db.delete(synthetic)
    await db.commit()
    db.expunge_all()

    for member_id in (a.id, b.id):
        assert (await db.get(Post, member_id)).absorbed_by_post_id is None
    feed_ids = [
        i["id"] for i in
        (await PostFeedService(db).scroll(cursor=None, artist_id=artist.id, limit=50))["items"]
    ]
    assert a.id in feed_ids and b.id in feed_ids


@pytest.mark.asyncio
async def test_a_post_with_no_embedding_yet_is_left_alone(db):
    """Embeddings land asynchronously after import, so the sweep must skip what
    it cannot place rather than grouping on a null — and must not hide it."""
    artist, source = await _seed(db, name="pending-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    post = Post(
        source_id=source.id, artist_id=artist.id,
        external_post_id="pending", post_date=long_ago,
    )
    db.add(post)
    await db.flush()
    db.add(ImageRecord(
        path="/images/pending.jpg", sha256="p" * 64, size_bytes=10,
        mime="image/jpeg", width=10, height=10, origin="downloaded",
        primary_post_id=post.id, artist_id=artist.id, siglip_embedding=None,
    ))
    await db.commit()

    assert await group_source(db, source, max_distance=0.10, window_minutes=60) == 0
    await db.commit()
    db.expunge_all()
    assert (await db.get(Post, post.id)).absorbed_by_post_id is None


@pytest.mark.asyncio
async def test_a_drop_still_inside_the_window_is_left_open(db):
    """A sweep landing mid-drop must not cut it in half and call the second
    half its own drop. Waiting one window costs nothing — the sweep re-runs."""
    artist, source = await _seed(db, name="open-artist")
    now = datetime.now(UTC)
    await _message(db, source, artist, ext="o1", at=now - timedelta(minutes=5), vec=_vec(0.0))
    await db.commit()

    assert await group_source(db, source, max_distance=0.10, window_minutes=60) == 0


@pytest.mark.asyncio
async def test_re_running_the_sweep_does_not_re_group_what_it_already_took(db):
    artist, source = await _seed(db, name="idempotent-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    await _message(db, source, artist, ext="i1", at=long_ago, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="i2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    await db.commit()

    assert await group_source(db, source, max_distance=0.10, window_minutes=60) == 1
    await db.commit()
    assert await group_source(db, source, max_distance=0.10, window_minutes=60) == 0
    await db.commit()

    count = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalars().all()
    assert len(count) == 1


@pytest.mark.asyncio
async def test_a_synthetic_post_is_never_itself_absorbed(db):
    """Groups of groups would compound every mistake the grouper makes."""
    artist, source = await _seed(db, name="no-nesting-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    await _message(db, source, artist, ext="g1", at=long_ago, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="g2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    synthetic = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    assert synthetic.absorbed_by_post_id is None
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()
    db.expunge_all()
    assert (await db.get(Post, synthetic.id)).absorbed_by_post_id is None


@pytest.mark.asyncio
async def test_the_sweep_is_a_no_op_when_the_switch_is_off(db):
    artist, source = await _seed(db, name="switched-off-artist")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    await _message(db, source, artist, ext="s1", at=long_ago, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="s2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    settings = await MLSettings.load(db)
    settings.discord_grouping_enabled = False
    await db.commit()

    result = await sweep(db)
    # Exact equality on purpose, not a subset check: a disabled sweep reports a
    # COMPLETE zeroed shape rather than a shorter one, so a caller can read any
    # counter unconditionally. (Today's caller short-circuits on `enabled`, so
    # nothing would KeyError — the point is that it does not HAVE to.) E3 adding
    # `images_joined` broke this assertion, which is exactly what it is for.
    assert result == {
        "enabled": False, "sources": 0, "posts_created": 0, "images_joined": 0,
    }


@pytest.mark.asyncio
async def test_the_sweep_only_touches_discord_sources(db):
    """Patreon posts ARE authored. Grouping them would be FC rewriting a
    creator's own publishing decisions."""
    artist, source = await _seed(db, name="patreon-artist", platform="patreon")
    long_ago = datetime.now(UTC) - timedelta(days=2)
    a, _ = await _message(db, source, artist, ext="p1", at=long_ago, vec=_vec(0.0))
    b, _ = await _message(
        db, source, artist, ext="p2", at=long_ago + timedelta(minutes=1),
        vec=_vec(0.10),
    )
    settings = await MLSettings.load(db)
    settings.discord_grouping_enabled = True
    await db.commit()

    await sweep(db)
    await db.commit()
    db.expunge_all()
    assert (await db.get(Post, a.id)).absorbed_by_post_id is None
    assert (await db.get(Post, b.id)).absorbed_by_post_id is None


# --- E3: an open grouping -------------------------------------------------
#
# Three hard problems, one test class each. The anti-thrash rule is ASSERTED
# rather than assumed, because "a group that grows daily monopolises the feed"
# is the kind of defect nobody notices until they have lived with it.


def test_growth_below_the_minimum_never_moves_the_post():
    """One image a day must not chain bumps together."""
    now = datetime.now(UTC)
    assert not should_resurface(
        images_since_surface=1, last_surface_at=now - timedelta(days=30),
        now=now, min_images=2, cooldown=timedelta(hours=24),
    )


def test_growth_inside_the_cooldown_never_moves_the_post():
    """Even a big second wave waits — otherwise a burst bumps once per sweep."""
    now = datetime.now(UTC)
    assert not should_resurface(
        images_since_surface=10, last_surface_at=now - timedelta(hours=1),
        now=now, min_images=2, cooldown=timedelta(hours=24),
    )


def test_a_real_second_wave_moves_the_post_once():
    now = datetime.now(UTC)
    assert should_resurface(
        images_since_surface=3, last_surface_at=now - timedelta(days=2),
        now=now, min_images=2, cooldown=timedelta(hours=24),
    )


def _group(post_id: int, angle: float):
    return (Post(id=post_id), _vec(angle))


def test_a_candidate_near_two_groups_joins_neither():
    """Bridging. Merging would rewrite history — two posts the operator may
    already have seen become one — so the recoverable answer is to leave it and
    let it found its own group."""
    left, right = _group(1, -0.30), _group(2, 0.30)
    assert assign_to_group(_vec(0.0), [left, right], max_distance=0.5) is None


def test_a_candidate_clearly_nearer_one_group_joins_it():
    near, far = _group(1, 0.02), _group(2, 1.2)
    chosen = assign_to_group(_vec(0.0), [near, far], max_distance=0.5)
    assert chosen is not None and chosen.id == 1


def test_a_candidate_near_nothing_joins_nothing():
    assert assign_to_group(
        _vec(0.0), [_group(1, math.pi / 2)], max_distance=0.10,
    ) is None


@pytest.mark.asyncio
async def test_a_later_variant_joins_the_existing_post_instead_of_founding_one(db):
    artist, source = await _seed(db, name="rejoin-artist")
    start = datetime.now(UTC) - timedelta(days=3)
    await _message(db, source, artist, ext="r1", at=start, vec=_vec(0.0), text="first")
    await _message(
        db, source, artist, ext="r2", at=start + timedelta(minutes=2),
        vec=_vec(0.05), text="second",
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    group = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    group_id = group.id

    # A day later — far outside the drop window, well inside the open period.
    await _message(
        db, source, artist, ext="r3", at=start + timedelta(days=1),
        vec=_vec(0.06), text="next day",
    )
    await db.commit()

    joined = await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2,
        resurface_cooldown_hours=24,
    )
    await db.commit()
    assert joined == 1

    # No second post — that is the point of the step.
    posts = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalars().all()
    assert len(posts) == 1

    db.expunge_all()
    grown = await db.get(Post, group_id)
    assert grown.last_grew_at is not None
    assert grown.synthesis_details["message_count"] == 3
    # The new message's text joined the body, in arrival order.
    assert grown.description == "first\n\nsecond\n\nnext day"


@pytest.mark.asyncio
async def test_a_variant_after_the_group_closed_starts_a_new_post(db):
    """Groups must not stay open forever — artists reuse characters for years."""
    artist, source = await _seed(db, name="closed-artist")
    start = datetime.now(UTC) - timedelta(days=60)
    await _message(db, source, artist, ext="c1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="c2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    await _message(
        db, source, artist, ext="c3", at=datetime.now(UTC) - timedelta(days=1),
        vec=_vec(0.06),
    )
    await db.commit()

    joined = await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2,
        resurface_cooldown_hours=24,
    )
    await db.commit()
    assert joined == 0, "a closed group must not absorb a two-month-later drop"

    created = await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()
    assert created == 1


@pytest.mark.asyncio
async def test_joining_does_not_duplicate_images(db):
    """The unique (image, post) constraint is the backstop, not the argument —
    a second join pass must add nothing at all."""
    artist, source = await _seed(db, name="nodupe-artist")
    start = datetime.now(UTC) - timedelta(days=3)
    await _message(db, source, artist, ext="d1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="d2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    await _message(
        db, source, artist, ext="d3", at=start + timedelta(days=1), vec=_vec(0.06),
    )
    await db.commit()

    kwargs = {
        "max_distance": 0.10, "window_minutes": 60, "close_after_hours": 168,
        "resurface_min_images": 2, "resurface_cooldown_hours": 24,
    }
    assert await join_open_groups(db, source, **kwargs) == 1
    await db.commit()
    assert await join_open_groups(db, source, **kwargs) == 0
    await db.commit()

    group = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    links = (await db.execute(
        select(ImageProvenance.image_record_id).where(ImageProvenance.post_id == group.id)
    )).scalars().all()
    assert len(links) == len(set(links)) == 3


@pytest.mark.asyncio
async def test_a_group_that_grows_by_one_does_not_move_in_the_feed(db):
    """The anti-thrash rule, end to end. A drip-feed updates IN PLACE."""
    artist, source = await _seed(db, name="dripfeed-artist")
    start = datetime.now(UTC) - timedelta(days=3)
    await _message(db, source, artist, ext="t1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="t2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    await _message(
        db, source, artist, ext="t3", at=start + timedelta(days=1), vec=_vec(0.06),
    )
    await db.commit()
    await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24,
    )
    await db.commit()

    group = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    assert group.last_grew_at is not None, "it grew"
    assert group.resurfaced_at is None, "but one image must not move it"
    assert group.synthesis_details["images_since_surface"] == 1


@pytest.mark.asyncio
async def test_a_second_wave_resurfaces_the_post(db):
    artist, source = await _seed(db, name="secondwave-artist")
    start = datetime.now(UTC) - timedelta(days=3)
    await _message(db, source, artist, ext="w1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="w2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    for i, ext in enumerate(("w3", "w4", "w5")):
        await _message(
            db, source, artist, ext=ext,
            at=start + timedelta(days=1, minutes=i), vec=_vec(0.06),
        )
    await db.commit()
    await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24,
    )
    await db.commit()

    group = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    assert group.resurfaced_at is not None
    # The counter resets, so the NEXT trickle starts from zero rather than
    # riding the same three images into a second bump.
    assert group.synthesis_details["images_since_surface"] == 0


@pytest.mark.asyncio
async def test_a_resurfaced_group_sorts_by_when_it_moved_not_when_it_started(db):
    """The feed ordering decision, pinned. The group's drop began three days
    ago — older than the decoy post — but it resurfaced just now, so it leads."""
    artist, source = await _seed(db, name="feedorder-artist")
    start = datetime.now(UTC) - timedelta(days=3)
    await _message(db, source, artist, ext="f1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="f2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    # An ordinary post from yesterday: newer than the drop's start, older than
    # the moment the group resurfaces.
    decoy = Post(
        source_id=source.id, artist_id=artist.id, external_post_id="decoy",
        post_date=datetime.now(UTC) - timedelta(days=1), post_title="Decoy",
    )
    db.add(decoy)
    await db.flush()
    decoy_id = decoy.id

    for i, ext in enumerate(("f3", "f4", "f5")):
        await _message(
            db, source, artist, ext=ext,
            at=start + timedelta(days=1, minutes=i), vec=_vec(0.06),
        )
    await db.commit()
    await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24,
    )
    await db.commit()

    group = (await db.execute(
        select(Post).where(Post.synthesized_by == DROP_GROUPER)
    )).scalar_one()
    page = await PostFeedService(db).scroll(
        cursor=None, artist_id=artist.id, limit=50,
    )
    ids = [i["id"] for i in page["items"]]
    assert ids.index(group.id) < ids.index(decoy_id), (
        "a resurfaced group must lead a post published after the drop STARTED"
    )
    # And it is still reported as a grouping that grew.
    item = next(i for i in page["items"] if i["id"] == group.id)
    assert item["last_grew_at"] is not None


@pytest.mark.asyncio
async def test_pagination_stays_stable_when_a_group_has_resurfaced(db):
    """The cursor is built in Python and the ORDER BY in SQL. If they disagree
    on resurfaced_at's precedence, rows are silently skipped or repeated at
    every page boundary — so page through one at a time and check."""
    artist, source = await _seed(db, name="cursor-artist")
    start = datetime.now(UTC) - timedelta(days=5)
    await _message(db, source, artist, ext="x1", at=start, vec=_vec(0.0))
    await _message(
        db, source, artist, ext="x2", at=start + timedelta(minutes=2), vec=_vec(0.05),
    )
    await db.commit()
    await group_source(db, source, max_distance=0.10, window_minutes=60)
    await db.commit()

    for i in range(4):
        db.add(Post(
            source_id=source.id, artist_id=artist.id,
            external_post_id=f"plain{i}",
            post_date=datetime.now(UTC) - timedelta(days=i + 1),
        ))
    for i, ext in enumerate(("x3", "x4", "x5")):
        await _message(
            db, source, artist, ext=ext,
            at=start + timedelta(days=1, minutes=i), vec=_vec(0.06),
        )
    await db.commit()
    await join_open_groups(
        db, source, max_distance=0.10, window_minutes=60,
        close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24,
    )
    await db.commit()

    svc = PostFeedService(db)
    everything = [i["id"] for i in (
        await svc.scroll(cursor=None, artist_id=artist.id, limit=100)
    )["items"]]

    walked, cursor = [], None
    while True:
        page = await svc.scroll(
            cursor=cursor, artist_id=artist.id, limit=1,
        )
        walked.extend(i["id"] for i in page["items"])
        cursor = page["next_cursor"]
        if cursor is None:
            break
    assert walked == everything