feat: FC authors the post that Discord never wrote (milestone 388 step E2)

Discord is a delivery channel, not a publisher. One message is not one post,
and today every message lands as its own `post` row, so chat lines compete
with authored work for the same surface. Rather than demote them into a
second-class feed, FC now writes the post itself: one row per DROP, its
images the drop's images, its body the messages' text in arrival order.

Synthesising a `Post` (rather than inventing a parallel entity) is the whole
point — the result is post-shaped by construction, so feed, provenance,
translation, attachments and series keep working on it unchanged.

The predicate is three axes ANDed, and the time one does the real work:

    same source  AND  cosine distance <= threshold  AND  no gap > window

Similarity alone over-groups, and that is the failure that would make this
useless: any two pieces of the same character by the same artist sit close in
SigLIP space, so a cosine-only rule collapses a month of one character into a
single "post". Two details inside the predicate are load-bearing —

* distance is measured to the group's SEED, never to the previous member,
  because chaining lets a group DRIFT: twenty small steps walk from one piece
  to a completely different one, each hop individually within threshold;
* the window is measured between CONSECUTIVE messages, not from the first, so
  an artist trickling variants out over an evening stays one drop.

Why a post-import sweep and not part of ingest. The obvious alternative was to
migrate Discord to the native post-first ingester (#1266) and group at capture
time. That cannot work: the grouping signal is `siglip_embedding`, which is
produced asynchronously AFTER import (tasks/ml.py, the GPU backfill), so at
capture time there is nothing to group on. Grouping is necessarily something
that happens once the vectors catch up — hence a re-runnable sweep that skips
what it cannot yet place, and an hourly (not daily) cadence.

The honesty rule, enforced in the schema. `post.synthesized_by` names the
grouper; `synthesis_details` records the members, the count, and the
thresholds AS THEY WERE (they are operator-tunable, so without that "why did
it group these" is unanswerable a month later). Member posts are absorbed, not
destroyed — they remain the images' true origin and the audit trail — and
`absorbed_by_post_id` is ON DELETE SET NULL, so deleting a synthetic post
releases its members back into the feed in one DELETE with no repair step.
`post_title` stays NULL deliberately: a synthesised title is the one place
this could put words in a creator's mouth.

Two guards the first draft would have failed:

* the per-run cap took the lowest post IDs, not the oldest posts — DISTINCT ON
  forces its own ORDER BY, so the sort now happens outside the subquery;
* a cap landing mid-drop would have published a truncated group claiming to be
  a whole drop, so the last group is left for the next run.

And one vacuous test caught before it shipped: the support vector perturbed a
single component of an all-ones vector, moving it ~1e-6, so every distance
assertion passed regardless of what the predicate did. `_vec` now builds a
unit vector at a stated angle, where distance is exactly 1 - cos(delta) —
rule 167, a guard has to be able to fail.

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/0092_synthetic_posts.py

@@ -0,0 +1,92 @@
+"""Synthetic posts — FC authors a post for content that arrived as chat.
+
+Milestone 388, step E2. Discord is a delivery channel, not a publisher: one
+message is not one post, and today every message becomes its own `post` row
+competing with authored work for the same surface. This adds the three columns
+that let FC group a creator's variant drop into a post it wrote itself, while
+keeping that fact visible and the grouping reversible.
+
+## Why a flag and a back-pointer rather than a separate table
+
+A synthetic post has to BE a post — same row, same columns — or every existing
+surface (feed, provenance, translation, attachments, series) would need a
+second code path for it. `synthesized_by` marks the ones FC authored;
+`absorbed_by_post_id` points a member message-post at the post that replaced
+it in the feed. The members are not deleted: they remain the images' true
+origin, and destroying them would make the grouping un-auditable at exactly
+the moment somebody wants to check it.
+
+Reversal is one DELETE. `absorbed_by_post_id` is ON DELETE SET NULL, so
+removing a synthetic post releases its members and they return to the feed
+unaided.
+
+Revision ID: 0092
+Revises: 0091
+Create Date: 2026-09-10
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0092"
+down_revision: Union[str, None] = "0091"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # No CHECK on synthesized_by (rule 36 considered and declined): there is one
+    # grouper today and a second would be a new VALUE, not a new invariant —
+    # matching source.error_type and service_seen.kind.
+    op.add_column("post", sa.Column("synthesized_by", sa.String(length=32), nullable=True))
+    op.add_column("post", sa.Column("synthesis_details", sa.JSON(), nullable=True))
+    op.add_column(
+        "post", sa.Column("absorbed_by_post_id", sa.Integer(), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_post_absorbed_by_post_id"), "post", ["absorbed_by_post_id"],
+    )
+    # SET NULL, not CASCADE: deleting the synthetic post must RELEASE its
+    # members, never take them with it. The members are the real capture.
+    op.create_foreign_key(
+        "fk_post_absorbed_by_post_id_post", "post", "post",
+        ["absorbed_by_post_id"], ["id"], ondelete="SET NULL",
+    )
+
+    # Grouping tunables. Every one of these is operator-facing (project rule
+    # 25) because the quality bar here is a judgement call no test can settle:
+    # too greedy merges distinct pieces, too shy leaves a drop scattered.
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "discord_grouping_enabled", sa.Boolean(),
+            server_default="true", nullable=False,
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "discord_group_max_distance", sa.Float(),
+            server_default=sa.text("0.10"), nullable=False,
+        ),
+    )
+    op.add_column(
+        "ml_settings",
+        sa.Column(
+            "discord_group_window_minutes", sa.Float(),
+            server_default=sa.text("60"), nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("ml_settings", "discord_group_window_minutes")
+    op.drop_column("ml_settings", "discord_group_max_distance")
+    op.drop_column("ml_settings", "discord_grouping_enabled")
+    op.drop_constraint("fk_post_absorbed_by_post_id_post", "post", type_="foreignkey")
+    op.drop_index(op.f("ix_post_absorbed_by_post_id"), table_name="post")
+    op.drop_column("post", "absorbed_by_post_id")
+    op.drop_column("post", "synthesis_details")
+    op.drop_column("post", "synthesized_by")

backend/app/api/ml_admin.py

@@ -48,6 +48,12 @@ _EDITABLE = (
     "process_conflict_threshold",
     "embedder_model_name",
     "embedder_model_version",
+    # Discord drop grouping (#388 E2). Operator-facing because the quality bar
+    # is a judgement no test can settle: too greedy merges distinct pieces, too
+    # shy leaves a drop scattered.
+    "discord_grouping_enabled",
+    "discord_group_max_distance",
+    "discord_group_window_minutes",
     *_DETECTOR_FIELDS,
 )
 
@@ -148,6 +154,14 @@ def _validate(p: dict) -> str | None:
         return f"process_auto_apply_threshold must be between {AUTO_APPLY_THRESHOLD_MIN} and {AUTO_APPLY_THRESHOLD_MAX}"
     if not (0.0 <= float(p["process_conflict_threshold"]) <= 1.0):
         return "process_conflict_threshold must be between 0 and 1"
+    # Discord drop grouping (#388 E2). max_distance is a cosine DISTANCE, so
+    # unlike the *_threshold family above it is not on the auto-apply scale:
+    # 0 is identical and 1 is unrelated, and both ends are legal. The upper
+    # bound is 1.0 rather than AUTO_APPLY_THRESHOLD_MAX for that reason.
+    if not (0.0 <= float(p["discord_group_max_distance"]) <= 1.0):
+        return "discord_group_max_distance must be between 0 and 1"
+    if float(p["discord_group_window_minutes"]) <= 0:
+        return "discord_group_window_minutes must be > 0"
     # Embedder model swap (#1190): both must be non-empty. Changing them means a
     # different embedding space — the operator must re-embed + retrain after.
     for key in ("embedder_model_name", "embedder_model_version"):

backend/app/celery_app.py

@@ -200,6 +200,13 @@ def make_celery() -> Celery:
                 "task": "backend.app.tasks.maintenance.snapshot_head_metrics",
                 "schedule": 86400.0,
             },
+            "group-discord-drops-hourly": {
+                "task": "backend.app.tasks.maintenance.group_discord_drops",
+                "schedule": 3600.0,  # hourly. Not daily: the grouping signal is
+                # the SigLIP embedding, which lands asynchronously AFTER import
+                # (#388 E2), so this sweep is what picks up a drop once its
+                # vectors have caught up. No-op unless discord_grouping_enabled.
+            },
             "integrity-verify-weekly": {
                 "task": "backend.app.tasks.maintenance.verify_integrity",
                 "schedule": 604800.0,  # weekly

backend/app/models/ml_settings.py

@@ -252,6 +252,40 @@ class MLSettings(Base):
         Integer, nullable=False, default=64,
         server_default="64",
     )
+    # -- Discord drop grouping (milestone 388) -----------------------------
+    # FC authors a post out of a creator's variant drop. The predicate is three
+    # axes ANDed together, and the time one does the real work: SIMILARITY
+    # ALONE OVER-GROUPS. Any two pieces of the same character by the same
+    # artist sit close in SigLIP space, so a cosine-only rule collapses a month
+    # of one character into a single "post". What makes a variant set a set is
+    # that it was dropped TOGETHER.
+    discord_grouping_enabled: Mapped[bool] = mapped_column(
+        # ON by default, matching the operator's standing opt-OUT preference for
+        # automatic behaviour (2026-06-29, recorded on the head/ccip auto-apply
+        # switches). Safe to default on because the act is reversible by one
+        # DELETE: removing a synthetic post un-absorbs its members.
+        Boolean, nullable=False, default=True,
+        server_default="true",
+    )
+    # Cosine DISTANCE, not similarity — this is the units gallery_service's
+    # `cosine_distance` already speaks, and converting at the query site is a
+    # step to get backwards. Lower = stricter. 0.10 is deliberately TIGHT: the
+    # two failure modes are not symmetric. Grouping too shy leaves a drop
+    # scattered, which is visible and fixable by raising this; grouping too
+    # greedy merges distinct pieces into a post that claims they belong
+    # together, which is the failure that would discredit the feature.
+    discord_group_max_distance: Mapped[float] = mapped_column(
+        Float, nullable=False, default=0.10,
+        server_default=text("0.10"),
+    )
+    # The gap that ENDS a drop, measured between CONSECUTIVE messages rather
+    # than from the first — an artist trickling variants out over an evening is
+    # one drop, and a window anchored on the first message would cut it in half
+    # at an arbitrary point.
+    discord_group_window_minutes: Mapped[float] = mapped_column(
+        Float, nullable=False, default=60.0,
+        server_default=text("60"),
+    )
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )

backend/app/models/post.py

@@ -102,3 +102,37 @@ class Post(Base):
     downloaded_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), nullable=False, server_default=func.now()
     )
+
+    # -- Synthetic posts (milestone 388). ----------------------------------
+    # Discord is a delivery CHANNEL, not a publisher: one message is not one
+    # post. So FC authors the post itself, grouping a creator's variant drop
+    # into a single row (services/discord_grouping.py).
+    #
+    # NULL for every post a creator actually wrote — which is all of them until
+    # a grouper runs. Non-NULL names the grouper that authored this row, and is
+    # the ONE flag the UI keys off to say so. The honesty rule is the whole
+    # point: a synthetic post must never present itself as authored, and a
+    # column that is absent-or-a-name makes "was this us?" answerable from the
+    # row rather than inferred from its shape.
+    #
+    # Plain String, no CHECK (rule 36 considered and declined) — same reasoning
+    # as source.error_type and service_seen.kind. There is exactly one grouper
+    # today; a second would be a value, not an invariant.
+    synthesized_by: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # What it was built from, so the operator can audit a grouping FC invented:
+    # member post ids, message count, and the thresholds in force when the
+    # decision was made. That last part matters — the thresholds are operator-
+    # tunable, so "why did it group these" is unanswerable a month later
+    # without recording the values that produced it.
+    synthesis_details: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    # Set on a MEMBER post, pointing at the synthetic post that absorbed it.
+    # The feed hides absorbed posts (they are the chat lines the synthetic post
+    # replaced); every other surface still reaches them by id, because they
+    # remain the image's true origin and the grouping has to be inspectable.
+    #
+    # Self-FK, ON DELETE SET NULL: deleting a synthetic post un-absorbs its
+    # members and they return to the feed on their own. That is the reversal
+    # path, and it is one DELETE — nothing to undo by hand.
+    absorbed_by_post_id: Mapped[int | None] = mapped_column(
+        ForeignKey("post.id", ondelete="SET NULL"), nullable=True, index=True
+    )

backend/app/services/discord_grouping.py

@@ -0,0 +1,357 @@
+"""Discord drop grouping — FC authors the post that Discord never wrote.
+
+Milestone 388, step E2.
+
+Discord is a delivery CHANNEL, not a publisher. A creator drops a set of
+near-variants — the same piece with different hair colour, accessories, an
+outfit swap — across a handful of messages, and today each of those messages
+lands as its own `post` row, so chat lines compete with authored work for the
+same surface. The fix is not to demote them into a second-class feed; it is to
+let FC write the post: one row per DROP, its images the drop's images, its body
+the messages' text in arrival order.
+
+The result is post-shaped by construction, which is the entire reason to
+synthesise a `Post` rather than invent a parallel entity — feed, provenance,
+translation, attachments and series all keep working on it unchanged.
+
+## The predicate: three axes, ANDed, and the time one does the real work
+
+**Similarity alone over-groups, and that is the failure that would make this
+useless.** Any two pieces of the same character by the same artist sit close in
+SigLIP space; a cosine-only rule collapses a month of one character into a
+single "post". What makes a variant set a set is that it was dropped TOGETHER.
+
+    same source  AND  cosine distance <= threshold  AND  no gap > window
+
+Two details in there are load-bearing:
+
+* **Distance is measured to the group's SEED, never to the previous member.**
+  Chaining to the previous member lets a group DRIFT: twenty small steps walk
+  from one piece to a completely different one, each hop individually within
+  threshold. Anchoring on the seed bounds the whole group to one neighbourhood.
+* **The window is measured between CONSECUTIVE messages, not from the first.**
+  An artist trickling variants out over an evening is one drop; a window
+  anchored on the first message would cut it in half at an arbitrary point.
+
+## Why this is a post-import sweep and not part of ingest
+
+The obvious alternative was to migrate Discord to the native post-first
+ingester (#1266) and group at capture time. **That cannot work**, and the
+reason is worth recording: the grouping signal is `siglip_embedding`, which is
+produced ASYNCHRONOUSLY after import (`tasks/ml.py`, the GPU queue backfill).
+At capture time the embedding does not exist yet, so an ingester has nothing to
+group on. Grouping is necessarily something that happens once the vectors have
+caught up — which also means this sweep must be re-runnable and must simply
+skip what it cannot yet place. It does: a post whose image has no embedding is
+left alone and picked up on a later run.
+
+## The honesty rule
+
+A synthetic post must never pretend an artist authored it. It carries
+`synthesized_by`, records what it was built from in `synthesis_details`
+(members, count, and the thresholds in force at the time), and leaves its
+member posts intact and reachable. Deleting the synthetic post releases the
+members back into the feed — one DELETE, no repair step. FC invented this
+grouping; the operator has to be able to see that, inspect it, and undo it.
+"""
+
+from __future__ import annotations
+
+import logging
+import math
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import Select, func, select, update
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..models import ImageProvenance, ImageRecord, MLSettings, Post, Source
+
+log = logging.getLogger(__name__)
+
+# The value that lands in `post.synthesized_by`. One grouper today; a second
+# would be another value here, which is exactly why the column has no CHECK.
+DROP_GROUPER = "discord_drop"
+
+PLATFORM = "discord"
+
+# Ceiling on member posts examined per source per run. A first sweep over an
+# established library would otherwise pull every Discord message's 1152-float
+# vector into memory at once. The sweep is re-runnable and works oldest-first,
+# so a backlog simply drains over successive runs rather than needing one
+# heroic pass.
+MAX_CANDIDATES_PER_SOURCE = 500
+
+
+@dataclass
+class DropGroup:
+    """One drop: the member posts, in arrival order, that will become a post."""
+
+    member_ids: list[int] = field(default_factory=list)
+    seed: list[float] | None = None
+    last_at: datetime | None = None
+
+
+def cosine_distance(a, b) -> float:
+    """Cosine distance between two embeddings, in the same units pgvector's
+    `cosine_distance` operator returns (0 = identical, 1 = orthogonal).
+
+    Computed in Python rather than SQL because the comparison is against a
+    group seed held in a loop, not against a column — and pure arithmetic keeps
+    numpy off this path entirely. pgvector may hand back a numpy array or a
+    list depending on driver version, so both are coerced.
+    """
+    va = [float(x) for x in a]
+    vb = [float(x) for x in b]
+    # strict=True: two embeddings of different length is a corrupted row or a
+    # model swap that skipped the re-embed, and silently truncating to the
+    # shorter one would score it as a near match.
+    dot = sum(x * y for x, y in zip(va, vb, strict=True))
+    na = math.sqrt(sum(x * x for x in va))
+    nb = math.sqrt(sum(y * y for y in vb))
+    if na == 0.0 or nb == 0.0:
+        # A zero vector has no direction, so no meaningful distance. Return the
+        # maximum so it can never pull anything into a group.
+        return 1.0
+    return 1.0 - (dot / (na * nb))
+
+
+def _candidate_stmt(source_id: int, *, not_after: datetime) -> Select:
+    """Ungrouped Discord message-posts, one representative image each, OLDEST
+    FIRST — which is the order `build_groups` requires.
+
+    DISTINCT ON the post picks the lowest-id embedded image as that post's
+    representative: a Discord message carrying several attachments is still one
+    point in the drop, and comparing every attachment would let one incidental
+    image drag an unrelated message into the group.
+
+    The DISTINCT ON is wrapped in a subquery rather than ordered directly,
+    because Postgres requires a DISTINCT ON query's ORDER BY to LEAD with the
+    distinct expression — so the inner query must sort by `post.id`, which is
+    insertion order and not arrival order at all once a backfill has imported
+    anything out of sequence. Sorting outside is what makes the caller's LIMIT
+    take the OLDEST candidates instead of the lowest-numbered ones.
+    """
+    sort_key = func.coalesce(Post.post_date, Post.downloaded_at)
+    inner = (
+        select(
+            Post.id.label("post_id"),
+            sort_key.label("occurred_at"),
+            ImageRecord.siglip_embedding.label("embedding"),
+        )
+        .join(ImageRecord, ImageRecord.primary_post_id == Post.id)
+        .where(
+            Post.source_id == source_id,
+            # Never absorb a post FC wrote, and never re-absorb one already
+            # taken — both would build groups out of groups.
+            Post.synthesized_by.is_(None),
+            Post.absorbed_by_post_id.is_(None),
+            ImageRecord.siglip_embedding.is_not(None),
+            sort_key <= not_after,
+        )
+        .distinct(Post.id)
+        .order_by(Post.id, ImageRecord.id)
+        .subquery()
+    )
+    return (
+        select(inner.c.post_id, inner.c.occurred_at, inner.c.embedding)
+        .order_by(inner.c.occurred_at, inner.c.post_id)
+    )
+
+
+def build_groups(
+    rows: list[tuple[int, datetime, list[float]]],
+    *,
+    max_distance: float,
+    window: timedelta,
+) -> list[DropGroup]:
+    """Walk candidates in arrival order and cut them into drops.
+
+    `rows` must be sorted oldest-first — the whole predicate is about
+    adjacency in time, so an unsorted input would silently produce nonsense
+    rather than fail.
+    """
+    groups: list[DropGroup] = []
+    current: DropGroup | None = None
+
+    for post_id, occurred_at, embedding in rows:
+        if current is not None:
+            gap_ok = occurred_at - current.last_at <= window
+            # Distance to the SEED, not to the previous member — see the module
+            # docstring on drift.
+            near = cosine_distance(current.seed, embedding) <= max_distance
+            if gap_ok and near:
+                current.member_ids.append(post_id)
+                current.last_at = occurred_at
+                continue
+            groups.append(current)
+        current = DropGroup(
+            member_ids=[post_id], seed=embedding, last_at=occurred_at,
+        )
+
+    if current is not None:
+        groups.append(current)
+    return groups
+
+
+async def _synthesize(
+    session: AsyncSession,
+    *,
+    source: Source,
+    group: DropGroup,
+    max_distance: float,
+    window_minutes: float,
+) -> Post | None:
+    """Write one synthetic post for `group` and absorb its members."""
+    members = (await session.execute(
+        select(Post)
+        .where(Post.id.in_(group.member_ids))
+        .order_by(func.coalesce(Post.post_date, Post.downloaded_at), Post.id)
+    )).scalars().all()
+    if not members:
+        return None
+
+    first = members[0]
+    # Deterministic key, so a re-run cannot mint a second post for the same
+    # drop: the unique (source_id, external_post_id) constraint would reject it
+    # even if the member filter somehow let the drop through twice.
+    external_id = f"fc-drop:{first.external_post_id}"[:128]
+
+    # The messages' own text, in arrival order, IS the post's body — that is
+    # what the operator asked for and it is the only text a drop has. Blank
+    # messages (an attachment with no caption) contribute nothing rather than a
+    # run of empty lines.
+    body = "\n\n".join(m.description.strip() for m in members if m.description and m.description.strip())
+
+    post = Post(
+        source_id=source.id,
+        artist_id=source.artist_id,
+        external_post_id=external_id,
+        # post_title stays NULL DELIBERATELY. A synthesised title is the one
+        # place this feature could accidentally put words in a creator's mouth;
+        # the UI labels the row from `synthesized_by` instead, which cannot be
+        # mistaken for something the artist wrote.
+        post_title=None,
+        post_url=first.post_url,
+        post_date=first.post_date or first.downloaded_at,
+        description=body or None,
+        synthesized_by=DROP_GROUPER,
+        synthesis_details={
+            "member_post_ids": [m.id for m in members],
+            "message_count": len(members),
+            # The thresholds AS THEY WERE. They are operator-tunable, so
+            # without this "why did it group these" is unanswerable later.
+            "max_distance": max_distance,
+            "window_minutes": window_minutes,
+            "grouped_at": datetime.now(timezone.utc).isoformat(),
+        },
+    )
+    session.add(post)
+    await session.flush()
+
+    await session.execute(
+        update(Post)
+        .where(Post.id.in_([m.id for m in members]))
+        .values(absorbed_by_post_id=post.id)
+    )
+
+    # Link every member image to the synthetic post via provenance. The feed
+    # and detail views already union provenance with primary_post_id
+    # (post_feed_service._thumbnails_for), so this alone makes the drop's
+    # images show up under the post FC wrote — no second render path.
+    #
+    # primary_post_id is deliberately NOT rewritten: the message post remains
+    # the image's true origin, and the synthetic post is an ADDITIONAL claim on
+    # it, which is what keeps the grouping reversible.
+    image_rows = (await session.execute(
+        select(ImageRecord.id).where(ImageRecord.primary_post_id.in_([m.id for m in members]))
+    )).scalars().all()
+    if image_rows:
+        await session.execute(
+            pg_insert(ImageProvenance)
+            .values([
+                {"image_record_id": iid, "post_id": post.id, "source_id": source.id}
+                for iid in image_rows
+            ])
+            # (image, post) is unique; a re-run that raced itself is a no-op
+            # rather than an IntegrityError that loses the whole sweep.
+            .on_conflict_do_nothing(constraint="uq_image_provenance_image_post")
+        )
+    return post
+
+
+async def group_source(
+    session: AsyncSession,
+    source: Source,
+    *,
+    max_distance: float,
+    window_minutes: float,
+    now: datetime | None = None,
+) -> int:
+    """Group one Discord source's ungrouped messages. Returns posts created."""
+    window = timedelta(minutes=window_minutes)
+    now = now or datetime.now(timezone.utc)
+    # Leave the most recent window alone: a drop that is still arriving would
+    # otherwise be cut in half by whichever sweep happened to land mid-drop,
+    # and the second half would become a separate post claiming to be its own
+    # drop. Waiting one window costs nothing (the sweep re-runs) and is the E2
+    # side of "keep the grouping open"; E3 handles the harder case where a
+    # matching drop resumes after the gap has already passed.
+    rows = (await session.execute(
+        _candidate_stmt(source.id, not_after=now - window)
+        .limit(MAX_CANDIDATES_PER_SOURCE)
+    )).all()
+    if not rows:
+        return 0
+
+    groups = build_groups(
+        [(pid, occurred, emb) for pid, occurred, emb in rows],
+        max_distance=max_distance, window=window,
+    )
+    if len(rows) == MAX_CANDIDATES_PER_SOURCE and len(groups) > 1:
+        # The cap may have fallen INSIDE the last drop, and synthesising a
+        # truncated group would publish a post that claims to be the whole drop
+        # while the rest of it sits one row past the limit. Leave it for the
+        # next run, which starts from the same place and sees the remainder.
+        # Guarded on len > 1 so a single oversized group is not dropped
+        # forever — it would make no progress at all.
+        groups = groups[:-1]
+
+    created = 0
+    for group in groups:
+        post = await _synthesize(
+            session, source=source, group=group,
+            max_distance=max_distance, window_minutes=window_minutes,
+        )
+        if post is not None:
+            created += 1
+    return created
+
+
+async def sweep(session: AsyncSession, *, now: datetime | None = None) -> dict:
+    """Group every enabled Discord source. No-op when the switch is off."""
+    settings = await MLSettings.load(session)
+    if not settings.discord_grouping_enabled:
+        return {"enabled": False, "sources": 0, "posts_created": 0}
+
+    sources = (await session.execute(
+        select(Source).where(
+            Source.platform == PLATFORM,
+            Source.enabled.is_(True),
+        )
+    )).scalars().all()
+
+    created = 0
+    for source in sources:
+        created += await group_source(
+            session, source,
+            max_distance=float(settings.discord_group_max_distance),
+            window_minutes=float(settings.discord_group_window_minutes),
+            now=now,
+        )
+    log.info(
+        "discord drop grouping: %d source(s), %d synthetic post(s) created",
+        len(sources), created,
+    )
+    return {"enabled": True, "sources": len(sources), "posts_created": created}

backend/app/services/post_feed_service.py

@@ -86,6 +86,13 @@ class PostFeedService:
             .join(Artist, Post.artist_id == Artist.id)
             .outerjoin(Source, Post.source_id == Source.id)
         )
+        # Absorbed posts are the individual chat messages a synthetic post
+        # replaced (milestone 388 E2). They stay in the table — they are the
+        # images' true origin and the grouping has to be auditable — but the
+        # feed shows the post FC authored, not the dozen lines it was built
+        # from. `around` and `get_post` deliberately do NOT apply this: reaching
+        # a member by id is how you inspect a grouping.
+        stmt = stmt.where(Post.absorbed_by_post_id.is_(None))
         if artist_id is not None:
             stmt = stmt.where(Post.artist_id == artist_id)
         if platform is not None:
@@ -400,6 +407,18 @@ class PostFeedService:
             "translated_source_lang": post.translated_source_lang,
             # Sticky per-post translation choice (auto/force/original, #155).
             "translation_override": post.translation_override,
+            # Milestone 388 E2. Non-null means FC AUTHORED this post by grouping
+            # a creator's drop — the UI must say so wherever the post appears,
+            # and `synthesis` carries what it was built from so the operator can
+            # audit a grouping FC invented. Null for every real post; the two
+            # keys are always present so the frontend never branches on absence.
+            "synthesized_by": post.synthesized_by,
+            "synthesis": post.synthesis_details,
+            # 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
+            # stream.
+            "absorbed_by_post_id": post.absorbed_by_post_id,
             "artist": {"id": artist.id, "name": artist.name, "slug": artist.slug},
             "source": (
                 {"id": source.id, "platform": source.platform}

backend/app/tasks/maintenance.py

@@ -1131,3 +1131,40 @@ def vacuum_analyze() -> dict:
             done.append(table)
     log.info("vacuum_analyze complete: %s", done)
     return {"vacuumed": done}
+
+
+@celery.task(
+    name="backend.app.tasks.maintenance.group_discord_drops",
+    soft_time_limit=1800, time_limit=2100,
+)
+def group_discord_drops() -> str:
+    """Milestone 388 E2: group Discord message-posts into the drops FC authors.
+
+    Lives on the MAINTENANCE lane, not the ml lane, even though it reads SigLIP
+    vectors — it does no inference and imports no ML library, and the ml-worker
+    is an OPTIONAL container (B3). Routing it to 'ml' would silently disable
+    grouping on every stack that runs a GPU agent and drops that container,
+    which is the same trap gpu_queue.py was moved here to avoid.
+
+    Async body under its own loop, per the _async_session contract: the sweep
+    needs pgvector column reads and the shared services are async.
+    """
+    import asyncio
+
+    from ..services.discord_grouping import sweep
+    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 sweep(session)
+                await session.commit()
+            return result
+        finally:
+            await engine.dispose()
+
+    res = asyncio.run(_run())
+    if not res["enabled"]:
+        return "disabled"
+    return f"sources={res['sources']} created={res['posts_created']}"

tests/test_discord_grouping.py

@@ -0,0 +1,389 @@
+"""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,
+    build_groups,
+    cosine_distance,
+    group_source,
+    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)
+    assert result == {"enabled": False, "sources": 0, "posts_created": 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