feat: FC authors the post that Discord never wrote (milestone 388 step E2)
CI / lint (push) Failing after 3s
CI / extension-version (push) Successful in 3s
Build images / sign-extension (push) Successful in 5s
Build images / build-agent (push) Successful in 9s
CI / frontend-build (push) Successful in 25s
CI / backend-lint-and-test (push) Successful in 31s
Build images / build-web (push) Successful in 1m5s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 1m53s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m5s
CI / lint (push) Failing after 3s
CI / extension-version (push) Successful in 3s
Build images / sign-extension (push) Successful in 5s
Build images / build-agent (push) Successful in 9s
CI / frontend-build (push) Successful in 25s
CI / backend-lint-and-test (push) Successful in 31s
Build images / build-web (push) Successful in 1m5s
Build images / smoke-web (push) Skipped
Build images / build-ml (push) Successful in 1m53s
Build images / promote (push) Skipped
CI / integration (push) Successful in 2m5s
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
This commit is contained in:
@@ -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")
|
||||
@@ -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"):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
)
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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}
|
||||
@@ -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}
|
||||
|
||||
@@ -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']}"
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user