fix: the matcher sees a drop's names, finds merged trickles, and catches up on its own (4390, 4392)

Three defects found while answering "is the linking automatic":

1. The name and image checks never worked on a live drop. The corpus keyed
   every image by primary_post_id, but a drop's images belong to its member
   messages and the drop claims them only through provenance, so a drop
   looked nameless and hashless. The tests attached images to the drop
   itself, which discord_grouping never does. Images now count under the
   post a reader sees them on: a message's absorbing drop, else the post
   itself.

2. The trickle merge (e08401c) dates a drop by its first stage, days before
   the release a teaser announces, which put merged trickles outside the 24h
   window. Candidates are now found and timed by their closest member
   message. The sweep follows recently grown drops by their messages' times
   the same way.

3. A pair left pending was skipped forever: the matcher skipped every
   recorded pair, not only decided ones. Pending pairs are now re-scored in
   place and linked once conclusive. Linked and dismissed pairs are still
   never touched.

Also, "Scan now" shared the sweep's 48-hour horizon, so it could not reach
the history it is described as being for. It now scores every post by an
artist with Discord drops (rescan(full=True)).

New tests build drops the way the grouper does, with images owned by the
member messages.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/api/posts.py

@@ -207,6 +207,8 @@ async def rescan_associations():
     be within the window to exist at all); this is the button for a first run
     over a library that predates the feature."""
     async with get_session() as session:
-        result = await association_rescan(session)
+        # full=True: the button reaches the whole history, which the hourly
+        # sweep's 48-hour horizon never does.
+        result = await association_rescan(session, full=True)
         await session.commit()
     return jsonify(result)

backend/app/services/post_association_service.py

@@ -286,10 +286,28 @@ class PostAssociationService:
 
         paths_by_post: dict[int, list[str]] = {}
         hashes_by_post: dict[int, list[int]] = {}
+        # An image is counted under the post a reader SEES it on: a Discord
+        # message absorbed into a drop contributes to the drop, not to itself.
+        #
+        # Keyed on `primary_post_id` alone until 2026-09-24, which on the live
+        # instance meant a drop never had a name or a hash at all — its images
+        # are owned by its member messages, and the drop only claims them
+        # through provenance. The identity route therefore never fired there;
+        # the tests missed it because they attached images to the drop itself,
+        # which discord_grouping never does.
+        #
+        # Counting by the drop also makes the span honest: five wips of one
+        # piece posted as five messages and grouped into one drop are ONE post
+        # as far as "how many posts carry this name" is concerned.
+        owner = Post.__table__.alias("owner")
         rows = await self.session.execute(
             select(
-                ImageRecord.primary_post_id, ImageRecord.path, ImageRecord.phash
-            ).where(
+                func.coalesce(owner.c.absorbed_by_post_id, ImageRecord.primary_post_id),
+                ImageRecord.path, ImageRecord.phash,
+            )
+            .select_from(ImageRecord)
+            .join(owner, owner.c.id == ImageRecord.primary_post_id)
+            .where(
                 ImageRecord.artist_id == artist_id,
                 ImageRecord.primary_post_id.is_not(None),
             )
@@ -333,44 +351,81 @@ class PostAssociationService:
         self._corpora[artist_id] = corpus
         return corpus
 
-    async def _decided(self, announcement_id: int) -> set[int]:
-        """Payload posts already proposed for this announcement, in ANY status.
+    async def _decided(self, announcement_id: int) -> dict[int, PostAssociation]:
+        """Pairs already recorded for this announcement, keyed by payload.
 
-        Dismissed pairs are included deliberately: re-proposing a pair the
-        operator has already rejected on every subsequent scan is the single
-        behaviour that makes a review queue get ignored.
+        Linked and dismissed pairs are never touched again: re-proposing a pair
+        the operator has already rejected on every subsequent scan is the
+        single behaviour that makes a review queue get ignored.
+
+        A PENDING pair is different — nobody has decided it — so the caller
+        re-scores it. Otherwise a pair queued by an older, weaker matcher sits
+        in the queue forever even once the evidence is conclusive, which is the
+        chore the operator asked FC not to hand them.
         """
         rows = (await self.session.execute(
-            select(PostAssociation.payload_post_id)
+            select(PostAssociation)
             .where(PostAssociation.announcement_post_id == announcement_id)
         )).scalars().all()
-        return set(rows)
+        return {a.payload_post_id: a for a in rows}
 
     async def _candidate_groups(
         self, announcement: Post, *, window: timedelta,
-    ) -> list[Post]:
-        """Synthetic Discord groupings by the SAME artist, inside the window.
+    ) -> list[tuple[Post, datetime]]:
+        """Synthetic Discord groupings by the SAME artist with a message inside
+        the window — each with the time of its message CLOSEST to the teaser.
 
         Same-artist is the identity signal and it is free (see the module
         docstring on E4). It is also a hard filter rather than a scored one:
         two different creators posting minutes apart is a coincidence, not
         evidence, and letting it score at all would mean a busy hour across the
         library could out-vote everything else.
+
+        Measured on the MESSAGES, not the drop's own date. A drop is dated by
+        its first message, and since #4390 merges a creator's trickle into one
+        drop, that can be days before the release the teaser announces —
+        "Very early Marin" on Sep 7, `MarinaraSauce_base` on Sep 11. Matching
+        on the drop's date would put every merged trickle outside the window.
         """
         at = _post_time(announcement)
-        sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
-        return (await self.session.execute(
-            select(Post)
+        member = Post.__table__.alias("member")
+        member_at = func.coalesce(member.c.post_date, member.c.downloaded_at)
+        rows = list((await self.session.execute(
+            select(member.c.absorbed_by_post_id, member_at)
             .where(
+                member.c.artist_id == announcement.artist_id,
+                member.c.absorbed_by_post_id.is_not(None),
+                member_at >= at - window,
+                member_at <= at + window,
+            )
+        )).all())
+        # The drop's own date counts too — the first message's, so it adds
+        # nothing for a real drop, but it keeps a drop with no member rows
+        # (hand-built, or one whose messages were removed) matchable.
+        own_at = func.coalesce(Post.post_date, Post.downloaded_at)
+        rows += (await self.session.execute(
+            select(Post.id, own_at).where(
                 Post.artist_id == announcement.artist_id,
                 Post.synthesized_by == DROP_GROUPER,
+                own_at >= at - window,
+                own_at <= at + window,
+            )
+        )).all()
+        closest: dict[int, datetime] = {}
+        for group_id, when in rows:
+            if group_id not in closest or abs(when - at) < abs(closest[group_id] - at):
+                closest[group_id] = when
+        if not closest:
+            return []
+        groups = (await self.session.execute(
+            select(Post).where(
+                Post.id.in_(list(closest)),
+                Post.synthesized_by == DROP_GROUPER,
                 Post.id != announcement.id,
-                sort_key >= at - window,
-                sort_key <= at + window,
             )
-            .order_by(sort_key)
-            .limit(MAX_CANDIDATES)
         )).scalars().all()
+        ranked = sorted(groups, key=lambda g: (abs(closest[g.id] - at), g.id))
+        return [(g, closest[g.id]) for g in ranked[:MAX_CANDIDATES]]
 
     async def _claimed(self, announcement_id: int, payload_id: int) -> bool:
         """Is either end of this pair already spoken for by an accepted link?
@@ -415,8 +470,9 @@ class PostAssociationService:
 
         made = 0
         scored: list[tuple[Post, float, dict, float]] = []
-        for group in await self._candidate_groups(announcement, window=window):
-            if group.id in already:
+        for group, group_at in await self._candidate_groups(announcement, window=window):
+            prior = already.get(group.id)
+            if prior is not None and prior.status != "pending":
                 continue
             named, token = shared_identity(
                 here,
@@ -435,7 +491,7 @@ class PostAssociationService:
             identity = max(named, copied)
             circumstantial = {
                 "proximity": proximity_signal(
-                    _post_time(group) - _post_time(announcement), window,
+                    group_at - _post_time(announcement), window,
                 ),
                 "declared": declared,
                 "marker": marker_overlap(
@@ -494,6 +550,16 @@ class PostAssociationService:
             status = "linked" if group.id == auto_id else "pending"
             if status == "linked":
                 linked += 1
+            prior = already.get(group.id)
+            if prior is not None:
+                # Re-scored in place: a pending pair keeps its row (and id),
+                # and only an upgrade to linked counts as news.
+                prior.score = score
+                prior.signals = signals
+                if status == "linked":
+                    prior.status = "linked"
+                    prior.linked_by = "fc"
+                continue
             self.session.add(PostAssociation(
                 announcement_post_id=announcement.id,
                 payload_post_id=group.id,
@@ -573,8 +639,17 @@ class PostAssociationService:
         return out
 
 
-async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
-    """Score every recent non-synthetic post against nearby groupings."""
+async def rescan(
+    session: AsyncSession, *, now: datetime | None = None, full: bool = False,
+) -> dict:
+    """Score recent non-synthetic posts against nearby groupings.
+
+    `full=True` scores EVERY post by an artist who has Discord drops at all —
+    the manual button's job, for history that predates the feature or that a
+    trickle merge (#4390) has just rearranged. It used to share the sweep's
+    48-hour horizon, so the button described as "a first run over a library
+    that predates the feature" could not reach that library.
+    """
     settings = await ImportSettings.load(session)
     if not settings.discord_link_enabled:
         return {"enabled": False, "scanned": 0, "proposed": 0, "linked": 0}
@@ -586,6 +661,18 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
     # a full-library rescan is the manual button's job, not the sweep's.
     horizon = now - timedelta(hours=window_hours * 2)
     sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
+    if full:
+        with_drops = select(Post.artist_id).where(
+            Post.synthesized_by == DROP_GROUPER
+        ).distinct()
+        ids = set((await session.execute(
+            select(Post.id).where(
+                Post.synthesized_by.is_(None),
+                Post.absorbed_by_post_id.is_(None),
+                Post.artist_id.in_(with_drops),
+            )
+        )).scalars().all())
+        return await _score(session, settings, ids, window_hours)
     ids = set((await session.execute(
         select(Post.id).where(
             Post.synthesized_by.is_(None),
@@ -605,14 +692,28 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
     # That is #4392's third cause, and it is the one that left a measured 0.800
     # pair with an empty review queue on the live instance. The other two were
     # about scoring; this one meant nothing was scored at all.
-    drop_times = (await session.execute(
-        select(sort_key).where(
+    #
+    # The times are the drops' MESSAGES, not the drops' own dates: a drop that
+    # grew today by a trickle merge (#4390) is dated by its first stage, days
+    # earlier, while the teaser sits beside the message that just joined.
+    recent = (
+        select(Post.id).where(
             Post.synthesized_by == DROP_GROUPER,
             func.coalesce(Post.last_grew_at, Post.downloaded_at) >= horizon,
         )
         .order_by(func.coalesce(Post.last_grew_at, Post.downloaded_at).desc())
         .limit(MAX_RECENT_DROPS)
-    )).scalars().all()
+    )
+    member = Post.__table__.alias("member")
+    drop_times = set((await session.execute(
+        select(func.coalesce(member.c.post_date, member.c.downloaded_at))
+        .where(member.c.absorbed_by_post_id.in_(recent))
+    )).scalars().all())
+    # Plus each drop's own date — its first message's, so a duplicate for a
+    # real drop, but what a drop with no member rows is placed by.
+    drop_times |= set((await session.execute(
+        select(sort_key).where(Post.id.in_(recent))
+    )).scalars().all())
     # The interval arithmetic is done in Python rather than SQL: a handful of
     # literal ranges is portable, and `now - INTERVAL` is not.
     ranges = [
@@ -629,6 +730,12 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
             )
         )).scalars().all())
 
+    return await _score(session, settings, ids, window_hours)
+
+
+async def _score(
+    session: AsyncSession, settings: ImportSettings, ids: set[int], window_hours: float,
+) -> dict:
     svc = PostAssociationService(session)
     proposed = 0
     linked = 0

tests/test_post_association.py

@@ -950,3 +950,156 @@ def test_the_duplicate_threshold_sits_inside_the_measured_gap():
     justifies it: 20 bits was the widest true pair, 108 the nearest unrelated
     one."""
     assert 20 < DUPLICATE_MAX_DISTANCE < 108
+
+
+# --- drops shaped as the grouper actually writes them ----------------------
+#
+# Every test above hand-builds a drop that OWNS its images. discord_grouping
+# never does that: a drop's images belong to its member messages and the drop
+# claims them through provenance. Against that shape the matcher saw no names
+# and no hashes at all, so the identity route could not fire on the live
+# instance — and nothing here could have noticed.
+
+
+async def _real_drop(db, artist, discord, *, stages):
+    """Messages, each owning one named image, grouped by the real grouper and
+    merged by the real trickle pass. `stages` is [(at, name), ...]."""
+    from backend.app.services.discord_grouping import group_source, merge_trickles
+
+    for i, (at, name) in enumerate(stages):
+        msg = Post(source_id=discord.id, artist_id=artist.id,
+                   external_post_id=f"{artist.slug}-msg-{i}", post_date=at)
+        db.add(msg)
+        await db.flush()
+        vec = [0.0] * 1152
+        vec[i % 1152] = 1.0
+        db.add(ImageRecord(
+            path=f"/images/{artist.slug}/20260901_{1234567890000 + i}_01_{name}.png",
+            sha256=f"{artist.id:08d}{i:056d}", size_bytes=10, mime="image/png",
+            width=10, height=10, origin="downloaded", primary_post_id=msg.id,
+            artist_id=artist.id, siglip_embedding=vec,
+        ))
+    await db.flush()
+    await group_source(db, discord, max_distance=0.10, window_minutes=60)
+    await merge_trickles(db, discord, gap=timedelta(hours=168), min_images=2,
+                         cooldown=timedelta(hours=24))
+    await db.commit()
+    return (await db.execute(
+        select(Post).where(Post.source_id == discord.id,
+                           Post.synthesized_by == DROP_GROUPER,
+                           Post.absorbed_by_post_id.is_(None))
+    )).scalars().all()
+
+
+@pytest.mark.asyncio
+async def test_a_drop_whose_images_belong_to_its_messages_still_links_on_its_name(db):
+    """The live shape. `0-k` on the teaser and on the drop's MESSAGE, nowhere
+    else — conclusive, so FC links it without asking."""
+    artist, patreon, discord = await _artist_with_channels(db, "liveshape")
+    now = datetime.now(UTC) - timedelta(days=5)
+    (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")])
+    teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=20),
+                           body="new one", names=["0-k"])
+    await db.commit()
+
+    made = await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True,
+    )
+    await db.commit()
+
+    assert made == (1, 1)
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    assert (assoc.payload_post_id, assoc.signals["identity_token"]) == (drop.id, "0-k")
+
+
+@pytest.mark.asyncio
+async def test_a_merged_trickle_is_found_by_its_latest_stage(db):
+    """A trickle merged by #4390 is dated by its FIRST stage — four days before
+    the release the teaser announces. Matching on the drop's own date would put
+    it outside the 24h window."""
+    artist, patreon, discord = await _artist_with_channels(db, "trickleshape")
+    start = datetime.now(UTC) - timedelta(days=10)
+    (drop,) = await _real_drop(db, artist, discord, stages=[
+        (start, "svtt_wip1"), (start + timedelta(days=2), "svtt_wip3"),
+        (start + timedelta(days=4), "svtt_drench_b"),
+    ])
+    teaser = await _teaser(db, artist, patreon, at=start + timedelta(days=4, hours=1),
+                           body="new one", names=["svtt_teaser"])
+    await db.commit()
+
+    made = await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True,
+    )
+    await db.commit()
+
+    assert made[0] == 1
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    assert assoc.payload_post_id == drop.id
+    # Timed against the closest message, an hour away — not the first, four
+    # days away.
+    assert assoc.signals["proximity"] > 0.9
+
+
+@pytest.mark.asyncio
+async def test_a_pair_left_pending_is_linked_once_the_evidence_is_conclusive(db):
+    """Nobody decided a pending pair, so the matcher re-scores it. Otherwise a
+    pair queued by an older, weaker matcher waits for a click forever."""
+    artist, patreon, discord = await _artist_with_channels(db, "pendingshape")
+    now = datetime.now(UTC) - timedelta(days=5)
+    (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")])
+    teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=2),
+                           body="new one", names=["0-k"])
+    db.add(PostAssociation(announcement_post_id=teaser.id, payload_post_id=drop.id,
+                           score=0.61, signals={"proximity": 0.9}, status="pending"))
+    await db.commit()
+
+    await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True,
+    )
+    await db.commit()
+
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    assert (assoc.status, assoc.linked_by) == ("linked", "fc")
+
+
+@pytest.mark.asyncio
+async def test_a_dismissed_pair_is_never_rescored(db):
+    artist, patreon, discord = await _artist_with_channels(db, "dismissedshape")
+    now = datetime.now(UTC) - timedelta(days=5)
+    (drop,) = await _real_drop(db, artist, discord, stages=[(now, "0-k_base")])
+    teaser = await _teaser(db, artist, patreon, at=now + timedelta(hours=2),
+                           body="new one", names=["0-k"])
+    db.add(PostAssociation(announcement_post_id=teaser.id, payload_post_id=drop.id,
+                           score=0.61, signals={}, status="dismissed"))
+    await db.commit()
+
+    assert await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW, auto_link=True,
+    ) == (0, 0)
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    assert assoc.status == "dismissed"
+
+
+@pytest.mark.asyncio
+async def test_the_full_rescan_reaches_history_the_sweep_does_not(db):
+    """The "Scan now" button. It used to share the sweep's 48-hour horizon, so
+    it could not reach the library it was described as being for."""
+    artist, patreon, discord = await _artist_with_channels(db, "historyshape")
+    long_ago = datetime.now(UTC) - timedelta(days=400)
+    (drop,) = await _real_drop(db, artist, discord, stages=[(long_ago, "0-k_base")])
+    # Authored long ago too. A drop FC wrote TODAY is exactly what the sweep
+    # follows back to its teaser (#4392's third cause), so it would find this
+    # pair without the button.
+    drop.downloaded_at = long_ago
+    await _teaser(db, artist, patreon, at=long_ago + timedelta(hours=2),
+                  body="new one", names=["0-k"])
+    settings = await ImportSettings.load(db)
+    settings.discord_link_enabled = True
+    await db.commit()
+
+    assert (await rescan(db))["proposed"] == 0
+    await db.commit()
+    result = await rescan(db, full=True)
+    await db.commit()
+
+    assert (result["proposed"], result["linked"]) == (1, 1)