fix: a sweep follows the drops FC has just authored, not only recent posts (4392)

#4392's third cause, and the only one of the three about whether a pair is
SCORED AT ALL rather than how well.

A drop's post_date is backdated to its first message, but FC cannot author the
drop until that message has an embedding and the hourly grouper has run. So a
drop created this minute lands wherever its messages were — days or weeks back
in the feed. The sweep only looked at announcements published within twice the
window of NOW, so by the time the drop existed its neighbours were already
outside the horizon, and nothing brought the sweep back to them.

Measured on the live instance: a pair scoring 0.800 with `associations: []`
and an empty queue. The two scoring fixes in the previous commit would not
have helped it, because nothing scored it.

So the sweep now also gathers announcements sitting beside any drop whose
`last_grew_at`/`downloaded_at` is inside the horizon — the drop's own clock
rather than its backdated position. Capped at MAX_RECENT_DROPS so a backfill
authoring thousands at once does not quietly become the full-library rescan
the manual button exists for, and the id set is sorted before the loop because
a sweep visiting posts in a different order each run is one whose failures
cannot be reproduced.

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

backend/app/services/post_association_service.py

@@ -77,7 +77,7 @@ from collections import Counter
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 
-from sqlalchemy import func, or_, select
+from sqlalchemy import and_, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..models import ImageRecord, ImportSettings, Post, PostAssociation
@@ -123,6 +123,12 @@ DECLARED_MENTION = 0.6
 
 MAX_CANDIDATES = 25
 
+# How many just-grouped drops one sweep will look around. A ceiling on the
+# `or_` the sweep builds, not a policy — a backfill that authors thousands of
+# drops at once should not turn one sweep into a full-library rescan, which is
+# the manual button's job.
+MAX_RECENT_DROPS = 200
+
 
 @dataclass(frozen=True)
 class _Corpus:
@@ -410,21 +416,60 @@ async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
 
     now = now or datetime.now(UTC)
     window_hours = float(settings.discord_link_window_hours)
+    window = timedelta(hours=window_hours)
     # Only look at announcements that could still have a partner in range —
     # 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)
-    ids = (await session.execute(
+    ids = set((await session.execute(
         select(Post.id).where(
             Post.synthesized_by.is_(None),
             Post.absorbed_by_post_id.is_(None),
             sort_key >= horizon,
         )
+    )).scalars().all())
+
+    # ...and announcements sitting next to a drop FC has only JUST authored.
+    #
+    # A drop's `post_date` is backdated to its first message, but FC cannot
+    # write the drop until the message has an embedding and the hourly grouper
+    # has run — so a drop created this minute can land weeks back in the feed.
+    # Its neighbours were last swept before it existed, and a sweep keyed only
+    # on how recent the ANNOUNCEMENT is will never look at them again.
+    #
+    # 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(
+            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()
+    # The interval arithmetic is done in Python rather than SQL: a handful of
+    # literal ranges is portable, and `now - INTERVAL` is not.
+    ranges = [
+        and_(sort_key >= at - window, sort_key <= at + window)
+        for at in drop_times
+        if at is not None
+    ]
+    if ranges:
+        ids |= set((await session.execute(
+            select(Post.id).where(
+                Post.synthesized_by.is_(None),
+                Post.absorbed_by_post_id.is_(None),
+                or_(*ranges),
+            )
+        )).scalars().all())
 
     svc = PostAssociationService(session)
     proposed = 0
-    for pid in ids:
+    # Sorted because `ids` is now a union of two queries: set iteration order
+    # is arbitrary, and a sweep that visits posts in a different order each
+    # run is one whose failures cannot be reproduced.
+    for pid in sorted(ids):
         proposed += await svc.match_post(
             pid,
             threshold=float(settings.discord_link_threshold),