diff --git a/alembic/versions/0092_synthetic_posts.py b/alembic/versions/0092_synthetic_posts.py new file mode 100644 index 0000000..dd551f5 --- /dev/null +++ b/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") diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 5660090..4529df1 100644 --- a/backend/app/api/ml_admin.py +++ b/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"): diff --git a/backend/app/celery_app.py b/backend/app/celery_app.py index d8983ed..393cd84 100644 --- a/backend/app/celery_app.py +++ b/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 diff --git a/backend/app/models/ml_settings.py b/backend/app/models/ml_settings.py index 4705d1a..e11c0a1 100644 --- a/backend/app/models/ml_settings.py +++ b/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() ) diff --git a/backend/app/models/post.py b/backend/app/models/post.py index e3f8e5e..b2f064f 100644 --- a/backend/app/models/post.py +++ b/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 + ) diff --git a/backend/app/services/discord_grouping.py b/backend/app/services/discord_grouping.py new file mode 100644 index 0000000..3eb0b56 --- /dev/null +++ b/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} diff --git a/backend/app/services/post_feed_service.py b/backend/app/services/post_feed_service.py index d4aec83..24d5cb6 100644 --- a/backend/app/services/post_feed_service.py +++ b/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} diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 10b01ec..7346841 100644 --- a/backend/app/tasks/maintenance.py +++ b/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']}" diff --git a/tests/test_discord_grouping.py b/tests/test_discord_grouping.py new file mode 100644 index 0000000..aa57900 --- /dev/null +++ b/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