From 1e45e2c56c77c125e952fc1b34ce5ce0ea981207 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 10 Sep 2026 11:30:17 -0400 Subject: [PATCH] =?UTF-8?q?feat:=20an=20open=20grouping=20=E2=80=94=20a=20?= =?UTF-8?q?later=20drop=20joins=20its=20post=20(milestone=20388=20step=20E?= =?UTF-8?q?3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A synthetic post is no longer sealed at creation. A creator who adds two more variants the next day extends the existing post, its body grows with the new messages, and no rival post appears. That is what makes chat capture read as content trickling in rather than as a stream of separate arrivals. The sweep now runs two passes per source and the ORDER is load-bearing: offer new messages to still-open groups BEFORE founding new ones, because whichever runs first claims a message. E3's three named problems, each answered rather than discovered later: **Bridging.** A candidate near two groups joins NEITHER. Nearest-wins would silently make an arbitrary choice between two posts the operator may already have seen; merging them is worse still, because a merge rewrites history and anything pointing at the absorbed post dangles. Leaving it to found its own group is the recoverable failure. AMBIGUITY_MARGIN is a module constant and deliberately not a setting — it is not a quality dial anyone would tune toward a better feed, and exposing it would invite turning it to zero, which is exactly the silent arbitrary choice it prevents. **Re-surfacing without thrashing.** A grouping has two dates, and which one orders the feed is a real decision, so the feed orders by neither directly. Ordering by when the drop STARTED buries a group that grows a week later under a week of other posts — defeating the point of keeping it open. Ordering by every growth lets a group gaining one image a day live permanently at the top, so chat out-competes authored posts for the front page — the opposite of "post pacing stays front and centre". Instead `resurfaced_at` moves only when growth clears BOTH a minimum-images bar and a cooldown, so a drip-feed updates in place and a genuine second wave resurfaces exactly once. It is NULL on every ordinary post, so the sort key COALESCEs through it without moving anything that is not a grouping. **Reopening forever.** Groups close after a quiet period — artists reuse characters for years, and a group left open indefinitely will eventually absorb something it shouldn't. Openness is DERIVED, not stored: a group is open if it grew (or started) within the window. Lowering the setting closes old groups and raising it reopens them, with nothing to repair either way; a stored closed_at would have needed a sweep to set it and a repair path to ever change the policy. Rule 89 is satisfied structurally rather than by a parallel mechanism: celery_signals writes a TaskRun for every task, which already supplies duration, the 5-minute stalled-run recovery, and retention pruning. What this step owed on top of that was a wall-clock limit (present) and idempotence — re-running the joiner adds nothing, asserted directly rather than left to the unique (image, post) constraint to catch. Two bugs fixed in the writing, one of which my own test would have hit: * `assign_to_group` sorted bare (distance, Post) tuples, which falls through to comparing Posts when two distances tie — and a perfectly symmetric bridge, the exact case the function exists for, would have raised TypeError instead of declining to choose. Now keyed on the distance alone. * The cursor was still built from `post_date or downloaded_at` while the ORDER BY had gained `resurfaced_at`. Two expressions that disagree at a page boundary don't error, they silently skip or repeat rows; both sites now go through one `_post_sort_value`, and a test pages through one row at a time to prove the walk matches the whole list. Image linking is now one shared helper rather than written twice, because creation and joining would otherwise be free to drift on exactly the detail (which post owns the image) that makes a grouping reversible. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01LNXXULQDjVZmbuNa2G9mD9 --- alembic/versions/0093_open_groupings.py | 86 +++++ backend/app/api/ml_admin.py | 15 + backend/app/models/ml_settings.py | 29 ++ backend/app/models/post.py | 24 ++ backend/app/services/discord_grouping.py | 356 ++++++++++++++++-- backend/app/services/post_feed_service.py | 47 ++- backend/app/tasks/maintenance.py | 5 +- frontend/src/components/posts/PostCard.vue | 27 +- .../settings/DiscordGroupingCard.vue | 52 +++ frontend/test/components/postCard.spec.js | 30 ++ tests/test_discord_grouping.py | 341 +++++++++++++++++ 11 files changed, 974 insertions(+), 38 deletions(-) create mode 100644 alembic/versions/0093_open_groupings.py diff --git a/alembic/versions/0093_open_groupings.py b/alembic/versions/0093_open_groupings.py new file mode 100644 index 0000000..4317375 --- /dev/null +++ b/alembic/versions/0093_open_groupings.py @@ -0,0 +1,86 @@ +"""An open grouping — a synthetic post that a later drop can still join. + +Milestone 388, step E3. E2's synthetic post was sealed at creation: a creator +who added two more variants the next day started a second post. These two +columns let the group stay open and absorb the follow-up, without the post +either freezing or thrashing the feed. + +## Why openness is derived rather than stored + +There is no `closed_at` here on purpose. A group is open if it grew (or +started) within `ml_settings.discord_group_close_after_hours`, so openness is a +comparison rather than a state — which means lowering the setting closes old +groups and raising it reopens them, with nothing to repair either way. A stored +flag would need its own sweep to set it and its own repair path to ever change +the policy, for no gain. + +## Why `resurfaced_at` is separate from `last_grew_at` + +They answer different questions. `last_grew_at` is when the group last +absorbed something — it decides how long the group stays joinable and is what +the card shows. `resurfaced_at` is the FEED POSITION, advanced only when the +anti-thrash rule fires, so a group that gains one image a day updates in place +while a genuine second wave moves once. Folding them together would make every +addition a bump, which is the annoyance this step exists to avoid. + +Both are NULL on every ordinary post, so the feed's sort key can COALESCE +through `resurfaced_at` without moving anything that is not a grouping. + +Revision ID: 0093 +Revises: 0092 +Create Date: 2026-09-10 + +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0093" +down_revision: Union[str, None] = "0092" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "post", sa.Column("last_grew_at", sa.DateTime(timezone=True), nullable=True), + ) + op.add_column( + "post", sa.Column("resurfaced_at", sa.DateTime(timezone=True), nullable=True), + ) + # No index on either. The feed already sorts on an unindexed + # COALESCE(post_date, downloaded_at) expression, so adding resurfaced_at to + # that COALESCE changes nothing about how the query plans — and inventing a + # functional index here would be guessing at the fix for a cost nobody has + # measured. Measuring it is step B2's job. + + op.add_column( + "ml_settings", + sa.Column( + "discord_group_close_after_hours", sa.Float(), + server_default=sa.text("168"), nullable=False, + ), + ) + op.add_column( + "ml_settings", + sa.Column( + "discord_group_resurface_min_images", sa.Integer(), + server_default="2", nullable=False, + ), + ) + op.add_column( + "ml_settings", + sa.Column( + "discord_group_resurface_cooldown_hours", sa.Float(), + server_default=sa.text("24"), nullable=False, + ), + ) + + +def downgrade() -> None: + op.drop_column("ml_settings", "discord_group_resurface_cooldown_hours") + op.drop_column("ml_settings", "discord_group_resurface_min_images") + op.drop_column("ml_settings", "discord_group_close_after_hours") + op.drop_column("post", "resurfaced_at") + op.drop_column("post", "last_grew_at") diff --git a/backend/app/api/ml_admin.py b/backend/app/api/ml_admin.py index 4529df1..ff3483f 100644 --- a/backend/app/api/ml_admin.py +++ b/backend/app/api/ml_admin.py @@ -54,6 +54,11 @@ _EDITABLE = ( "discord_grouping_enabled", "discord_group_max_distance", "discord_group_window_minutes", + # E3: how long a grouping stays open, and the anti-thrash rule that keeps + # a growing one from monopolising the feed. + "discord_group_close_after_hours", + "discord_group_resurface_min_images", + "discord_group_resurface_cooldown_hours", *_DETECTOR_FIELDS, ) @@ -162,6 +167,16 @@ def _validate(p: dict) -> str | None: 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" + # A group must stay open at least as long as the drop window it was cut + # with, or the joiner could never reach a message the grouper deferred — + # the two would fight, and the symptom (drops that never grow) would look + # like the predicate failing rather than a settings contradiction. + if float(p["discord_group_close_after_hours"]) * 60 < float(p["discord_group_window_minutes"]): + return "discord_group_close_after_hours must be at least the drop window" + if int(p["discord_group_resurface_min_images"]) < 1: + return "discord_group_resurface_min_images must be >= 1" + if float(p["discord_group_resurface_cooldown_hours"]) < 0: + return "discord_group_resurface_cooldown_hours 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/models/ml_settings.py b/backend/app/models/ml_settings.py index e11c0a1..021a56f 100644 --- a/backend/app/models/ml_settings.py +++ b/backend/app/models/ml_settings.py @@ -286,6 +286,35 @@ class MLSettings(Base): Float, nullable=False, default=60.0, server_default=text("60"), ) + # How long a synthetic post keeps accepting new members (#388 E3). This is + # NOT the drop window above: the window cuts one sweep's messages into + # drops, this decides how long a finished drop can still be REJOINED when a + # creator adds variants days later. A week by default — long enough for the + # "and here is the alt outfit" follow-up that motivated the feature, short + # enough that a group does not still be open when the same character comes + # round again months later and gets absorbed by mistake. + # + # Openness is DERIVED from this, not stored: a group is open if it grew (or + # started) within this period. So lowering it closes old groups and raising + # it reopens them, which is comprehensible and reversible — the alternative, + # a stored closed_at, would need its own repair path to ever change. + discord_group_close_after_hours: Mapped[float] = mapped_column( + Float, nullable=False, default=168.0, + server_default=text("168"), + ) + # Anti-thrash (#388 E3). An updated post SHOULD be visible — that is the + # point of keeping it open — but a group gaining one image a day must not + # monopolise the feed. Growth smaller than this never moves the post, and + # no group moves more than once per cooldown, so a drip-feed updates in + # place while a real second wave resurfaces exactly once. + discord_group_resurface_min_images: Mapped[int] = mapped_column( + Integer, nullable=False, default=2, + server_default="2", + ) + discord_group_resurface_cooldown_hours: Mapped[float] = mapped_column( + Float, nullable=False, default=24.0, + server_default=text("24"), + ) 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 b2f064f..8fa07fb 100644 --- a/backend/app/models/post.py +++ b/backend/app/models/post.py @@ -136,3 +136,27 @@ class Post(Base): absorbed_by_post_id: Mapped[int | None] = mapped_column( ForeignKey("post.id", ondelete="SET NULL"), nullable=True, index=True ) + # -- An OPEN grouping (milestone 388 E3) ------------------------------- + # A synthetic post is not sealed at creation: a creator who adds two more + # variants the next day extends the existing post rather than starting a + # new one. These two columns are what make that possible without the post + # either freezing or thrashing the feed. + # + # `last_grew_at` is when the group last absorbed something. It answers two + # questions: how long the group stays JOINABLE (a group closes after a + # quiet period — artists reuse characters for years, and a group left open + # forever will eventually absorb something it shouldn't), and what the card + # shows as "updated N ago". NULL means it has never grown since creation. + last_grew_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + # The feed position, and ONLY set when the anti-thrash rule fires — see + # discord_grouping.should_resurface. A group that gains one image a day + # must not sit permanently at the top of the feed, so growth updates the + # post without necessarily moving it; a genuine second wave moves it once. + # + # NULL on every ordinary post, which is why the feed's sort key can + # COALESCE through it without changing where anything else lands. + resurfaced_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) diff --git a/backend/app/services/discord_grouping.py b/backend/app/services/discord_grouping.py index b224e78..f0f2b6a 100644 --- a/backend/app/services/discord_grouping.py +++ b/backend/app/services/discord_grouping.py @@ -245,6 +245,10 @@ async def _synthesize( "max_distance": max_distance, "window_minutes": window_minutes, "grouped_at": datetime.now(UTC).isoformat(), + # Growth accumulated since the post last moved in the feed (E3). + # Seeded here so the joiner never meets its absence — creation IS + # a surfacing, so the count starts at zero. + "images_since_surface": 0, }, ) session.add(post) @@ -256,31 +260,52 @@ async def _synthesize( .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") - ) + await _link_member_images( + session, post_id=post.id, member_ids=[m.id for m in members], + source_id=source.id, + ) return post +async def _link_member_images( + session: AsyncSession, *, post_id: int, member_ids: list[int], source_id: int | None, +) -> int: + """Attach every member's images to the synthetic post. Returns how many. + + 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. + + Shared by creation and by the E3 joiner rather than written twice, because + the two would otherwise be free to drift on exactly the detail (which post + owns the image) that makes a grouping reversible. + """ + if not member_ids: + return 0 + image_rows = (await session.execute( + select(ImageRecord.id).where(ImageRecord.primary_post_id.in_(member_ids)) + )).scalars().all() + if not image_rows: + return 0 + 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. This is a BACKSTOP against a re-run that + # raced itself, not the correctness argument: callers only ever pass + # members that were unabsorbed a moment ago, so a conflict here means + # concurrency, not a logic error. + .on_conflict_do_nothing(constraint="uq_image_provenance_image_post") + ) + return len(image_rows) + + async def group_source( session: AsyncSession, source: Source, @@ -329,11 +354,265 @@ async def group_source( return created +# --------------------------------------------------------------------------- +# E3: an open grouping — a later drop joins its post and updates it. +# --------------------------------------------------------------------------- +# +# A synthetic post is not sealed at creation. A creator who adds two more +# variants the next day extends the existing post rather than starting a new +# one, and its body grows with the new messages. That is what makes chat +# capture read as content TRICKLING IN rather than as a stream of separate +# arrivals. +# +# Three hard problems, each answered deliberately below: bridging (a candidate +# near two groups), re-surfacing without thrashing the feed, and groups that +# stay open forever. + +# How much closer the nearest group must be than the runner-up before a +# candidate is assigned to it at all. +# +# NOT a setting, deliberately. It is not a quality dial the operator would tune +# toward a better feed — it expresses "these two are too close to call", and +# exposing it would invite turning it to zero, which is precisely the silent +# arbitrary choice it exists to prevent. When a candidate is genuinely between +# two groups the recoverable answer is to leave it out and let it start its +# own; the unrecoverable one is to merge, because a merge rewrites history — +# two posts the operator may already have seen become one, and anything +# pointing at the absorbed post dangles. +AMBIGUITY_MARGIN = 0.02 + + +def should_resurface( + *, + images_since_surface: int, + last_surface_at: datetime, + now: datetime, + min_images: int, + cooldown: timedelta, +) -> bool: + """Has this group grown enough, and waited long enough, to move in the feed? + + An updated post SHOULD be visible — that is the point of keeping it open — + but a group gaining one image a day must not sit permanently at the top. + Both conditions have to hold: enough new images that the update is worth an + interruption, and enough time since the last one that a steady drip cannot + chain bumps together. + """ + if images_since_surface < min_images: + return False + return now - last_surface_at >= cooldown + + +async def _group_seed(session: AsyncSession, post_id: int) -> list[float] | None: + """The embedding a group is measured against — its FIRST member's image. + + Derived rather than stored, and derived by the same definition `build_groups` + used (earliest member, lowest-id embedded image). Storing it at creation + would have meant a backfill for groups already written and two definitions + free to disagree; this way there is one. + """ + sort_key = func.coalesce(Post.post_date, Post.downloaded_at) + return (await session.execute( + select(ImageRecord.siglip_embedding) + .join(Post, ImageRecord.primary_post_id == Post.id) + .where( + Post.absorbed_by_post_id == post_id, + ImageRecord.siglip_embedding.is_not(None), + ) + .order_by(sort_key, Post.id, ImageRecord.id) + .limit(1) + )).scalar_one_or_none() + + +async def open_groups( + session: AsyncSession, source_id: int, *, now: datetime, close_after: timedelta, +) -> list[tuple[Post, list[float]]]: + """This source's synthetic posts that are still accepting members. + + Openness is DERIVED, not stored: a group is open if it grew — or started — + within `close_after`. A group left open forever would eventually absorb + something it shouldn't, because artists reuse characters for years; a + stored `closed_at` would need a sweep to set it and a repair path to ever + change the policy. This way the policy IS the query. + """ + cutoff = now - close_after + posts = (await session.execute( + select(Post).where( + Post.source_id == source_id, + Post.synthesized_by == DROP_GROUPER, + # A synthetic post that was itself absorbed is not a thing today + # (nothing absorbs one), but joining into one would nest groups. + Post.absorbed_by_post_id.is_(None), + func.coalesce(Post.last_grew_at, Post.post_date, Post.downloaded_at) >= cutoff, + ) + )).scalars().all() + + out: list[tuple[Post, list[float]]] = [] + for post in posts: + seed = await _group_seed(session, post.id) + if seed is not None: + out.append((post, seed)) + return out + + +def assign_to_group( + embedding: list[float], + groups: list[tuple[Post, list[float]]], + *, + max_distance: float, +) -> Post | None: + """Pick the one group this image belongs to, or None to leave it alone. + + Returns None in two cases that mean different things and are deliberately + treated the same: nothing is close enough (so E2 will start a new group + from it), or two groups are BOTH close and too near each other to choose + between (so E2 will start a new group from it). The second is the bridging + case, and letting it start its own post is the recoverable failure — + merging two existing posts is not. + """ + # key= on the distance ALONE. Sorting bare tuples falls through to the + # second element when two distances tie, and `Post` has no ordering — so a + # perfectly symmetric bridge (the exact case this function exists for) + # would raise TypeError instead of declining to choose. + scored = sorted( + ((cosine_distance(seed, embedding), post) for post, seed in groups), + key=lambda pair: pair[0], + ) + within = [(d, p) for d, p in scored if d <= max_distance] + if not within: + return None + if len(within) >= 2 and (within[1][0] - within[0][0]) < AMBIGUITY_MARGIN: + return None + return within[0][1] + + +async def _absorb_into( + session: AsyncSession, + *, + group: Post, + member_ids: list[int], + source_id: int | None, + now: datetime, + min_images: int, + cooldown: timedelta, +) -> int: + """Extend an existing synthetic post with new members. Returns images added.""" + members = (await session.execute( + select(Post) + .where(Post.id.in_(member_ids)) + .order_by(func.coalesce(Post.post_date, Post.downloaded_at), Post.id) + )).scalars().all() + if not members: + return 0 + + added_images = await _link_member_images( + session, post_id=group.id, member_ids=[m.id for m in members], + source_id=source_id, + ) + await session.execute( + update(Post) + .where(Post.id.in_([m.id for m in members])) + .values(absorbed_by_post_id=group.id) + ) + + # The new messages' text joins the body, in arrival order, exactly as at + # creation — the post's body is the drop's text and the drop just grew. + new_text = "\n\n".join( + m.description.strip() for m in members if m.description and m.description.strip() + ) + if new_text: + group.description = f"{group.description}\n\n{new_text}" if group.description else new_text + + details = dict(group.synthesis_details or {}) + existing_ids = list(details.get("member_post_ids") or []) + details["member_post_ids"] = existing_ids + [ + m.id for m in members if m.id not in existing_ids + ] + details["message_count"] = len(details["member_post_ids"]) + since = int(details.get("images_since_surface") or 0) + added_images + details["last_grew_at"] = now.isoformat() + + # The feed position moves only when the anti-thrash rule fires. Measured + # from the last time the post actually MOVED (resurfaced_at), falling back + # to when the drop started — creation is itself a surfacing. + last_surface = group.resurfaced_at or group.post_date or group.downloaded_at + if should_resurface( + images_since_surface=since, last_surface_at=last_surface, now=now, + min_images=min_images, cooldown=cooldown, + ): + group.resurfaced_at = now + since = 0 + details["images_since_surface"] = since + + group.synthesis_details = details + group.last_grew_at = now + return added_images + + +async def join_open_groups( + session: AsyncSession, + source: Source, + *, + max_distance: float, + window_minutes: float, + close_after_hours: float, + resurface_min_images: int, + resurface_cooldown_hours: float, + now: datetime | None = None, +) -> int: + """Offer this source's ungrouped messages to its open groups. + + Runs BEFORE `group_source` in the sweep: a message that belongs to an + existing drop must join it rather than found a rival post, and whichever + runs first wins that message. + """ + now = now or datetime.now(UTC) + window = timedelta(minutes=window_minutes) + groups = await open_groups( + session, source.id, now=now, close_after=timedelta(hours=close_after_hours), + ) + if not groups: + return 0 + + # Same quarantine as E2: a drop still arriving is left for the next run. + rows = (await session.execute( + _candidate_stmt(source.id, not_after=now - window) + .limit(MAX_CANDIDATES_PER_SOURCE) + )).all() + if not rows: + return 0 + + claimed: dict[int, list[int]] = {} + for post_id, _occurred_at, embedding in rows: + target = assign_to_group(embedding, groups, max_distance=max_distance) + if target is not None: + claimed.setdefault(target.id, []).append(post_id) + + by_id = {post.id: post for post, _seed in groups} + joined = 0 + for group_id, member_ids in claimed.items(): + joined += await _absorb_into( + session, group=by_id[group_id], member_ids=member_ids, + source_id=source.id, now=now, + min_images=resurface_min_images, + cooldown=timedelta(hours=resurface_cooldown_hours), + ) + return joined + + async def sweep(session: AsyncSession, *, now: datetime | None = None) -> dict: - """Group every enabled Discord source. No-op when the switch is off.""" + """Group every enabled Discord source. No-op when the switch is off. + + Two passes per source, and the ORDER is load-bearing: offer new messages to + the groups that are still open (E3) BEFORE founding new ones (E2). Whichever + runs first claims a message, and a variant that belongs to yesterday's drop + must extend that post rather than found a rival to it. + """ settings = await MLSettings.load(session) if not settings.discord_grouping_enabled: - return {"enabled": False, "sources": 0, "posts_created": 0} + return { + "enabled": False, "sources": 0, "posts_created": 0, "images_joined": 0, + } sources = (await session.execute( select(Source).where( @@ -342,16 +621,35 @@ async def sweep(session: AsyncSession, *, now: datetime | None = None) -> dict: ) )).scalars().all() + max_distance = float(settings.discord_group_max_distance) + window_minutes = float(settings.discord_group_window_minutes) + created = 0 + joined = 0 for source in sources: + joined += await join_open_groups( + session, source, + max_distance=max_distance, + window_minutes=window_minutes, + close_after_hours=float(settings.discord_group_close_after_hours), + resurface_min_images=int(settings.discord_group_resurface_min_images), + resurface_cooldown_hours=float( + settings.discord_group_resurface_cooldown_hours + ), + now=now, + ) created += await group_source( session, source, - max_distance=float(settings.discord_group_max_distance), - window_minutes=float(settings.discord_group_window_minutes), + max_distance=max_distance, + window_minutes=window_minutes, now=now, ) log.info( - "discord drop grouping: %d source(s), %d synthetic post(s) created", - len(sources), created, + "discord drop grouping: %d source(s), %d synthetic post(s) created, " + "%d image(s) joined to open groups", + len(sources), created, joined, ) - return {"enabled": True, "sources": len(sources), "posts_created": created} + return { + "enabled": True, "sources": len(sources), + "posts_created": created, "images_joined": joined, + } diff --git a/backend/app/services/post_feed_service.py b/backend/app/services/post_feed_service.py index 24d5cb6..80c4068 100644 --- a/backend/app/services/post_feed_service.py +++ b/backend/app/services/post_feed_service.py @@ -41,8 +41,40 @@ THUMBNAIL_LIMIT = 6 def _sort_key(): - """Postgres COALESCE expression used in ORDER BY and WHERE clauses.""" - return func.coalesce(Post.post_date, Post.downloaded_at) + """Postgres COALESCE expression used in ORDER BY and WHERE clauses. + + `resurfaced_at` leads (milestone 388 E3). A synthetic post stays OPEN — a + creator who adds variants the next day extends the existing post — so such + a post has two dates, and which one orders the feed is a real decision: + + * ordering by when the drop STARTED buries a group that grows a week later + under a week of other posts, so the operator never sees the new content — + which defeats keeping the group open at all; + * ordering by every growth lets a group that gains one image a day sit + permanently at the top, so chat out-competes authored posts for the front + page — the opposite of "post pacing stays front and centre". + + So the feed orders by neither directly. `resurfaced_at` moves only when the + anti-thrash rule fires (discord_grouping.should_resurface: enough new + images AND enough time since the last move), which means a drip-feed + updates IN PLACE and a genuine second wave resurfaces exactly once. + + It is NULL on every ordinary post, so this COALESCE cannot move anything + that is not a grouping. Used identically in ORDER BY and in the cursor's + WHERE, which is what keeps pagination stable across the change. + """ + return func.coalesce(Post.resurfaced_at, Post.post_date, Post.downloaded_at) + + +def _post_sort_value(post: Post): + """The Python twin of `_sort_key()`, for building a cursor from a loaded row. + + Kept next to it on purpose: these two are one expression in two languages, + and the failure when they disagree is not an error but a quiet one — rows + skipped or repeated at page boundaries, which reads as a backend bug + anywhere but here. + """ + return post.resurfaced_at or post.post_date or post.downloaded_at class PostFeedService: @@ -134,7 +166,10 @@ class PostFeedService: # Far edge in the travel direction: oldest row going older, # newest row going newer (rows is descending for display). edge_post = rows[-1][0] if direction == "older" else rows[0][0] - edge_key = edge_post.post_date or edge_post.downloaded_at + # Must match _sort_key() exactly, including resurfaced_at's + # precedence: a cursor built from a different expression than the + # ORDER BY silently skips or repeats rows at every page boundary. + edge_key = _post_sort_value(edge_post) next_cursor = encode_cursor(edge_key, edge_post.id) post_ids = [p.id for p, _, _ in rows] @@ -168,7 +203,7 @@ class PostFeedService: if anchor is None: return None anchor_post, anchor_artist, anchor_source = anchor - anchor_key = anchor_post.post_date or anchor_post.downloaded_at + anchor_key = _post_sort_value(anchor_post) anchor_cursor = encode_cursor(anchor_key, anchor_post.id) older = await self.scroll( @@ -414,6 +449,10 @@ class PostFeedService: # keys are always present so the frontend never branches on absence. "synthesized_by": post.synthesized_by, "synthesis": post.synthesis_details, + # #388 E3. A grouping stays open, so the card can say "updated N + # ago" — which is the whole signal that chat content is trickling + # in. NULL means it has not grown since it was created. + "last_grew_at": post.last_grew_at.isoformat() if post.last_grew_at else None, # 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 diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 7346841..acbb3b1 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -1167,4 +1167,7 @@ def group_discord_drops() -> str: res = asyncio.run(_run()) if not res["enabled"]: return "disabled" - return f"sources={res['sources']} created={res['posts_created']}" + return ( + f"sources={res['sources']} created={res['posts_created']} " + f"joined={res['images_joined']}" + ) diff --git a/frontend/src/components/posts/PostCard.vue b/frontend/src/components/posts/PostCard.vue index 8ec2d5e..90bc754 100644 --- a/frontend/src/components/posts/PostCard.vue +++ b/frontend/src/components/posts/PostCard.vue @@ -27,6 +27,12 @@ · {{ totalImages }} image{{ totalImages === 1 ? '' : 's' }} + + + · updated {{ grewRelative }} + { const railCols = computed(() => rail.value.length + (moreCount.value > 0 ? 1 : 0)) const sortDateIso = computed(() => props.post.post_date || props.post.downloaded_at) + +// #388 E3. A grouping stays OPEN, so its own date and its latest activity are +// different facts. The card keeps showing when the drop STARTED — that is the +// post's identity — and reports growth separately, because "this post is from +// Tuesday but gained images this morning" is the whole signal that chat +// content is trickling in. +const grewAt = computed(() => (synthesized.value ? props.post.last_grew_at : null)) +const grewRelative = computed(() => (grewAt.value ? relativeFrom(grewAt.value) : '')) const absoluteDate = computed(() => new Date(sortDateIso.value).toLocaleString()) -const relativeDate = computed(() => { - const then = new Date(sortDateIso.value).getTime() +function relativeFrom (iso) { + const then = new Date(iso).getTime() const diff = (Date.now() - then) / 1000 if (diff < 60) return `${Math.floor(diff)}s ago` if (diff < 3600) return `${Math.floor(diff / 60)}m ago` if (diff < 86400) return `${Math.floor(diff / 3600)}h ago` if (diff < 86400 * 30) return `${Math.floor(diff / 86400)}d ago` - return new Date(sortDateIso.value).toLocaleDateString() -}) + return new Date(iso).toLocaleDateString() +} +const relativeDate = computed(() => relativeFrom(sortDateIso.value)) // --- images → post-scoped modal --------------------------------------- async function fullImageIds () { @@ -375,6 +390,10 @@ function formatBytes (n) { .fc-post-card__synthetic { color: rgb(var(--v-theme-on-surface-variant)); } + +/* Growth is news, so it gets the accent the rest of the meta line doesn't — + but it is still the meta line, not a badge competing with the artwork. */ +.fc-post-card__grew { color: rgb(var(--v-theme-accent)); } .fc-post-card__date, .fc-post-card__meta { white-space: nowrap; } diff --git a/frontend/src/components/settings/DiscordGroupingCard.vue b/frontend/src/components/settings/DiscordGroupingCard.vue index 6055c0d..c7a91d1 100644 --- a/frontend/src/components/settings/DiscordGroupingCard.vue +++ b/frontend/src/components/settings/DiscordGroupingCard.vue @@ -48,6 +48,54 @@ + + +
+ A grouped post stays open: variants the creator adds + later join the existing post instead of starting a new one, and its + text grows with them. +
+ + + +
+ How long after its last addition a post still accepts new variants. + Not the drop window above — that cuts one session into drops; this + decides how late a follow-up can still join. Too long and the same + character coming round again months later gets absorbed by mistake. +
+ + + +
+ Growth smaller than this updates the post where it sits instead of + moving it back to the top of the feed. +
+ + + +
+ Together with the count above, this is what stops a post that gains + an image a day from living permanently at the top of the feed. +
+ +
@@ -77,6 +125,10 @@ function onSave() { discord_grouping_enabled: Boolean(local.discord_grouping_enabled), discord_group_max_distance: Number(local.discord_group_max_distance), discord_group_window_minutes: Number(local.discord_group_window_minutes), + discord_group_close_after_hours: Number(local.discord_group_close_after_hours), + discord_group_resurface_min_images: Number(local.discord_group_resurface_min_images), + discord_group_resurface_cooldown_hours: + Number(local.discord_group_resurface_cooldown_hours), }) } diff --git a/frontend/test/components/postCard.spec.js b/frontend/test/components/postCard.spec.js index 68c259b..8758cbb 100644 --- a/frontend/test/components/postCard.spec.js +++ b/frontend/test/components/postCard.spec.js @@ -84,6 +84,36 @@ describe('PostCard', () => { expect(w.find('.fc-post-card__synthetic').exists()).toBe(false) }) + it('reports growth separately from the post date', () => { + // The drop's own date is its identity; growth is news about it. "From + // Tuesday, gained images this morning" is the trickling-in signal, and + // collapsing the two would erase it. + const w = mountComponent(PostCard, { + props: { + post: { ...SYNTH, last_grew_at: new Date(Date.now() - 3600e3).toISOString() }, + }, + pinia: freshPinia(), + }) + expect(w.text()).toContain('updated 1h ago') + }) + + it('says nothing about growth on a grouping that has not grown', () => { + // An "updated" label that is always there teaches you to ignore it. + const w = mountComponent(PostCard, { + props: { post: { ...SYNTH, last_grew_at: null } }, + pinia: freshPinia(), + }) + expect(w.text()).not.toContain('updated') + }) + + it('never claims an ordinary post grew, even if the field leaks in', () => { + const w = mountComponent(PostCard, { + props: { post: { ...BASE, last_grew_at: new Date().toISOString() } }, + pinia: freshPinia(), + }) + expect(w.text()).not.toContain('updated') + }) + it('singularises a one-message drop', () => { const w = mountComponent(PostCard, { props: { post: { ...SYNTH, synthesis: { message_count: 1 } } }, diff --git a/tests/test_discord_grouping.py b/tests/test_discord_grouping.py index aa57900..f5e3fe3 100644 --- a/tests/test_discord_grouping.py +++ b/tests/test_discord_grouping.py @@ -16,9 +16,12 @@ from sqlalchemy import select from backend.app.models import Artist, ImageProvenance, ImageRecord, MLSettings, Post, Source from backend.app.services.discord_grouping import ( DROP_GROUPER, + assign_to_group, build_groups, cosine_distance, group_source, + join_open_groups, + should_resurface, sweep, ) from backend.app.services.post_feed_service import PostFeedService @@ -387,3 +390,341 @@ async def test_the_sweep_only_touches_discord_sources(db): 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 + + +# --- E3: an open grouping ------------------------------------------------- +# +# Three hard problems, one test class each. The anti-thrash rule is ASSERTED +# rather than assumed, because "a group that grows daily monopolises the feed" +# is the kind of defect nobody notices until they have lived with it. + + +def test_growth_below_the_minimum_never_moves_the_post(): + """One image a day must not chain bumps together.""" + now = datetime.now(UTC) + assert not should_resurface( + images_since_surface=1, last_surface_at=now - timedelta(days=30), + now=now, min_images=2, cooldown=timedelta(hours=24), + ) + + +def test_growth_inside_the_cooldown_never_moves_the_post(): + """Even a big second wave waits — otherwise a burst bumps once per sweep.""" + now = datetime.now(UTC) + assert not should_resurface( + images_since_surface=10, last_surface_at=now - timedelta(hours=1), + now=now, min_images=2, cooldown=timedelta(hours=24), + ) + + +def test_a_real_second_wave_moves_the_post_once(): + now = datetime.now(UTC) + assert should_resurface( + images_since_surface=3, last_surface_at=now - timedelta(days=2), + now=now, min_images=2, cooldown=timedelta(hours=24), + ) + + +def _group(post_id: int, angle: float): + return (Post(id=post_id), _vec(angle)) + + +def test_a_candidate_near_two_groups_joins_neither(): + """Bridging. Merging would rewrite history — two posts the operator may + already have seen become one — so the recoverable answer is to leave it and + let it found its own group.""" + left, right = _group(1, -0.30), _group(2, 0.30) + assert assign_to_group(_vec(0.0), [left, right], max_distance=0.5) is None + + +def test_a_candidate_clearly_nearer_one_group_joins_it(): + near, far = _group(1, 0.02), _group(2, 1.2) + chosen = assign_to_group(_vec(0.0), [near, far], max_distance=0.5) + assert chosen is not None and chosen.id == 1 + + +def test_a_candidate_near_nothing_joins_nothing(): + assert assign_to_group( + _vec(0.0), [_group(1, math.pi / 2)], max_distance=0.10, + ) is None + + +@pytest.mark.asyncio +async def test_a_later_variant_joins_the_existing_post_instead_of_founding_one(db): + artist, source = await _seed(db, name="rejoin-artist") + start = datetime.now(UTC) - timedelta(days=3) + await _message(db, source, artist, ext="r1", at=start, vec=_vec(0.0), text="first") + await _message( + db, source, artist, ext="r2", at=start + timedelta(minutes=2), + vec=_vec(0.05), text="second", + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + group = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalar_one() + group_id = group.id + + # A day later — far outside the drop window, well inside the open period. + await _message( + db, source, artist, ext="r3", at=start + timedelta(days=1), + vec=_vec(0.06), text="next day", + ) + await db.commit() + + joined = await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, + resurface_cooldown_hours=24, + ) + await db.commit() + assert joined == 1 + + # No second post — that is the point of the step. + posts = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalars().all() + assert len(posts) == 1 + + db.expunge_all() + grown = await db.get(Post, group_id) + assert grown.last_grew_at is not None + assert grown.synthesis_details["message_count"] == 3 + # The new message's text joined the body, in arrival order. + assert grown.description == "first\n\nsecond\n\nnext day" + + +@pytest.mark.asyncio +async def test_a_variant_after_the_group_closed_starts_a_new_post(db): + """Groups must not stay open forever — artists reuse characters for years.""" + artist, source = await _seed(db, name="closed-artist") + start = datetime.now(UTC) - timedelta(days=60) + await _message(db, source, artist, ext="c1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="c2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + await _message( + db, source, artist, ext="c3", at=datetime.now(UTC) - timedelta(days=1), + vec=_vec(0.06), + ) + await db.commit() + + joined = await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, + resurface_cooldown_hours=24, + ) + await db.commit() + assert joined == 0, "a closed group must not absorb a two-month-later drop" + + created = await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + assert created == 1 + + +@pytest.mark.asyncio +async def test_joining_does_not_duplicate_images(db): + """The unique (image, post) constraint is the backstop, not the argument — + a second join pass must add nothing at all.""" + artist, source = await _seed(db, name="nodupe-artist") + start = datetime.now(UTC) - timedelta(days=3) + await _message(db, source, artist, ext="d1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="d2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + await _message( + db, source, artist, ext="d3", at=start + timedelta(days=1), vec=_vec(0.06), + ) + await db.commit() + + kwargs = { + "max_distance": 0.10, "window_minutes": 60, "close_after_hours": 168, + "resurface_min_images": 2, "resurface_cooldown_hours": 24, + } + assert await join_open_groups(db, source, **kwargs) == 1 + await db.commit() + assert await join_open_groups(db, source, **kwargs) == 0 + await db.commit() + + group = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalar_one() + links = (await db.execute( + select(ImageProvenance.image_record_id).where(ImageProvenance.post_id == group.id) + )).scalars().all() + assert len(links) == len(set(links)) == 3 + + +@pytest.mark.asyncio +async def test_a_group_that_grows_by_one_does_not_move_in_the_feed(db): + """The anti-thrash rule, end to end. A drip-feed updates IN PLACE.""" + artist, source = await _seed(db, name="dripfeed-artist") + start = datetime.now(UTC) - timedelta(days=3) + await _message(db, source, artist, ext="t1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="t2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + await _message( + db, source, artist, ext="t3", at=start + timedelta(days=1), vec=_vec(0.06), + ) + await db.commit() + await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24, + ) + await db.commit() + + group = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalar_one() + assert group.last_grew_at is not None, "it grew" + assert group.resurfaced_at is None, "but one image must not move it" + assert group.synthesis_details["images_since_surface"] == 1 + + +@pytest.mark.asyncio +async def test_a_second_wave_resurfaces_the_post(db): + artist, source = await _seed(db, name="secondwave-artist") + start = datetime.now(UTC) - timedelta(days=3) + await _message(db, source, artist, ext="w1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="w2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + for i, ext in enumerate(("w3", "w4", "w5")): + await _message( + db, source, artist, ext=ext, + at=start + timedelta(days=1, minutes=i), vec=_vec(0.06), + ) + await db.commit() + await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24, + ) + await db.commit() + + group = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalar_one() + assert group.resurfaced_at is not None + # The counter resets, so the NEXT trickle starts from zero rather than + # riding the same three images into a second bump. + assert group.synthesis_details["images_since_surface"] == 0 + + +@pytest.mark.asyncio +async def test_a_resurfaced_group_sorts_by_when_it_moved_not_when_it_started(db): + """The feed ordering decision, pinned. The group's drop began three days + ago — older than the decoy post — but it resurfaced just now, so it leads.""" + artist, source = await _seed(db, name="feedorder-artist") + start = datetime.now(UTC) - timedelta(days=3) + await _message(db, source, artist, ext="f1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="f2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + # An ordinary post from yesterday: newer than the drop's start, older than + # the moment the group resurfaces. + decoy = Post( + source_id=source.id, artist_id=artist.id, external_post_id="decoy", + post_date=datetime.now(UTC) - timedelta(days=1), post_title="Decoy", + ) + db.add(decoy) + await db.flush() + decoy_id = decoy.id + + for i, ext in enumerate(("f3", "f4", "f5")): + await _message( + db, source, artist, ext=ext, + at=start + timedelta(days=1, minutes=i), vec=_vec(0.06), + ) + await db.commit() + await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24, + ) + await db.commit() + + group = (await db.execute( + select(Post).where(Post.synthesized_by == DROP_GROUPER) + )).scalar_one() + page = await PostFeedService(db).scroll( + cursor=None, artist_id=artist.id, limit=50, + ) + ids = [i["id"] for i in page["items"]] + assert ids.index(group.id) < ids.index(decoy_id), ( + "a resurfaced group must lead a post published after the drop STARTED" + ) + # And it is still reported as a grouping that grew. + item = next(i for i in page["items"] if i["id"] == group.id) + assert item["last_grew_at"] is not None + + +@pytest.mark.asyncio +async def test_pagination_stays_stable_when_a_group_has_resurfaced(db): + """The cursor is built in Python and the ORDER BY in SQL. If they disagree + on resurfaced_at's precedence, rows are silently skipped or repeated at + every page boundary — so page through one at a time and check.""" + artist, source = await _seed(db, name="cursor-artist") + start = datetime.now(UTC) - timedelta(days=5) + await _message(db, source, artist, ext="x1", at=start, vec=_vec(0.0)) + await _message( + db, source, artist, ext="x2", at=start + timedelta(minutes=2), vec=_vec(0.05), + ) + await db.commit() + await group_source(db, source, max_distance=0.10, window_minutes=60) + await db.commit() + + for i in range(4): + db.add(Post( + source_id=source.id, artist_id=artist.id, + external_post_id=f"plain{i}", + post_date=datetime.now(UTC) - timedelta(days=i + 1), + )) + for i, ext in enumerate(("x3", "x4", "x5")): + await _message( + db, source, artist, ext=ext, + at=start + timedelta(days=1, minutes=i), vec=_vec(0.06), + ) + await db.commit() + await join_open_groups( + db, source, max_distance=0.10, window_minutes=60, + close_after_hours=168, resurface_min_images=2, resurface_cooldown_hours=24, + ) + await db.commit() + + svc = PostFeedService(db) + everything = [i["id"] for i in ( + await svc.scroll(cursor=None, artist_id=artist.id, limit=100) + )["items"]] + + walked, cursor = [], None + while True: + page = await svc.scroll( + cursor=cursor, artist_id=artist.id, limit=1, + ) + walked.extend(i["id"] for i in page["items"]) + cursor = page["next_cursor"] + if cursor is None: + break + assert walked == everything