feat: link the Patreon teaser to the Discord drop it announced (388 step E5)

The point of the milestone rather than its tail. Two of the operator's artists
post a deliberately cropped fragment on Patreon to signal that the real thing
has landed in their Discord; this proposes those pairs.

Confirm-only, following the FC-6.3 series matcher. A wrongly-asserted
association tells the operator two different pieces are one, which is strictly
worse than no link: no link leaves them where they already were, a wrong one
actively misinforms and then propagates into whatever reads it. So the
matcher's job is a SHORT list worth reading, not a long list worth trusting.

**The threshold sits above every single signal weight, and that is the
design.** Proximity is 0.55, declaration 0.45, the cut 0.60 — so neither
signal can carry a pair alone. That makes "time proximity alone is never
sufficient" an arithmetic property rather than an aspiration: on a busy day an
artist posts several times, and a matcher that could pair on proximity alone
would turn every one of those days into false pairs until the review queue got
abandoned. A guard test asserts the relationship against WEIGHTS directly, so
it survives any refactor of the scorer, and says in its own failure message not
to fix it by lowering the assertion.

**Crop-to-source matching is HELD, on the plan's instruction** — real work with
real false-positive risk, worth building only once signals 1 and 2 are shown
insufficient against the operator's actual artists. Worth stating: a naive
whole-image SigLIP similarity is NOT that signal. A cropped teaser and its full
version are precisely the pair a whole-image comparison handles worst, so
adding one as a "bonus" would mostly add noise while looking like progress.

Two premises in the plan corrected in the building:

* **E4 is not actually a prerequisite.** A Patreon Source and a Discord Source
  the operator has added under one Artist already share `Post.artist_id`, and
  the synthetic grouping inherits it. E4 EXTENDS this to creators FC has to
  learn the association for; it is not needed to represent one FC was told.
  Same-artist is then a hard filter, not a scored signal — two different
  creators posting minutes apart is a coincidence, not evidence.
* **`link_extract` cannot supply the declaration signal.** It exists, but
  `SUPPORTED_HOSTS` is file hosts only and `host_for()` returns None for a
  Discord URL, so no ExternalLink row is ever written for one. The signal
  reads the post body directly instead.

And a bug my own test would have caught: `declared_signal` stripped the HTML
before looking for an invite, but `html_to_plain` discards attributes and
these creators put the invite in an anchor's `href` — so the strongest form of
the signal was being thrown away, leaving only whatever the link text said.
The invite now matches the raw body; the bare mention still matches stripped
text, so `\bdiscord\b` is tested against prose rather than against markup.

Dismissed rows are kept, not deleted: the row is what remembers the rejection,
and re-proposing a rejected pair on every scan is the one behaviour that makes
a review queue get ignored. Both FKs CASCADE, so E3's one-DELETE reversal
cannot leave a proposal pointing at a post that no longer exists.

Only ACCEPTED links reach the post payload. A pending proposal is a question
for the review queue, not a claim to render beside the artwork.

UI (rule 27) follows in the next commit.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LNXXULQDjVZmbuNa2G9mD9

alembic/versions/0094_post_association.py

@@ -0,0 +1,124 @@
+"""post_association — "this Patreon post announced that Discord drop".
+
+Milestone 388, step E5.
+
+Two of the operator's artists post a deliberately cropped fragment on Patreon
+to signal that the real thing has landed in their Discord. This table holds the
+proposed and accepted links between the announcement and the drop.
+
+Directional and confirm-only. The pair is asymmetric (the teaser announces the
+drop, not the reverse), the two posts are never merged (the creator published
+twice, deliberately — flattening that hides the behaviour being modelled), and
+nothing is linked until the operator accepts, following the FC-6.3 series
+matcher. A wrongly-asserted association tells them two different pieces are
+one, which is worse than no link at all.
+
+Dismissed rows are KEPT. The row is what remembers the rejection, and
+re-proposing a rejected pair on every scan is what makes a review queue get
+ignored.
+
+Revision ID: 0094
+Revises: 0093
+Create Date: 2026-09-10
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0094"
+down_revision: Union[str, None] = "0093"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "post_association",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("announcement_post_id", sa.Integer(), nullable=False),
+        sa.Column("payload_post_id", sa.Integer(), nullable=False),
+        sa.Column("score", sa.Float(), nullable=False),
+        sa.Column("signals", sa.JSON(), nullable=True),
+        # No CHECK on status (rule 36 considered and declined), matching
+        # series_suggestion.status — the same review-queue vocabulary, and the
+        # same check-existing-enums lesson.
+        sa.Column(
+            "status", sa.String(length=16), server_default="pending", nullable=False,
+        ),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True),
+            server_default=sa.text("now()"), nullable=False,
+        ),
+        sa.Column(
+            "updated_at", sa.DateTime(timezone=True),
+            server_default=sa.text("now()"), nullable=False,
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_post_association")),
+        # CASCADE on both sides: an association to a post that no longer exists
+        # is not a fact worth keeping, and E3's reversal path (delete the
+        # grouping) must not leave a dangling proposal behind.
+        sa.ForeignKeyConstraint(
+            ["announcement_post_id"], ["post.id"], ondelete="CASCADE",
+            name=op.f("fk_post_association_announcement_post_id_post"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["payload_post_id"], ["post.id"], ondelete="CASCADE",
+            name=op.f("fk_post_association_payload_post_id_post"),
+        ),
+        sa.UniqueConstraint(
+            "announcement_post_id", "payload_post_id",
+            name="uq_post_association_pair",
+        ),
+    )
+    op.create_index(
+        op.f("ix_post_association_announcement_post_id"),
+        "post_association", ["announcement_post_id"],
+    )
+    op.create_index(
+        op.f("ix_post_association_payload_post_id"),
+        "post_association", ["payload_post_id"],
+    )
+    op.create_index(
+        op.f("ix_post_association_status"), "post_association", ["status"],
+    )
+
+    op.add_column(
+        "import_settings",
+        sa.Column(
+            "discord_link_enabled", sa.Boolean(),
+            server_default="true", nullable=False,
+        ),
+    )
+    # 0.60 sits ABOVE the largest single signal weight on purpose — see
+    # post_association_service.WEIGHTS. That is what makes "time proximity
+    # alone is never sufficient" arithmetic rather than aspirational.
+    op.add_column(
+        "import_settings",
+        sa.Column(
+            "discord_link_threshold", sa.Float(),
+            server_default="0.60", nullable=False,
+        ),
+    )
+    op.add_column(
+        "import_settings",
+        sa.Column(
+            "discord_link_window_hours", sa.Float(),
+            server_default="24", nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("import_settings", "discord_link_window_hours")
+    op.drop_column("import_settings", "discord_link_threshold")
+    op.drop_column("import_settings", "discord_link_enabled")
+    op.drop_index(op.f("ix_post_association_status"), table_name="post_association")
+    op.drop_index(
+        op.f("ix_post_association_payload_post_id"), table_name="post_association",
+    )
+    op.drop_index(
+        op.f("ix_post_association_announcement_post_id"), table_name="post_association",
+    )
+    op.drop_table("post_association")

backend/app/api/posts.py

@@ -5,6 +5,8 @@ from quart import Blueprint, jsonify, request
 from ..extensions import get_session
 from ..models import ImportSettings, Post
 from ..services import interpreter_client as ic
+from ..services.post_association_service import PostAssociationService
+from ..services.post_association_service import rescan as association_rescan
 from ..services.post_feed_service import PostFeedService
 from ..services.source_service import KNOWN_PLATFORMS
 from ..utils.text import html_to_plain
@@ -165,3 +167,46 @@ async def set_translation_override(post_id: int):
             "translated_source_lang": post.translated_source_lang,
             "applied": applied,
         })
+
+
+# --- #388 E5: the announcement review queue -------------------------------
+#
+# Confirm-only, following the series-suggestion routes (api/tags.py). Nothing
+# here links anything on its own: the matcher proposes, the operator decides.
+
+
+@posts_bp.route("/associations", methods=["GET"])
+async def list_associations():
+    async with get_session() as session:
+        return jsonify({"items": await PostAssociationService(session).list_pending()})
+
+
+@posts_bp.route("/associations/<int:association_id>/accept", methods=["POST"])
+async def accept_association(association_id: int):
+    async with get_session() as session:
+        result = await PostAssociationService(session).accept(association_id)
+        if result is None:
+            return _bad("association not found", 404)
+        await session.commit()
+    return jsonify(result)
+
+
+@posts_bp.route("/associations/<int:association_id>/dismiss", methods=["POST"])
+async def dismiss_association(association_id: int):
+    async with get_session() as session:
+        result = await PostAssociationService(session).dismiss(association_id)
+        if result is None:
+            return _bad("association not found", 404)
+        await session.commit()
+    return jsonify(result)
+
+
+@posts_bp.route("/associations/rescan", methods=["POST"])
+async def rescan_associations():
+    """Manual re-scan. The beat sweep only looks at recent posts (a pair has to
+    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)
+        await session.commit()
+    return jsonify(result)

backend/app/api/settings.py

@@ -39,6 +39,10 @@ _EDITABLE_FIELDS = (
     "download_failure_warning_threshold",
     "series_suggest_enabled",
     "series_suggest_threshold",
+    # #388 E5 — the announcement matcher (Patreon teaser ↔ Discord drop).
+    "discord_link_enabled",
+    "discord_link_threshold",
+    "discord_link_window_hours",
     "extdl_mega_enabled",
     "extdl_gdrive_enabled",
     "extdl_mediafire_enabled",
@@ -150,6 +154,22 @@ async def update_import_settings():
             return jsonify(
                 {"error": "series_suggest_threshold must be a number in [0, 1]"}
             ), 400
+    if "discord_link_enabled" in body and not isinstance(
+        body["discord_link_enabled"], bool
+    ):
+        return jsonify({"error": "discord_link_enabled must be a boolean"}), 400
+    if "discord_link_threshold" in body:
+        v = body["discord_link_threshold"]
+        if not isinstance(v, (int, float)) or isinstance(v, bool) or v < 0 or v > 1:
+            return jsonify(
+                {"error": "discord_link_threshold must be a number in [0, 1]"}
+            ), 400
+    if "discord_link_window_hours" in body:
+        v = body["discord_link_window_hours"]
+        if not isinstance(v, (int, float)) or isinstance(v, bool) or v <= 0:
+            return jsonify(
+                {"error": "discord_link_window_hours must be a positive number"}
+            ), 400
     if "wip_title_tagging_enabled" in body and not isinstance(
         body["wip_title_tagging_enabled"], bool
     ):

backend/app/celery_app.py

@@ -207,6 +207,13 @@ def make_celery() -> Celery:
                 # (#388 E2), so this sweep is what picks up a drop once its
                 # vectors have caught up. No-op unless discord_grouping_enabled.
             },
+            "match-post-associations-hourly": {
+                "task": "backend.app.tasks.maintenance.match_post_associations",
+                "schedule": 3600.0,  # hourly, and AFTER the grouper's own cadence
+                # by construction: a pair cannot be proposed until the drop it
+                # points at exists as a grouping (#388 E5). No-op unless
+                # discord_link_enabled.
+            },
             "integrity-verify-weekly": {
                 "task": "backend.app.tasks.maintenance.verify_integrity",
                 "schedule": 604800.0,  # weekly

backend/app/models/__init__.py

@@ -28,6 +28,7 @@ from .pixiv_failed_media import PixivFailedMedia
 from .pixiv_seen_media import PixivSeenMedia
 from .platform_membership import PlatformMembership
 from .post import Post
+from .post_association import PostAssociation
 from .post_attachment import PostAttachment, attachment_download_url
 from .presentation_review import PresentationReview
 from .series_chapter import SeriesChapter
@@ -59,6 +60,7 @@ __all__ = [
     "SubscribeStarFailedMedia",
     "SubscribeStarSeenMedia",
     "Post",
+    "PostAssociation",
     "PostAttachment",
     "attachment_download_url",
     "PresentationReview",

backend/app/models/import_settings.py

@@ -97,6 +97,33 @@ class ImportSettings(Base):
         server_default="0.5",
     )
 
+    # Milestone 388 E5 — the announcement matcher: "this Patreon post announced
+    # that Discord drop". Lives here rather than in MLSettings, with the series
+    # matcher it is modelled on, because it runs no inference: the signals are
+    # time proximity and whether the post says so.
+    discord_link_enabled: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True,
+        server_default="true",
+    )
+    # The weighted-score cut-off. 0.60 is not arbitrary: it is deliberately set
+    # ABOVE the largest single signal weight, which is what makes "time
+    # proximity alone must never be sufficient" an ARITHMETIC property rather
+    # than a hope. On a busy day an artist posts several times; if proximity
+    # could carry a pair by itself, every one of those days would produce false
+    # pairs and the review queue would be abandoned. See
+    # post_association_service.WEIGHTS — a guard test pins the relationship.
+    discord_link_threshold: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.60,
+        server_default="0.60",
+    )
+    # How far apart the announcement and the drop may be. The Patreon post
+    # exists IN ORDER TO announce the drop, so they are minutes-to-hours apart;
+    # a day is generous and still excludes "same week".
+    discord_link_window_hours: Mapped[float] = mapped_column(
+        Float, nullable=False, default=24.0,
+        server_default="24",
+    )
+
     # #830 off-platform file-host downloads — per-host enable lever (default on,
     # rule #26). Column names are extdl_<host>_enabled so the worker reads them
     # via getattr(settings, f"extdl_{host}_enabled", True).

backend/app/models/post_association.py

@@ -0,0 +1,100 @@
+"""PostAssociation — "this Patreon post announced that Discord drop".
+
+Milestone 388, step E5, and the point of the milestone rather than its tail.
+
+Two of the operator's artists post a deliberately CROPPED fragment on Patreon
+to signal that the real thing has landed in their Discord. The Patreon post is
+the announcement; the Discord grouping (milestone 388 E2) is the payload. This
+row is the link between them.
+
+## Directional, and NOT a merge
+
+`announcement` → `payload` is asymmetric on purpose. The teaser announces the
+drop; the drop does not announce the teaser, and a symmetric "related posts"
+edge would lose the only thing that makes the pair interesting.
+
+Nor are the two collapsed into one post. The creator published twice,
+deliberately, on two platforms with different audiences — flattening that
+hides the very behaviour being modelled, and would destroy the operator's
+ability to see that the Patreon post is a teaser at all.
+
+## Confirm-only, following FC-6.3 (task 737)
+
+`status` starts at `pending` and nothing is linked until the operator accepts.
+A wrongly-asserted association tells them two different pieces are one, which
+is worse than no link: no link leaves them where they already are, a wrong one
+actively misinforms. Same reason the series matcher writes to a review queue
+instead of filing posts on its own.
+
+`status` is a plain String, no CHECK — matching `series_suggestion.status`,
+which records the same check-existing-enums lesson.
+
+## Why pHash could not do this, and the correction matters
+
+The original plan claimed this link was already sitting in `image_provenance`
+via pHash dedup. It is not. `compute_phash` is `imagehash.phash` at
+`hash_size=8` — a DCT hash over the WHOLE image, robust to rescaling and
+recompression but NOT to cropping, because a crop changes the global
+signature. Cross-platform provenance still links straight re-posts; it does
+nothing for a cropped teaser and its full version, which is precisely the pair
+the operator described. Hence a scored proposal rather than a lookup.
+"""
+
+from datetime import datetime
+
+from sqlalchemy import (
+    JSON,
+    DateTime,
+    Float,
+    ForeignKey,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .base import Base
+
+
+class PostAssociation(Base):
+    __tablename__ = "post_association"
+    __table_args__ = (
+        UniqueConstraint(
+            "announcement_post_id", "payload_post_id",
+            name="uq_post_association_pair",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True)
+    # The teaser — a real post the creator wrote (Patreon, today).
+    announcement_post_id: Mapped[int] = mapped_column(
+        ForeignKey("post.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+    # What it announced — a synthetic Discord grouping, today. CASCADE on both
+    # sides: an association to a post that no longer exists is not a fact worth
+    # keeping, and E3's reversal path (delete the grouping) should not leave a
+    # dangling proposal behind.
+    payload_post_id: Mapped[int] = mapped_column(
+        ForeignKey("post.id", ondelete="CASCADE"), nullable=False, index=True
+    )
+
+    score: Mapped[float] = mapped_column(Float, nullable=False)
+    # Per-signal strengths as scored, so a proposal stays explicable after the
+    # weights or the threshold are tuned. Without it, "why was this suggested"
+    # is unanswerable the moment anything moves.
+    signals: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    # pending | linked | dismissed. A DISMISSED row is kept, not deleted — it
+    # is what stops the matcher proposing the same rejected pair on every
+    # subsequent scan, which is the behaviour that makes a review queue
+    # unusable.
+    status: Mapped[str] = mapped_column(
+        String(16), nullable=False, server_default="pending", index=True
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False,
+        server_default=func.now(), onupdate=func.now(),
+    )

backend/app/services/post_association_service.py

@@ -0,0 +1,308 @@
+"""The announcement matcher: which Patreon post announced which Discord drop.
+
+Milestone 388, step E5.
+
+Two of the operator's artists post a deliberately CROPPED fragment on Patreon
+to signal that the real thing has landed in their Discord. This service
+proposes those pairs, and proposes only — the operator accepts or dismisses,
+following the FC-6.3 series matcher (task 737) rather than linking on its own.
+
+## Why confirm-only is not caution for its own sake
+
+A wrongly-asserted association tells the operator that two different pieces are
+one. That is strictly worse than no link at all: no link leaves them exactly
+where they already were, a wrong one actively misinforms and then propagates
+into whatever reads the association. So the matcher's job is to make a SHORT
+list worth reading, not a long list worth trusting.
+
+## Signals, and the one deliberately NOT built
+
+1. **Time proximity.** The Patreon post exists in order to announce the drop,
+   so the two are minutes-to-hours apart. Nearly free, and strong.
+2. **The post says so.** These announcements routinely name Discord or carry
+   an invite link, which is close to a declaration.
+
+3. **Crop-to-source matching is HELD, on the plan's own instruction** — it is
+   real work with real false-positive risk, and it is only worth building once
+   1 and 2 are shown to be insufficient against the operator's actual artists.
+   Nothing here should be read as evidence it is unnecessary; it is deferred,
+   and the thing that would justify it is an empty review queue on a pair the
+   operator can see with their own eyes.
+
+   Note also that a naive whole-image SigLIP similarity is NOT that signal. A
+   cropped teaser and its full version are exactly the pair a whole-image
+   comparison handles worst, so adding one as a "bonus" would mostly add noise
+   while looking like progress.
+
+## Creator identity comes free, so E4 is not actually a prerequisite
+
+The plan listed E4 (creator identity across the two channels) as a dependency.
+It is not one for the pairs that matter today: a Patreon `Source` and a Discord
+`Source` the operator has added under the same `Artist` already share
+`Post.artist_id`, and the synthetic grouping inherits it (E2). E4 EXTENDS this
+to creators whose association FC has to learn rather than being told; it is not
+needed to represent an association FC already knows.
+
+## A correction the plan carried, worth restating
+
+`link_extract.py` does exist and does capture off-platform links — but only
+file hosts (`SUPPORTED_HOSTS` is mega/gdrive/mediafire/dropbox/pixeldrain).
+`host_for()` returns None for a Discord URL, so no `ExternalLink` row is ever
+written for one. The declaration signal therefore reads the post body itself
+rather than the extracted-links table the plan assumed it could use.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import ImportSettings, Post, PostAssociation
+from ..utils.text import html_to_plain
+from .discord_grouping import DROP_GROUPER
+
+log = logging.getLogger(__name__)
+
+# Additive weights, summing to 1.0. Kept as constants rather than settings —
+# the sensitivity knob that matters is the threshold, and per-signal weights
+# are an over-tune (same call as series_match_service.WEIGHTS).
+#
+# THE RELATIONSHIP TO THE THRESHOLD IS THE DESIGN. No single weight may reach
+# the default threshold, which is what makes "time proximity alone is never
+# enough" arithmetic rather than aspirational: on a busy day an artist posts
+# several times, and a matcher that could pair on proximity alone would turn
+# every busy day into false pairs. A guard test pins this.
+WEIGHTS = {"proximity": 0.55, "declared": 0.45}
+
+# A Discord INVITE in the body is close to a declaration; the bare word is
+# weaker but still meaningful, because these posts are short and on-topic.
+_INVITE = re.compile(r"discord\.(?:gg|com/invite)/", re.I)
+_MENTION = re.compile(r"\bdiscord\b", re.I)
+
+DECLARED_INVITE = 1.0
+DECLARED_MENTION = 0.6
+
+MAX_CANDIDATES = 25
+
+
+def proximity_signal(gap: timedelta, window: timedelta) -> float:
+    """1.0 when the two posts are simultaneous, decaying linearly to 0 at the
+    window's edge. Linear rather than a step, so a pair an hour outside a
+    hand-tuned window degrades instead of vanishing."""
+    if window <= timedelta(0):
+        return 0.0
+    seconds = abs(gap.total_seconds())
+    if seconds >= window.total_seconds():
+        return 0.0
+    return round(1.0 - (seconds / window.total_seconds()), 4)
+
+
+def declared_signal(description: str | None) -> float:
+    """Does the announcement say, in its own body, that this is about Discord?
+
+    The invite is matched against the RAW body and the bare mention against the
+    stripped text, which is not fussiness — post bodies are HTML, and these
+    creators put the invite in an anchor's `href`. `html_to_plain` discards
+    attributes, so stripping first would have thrown away the strongest form of
+    the signal and left only whatever the link text happened to say.
+
+    The mention still reads stripped text, so `\bdiscord\b` is matched against
+    prose rather than against markup and URLs, where it would fire on any link
+    that merely passes through a discord domain.
+    """
+    if not description:
+        return 0.0
+    if _INVITE.search(description):
+        return DECLARED_INVITE
+    if _MENTION.search(html_to_plain(description) or ""):
+        return DECLARED_MENTION
+    return 0.0
+
+
+def weighted_score(signals: dict) -> float:
+    return round(sum(WEIGHTS[k] * signals.get(k, 0.0) for k in WEIGHTS), 4)
+
+
+def _post_time(post: Post) -> datetime:
+    return post.post_date or post.downloaded_at
+
+
+class PostAssociationService:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    async def _decided(self, announcement_id: int) -> set[int]:
+        """Payload posts already proposed for this announcement, in ANY status.
+
+        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.
+        """
+        rows = (await self.session.execute(
+            select(PostAssociation.payload_post_id)
+            .where(PostAssociation.announcement_post_id == announcement_id)
+        )).scalars().all()
+        return set(rows)
+
+    async def _candidate_groups(
+        self, announcement: Post, *, window: timedelta,
+    ) -> list[Post]:
+        """Synthetic Discord groupings by the SAME artist, inside the window.
+
+        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.
+        """
+        at = _post_time(announcement)
+        sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
+        return (await self.session.execute(
+            select(Post)
+            .where(
+                Post.artist_id == announcement.artist_id,
+                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()
+
+    async def match_post(
+        self, announcement_id: int, *, threshold: float, window_hours: float,
+    ) -> int:
+        """Score one announcement against nearby groupings. Returns proposals made."""
+        announcement = await self.session.get(Post, announcement_id)
+        if announcement is None or announcement.synthesized_by is not None:
+            # A synthetic post cannot announce anything — FC wrote it.
+            return 0
+
+        window = timedelta(hours=window_hours)
+        declared = declared_signal(announcement.description)
+        already = await self._decided(announcement_id)
+
+        made = 0
+        for group in await self._candidate_groups(announcement, window=window):
+            if group.id in already:
+                continue
+            signals = {
+                "proximity": proximity_signal(
+                    _post_time(group) - _post_time(announcement), window,
+                ),
+                "declared": declared,
+            }
+            score = weighted_score(signals)
+            if score < threshold:
+                continue
+            self.session.add(PostAssociation(
+                announcement_post_id=announcement.id,
+                payload_post_id=group.id,
+                score=score,
+                signals=signals,
+                status="pending",
+            ))
+            made += 1
+        return made
+
+    async def list_pending(self) -> list[dict]:
+        rows = (await self.session.execute(
+            select(PostAssociation)
+            .where(PostAssociation.status == "pending")
+            .order_by(PostAssociation.score.desc(), PostAssociation.id.desc())
+        )).scalars().all()
+        return [
+            {
+                "id": a.id,
+                "announcement_post_id": a.announcement_post_id,
+                "payload_post_id": a.payload_post_id,
+                "score": a.score,
+                "signals": a.signals,
+            }
+            for a in rows
+        ]
+
+    async def accept(self, association_id: int) -> dict | None:
+        a = await self.session.get(PostAssociation, association_id)
+        if a is None:
+            return None
+        a.status = "linked"
+        return {"id": a.id, "status": a.status}
+
+    async def dismiss(self, association_id: int) -> dict | None:
+        a = await self.session.get(PostAssociation, association_id)
+        if a is None:
+            return None
+        # Kept, not deleted — the row is what remembers the rejection.
+        a.status = "dismissed"
+        return {"id": a.id, "status": a.status}
+
+    async def linked_for(self, post_ids: list[int]) -> dict[int, list[dict]]:
+        """Accepted links touching these posts, keyed by post id, BOTH ways.
+
+        A post is either end of the relationship, and each end wants the other
+        one: the teaser wants "the full set is over here", the grouping wants
+        "this is what announced me". One query, both directions.
+        """
+        if not post_ids:
+            return {}
+        rows = (await self.session.execute(
+            select(PostAssociation).where(
+                PostAssociation.status == "linked",
+                or_(
+                    PostAssociation.announcement_post_id.in_(post_ids),
+                    PostAssociation.payload_post_id.in_(post_ids),
+                ),
+            )
+        )).scalars().all()
+        out: dict[int, list[dict]] = {}
+        for a in rows:
+            if a.announcement_post_id in post_ids:
+                out.setdefault(a.announcement_post_id, []).append(
+                    {"role": "announces", "post_id": a.payload_post_id, "id": a.id}
+                )
+            if a.payload_post_id in post_ids:
+                out.setdefault(a.payload_post_id, []).append(
+                    {"role": "announced_by", "post_id": a.announcement_post_id, "id": a.id}
+                )
+        return out
+
+
+async def rescan(session: AsyncSession, *, now: datetime | None = None) -> dict:
+    """Score every recent non-synthetic post against nearby groupings."""
+    settings = await ImportSettings.load(session)
+    if not settings.discord_link_enabled:
+        return {"enabled": False, "scanned": 0, "proposed": 0}
+
+    now = now or datetime.now(UTC)
+    window_hours = float(settings.discord_link_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(
+        select(Post.id).where(
+            Post.synthesized_by.is_(None),
+            Post.absorbed_by_post_id.is_(None),
+            sort_key >= horizon,
+        )
+    )).scalars().all()
+
+    svc = PostAssociationService(session)
+    proposed = 0
+    for pid in ids:
+        proposed += await svc.match_post(
+            pid,
+            threshold=float(settings.discord_link_threshold),
+            window_hours=window_hours,
+        )
+    log.info(
+        "discord announcement matcher: scanned %d post(s), proposed %d pair(s)",
+        len(ids), proposed,
+    )
+    return {"enabled": True, "scanned": len(ids), "proposed": proposed}

backend/app/services/post_feed_service.py

@@ -175,9 +175,10 @@ class PostFeedService:
         post_ids = [p.id for p, _, _ in rows]
         thumbs_map = await self._thumbnails_for(post_ids)
         atts_map = await self._attachments_for(post_ids)
+        links_map = await self._links_for(post_ids)
 
         items = [
-            self._to_dict(post, artist, source, thumbs_map, atts_map)
+            self._to_dict(post, artist, source, thumbs_map, atts_map, links_map)
             for post, artist, source in rows
         ]
         return {"items": items, "next_cursor": next_cursor}
@@ -218,6 +219,7 @@ class PostFeedService:
         atts_map = await self._attachments_for([anchor_post.id])
         anchor_item = self._to_dict(
             anchor_post, anchor_artist, anchor_source, thumbs_map, atts_map,
+            await self._links_for([anchor_post.id]),
         )
         return {
             "items": newer["items"] + [anchor_item] + older["items"],
@@ -241,7 +243,10 @@ class PostFeedService:
         # the default arg.
         thumbs_map = await self._thumbnails_for([post.id], limit=None)
         atts_map = await self._attachments_for([post.id])
-        item = self._to_dict(post, artist, source, thumbs_map, atts_map)
+        item = self._to_dict(
+            post, artist, source, thumbs_map, atts_map,
+            await self._links_for([post.id]),
+        )
         item["description_full"] = html_to_plain(post.description)
         # Full (uncapped) translated description for the detail view (#143).
         item["description_translated_full"] = post.description_translated
@@ -407,9 +412,25 @@ class PostFeedService:
             })
         return out
 
+    async def _links_for(self, post_ids: list[int]) -> dict[int, list[dict]]:
+        """Accepted announcement links touching these posts (#388 E5).
+
+        Only ACCEPTED ones. A pending proposal is a question for the review
+        queue, not a claim to render beside the artwork — showing one here
+        would assert a link the operator has not agreed to, which is the exact
+        failure the confirm-only design exists to prevent.
+        """
+        # Imported here rather than at module scope: post_association_service
+        # imports discord_grouping, which imports the models, and the feed
+        # service is imported by the API at startup. A local import keeps that
+        # chain out of the import graph for a purely optional read.
+        from .post_association_service import PostAssociationService
+
+        return await PostAssociationService(self.session).linked_for(post_ids)
+
     def _to_dict(
         self, post: Post, artist: Artist, source: Source | None,
-        thumbs_map: dict, atts_map: dict,
+        thumbs_map: dict, atts_map: dict, links_map: dict | None = None,
     ) -> dict:
         plain_full = html_to_plain(post.description) if post.description else None
         if plain_full is None:
@@ -453,6 +474,11 @@ class PostFeedService:
             # ago" — which is the whole signal that chat content is trickling
             # in. NULL means it has not grown since it was created.
             "last_grew_at": post.last_grew_at.isoformat() if post.last_grew_at else None,
+            # Accepted links only (#388 E5): [{role, post_id, id}], where role
+            # is "announces" (this post is the teaser) or "announced_by" (this
+            # post is the drop). Always a list so the UI never branches on
+            # absence.
+            "associations": (links_map or {}).get(post.id, []),
             # Non-null on a chat message a synthetic post absorbed. The feed
             # filters these out, but `around`/`get_post` still reach them, and
             # the UI uses this to explain why a post it linked to is not in the

backend/app/tasks/maintenance.py

@@ -1171,3 +1171,36 @@ def group_discord_drops() -> str:
         f"sources={res['sources']} created={res['posts_created']} "
         f"joined={res['images_joined']}"
     )
+
+
+@celery.task(
+    name="backend.app.tasks.maintenance.match_post_associations",
+    soft_time_limit=900, time_limit=1200,
+)
+def match_post_associations() -> str:
+    """Milestone 388 E5: propose which Patreon post announced which Discord drop.
+
+    Proposes only — every pair lands in a review queue and nothing is linked
+    until the operator accepts. Maintenance lane for the same reason as the
+    grouper: no inference, no ML library, and it must not depend on the
+    optional ml-worker being present.
+    """
+    import asyncio
+
+    from ..services.post_association_service import rescan
+    from ._async_session import async_session_factory
+
+    async def _run() -> dict:
+        async_factory, engine = async_session_factory()
+        try:
+            async with async_factory() as session:
+                result = await rescan(session)
+                await session.commit()
+            return result
+        finally:
+            await engine.dispose()
+
+    res = asyncio.run(_run())
+    if not res["enabled"]:
+        return "disabled"
+    return f"scanned={res['scanned']} proposed={res['proposed']}"

tests/test_post_association.py

@@ -0,0 +1,374 @@
+"""Milestone 388 E5: which Patreon post announced which Discord drop.
+
+The failure this step must not have is a WRONG link. Telling the operator that
+two different pieces are one is worse than telling them nothing — no link
+leaves them where they already were, a wrong one actively misinforms. So most
+of what follows pins refusals, and the central one is structural rather than
+behavioural: the threshold sits above every single signal weight, which is what
+makes "time proximity alone is never sufficient" arithmetic instead of a hope.
+"""
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from sqlalchemy import select
+
+from backend.app.models import (
+    Artist,
+    ImageRecord,
+    ImportSettings,
+    Post,
+    PostAssociation,
+    Source,
+)
+from backend.app.services.discord_grouping import DROP_GROUPER
+from backend.app.services.post_association_service import (
+    WEIGHTS,
+    PostAssociationService,
+    declared_signal,
+    proximity_signal,
+    rescan,
+    weighted_score,
+)
+from backend.app.services.post_feed_service import PostFeedService
+
+pytestmark = pytest.mark.integration
+
+DEFAULT_THRESHOLD = 0.60
+WINDOW = 24.0
+
+
+# --- the structural guard -------------------------------------------------
+
+
+def test_no_single_signal_can_reach_the_threshold():
+    """THE load-bearing property of this matcher.
+
+    On a busy day an artist posts several times, so a matcher that could pair
+    on proximity alone would turn every busy day into false pairs and the
+    review queue would be abandoned. Requiring two signals is what prevents
+    that — and it is a fact about the WEIGHTS, not about any code path, so it
+    survives every refactor of the scorer.
+
+    If this fails, either a weight grew or the default threshold dropped. Do
+    not "fix" it by lowering the assertion; the arithmetic IS the safeguard.
+    """
+    assert max(WEIGHTS.values()) < DEFAULT_THRESHOLD
+    assert sum(WEIGHTS.values()) == pytest.approx(1.0)
+
+
+def test_perfect_proximity_alone_does_not_propose():
+    """The same property, expressed through the scorer."""
+    score = weighted_score({"proximity": 1.0, "declared": 0.0})
+    assert score < DEFAULT_THRESHOLD
+
+
+def test_an_explicit_declaration_alone_does_not_propose():
+    score = weighted_score({"proximity": 0.0, "declared": 1.0})
+    assert score < DEFAULT_THRESHOLD
+
+
+def test_both_signals_together_do_propose():
+    assert weighted_score({"proximity": 1.0, "declared": 1.0}) >= DEFAULT_THRESHOLD
+
+
+# --- the signals ----------------------------------------------------------
+
+
+def test_proximity_decays_to_zero_at_the_window_edge():
+    window = timedelta(hours=24)
+    assert proximity_signal(timedelta(0), window) == 1.0
+    assert proximity_signal(timedelta(hours=24), window) == 0.0
+    assert proximity_signal(timedelta(hours=48), window) == 0.0
+    assert 0.4 < proximity_signal(timedelta(hours=12), window) < 0.6
+
+
+def test_proximity_is_symmetric_because_either_can_land_first():
+    """The teaser usually goes up around the drop, not reliably before it."""
+    window = timedelta(hours=24)
+    assert proximity_signal(timedelta(hours=-2), window) == proximity_signal(
+        timedelta(hours=2), window
+    )
+
+
+def test_an_invite_link_scores_higher_than_a_bare_mention():
+    assert declared_signal("Full set on my discord.gg/abc123 now!") == 1.0
+    assert 0 < declared_signal("Posted the rest on discord earlier") < 1.0
+    assert declared_signal("New piece, hope you like it") == 0.0
+    assert declared_signal(None) == 0.0
+
+
+def test_the_declaration_signal_reads_through_html():
+    """Post bodies are HTML; a link inside an anchor tag must still count."""
+    assert declared_signal(
+        '<p>Full set: <a href="https://discord.gg/xyz">here</a></p>'
+    ) > 0
+
+
+def test_a_word_containing_discord_is_not_a_mention():
+    """\"discordant\" is not a declaration. Without the word boundary the
+    signal fires on ordinary prose and drags pairs over the threshold."""
+    assert declared_signal("a discordant palette, deliberately") == 0.0
+
+
+# --- end to end -----------------------------------------------------------
+
+
+async def _artist_with_channels(db, name: str):
+    artist = Artist(name=name, slug=name.lower().replace(" ", "-"))
+    db.add(artist)
+    await db.flush()
+    patreon = Source(
+        artist_id=artist.id, platform="patreon",
+        url=f"https://patreon.com/{name}", enabled=True,
+    )
+    discord = Source(
+        artist_id=artist.id, platform="discord",
+        url=f"https://discord.com/channels/1/{name}", enabled=True,
+    )
+    db.add_all([patreon, discord])
+    await db.flush()
+    return artist, patreon, discord
+
+
+async def _teaser(db, artist, source, *, at, body, ext="teaser"):
+    post = Post(
+        source_id=source.id, artist_id=artist.id, external_post_id=ext,
+        post_date=at, post_title="New piece", description=body,
+    )
+    db.add(post)
+    await db.flush()
+    db.add(ImageRecord(
+        path=f"/images/{source.id}-{ext}.jpg", sha256=f"{ext:0>64}"[:64],
+        size_bytes=10, mime="image/jpeg", width=10, height=10,
+        origin="downloaded", primary_post_id=post.id, artist_id=artist.id,
+    ))
+    await db.flush()
+    return post
+
+
+async def _drop(db, artist, source, *, at, ext="fc-drop:1"):
+    post = Post(
+        source_id=source.id, artist_id=artist.id, external_post_id=ext,
+        post_date=at, synthesized_by=DROP_GROUPER,
+        synthesis_details={"message_count": 4, "images_since_surface": 0},
+    )
+    db.add(post)
+    await db.flush()
+    return post
+
+
+@pytest.mark.asyncio
+async def test_a_teaser_and_its_drop_an_hour_apart_are_proposed(db):
+    artist, patreon, discord = await _artist_with_channels(db, "pairartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(hours=3),
+        body="Full set is up on discord.gg/abc now",
+    )
+    drop = await _drop(db, artist, discord, at=now - timedelta(hours=2))
+    await db.commit()
+
+    made = await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    )
+    await db.commit()
+    assert made == 1
+
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    assert assoc.announcement_post_id == teaser.id
+    assert assoc.payload_post_id == drop.id
+    assert assoc.status == "pending", "nothing is linked without the operator"
+    # The per-signal breakdown survives, so the proposal stays explicable
+    # after the weights or threshold move.
+    assert assoc.signals["declared"] == 1.0
+    assert assoc.signals["proximity"] > 0.9
+
+
+@pytest.mark.asyncio
+async def test_two_unrelated_posts_the_same_day_are_not_proposed(db):
+    """Time proximity alone must not be sufficient, or every busy day becomes
+    a false pair. The teaser here says nothing about Discord."""
+    artist, patreon, discord = await _artist_with_channels(db, "busyartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(hours=3),
+        body="Just a sketch I liked",
+    )
+    await _drop(db, artist, discord, at=now - timedelta(hours=2))
+    await db.commit()
+
+    made = await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    )
+    await db.commit()
+    assert made == 0
+    assert (await db.execute(select(PostAssociation))).scalars().all() == []
+
+
+@pytest.mark.asyncio
+async def test_a_drop_outside_the_window_is_not_proposed(db):
+    artist, patreon, discord = await _artist_with_channels(db, "farapartartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(days=10),
+        body="Everything is on discord.gg/abc",
+    )
+    await _drop(db, artist, discord, at=now)
+    await db.commit()
+
+    assert await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 0
+
+
+@pytest.mark.asyncio
+async def test_another_artists_drop_is_never_proposed(db):
+    """Same-artist is a HARD filter, not a scored signal: two different
+    creators posting minutes apart is a coincidence, not evidence."""
+    artist_a, patreon_a, _ = await _artist_with_channels(db, "artista")
+    _artist_b, _patreon_b, discord_b = await _artist_with_channels(db, "artistb")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist_a, patreon_a, at=now - timedelta(hours=1),
+        body="new drop on discord.gg/abc",
+    )
+    await _drop(db, _artist_b, discord_b, at=now)
+    await db.commit()
+
+    assert await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 0
+
+
+@pytest.mark.asyncio
+async def test_an_artist_with_no_discord_source_produces_nothing_and_no_error(db):
+    artist = Artist(name="soloartist", slug="soloartist")
+    db.add(artist)
+    await db.flush()
+    patreon = Source(
+        artist_id=artist.id, platform="patreon",
+        url="https://patreon.com/solo", enabled=True,
+    )
+    db.add(patreon)
+    await db.flush()
+    teaser = await _teaser(
+        db, artist, patreon, at=datetime.now(UTC),
+        body="on discord.gg/abc", ext="solo",
+    )
+    await db.commit()
+
+    assert await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 0
+
+
+@pytest.mark.asyncio
+async def test_a_synthetic_post_cannot_announce_anything(db):
+    """FC wrote it, so it announces nothing — and a grouping proposing itself
+    as the teaser for another grouping would be pure noise."""
+    artist, _patreon, discord = await _artist_with_channels(db, "noselfannounce")
+    now = datetime.now(UTC)
+    drop_a = await _drop(db, artist, discord, at=now - timedelta(hours=1), ext="fc-drop:a")
+    await _drop(db, artist, discord, at=now, ext="fc-drop:b")
+    await db.commit()
+
+    assert await PostAssociationService(db).match_post(
+        drop_a.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 0
+
+
+@pytest.mark.asyncio
+async def test_a_dismissed_pair_is_never_proposed_again(db):
+    """The row is what remembers the rejection. Re-proposing a rejected pair on
+    every scan is the single behaviour that makes a review queue get ignored."""
+    artist, patreon, discord = await _artist_with_channels(db, "dismissartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(hours=2),
+        body="discord.gg/abc has the rest",
+    )
+    await _drop(db, artist, discord, at=now - timedelta(hours=1))
+    await db.commit()
+
+    svc = PostAssociationService(db)
+    assert await svc.match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 1
+    await db.commit()
+
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    await svc.dismiss(assoc.id)
+    await db.commit()
+
+    assert await svc.match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    ) == 0
+    await db.commit()
+    assert len((await db.execute(select(PostAssociation))).scalars().all()) == 1
+
+
+@pytest.mark.asyncio
+async def test_only_an_accepted_link_reaches_the_post_payload(db):
+    """A pending proposal is a question for the review queue, not a claim to
+    render beside the artwork."""
+    artist, patreon, discord = await _artist_with_channels(db, "payloadartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(hours=2),
+        body="rest is on discord.gg/abc",
+    )
+    drop = await _drop(db, artist, discord, at=now - timedelta(hours=1))
+    await db.commit()
+
+    svc = PostAssociationService(db)
+    await svc.match_post(teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW)
+    await db.commit()
+
+    feed = PostFeedService(db)
+    assert (await feed.get_post(teaser.id))["associations"] == []
+
+    assoc = (await db.execute(select(PostAssociation))).scalar_one()
+    await svc.accept(assoc.id)
+    await db.commit()
+
+    # Both ends see it, and each sees the OTHER post with its own role.
+    teaser_item = await feed.get_post(teaser.id)
+    assert teaser_item["associations"] == [
+        {"role": "announces", "post_id": drop.id, "id": assoc.id}
+    ]
+    drop_item = await feed.get_post(drop.id)
+    assert drop_item["associations"] == [
+        {"role": "announced_by", "post_id": teaser.id, "id": assoc.id}
+    ]
+
+
+@pytest.mark.asyncio
+async def test_deleting_the_grouping_takes_its_proposals_with_it(db):
+    """E3's reversal path is one DELETE; it must not leave a dangling proposal
+    pointing at a post that no longer exists."""
+    artist, patreon, discord = await _artist_with_channels(db, "cascadeartist")
+    now = datetime.now(UTC)
+    teaser = await _teaser(
+        db, artist, patreon, at=now - timedelta(hours=2),
+        body="discord.gg/abc",
+    )
+    drop = await _drop(db, artist, discord, at=now - timedelta(hours=1))
+    await db.commit()
+
+    await PostAssociationService(db).match_post(
+        teaser.id, threshold=DEFAULT_THRESHOLD, window_hours=WINDOW,
+    )
+    await db.commit()
+    assert len((await db.execute(select(PostAssociation))).scalars().all()) == 1
+
+    await db.delete(drop)
+    await db.commit()
+    assert (await db.execute(select(PostAssociation))).scalars().all() == []
+
+
+@pytest.mark.asyncio
+async def test_the_rescan_is_a_no_op_when_the_switch_is_off(db):
+    settings = await ImportSettings.load(db)
+    settings.discord_link_enabled = False
+    await db.commit()
+    assert await rescan(db) == {"enabled": False, "scanned": 0, "proposed": 0}