From dec71386c3d26a66cefa8b3fad29674f5fff01cb Mon Sep 17 00:00:00 2001 From: TheMeinerLP Date: Fri, 21 Aug 2026 00:16:39 +0200 Subject: [PATCH 1/3] feat(discord): add /queue and let an admin re-run a finished session Every session recorded before the VAD fix on this branch has a document built from an empty or hallucinated transcript, and nothing in the system can produce a better one: JobQueue.claim selects only pending jobs, never done and never dead, so a finished job stays finished no matter how wrong its transcript is. /queue requeue is the explicit write that resurrects one. A re-queue is a state reset and nothing else. It returns the session to exactly the state it was in immediately after close_session and before documentation, and the machinery that already exists -- claim, complete, _create_session_document, announce_ready_sessions -- carries it forward a second time on its own. The decision of which jobs may be reset is a pure function over plain dicts in sturnus.application.requeue, beside sessions_to_announce and expired_jobs and tested the same way; the write is inline ORM in the cog, exactly as audio_cog._erase_audio is and for the same reason -- the selection is used by nothing else. Two rules keep the command from being destructive. A session with any job still pending or running is refused outright: the worker holding a running job will call complete() when it finishes, writing the old run's transcript back over the reset while the admin has already been told the reset happened. And a job whose audio_deleted_at is set is skipped rather than reset, because re-queueing it would only hand a worker a key it cannot download; its done status and its old transcript stay, so the new document is the redone speakers plus the untouched old text of the ones whose audio is gone -- which the reply says in as many words. Both of those are decided inside the same SELECT ... FOR UPDATE the JobQueue.complete takes, ORDER BY id included so the two cannot deadlock, and re-derived there rather than reused from the plan the confirmation was rendered from. Under READ COMMITTED a plan built outside that lock would see a snapshot predating a complete() already in flight. The job resets and the session reset share one transaction and one commit: half of that write landing would strand the session with no document and no error, since complete() and retry_pending_documents both require status == "closed". announced_at is cleared deliberately, and that is the choice to post a second link into the voice channel. mark_documented never touches that column and sessions_to_announce selects only sessions where it is null, so leaving it set would produce a session that transcribes, re-documents with a fresh URL and is then never announced -- nothing logged, nothing raised, the channel simply never told. A corrected transcript nobody hears about is indistinguishable from no transcript. The old document is orphaned rather than updated: DocumentSink exposes create and nothing else, anywhere in the codebase. That is accepted, not stumbled into, so the confirmation names the existing document_url and says it stays and will not be updated -- deleting or cross-linking it remains a human act. Confirmation is a discord.ui.View showing the plan, not a force flag: the three things an admin cannot undo -- discarded transcripts, a second document, a second public post -- are all on screen before a button exists. A second press finds the jobs already pending and is refused on that alone, so it cannot produce a second document or a second announcement. All three subcommands are admin-gated, including the read-only ones: even /queue status reports who was recorded and how much they said. Every query joins to session and filters on guild_id, and a session belonging to another guild gets the same reply as one that exists nowhere, so the command cannot be used to probe ids elsewhere. /queue session reports the length of a stored transcript and never its text -- 24 characters for a 100-minute recording is the tell, and that is all an admin needs to decide. --- src/sturnus/application/requeue.py | 142 +++ src/sturnus/infrastructure/discord/client.py | 8 + .../infrastructure/discord/queue_cog.py | 894 ++++++++++++++++ tests/application/test_requeue.py | 142 +++ .../discord/test_client_cogs.py | 10 +- .../infrastructure/discord/test_queue_cog.py | 992 ++++++++++++++++++ 6 files changed, 2187 insertions(+), 1 deletion(-) create mode 100644 src/sturnus/application/requeue.py create mode 100644 src/sturnus/infrastructure/discord/queue_cog.py create mode 100644 tests/application/test_requeue.py create mode 100644 tests/infrastructure/discord/test_queue_cog.py diff --git a/src/sturnus/application/requeue.py b/src/sturnus/application/requeue.py new file mode 100644 index 0000000..11a7fd8 --- /dev/null +++ b/src/sturnus/application/requeue.py @@ -0,0 +1,142 @@ +"""Which of a finished session's jobs may be transcribed a second time. + +Re-running a session means returning it to the exact state it was in +immediately after `close_session` and before documentation, so that the +machinery which already exists -- `JobQueue.claim`, `JobQueue.complete`, +`sturnus.application.worker._create_session_document`, +`sturnus.application.publishing.announce_ready_sessions` -- carries it +forward a second time on its own. Nothing orchestrates the redo; it is a +state reset and nothing else. + +This module holds the *decision* half of that: given the job rows of one +session, which of them may be reset, which must be left alone, and whether +the session may be touched at all. It is a pure function over plain dicts +and is tested without a database, sitting beside +`sturnus.application.publishing.sessions_to_announce` and +`sturnus.application.retention.expired_jobs` and following their style for +the same reason -- there is then exactly one definition of the rule, and +every sentence `/queue requeue` says to an administrator is derived from +this value rather than re-decided while rendering a reply. + +The write that acts on a plan lives in +`sturnus.infrastructure.discord.queue_cog`, which is also the only caller. + +Two rules carry all the weight, and both exist because getting them wrong +turns a helpful command into a destructive one. + +**A job that is not terminal blocks the whole session.** A `pending` job is +already going to be transcribed, so there is nothing to re-queue; a +`running` job is worse than pointless to reset, because the worker holding +it will still call `complete()` when it finishes, writing the old run's +transcript and flipping the row back to `done` -- silently undoing a reset +the administrator has already been told about. Refusing the session outright +is simple, safe, and easy to explain in a reply. + +**A job whose audio has been erased is skipped, not reset.** +`audio_deleted_at` is the authoritative, already-durable record that the S3 +object is gone: it is stamped only after `store.delete` actually succeeded, +by either the retention sweep (`sturnus.application.retention`) or an +immediate erasure request (`/audio delete`, `/audio purge`). Re-queueing +such a job hands a worker a key it cannot download, `queue.fail` fires, +`attempts` climbs, the job goes `dead` again, and the only product is noise +in the log. Recoverability is never inferred from `retention_until`, which +is a plan and not a fact -- the hourly sweep may simply not have run yet. + +A skipped job keeps its `done` status and its existing transcript, so it +still counts as terminal for `JobQueue.complete` and still contributes its +old text to `sturnus.application.assembly.assemble`. That composes +correctly: the new document is the redone speakers plus the untouched old +text of the speakers whose audio is gone. It also means the reply must say +so plainly -- an administrator told "3 speakers re-queued" and not told +"1 speaker's audio is erased, their old transcript is carried over" would +reasonably assume the whole document had been regenerated. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime +from typing import cast + +#: Job statuses `JobQueue` treats as finished. `claim` never selects +#: either of them, so nothing short of an explicit write resurrects such a +#: job -- which is exactly what makes them the ones a re-queue may reset. +#: `dead` belongs here as much as `done` does: it means "gave up after +#: `max_attempts`", never "unusable", and a job that died because the old +#: code path could not make sense of the audio is precisely the job an +#: administrator wants re-run against the new one. +TERMINAL_STATUSES = frozenset({"done", "dead"}) + + +@dataclass(frozen=True) +class RequeuePlan: + """What a re-queue of one session would do, before anything is written. + + The three tuples are disjoint and, together, cover every job of the + session exactly once. All three are always computed, even when the + plan turns out to be blocked: the plan describes the session, and the + caller decides what to do about it -- which keeps the refusal reply + able to say what *would* have happened as well as why it did not. + """ + + #: Jobs to reset to `pending`, ascending by id. + resettable_job_ids: tuple[int, ...] + #: Speakers left untouched because their audio no longer exists. Their + #: old transcript is carried into the new document unchanged. + erased_user_ids: tuple[int, ...] + #: Speakers whose job is `pending` or `running`. Any at all means the + #: session must be refused; see the module docstring. + active_user_ids: tuple[int, ...] + + @property + def is_blocked(self) -> bool: + """Whether a worker may still act on this session's jobs.""" + return bool(self.active_user_ids) + + @property + def is_empty(self) -> bool: + """Whether there is nothing left to reset. + + True for a session with no jobs at all and for one whose every + recording has been erased alike. Both must be refused rather than + confirmed: making no change while reporting success is the failure + mode this command has to avoid. + """ + return not self.resettable_job_ids + + +def plan_requeue(jobs: list[dict[str, object]]) -> RequeuePlan: + """Sorts one session's jobs into reset / skip / blocking. + + `jobs` are the rows of a single session, each carrying `id`, + `discord_user_id`, `status` and `audio_deleted_at`. Row order is not + trusted: the result is sorted by job id, because the confirmation text + built from it is read by a human and two runs against an unchanged + session must produce the same sentence. + + The classification is checked in the order blocking, then erased, then + resettable, and that order matters. A job can legitimately be + `pending` *and* have `audio_deleted_at` set -- the retention sweep does + not consult job status -- and reporting it as merely "skipped" would + let the session through while a worker can still claim one of its jobs, + which is the exact case the refusal exists for. + """ + resettable: list[int] = [] + erased: list[int] = [] + active: list[int] = [] + for candidate in sorted(jobs, key=lambda row: cast(int, row["id"])): + job_id = cast(int, candidate["id"]) + user_id = cast(int, candidate["discord_user_id"]) + status = cast(str, candidate["status"]) + audio_deleted_at = cast("datetime | None", candidate["audio_deleted_at"]) + if status not in TERMINAL_STATUSES: + active.append(user_id) + elif audio_deleted_at is not None: + erased.append(user_id) + else: + resettable.append(job_id) + return RequeuePlan( + resettable_job_ids=tuple(resettable), + erased_user_ids=tuple(erased), + active_user_ids=tuple(active), + ) diff --git a/src/sturnus/infrastructure/discord/client.py b/src/sturnus/infrastructure/discord/client.py index 96c8373..037878d 100644 --- a/src/sturnus/infrastructure/discord/client.py +++ b/src/sturnus/infrastructure/discord/client.py @@ -65,6 +65,7 @@ from sturnus.infrastructure.discord.config_cog import ConfigCog from sturnus.infrastructure.discord.consent_cog import ConsentCog from sturnus.infrastructure.discord.link_cog import LinkCog +from sturnus.infrastructure.discord.queue_cog import QueueCog from sturnus.infrastructure.discord.setup_cog import SetupCog from sturnus.infrastructure.discord.voice import VoiceReceiveAdapter from sturnus.infrastructure.documents.outline_oauth import OutlineOAuth @@ -227,6 +228,13 @@ async def setup_hook(self) -> None: await self.add_cog(ConfigCog(self._config_store, self.reconcile_guild, self.running_state)) await self.add_cog(SetupCog(self._config_store, self._clock, self.reconcile_guild)) await self.add_cog(AudioCog(self._session_factory, self._audio_store, self._clock)) + # Reads and re-queues transcription jobs. It takes the session + # factory rather than `JobQueue`/`SessionRepository` on purpose: + # its selections are specific to these three admin commands and + # used nowhere else, so neither of those grows a method only a + # slash command calls -- the same shape `AudioCog` above already + # has, and for the same reason. + await self.add_cog(QueueCog(self._session_factory, self._clock)) await self.add_cog( LinkCog(self._outline_oauth, self._link_states, self._account_links, self._clock) ) diff --git a/src/sturnus/infrastructure/discord/queue_cog.py b/src/sturnus/infrastructure/discord/queue_cog.py new file mode 100644 index 0000000..7d37635 --- /dev/null +++ b/src/sturnus/infrastructure/discord/queue_cog.py @@ -0,0 +1,894 @@ +"""Admin commands for looking at the transcription queue and re-running a session. + +Three subcommands, and no more. `/queue status` and `/queue session` are +read-only; `/queue requeue` is the only one that writes, and it is the +reason this cog exists: a session transcribed by a code path that has since +been fixed is worthless until something puts its jobs back on the queue, +and `JobQueue.claim` selects only `pending` jobs -- never `done`, never +`dead` -- so nothing short of an explicit write resurrects one. + +**What a re-queue is.** A state reset and nothing else. It returns the +session to exactly the state it was in immediately after `close_session` +and before documentation, so that `claim`, `complete`, +`sturnus.application.worker._create_session_document` and +`sturnus.application.publishing.announce_ready_sessions` carry it forward a +second time on their own. Nothing here orchestrates the redo. Which jobs +may be reset is decided by `sturnus.application.requeue.plan_requeue`, a +pure function tested without a database; read its module docstring first, +because the two rules that keep this command from being destructive live +there rather than here. + +**Why the SQL is inline instead of in a repository.** This selection is +specific to these three commands and used nowhere else, so +`SessionRepository` and `JobQueue` do not grow a method that only an admin +slash command calls. `sturnus.infrastructure.discord.audio_cog._erase_audio` +does the same thing for the same stated reason, and this module follows it +deliberately. + +**Everything is scoped to `interaction.guild_id`.** Every query joins +`transcription_job` to `session` and filters on `session.guild_id`, and a +session id belonging to another guild produces the *same* reply as one that +does not exist anywhere, so the command cannot be used to probe whether an +id exists elsewhere. Note the contrast with `audio_cog._erase_audio`, which +is deliberately cross-guild: that is a GDPR erasure serving a data subject +across the whole deployment, and a status readout is not. + +**Every reply is ephemeral, and every subcommand is admin-gated** -- +including the read-only ones. Even `/queue status` reports who was recorded +and how much they said, which is not an ordinary-member fact. The one +public thing a re-queue produces is the announcement +`announce_ready_sessions` posts once the redo is documented, and the +confirmation says so before anything is written, because that post is the +part of a re-queue an ordinary member sees. + +**The confirmation is a plan, not a yes/no prompt.** `/queue requeue` does +three things that are irreversible or publicly visible: it discards stored +transcripts that cannot be recovered if the redo fails; it causes a second +Outline document to be created, leaving the first orphaned (there is no +update path anywhere in the codebase -- `DocumentSink` exposes only +`create`); and it causes a second announcement in the voice channel. All +three are named in the confirmation text before the button is pressed. A +`force: bool = False` flag would be typed from muscle memory and presents +no plan to read, so this uses buttons instead. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +import discord +from discord import app_commands +from discord.ext import commands +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from sturnus.application.ports import Clock +from sturnus.application.requeue import TERMINAL_STATUSES, RequeuePlan, plan_requeue +from sturnus.infrastructure.db.models import Session, SessionParticipant, TranscriptionJob +from sturnus.infrastructure.db.queue import DEFAULT_LEASE_SECONDS +from sturnus.infrastructure.discord.permissions import require_admin + +log = logging.getLogger(__name__) + +#: The single answer for "that session is not yours to look at", used both +#: for an id that exists in another guild and for one that exists nowhere. +#: Identical on purpose: two different replies would turn this command into +#: a way to discover whether a session id is in use somewhere else. +NO_SUCH_SESSION = "No session with that id in this server." + +#: How long the Confirm/Cancel prompt stays live. Short, because the plan +#: it displays is a snapshot: a worker can claim a sibling job while it +#: sits on screen. The write re-derives the plan under a row lock anyway +#: (`_apply_requeue`), so a stale press is refused rather than obeyed -- +#: this timeout only keeps a forgotten prompt from lingering. +CONFIRM_TIMEOUT_SECONDS = 60.0 + +#: Job statuses reported by `/queue status`, in lifecycle order rather than +#: alphabetically: an administrator reads this line to see where work is +#: piling up, and `pending -> running -> done | dead` is the order that +#: makes that legible. +_REPORTED_STATUSES = ("pending", "running", "done", "dead") + + +@dataclass(frozen=True) +class SessionSummary: + """The session-level facts every `/queue` reply is rendered from. + + `channel_id` rather than the stored `channel_name`: `<#id>` renders as + the channel's *current* name in Discord and stays a working link if it + was renamed since, whereas `session.channel_name` is deliberately + frozen at the moment the session opened so old protocols do not get + rewritten by a later rename. + """ + + id: int + channel_id: int + status: str + ended_at: datetime | None + end_reason: str | None + document_url: str | None + announced_at: datetime | None + + +@dataclass(frozen=True) +class SessionView: + """One session, its re-queue plan, and the names to render it with. + + Read and returned as a unit so the reply cannot be built from a + session row read at one moment and job rows read at another. + """ + + summary: SessionSummary + plan: RequeuePlan + #: `discord_user_id` -> display name, from `session_participant`. + names: dict[int, str] + + +@dataclass(frozen=True) +class JobLine: + """One speaker's row in the `/queue session` readout.""" + + discord_user_id: int + status: str + attempts: int + audio_present: bool + error: str | None + #: The *length* of the stored transcript, never the transcript. This + #: reply has to be enough to decide whether a re-queue is warranted -- + #: a 100-minute recording whose transcript is 24 characters is the + #: tell -- without a slash command becoming a way to read meeting + #: content out of the document system. + transcript_length: int | None + + +@dataclass(frozen=True) +class QueueStatus: + """The guild-wide counts behind `/queue status`.""" + + counts: dict[str, int] + running_past_lease: int + #: When the session owning the oldest `pending` job ended. See + #: `_load_status` for why that is the closest thing to an enqueue time + #: this schema has. + oldest_pending_session_ended_at: datetime | None + closed_undocumented: int + + +# --------------------------------------------------------------------------- +# Reads. Plain ORM queries, all joined to `session` and filtered by guild. +# --------------------------------------------------------------------------- + + +def _summary_of(row: Session) -> SessionSummary: + return SessionSummary( + id=row.id, + channel_id=row.channel_id, + status=row.status, + ended_at=row.ended_at, + end_reason=row.end_reason, + document_url=row.document_url, + announced_at=row.announced_at, + ) + + +async def _participant_names(db: AsyncSession, session_id: int) -> dict[int, str]: + rows = await db.execute( + select(SessionParticipant.discord_user_id, SessionParticipant.discord_display_name).where( + SessionParticipant.session_id == session_id + ) + ) + return {user_id: name for user_id, name in rows} + + +async def _load_status( + session_factory: async_sessionmaker[AsyncSession], + guild_id: int, + now: datetime, + lease_seconds: float, +) -> QueueStatus: + """Counts this guild's jobs, its expired leases and its stuck sessions. + + The "oldest pending job" figure is derived from `session.ended_at` + rather than from the job row, because `transcription_job` has no + enqueue timestamp at all. `RecordingService.close` uploads and enqueues + every speaker and only then calls `close_session`, so `ended_at` is + within seconds of when the job was created -- close enough to answer + "has something been sitting in the queue for hours?", which is the only + question this line exists for. It is *not* the age of a re-queued job: + a reset job keeps its session's original `ended_at`, so a re-queue + makes this number read older than the job really is. `render_status` + therefore names the session's end rather than calling it the job's age, + and says outright that a re-queue skews it. + """ + lease_cutoff = now - timedelta(seconds=lease_seconds) + async with session_factory() as db: + rows = await db.execute( + select(TranscriptionJob.status, func.count()) + .join(Session, Session.id == TranscriptionJob.session_id) + .where(Session.guild_id == guild_id) + .group_by(TranscriptionJob.status) + ) + counts = {status: 0 for status in _REPORTED_STATUSES} + for status, count in rows: + counts[status] = counts.get(status, 0) + int(count) + + past_lease = await db.scalar( + select(func.count()) + .select_from(TranscriptionJob) + .join(Session, Session.id == TranscriptionJob.session_id) + .where( + Session.guild_id == guild_id, + TranscriptionJob.status == "running", + TranscriptionJob.claimed_at < lease_cutoff, + ) + ) + oldest = await db.scalar( + select(func.min(Session.ended_at)) + .join(TranscriptionJob, TranscriptionJob.session_id == Session.id) + .where(Session.guild_id == guild_id, TranscriptionJob.status == "pending") + ) + # The same condition `SessionRepository.closed_undocumented_sessions` + # uses for `retry_pending_documents`, restated here scoped to one + # guild: that method is deliberately guild-blind because the worker + # serves every guild from one process, and a status readout must + # not be. + has_jobs = select(TranscriptionJob.session_id).distinct() + unfinished = select(TranscriptionJob.session_id).where( + TranscriptionJob.status.not_in(tuple(TERMINAL_STATUSES)) + ) + stuck = await db.scalar( + select(func.count()) + .select_from(Session) + .where( + Session.guild_id == guild_id, + Session.status == "closed", + Session.id.in_(has_jobs), + Session.id.not_in(unfinished), + ) + ) + return QueueStatus( + counts=counts, + running_past_lease=int(past_lease or 0), + oldest_pending_session_ended_at=oldest, + closed_undocumented=int(stuck or 0), + ) + + +async def _load_session( + session_factory: async_sessionmaker[AsyncSession], guild_id: int, session_id: int +) -> tuple[SessionSummary, list[JobLine], dict[int, str]] | None: + """Reads one session's detail, or `None` if it is not this guild's.""" + async with session_factory() as db: + row = await db.scalar( + select(Session).where(Session.id == session_id, Session.guild_id == guild_id) + ) + if row is None: + return None + jobs = await db.execute( + select(TranscriptionJob) + .where(TranscriptionJob.session_id == session_id) + .order_by(TranscriptionJob.id) + ) + lines = [ + JobLine( + discord_user_id=job.discord_user_id, + status=job.status, + attempts=job.attempts, + audio_present=job.audio_deleted_at is None, + error=job.error, + transcript_length=None if job.transcript is None else len(job.transcript), + ) + for job in jobs.scalars() + ] + return _summary_of(row), lines, await _participant_names(db, session_id) + + +async def _load_requeue_view( + session_factory: async_sessionmaker[AsyncSession], guild_id: int, session_id: int +) -> SessionView | None: + """Builds the plan shown in the confirmation, without locking anything. + + Deliberately lock-free: this read only decides what to *offer*, and + holding a row lock across the seconds a human takes to read a prompt + would block every worker completing a sibling job of the same session + meanwhile. The plan that actually gets applied is re-derived inside the + lock by `_apply_requeue`, so a session that changes while the prompt is + on screen is refused rather than acted on from a stale snapshot. + """ + async with session_factory() as db: + row = await db.scalar( + select(Session).where(Session.id == session_id, Session.guild_id == guild_id) + ) + if row is None: + return None + return SessionView( + summary=_summary_of(row), + plan=plan_requeue(await _job_dicts(db, session_id)), + names=await _participant_names(db, session_id), + ) + + +async def _job_dicts(db: AsyncSession, session_id: int) -> list[dict[str, object]]: + """One session's jobs, shaped for `plan_requeue`. + + Deliberately unfiltered by status or `audio_deleted_at`: both checks + are that pure function's job alone, so there is exactly one definition + of the rule -- the same reasoning + `JobRepository.candidates_for_retention` follows for `expired_jobs`. + """ + rows = await db.execute( + select( + TranscriptionJob.id, + TranscriptionJob.discord_user_id, + TranscriptionJob.status, + TranscriptionJob.audio_deleted_at, + ).where(TranscriptionJob.session_id == session_id) + ) + return [ + { + "id": row.id, + "discord_user_id": row.discord_user_id, + "status": row.status, + "audio_deleted_at": row.audio_deleted_at, + } + for row in rows + ] + + +# --------------------------------------------------------------------------- +# The write. +# --------------------------------------------------------------------------- + + +async def _apply_requeue( + session_factory: async_sessionmaker[AsyncSession], guild_id: int, session_id: int +) -> SessionView | None: + """Resets a session's recoverable jobs, in one transaction and one commit. + + Returns the session as it was *decided* on -- the plan here is the one + that was actually applied, or the one that caused a refusal. `None` + means the session does not belong to this guild (or does not exist), + which the caller renders as `NO_SUCH_SESSION`. Whether the write + happened is derivable: it did exactly when the returned plan is neither + blocked nor empty. + + Two properties matter and neither is incidental. + + **The lock comes first.** `SELECT TranscriptionJob.id WHERE session_id + = ... ORDER BY id FOR UPDATE` is the same statement, with the same + ordering, that `JobQueue.complete` takes before recomputing its + remaining-jobs count. Taking it here is what serialises a re-queue + against a worker completing a sibling job of the same session instead + of letting the two interleave -- under READ COMMITTED this transaction + would otherwise happily build a plan from a snapshot that predates a + `complete()` already in flight, decide the session is entirely `done`, + and reset a job whose worker is about to write its old transcript back + over the reset. Keeping `ORDER BY id` identical to `complete`'s is what + stops the two statements deadlocking against each other. + + **One transaction, one commit, spanning both tables.** If the job + resets committed and the session reset did not, the redo would finish + against a still-`documented` session: `complete()` would return `False` + forever, because its last-job rule requires `status == "closed"`, and + `retry_pending_documents` -- which also looks for `"closed"` -- would + not sweep it either. The session would be stranded with no document and + no error anywhere. + """ + async with session_factory() as db: + # Before reading anything: a plan built outside this lock is a plan + # about a session that may already have moved. + await db.execute( + select(TranscriptionJob.id) + .where(TranscriptionJob.session_id == session_id) + .order_by(TranscriptionJob.id) + .with_for_update() + ) + row = await db.scalar( + select(Session).where(Session.id == session_id, Session.guild_id == guild_id) + ) + if row is None: + return None + view = SessionView( + summary=_summary_of(row), + plan=plan_requeue(await _job_dicts(db, session_id)), + names=await _participant_names(db, session_id), + ) + if view.plan.is_blocked or view.plan.is_empty: + return view + + await db.execute( + update(TranscriptionJob) + .where(TranscriptionJob.id.in_(view.plan.resettable_job_ids)) + .values( + status="pending", + # A lease timestamp on a `pending` row means nothing and + # would only make `/queue status` report a job as past its + # lease before any worker has looked at it. + claimed_at=None, + # A full budget: this is a new attempt at a new code path, + # not a continuation of the old one. + attempts=0, + # The old error described the old run. + error=None, + # `assemble` reads every job of the session, not only the + # one that finished last, so a reset job that kept its old + # text would put the very hallucinations this command + # exists to remove into the new document if the session + # were re-documented before the redo finished. Clearing it + # makes a half-done redo visibly incomplete instead of + # plausibly wrong. Losing the old text is intended: it + # being wrong is why we are here, and `complete` + # overwrites it on success anyway. + transcript=None, + ) + ) + await db.execute( + update(Session) + .where(Session.id == session_id) + .values( + # "closed", never "open": "open" would make + # `find_open_session` believe this guild has a live + # recording. "closed" reproduces the post-`close_session`, + # pre-documentation state exactly, which is what makes + # `complete`'s last-job rule fire again -- and while it + # sits there, `candidates_for_announcement`'s + # `status == "documented"` filter excludes it, so nothing + # announces mid-redo. + status="closed", + # Required, not optional. `sessions_to_announce` selects + # only sessions whose `announced_at` is still null, and + # `mark_documented` never touches this column -- so a + # re-queue that left it set would produce a session that + # transcribes, re-documents with a fresh URL, and is then + # never announced, with nothing logged and nothing raised. + # Clearing it is the deliberate choice to post again, + # exactly once, the same not-null guard preventing any + # further repeats. + announced_at=None, + # `document_provider`/`document_id`/`document_url` are + # deliberately untouched: the next `mark_documented` + # overwrites them, and clearing them now would only stop + # `/queue session` showing which document is superseded. + ) + ) + await db.commit() + return view + + +# --------------------------------------------------------------------------- +# Rendering. Pure functions over the values above, so the wording is +# testable without an `Interaction` -- the same reasoning `config_cog`'s +# `render_write_result` follows. +# --------------------------------------------------------------------------- + + +def _stamp(when: datetime | None) -> str: + if when is None: + return "never" + return when.astimezone(UTC).strftime("%Y-%m-%d %H:%M UTC") + + +def _age(delta: timedelta) -> str: + """A duration a human reads at a glance, e.g. `2d 3h` or `14m`.""" + minutes = int(delta.total_seconds() // 60) + if minutes < 1: + return "under a minute" + days, minutes = divmod(minutes, 1440) + hours, minutes = divmod(minutes, 60) + parts = [] + if days: + parts.append(f"{days}d") + if hours: + parts.append(f"{hours}h") + if minutes or not parts: + parts.append(f"{minutes}m") + return " ".join(parts) + + +def _named(user_ids: tuple[int, ...], names: dict[int, str]) -> str: + """Display names for a plan's speakers, falling back to the raw id. + + Never a `<@id>` mention: these replies list people who were recorded + and who may have asked for their audio to be erased, and a mention + would ping them into a conversation they are not part of. + """ + return ", ".join(names.get(user_id, f"user {user_id}") for user_id in user_ids) + + +def _speakers(count: int) -> str: + return "1 speaker" if count == 1 else f"{count} speakers" + + +def _running_jobs(count: int) -> str: + return "1 running job" if count == 1 else f"{count} running jobs" + + +def _closed_sessions(count: int) -> str: + return "1 closed session" if count == 1 else f"{count} closed sessions" + + +def render_status(status: QueueStatus, now: datetime, lease_seconds: float) -> str: + counts = ", ".join(f"{name}: {status.counts.get(name, 0)}" for name in _REPORTED_STATUSES) + lines = ["**Transcription queue for this server**", f"Jobs — {counts}"] + if status.running_past_lease: + # Stated as a possibility rather than a fact: the lease that + # actually applies is `job_lease_seconds` in the *worker's* + # environment, and this process cannot see it. A long-running job + # under a raised lease is perfectly healthy and must not be + # reported as abandoned. + lines.append( + f"⚠️ {_running_jobs(status.running_past_lease)} past the default " + f"{int(lease_seconds)}s lease — if the worker's `job_lease_seconds` is not " + "higher than that, another worker may already have reclaimed the job." + ) + if status.oldest_pending_session_ended_at is None: + lines.append("Oldest pending job: none — nothing is waiting.") + else: + ended = status.oldest_pending_session_ended_at + lines.append( + f"Oldest pending job: from a session that ended {_stamp(ended)} " + f"({_age(now - ended)} ago). A re-queued job keeps its session's original " + "end time, so this reads older than the job itself after a `/queue requeue`." + ) + if status.closed_undocumented: + lines.append( + f"{_closed_sessions(status.closed_undocumented)} with every job finished but " + "no document yet; the worker retries those on its own sweep." + ) + return "\n".join(lines) + + +def render_session(summary: SessionSummary, jobs: list[JobLine], names: dict[int, str]) -> str: + lines = [ + f"**Session {summary.id}** in <#{summary.channel_id}>", + f"Status: `{summary.status}` — ended {_stamp(summary.ended_at)}" + f" ({summary.end_reason or 'no reason recorded'})", + f"Document: {summary.document_url or '*(none)*'}", + f"Announced: {_stamp(summary.announced_at)}", + ] + if not jobs: + lines.append("No transcription jobs — nobody spoke in this session.") + return "\n".join(lines) + lines.append(f"{_speakers(len(jobs))}:") + for job in jobs: + name = names.get(job.discord_user_id, f"user {job.discord_user_id}") + length = ( + "transcript: none stored" + if job.transcript_length is None + else f"transcript: {job.transcript_length} characters" + ) + audio = "audio: present" if job.audio_present else "audio: erased" + line = f"- {name} — status: `{job.status}`, attempts: {job.attempts}, {audio}, {length}" + if job.error: + line += f", last error: {job.error}" + lines.append(line) + return "\n".join(lines) + + +def _document_line(summary: SessionSummary, *, future: bool) -> str: + """What happens to the document, stated before and after the write alike. + + `DocumentSink` has exactly one method, `create` -- no update, no patch, + nowhere in the codebase -- so a redo that reaches + `_create_session_document` always makes a brand-new Outline document + and `mark_documented` overwrites the session's `document_url`. The + previously published page stays in Outline, unlinked and unmentioned. + That is accepted and chosen, but an administrator who is not told about + it cannot make the cleanup a deliberate act, so it is spelled out here + both in the confirmation and in the result. + """ + verb = "will be created" if future else "is created when the redo finishes" + if summary.document_url is None: + return f"- A new document {verb}; this session has no document yet, so none is superseded." + return ( + f"- A **new** document {verb}. The existing one (<{summary.document_url}>) " + "stays where it is and is not updated or deleted." + ) + + +def render_requeue_confirmation( + summary: SessionSummary, plan: RequeuePlan, names: dict[int, str] +) -> str: + """The prompt shown above the Confirm/Cancel buttons. + + Deliberately the plan rather than a yes/no question: everything the + administrator cannot undo afterwards is on screen before they press + anything. + """ + lines = [ + f"Re-queue session {summary.id} (<#{summary.channel_id}>, " + f"ended {_stamp(summary.ended_at)})?", + f"- {_speakers(len(plan.resettable_job_ids))} will be re-transcribed; their " + "stored transcripts are discarded and cannot be recovered if the redo fails.", + ] + if plan.erased_user_ids: + lines.append( + f"- {_speakers(len(plan.erased_user_ids))} skipped because their audio has " + f"been erased ({_named(plan.erased_user_ids, names)}). Their existing " + "transcript is carried into the new document unchanged." + ) + lines.append(_document_line(summary, future=True)) + lines.append( + f"- A new link is posted in <#{summary.channel_id}> once it is ready. Everyone " + "who can see that channel sees the post, not only administrators." + ) + return "\n".join(lines) + + +def render_requeue_applied( + summary: SessionSummary, plan: RequeuePlan, names: dict[int, str] +) -> str: + total = len(plan.resettable_job_ids) + len(plan.erased_user_ids) + lines = [ + f"Re-queued {len(plan.resettable_job_ids)} of {_speakers(total)} in session " + f"{summary.id}. Their stored transcripts have been discarded; a worker picks " + "the jobs up on its next poll.", + ] + if plan.erased_user_ids: + lines.append( + f"- Skipped: {_named(plan.erased_user_ids, names)} — their audio has been " + "erased, so their existing transcript is carried into the new document " + "unchanged rather than re-transcribed." + ) + lines.append(_document_line(summary, future=False)) + lines.append(f"- A new link is posted in <#{summary.channel_id}> when it is ready.") + return "\n".join(lines) + + +def render_requeue_refusal( + summary: SessionSummary, plan: RequeuePlan, names: dict[int, str], *, rechecked: bool = False +) -> str: + """Why nothing was written. `rechecked` means the session moved under us. + + Split from the other two because a refusal is not a smaller success: + `config_cog` names four honest outcomes rather than one cheerful + confirmation, and this follows that. + """ + prefix = ( + "The session changed while the confirmation was on screen, so nothing was written. " + if rechecked + else "" + ) + if plan.is_blocked: + return ( + f"{prefix}Refused: {_speakers(len(plan.active_user_ids))} in session " + f"{summary.id} still have a job pending or running " + f"({_named(plan.active_user_ids, names)}). Those recordings are already " + "going to be transcribed, and resetting a job a worker is holding would be " + "undone the moment that worker finishes. Try again once the queue is idle." + ) + if plan.erased_user_ids: + return ( + f"{prefix}Nothing to do: every speaker in session {summary.id} has had " + f"their audio erased ({_named(plan.erased_user_ids, names)}). Re-queueing " + "would only hand a worker a key it cannot download. Their existing " + "transcripts are untouched." + ) + return ( + f"{prefix}Nothing to do: session {summary.id} has no transcription jobs at all " + "— nobody spoke, so there is nothing to transcribe again." + ) + + +# --------------------------------------------------------------------------- +# The confirmation view. +# --------------------------------------------------------------------------- + + +class RequeueConfirmView(discord.ui.View): + """Confirm / Cancel for one session's re-queue. + + Author-only (`interaction_check`), because Discord components are + clickable by anyone who can see the message; short-lived, and the + buttons disable themselves on timeout so a stale prompt cannot + re-queue a session tomorrow. The Discord rendering rules here are the + ones `sturnus.infrastructure.discord.views.ConsentView` already + established, which is why they look the same. + + The view holds ids, never a plan. Pressing Confirm re-derives the plan + inside the row lock (`_apply_requeue`), so the write can never be made + from the snapshot the prompt was rendered from -- and a second press, + which nothing stops Discord from delivering, finds the jobs already + `pending` and is refused on that alone rather than producing a second + document and a second public post. + """ + + def __init__( + self, + *, + author_id: int, + guild_id: int, + session_id: int, + session_factory: async_sessionmaker[AsyncSession], + timeout: float = CONFIRM_TIMEOUT_SECONDS, + ) -> None: + super().__init__(timeout=timeout) + self._author_id = author_id + self._guild_id = guild_id + self._session_id = session_id + self._session_factory = session_factory + #: Set by the cog right after sending, so the buttons can be + #: disabled on the message that actually carries them. + self.message: discord.Message | None = None + + async def interaction_check(self, interaction: discord.Interaction) -> bool: + if interaction.user.id != self._author_id: + await interaction.response.send_message( + f"Only <@{self._author_id}> can respond to this prompt.", ephemeral=True + ) + return False + return True + + async def on_timeout(self) -> None: + await self._disable() + + async def _disable(self) -> None: + """Greys out both buttons on the message they are attached to. + + A failed edit is logged and swallowed: the answer the + administrator is waiting for must not be lost because Discord + would not repaint a message, and a live button on a stopped view + is refused by `_apply_requeue`'s own re-check anyway. + """ + for item in self.children: + if isinstance(item, discord.ui.Button): + item.disabled = True + if self.message is None: + return + try: + await self.message.edit(view=self) + except discord.HTTPException as exc: + log.warning("Could not disable the /queue requeue buttons: %s", exc) + + @discord.ui.button(label="Confirm", style=discord.ButtonStyle.danger) + async def confirm( + self, interaction: discord.Interaction, _button: discord.ui.Button[RequeueConfirmView] + ) -> None: + # The write takes a row lock and can wait on a worker that is + # mid-`complete()`, which is easily more than the three seconds + # Discord gives a component interaction to answer. + await interaction.response.defer(ephemeral=True, thinking=True) + self.stop() + await self._disable() + view = await _apply_requeue(self._session_factory, self._guild_id, self._session_id) + if view is None: + await interaction.followup.send(NO_SUCH_SESSION, ephemeral=True) + return + if view.plan.is_blocked or view.plan.is_empty: + await interaction.followup.send( + render_requeue_refusal(view.summary, view.plan, view.names, rechecked=True), + ephemeral=True, + ) + return + log.info( + "Re-queued %d job(s) of session %d in guild %d (requested by %d)", + len(view.plan.resettable_job_ids), + self._session_id, + self._guild_id, + interaction.user.id, + ) + await interaction.followup.send( + render_requeue_applied(view.summary, view.plan, view.names), ephemeral=True + ) + + @discord.ui.button(label="Cancel", style=discord.ButtonStyle.secondary) + async def cancel( + self, interaction: discord.Interaction, _button: discord.ui.Button[RequeueConfirmView] + ) -> None: + self.stop() + await self._disable() + await interaction.response.send_message( + f"Cancelled. Session {self._session_id} is untouched.", ephemeral=True + ) + + +# --------------------------------------------------------------------------- +# The cog. +# --------------------------------------------------------------------------- + + +@app_commands.guild_only() +class QueueCog( + commands.GroupCog, name="queue", description="Inspect the transcription queue (admin only)." +): + """`/queue` command group: two read-only views and one re-queue.""" + + def __init__( + self, + session_factory: async_sessionmaker[AsyncSession], + clock: Clock, + lease_seconds: float = DEFAULT_LEASE_SECONDS, + ) -> None: + self._session_factory = session_factory + self._clock = clock + #: Only ever used to *describe* a running job as possibly abandoned. + #: The lease that actually applies is `job_lease_seconds` in the + #: worker's environment, which this process cannot read -- so this + #: is the documented default, and `render_status` says so rather + #: than presenting the count as a fact. + self._lease_seconds = lease_seconds + super().__init__() + + @app_commands.command(name="status", description="Counts of queued, running and failed jobs.") + @require_admin() + async def status(self, interaction: discord.Interaction) -> None: + guild_id = interaction.guild_id + if guild_id is None: + await interaction.response.send_message( + "This command can only be used in a server.", ephemeral=True + ) + return + # Four aggregates over two joined tables; on a cold connection + # pool that alone can miss Discord's three-second initial-response + # window, and then the administrator sees "The application did not + # respond" over a command that worked. + await interaction.response.defer(ephemeral=True, thinking=True) + now = self._clock.now() + status = await _load_status(self._session_factory, guild_id, now, self._lease_seconds) + await interaction.followup.send( + render_status(status, now, self._lease_seconds), ephemeral=True + ) + + @app_commands.command(name="session", description="One session's jobs, in detail.") + @app_commands.describe(session_id="The session's numeric id, as shown by /queue status") + @require_admin() + async def session(self, interaction: discord.Interaction, session_id: int) -> None: + guild_id = interaction.guild_id + if guild_id is None: + await interaction.response.send_message( + "This command can only be used in a server.", ephemeral=True + ) + return + await interaction.response.defer(ephemeral=True, thinking=True) + loaded = await _load_session(self._session_factory, guild_id, session_id) + if loaded is None: + await interaction.followup.send(NO_SUCH_SESSION, ephemeral=True) + return + summary, jobs, names = loaded + await interaction.followup.send(render_session(summary, jobs, names), ephemeral=True) + + @app_commands.command( + name="requeue", description="Transcribe a finished session again, from its stored audio." + ) + @app_commands.describe(session_id="The session's numeric id, as shown by /queue status") + @require_admin() + async def requeue(self, interaction: discord.Interaction, session_id: int) -> None: + """Offers a re-queue; the write happens only if Confirm is pressed. + + This half deliberately writes nothing. It reads the session, builds + the plan and renders it, and every consequence that cannot be + undone is named in that text before a button exists to press. + """ + guild_id = interaction.guild_id + if guild_id is None: + await interaction.response.send_message( + "This command can only be used in a server.", ephemeral=True + ) + return + await interaction.response.defer(ephemeral=True, thinking=True) + view = await _load_requeue_view(self._session_factory, guild_id, session_id) + if view is None: + await interaction.followup.send(NO_SUCH_SESSION, ephemeral=True) + return + if view.plan.is_blocked or view.plan.is_empty: + # No buttons at all: there is nothing to confirm, and offering + # a Confirm that would only be refused invites the + # administrator to press it and learn nothing new. + await interaction.followup.send( + render_requeue_refusal(view.summary, view.plan, view.names), ephemeral=True + ) + return + confirm = RequeueConfirmView( + author_id=interaction.user.id, + guild_id=guild_id, + session_id=session_id, + session_factory=self._session_factory, + ) + await interaction.followup.send( + render_requeue_confirmation(view.summary, view.plan, view.names), + view=confirm, + ephemeral=True, + ) + confirm.message = await interaction.original_response() diff --git a/tests/application/test_requeue.py b/tests/application/test_requeue.py new file mode 100644 index 0000000..ce31d32 --- /dev/null +++ b/tests/application/test_requeue.py @@ -0,0 +1,142 @@ +"""What `plan_requeue` decides, over plain dicts and without a database. + +Every sentence `/queue requeue` says to an administrator is derived from a +`RequeuePlan`, so the rules are pinned here -- once -- rather than through +an `Interaction` and a Postgres container. `tests/infrastructure/discord/ +test_queue_cog.py` covers the writes that follow from a plan; this file +covers the decision itself. +""" + +from datetime import UTC, datetime + +from sturnus.application.requeue import RequeuePlan, plan_requeue + +T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) +ANNA, BEN, CLARA = 100, 200, 300 + + +def job( + job_id: int, + user_id: int, + status: str = "done", + audio_deleted_at: datetime | None = None, +) -> dict[str, object]: + return { + "id": job_id, + "discord_user_id": user_id, + "status": status, + "audio_deleted_at": audio_deleted_at, + } + + +def test_every_finished_job_with_audio_is_resettable() -> None: + plan = plan_requeue([job(1, ANNA), job(2, BEN)]) + + assert plan.resettable_job_ids == (1, 2) + assert plan.erased_user_ids == () + assert plan.active_user_ids == () + assert plan.is_blocked is False + assert plan.is_empty is False + + +def test_a_dead_job_is_terminal_and_therefore_resettable() -> None: + """`dead` only means "gave up after `max_attempts`", never "unusable". + + A job that died because the *old* code path could not make sense of the + audio is precisely the job an administrator wants re-run against the + new one, so `dead` must not be mistaken for a reason to refuse. + """ + plan = plan_requeue([job(1, ANNA, status="dead")]) + + assert plan.resettable_job_ids == (1,) + assert plan.is_blocked is False + + +def test_a_job_whose_audio_is_erased_is_skipped_rather_than_reset() -> None: + """Re-queueing it would hand a worker an S3 key that no longer resolves. + + `queue.fail` would fire, `attempts` would climb, the job would go + `dead` again, and the only product would be log noise -- so the plan + names the speaker instead, and the reply tells the administrator their + old transcript is carried into the new document unchanged. + """ + plan = plan_requeue([job(1, ANNA), job(2, BEN, audio_deleted_at=T0)]) + + assert plan.resettable_job_ids == (1,) + assert plan.erased_user_ids == (BEN,) + assert plan.is_blocked is False + assert plan.is_empty is False + + +def test_a_running_job_blocks_the_whole_session() -> None: + """Resetting a `running` job would be undone by the worker holding it. + + That worker still calls `complete()` when it finishes, which writes the + old run's transcript and flips the row back to `done` -- silently + reverting the reset while the administrator has been told it happened. + """ + plan = plan_requeue([job(1, ANNA), job(2, BEN, status="running")]) + + assert plan.active_user_ids == (BEN,) + assert plan.is_blocked is True + + +def test_a_pending_job_blocks_the_whole_session() -> None: + """It is already going to be transcribed; there is nothing to re-queue.""" + plan = plan_requeue([job(1, ANNA, status="pending")]) + + assert plan.active_user_ids == (ANNA,) + assert plan.is_blocked is True + + +def test_an_active_job_is_reported_as_active_even_if_its_audio_is_gone() -> None: + """The blocking classification wins, because it is the one that refuses. + + A job can be `pending` with `audio_deleted_at` set -- the retention + sweep does not consult job status. Reporting it as merely "skipped" + would let the session through while a worker is still holding one of + its jobs, which is the case the refusal exists for. + """ + plan = plan_requeue([job(1, ANNA, status="running", audio_deleted_at=T0)]) + + assert plan.active_user_ids == (ANNA,) + assert plan.erased_user_ids == () + assert plan.is_blocked is True + + +def test_a_session_whose_audio_is_all_erased_has_nothing_to_reset() -> None: + """Reporting success while changing nothing is the failure to avoid.""" + plan = plan_requeue([job(1, ANNA, audio_deleted_at=T0), job(2, BEN, audio_deleted_at=T0)]) + + assert plan.resettable_job_ids == () + assert plan.erased_user_ids == (ANNA, BEN) + assert plan.is_empty is True + assert plan.is_blocked is False + + +def test_a_session_with_no_jobs_at_all_is_empty() -> None: + """Nobody ever spoke, so there is no recording to transcribe again.""" + plan = plan_requeue([]) + + assert plan == RequeuePlan(resettable_job_ids=(), erased_user_ids=(), active_user_ids=()) + assert plan.is_empty is True + assert plan.is_blocked is False + + +def test_the_plan_is_ordered_by_job_id_whatever_order_the_rows_arrive_in() -> None: + """The confirmation text is read by a human and must not reshuffle. + + Two runs of the same command against the same unchanged session have + to produce the same sentence, or an administrator comparing them + cannot tell a real change from row order. + """ + plan = plan_requeue( + [ + job(3, CLARA, audio_deleted_at=T0), + job(1, ANNA), + job(2, BEN, audio_deleted_at=T0), + ] + ) + + assert plan.resettable_job_ids == (1,) + assert plan.erased_user_ids == (BEN, CLARA) diff --git a/tests/infrastructure/discord/test_client_cogs.py b/tests/infrastructure/discord/test_client_cogs.py index 0a65d60..31b8f72 100644 --- a/tests/infrastructure/discord/test_client_cogs.py +++ b/tests/infrastructure/discord/test_client_cogs.py @@ -17,7 +17,15 @@ #: Every cog `setup_hook` is expected to register. Extend this set in the #: same commit that adds a new cog -- that is the whole point of this test. -EXPECTED_COGS = {"ConsentCog", "ConfigCog", "AboutCog", "SetupCog", "AudioCog", "LinkCog"} +EXPECTED_COGS = { + "ConsentCog", + "ConfigCog", + "AboutCog", + "SetupCog", + "AudioCog", + "LinkCog", + "QueueCog", +} def _registered_cog_names() -> set[str]: diff --git a/tests/infrastructure/discord/test_queue_cog.py b/tests/infrastructure/discord/test_queue_cog.py new file mode 100644 index 0000000..bb9e8cd --- /dev/null +++ b/tests/infrastructure/discord/test_queue_cog.py @@ -0,0 +1,992 @@ +"""What `/queue` shows and what `/queue requeue` actually writes. + +The command callbacks are invoked directly (`Command.callback`), the same +way `test_config_commands` and `test_link_cog` do: that is the coroutine +the cog defines, so calling it exercises the decision without a gateway. +The hand-rolled `_Response`/`_Followup`/`_Interaction` fakes are the same +shape as `test_config_commands`'s, and enforce the same contract Discord +really does -- an interaction is answered once, and a deferred one must +answer through `followup`. + +The writes run against a real ephemeral PostgreSQL through the +`clean_database` fixture, because everything worth pinning about a +re-queue is the exact column values it leaves behind and the row lock it +takes on the way -- neither survives a fake. These tests are not marked +`slow`; only tests that download a model are. + +The permission check is asserted by looking for `_has_admin_access` among +each command's installed checks rather than by driving Discord's check +machinery: `Command.callback` deliberately bypasses checks, so a test that +only ever calls the callback would keep passing if `@require_admin()` were +deleted from a command -- which is the regression that matters most here, +since even `/queue status` reports who was recorded and how much they said. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any, cast + +import discord +import pytest +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from sturnus.application.requeue import RequeuePlan +from sturnus.infrastructure.db.models import Base, Session, TranscriptionJob +from sturnus.infrastructure.db.repositories import JobRepository, SessionRepository +from sturnus.infrastructure.discord.permissions import _has_admin_access +from sturnus.infrastructure.discord.queue_cog import ( + NO_SUCH_SESSION, + QueueCog, + RequeueConfirmView, + SessionSummary, + _apply_requeue, + render_requeue_confirmation, +) + +T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) +T1 = T0 + timedelta(hours=1) +NOW = T0 + timedelta(hours=2) + +GUILD, OTHER_GUILD, CHANNEL = 1, 2, 77 +ANNA, BEN, CLARA = 100, 200, 300 +ADMIN = 999 +NAMES = {ANNA: "anna", BEN: "ben", CLARA: "clara"} +DOC_URL = "https://outline.example/doc/session-1" + + +class _Clock: + def __init__(self, now: datetime = NOW) -> None: + self._now = now + + def now(self) -> datetime: + return self._now + + +# --------------------------------------------------------------------------- +# Interaction fakes -- the same contract `test_config_commands` enforces. +# --------------------------------------------------------------------------- + + +class _Message: + """Just enough of `discord.Message` for a view to disable its buttons.""" + + def __init__(self) -> None: + self.edits: list[discord.ui.View | None] = [] + + async def edit(self, view: discord.ui.View | None = None) -> None: + self.edits.append(view) + + +class _Response: + """Discord's initial-response slot: usable exactly once, in one way.""" + + def __init__(self) -> None: + self.deferred = False + self.deferred_ephemeral: bool | None = None + self.messages: list[tuple[str, bool]] = [] + self.views: list[discord.ui.View | None] = [] + + async def defer(self, ephemeral: bool = False, thinking: bool = False) -> None: + assert not self.deferred, "an interaction can only be deferred once" + assert not self.messages, "already answered; there is nothing left to defer" + assert thinking, "a deferral with no thinking indicator shows the user nothing" + self.deferred = True + self.deferred_ephemeral = ephemeral + + async def send_message( + self, content: str, ephemeral: bool = False, view: discord.ui.View | None = None + ) -> None: + assert not self.deferred, "a deferred interaction must answer through followup" + self.messages.append((content, ephemeral)) + self.views.append(view) + + +class _Followup: + def __init__(self) -> None: + self.messages: list[tuple[str, bool]] = [] + self.views: list[discord.ui.View | None] = [] + + async def send( + self, content: str, ephemeral: bool = False, view: discord.ui.View | None = None + ) -> None: + self.messages.append((content, ephemeral)) + self.views.append(view) + + +class _User: + def __init__(self, user_id: int) -> None: + self.id = user_id + + +class _Interaction: + """Only the attributes these commands and the confirm view touch.""" + + def __init__(self, guild_id: int | None = GUILD, user_id: int = ADMIN) -> None: + self.guild_id = guild_id + self.user = _User(user_id) + self.response = _Response() + self.followup = _Followup() + self.message = _Message() + + async def original_response(self) -> _Message: + return self.message + + @property + def reply(self) -> str: + """The single answer the user actually saw, whichever way it went out.""" + answers = self.response.messages + self.followup.messages + assert len(answers) == 1, f"expected exactly one reply, got {answers}" + return answers[0][0] + + @property + def ephemeral(self) -> bool: + answers = self.response.messages + self.followup.messages + assert len(answers) == 1 + return answers[0][1] + + @property + def view(self) -> discord.ui.View | None: + """The components attached to the single answer, whichever slot sent it.""" + views = self.response.views + self.followup.views + assert len(views) == 1, f"expected exactly one reply, got {views}" + return views[0] + + +def _as_interaction(fake: _Interaction) -> discord.Interaction: + """The commands are typed against `discord.Interaction`; the fake is not. + + Kept in one place rather than repeated at every call site, exactly as + `test_config_commands._invoke` keeps its own signature mismatch in one + place. + """ + return cast(discord.Interaction, cast(object, fake)) + + +async def _invoke(cog: QueueCog, command: str, interaction: _Interaction, *args: object) -> None: + """Calls one command's own coroutine, bypassing Discord's dispatch.""" + callback = getattr(cog, command).callback + await callback(cog, _as_interaction(interaction), *args) + + +async def _press(view: discord.ui.View, label: str, interaction: _Interaction) -> None: + """Presses one of the view's buttons the way Discord dispatches it.""" + for item in view.children: + if isinstance(item, discord.ui.Button) and item.label == label: + await item.callback(_as_interaction(interaction)) + return + raise AssertionError(f"no button labelled {label!r} in {view.children}") + + +# --------------------------------------------------------------------------- +# Database fixtures -- copied from `tests/infrastructure/test_queue.py`. +# --------------------------------------------------------------------------- + + +@pytest.fixture +async def factory(clean_database: str) -> async_sessionmaker[AsyncSession]: + engine = create_async_engine(clean_database) + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + return async_sessionmaker(engine, expire_on_commit=False) + + +@dataclass +class Speaker: + """One seeded job row, in whatever state the test needs it.""" + + user_id: int + status: str = "done" + transcript: str | None = "old hallucinated text" + audio_deleted_at: datetime | None = None + attempts: int = 2 + error: str | None = "something went wrong last time" + claimed_at: datetime | None = T1 + + +async def seed( + factory: async_sessionmaker[AsyncSession], + speakers: list[Speaker], + *, + guild: int = GUILD, + session_status: str = "documented", + document_url: str | None = DOC_URL, + announced_at: datetime | None = T1, +) -> int: + """Builds one finished, documented, announced session and its jobs.""" + sessions = SessionRepository(factory) + jobs = JobRepository(factory) + session_id = await sessions.open_session(guild, CHANNEL, "meeting-raum", T0) + for speaker in speakers: + await sessions.add_participant(session_id, speaker.user_id, NAMES[speaker.user_id], T0) + job_id = await jobs.enqueue( + session_id=session_id, + discord_user_id=speaker.user_id, + s3_key=f"sessions/{session_id}/speakers/{speaker.user_id}.enc", + encryption_key_id="k1", + wrapped_data_key=b"wrapped", + retention_until=T0 + timedelta(days=30), + ) + async with factory() as db: + await db.execute( + update(TranscriptionJob) + .where(TranscriptionJob.id == job_id) + .values( + status=speaker.status, + transcript=speaker.transcript, + audio_deleted_at=speaker.audio_deleted_at, + attempts=speaker.attempts, + error=speaker.error, + claimed_at=speaker.claimed_at, + ) + ) + await db.commit() + await sessions.close_session(session_id, T1, "empty") + async with factory() as db: + await db.execute( + update(Session) + .where(Session.id == session_id) + .values( + status=session_status, + document_provider="outline", + document_id="doc-1", + document_url=document_url, + announced_at=announced_at, + ) + ) + await db.commit() + return session_id + + +async def read_jobs( + factory: async_sessionmaker[AsyncSession], session_id: int +) -> dict[int, TranscriptionJob]: + async with factory() as db: + rows = await db.execute( + select(TranscriptionJob).where(TranscriptionJob.session_id == session_id) + ) + return {job.discord_user_id: job for job in rows.scalars()} + + +async def read_session(factory: async_sessionmaker[AsyncSession], session_id: int) -> Session: + async with factory() as db: + row = await db.get(Session, session_id) + assert row is not None + return row + + +async def set_job_status( + factory: async_sessionmaker[AsyncSession], session_id: int, user_id: int, status: str +) -> None: + async with factory() as db: + await db.execute( + update(TranscriptionJob) + .where( + TranscriptionJob.session_id == session_id, + TranscriptionJob.discord_user_id == user_id, + ) + .values(status=status) + ) + await db.commit() + + +def cog(factory: async_sessionmaker[AsyncSession], now: datetime = NOW) -> QueueCog: + return QueueCog(factory, _Clock(now)) + + +# --------------------------------------------------------------------------- +# Permission gate. Every subcommand, including the read-only ones. +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("command", ["status", "session", "requeue"]) +def test_every_subcommand_is_admin_only(command: str) -> None: + """Even `/queue status` reports who was recorded and how much they said. + + `Command.callback` bypasses checks, so nothing else in this file would + notice `@require_admin()` disappearing from a command. + """ + checks = getattr(QueueCog, command).checks + assert _has_admin_access in checks, f"/queue {command} is not admin-gated" + + +# --------------------------------------------------------------------------- +# `/queue status` +# --------------------------------------------------------------------------- + + +async def test_status_counts_only_this_guilds_jobs( + factory: async_sessionmaker[AsyncSession], +) -> None: + """An administrator of one guild must not see another guild's queue.""" + await seed(factory, [Speaker(ANNA), Speaker(BEN, status="dead")]) + await seed( + factory, + [Speaker(CLARA, status="pending", claimed_at=None)], + guild=OTHER_GUILD, + ) + interaction = _Interaction() + + await _invoke(cog(factory), "status", interaction) + + assert "done: 1" in interaction.reply + assert "dead: 1" in interaction.reply + assert "pending: 0" in interaction.reply, "the other guild's pending job must not be counted" + assert interaction.ephemeral is True + + +async def test_status_reports_a_running_job_past_its_lease( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A worker killed mid-job leaves a `running` row nothing else reports. + + That is the state an administrator is looking for when a session has + quietly stopped moving, so the count is the reason this subcommand + exists at all. + """ + await seed(factory, [Speaker(ANNA, status="running", claimed_at=NOW - timedelta(hours=5))]) + interaction = _Interaction() + + await _invoke(cog(factory), "status", interaction) + + assert "running: 1" in interaction.reply + assert "1 running job past the default" in interaction.reply + + +async def test_status_reports_a_closed_session_that_never_got_documented( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The exact condition `retry_pending_documents` sweeps, per guild.""" + await seed(factory, [Speaker(ANNA)], session_status="closed", document_url=None) + interaction = _Interaction() + + await _invoke(cog(factory), "status", interaction) + + assert "1 closed session" in interaction.reply + + +async def test_status_dates_the_oldest_pending_job_by_its_sessions_end( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`transcription_job` has no enqueue timestamp, so this is the proxy. + + `RecordingService.close` enqueues every speaker and only then calls + `close_session`, so `ended_at` is within seconds of when the job was + created. It is not the same thing, and the reply must not claim it is + -- a re-queued job keeps its session's original end and would + otherwise be reported as hours old the instant it was reset. + """ + await seed(factory, [Speaker(ANNA, status="pending", claimed_at=None)]) + interaction = _Interaction() + + await _invoke(cog(factory), "status", interaction) + + assert "a session that ended 2026-08-19 21:00 UTC" in interaction.reply + assert "(1h ago)" in interaction.reply + assert "re-queued job keeps its session's original end time" in interaction.reply + + +async def test_status_says_plainly_when_nothing_is_waiting( + factory: async_sessionmaker[AsyncSession], +) -> None: + await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + + await _invoke(cog(factory), "status", interaction) + + assert "Oldest pending job: none" in interaction.reply + + +# --------------------------------------------------------------------------- +# `/queue session` +# --------------------------------------------------------------------------- + + +async def test_session_reports_the_transcript_length_and_never_its_text( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A 100-minute recording with a 24-character transcript is the tell. + + The length is enough to decide whether a re-queue is warranted; the + text is meeting content, and a slash command must not become a way to + read it out of the document system. + """ + session_id = await seed(factory, [Speaker(ANNA, transcript=" Copyright WDR 2021")]) + interaction = _Interaction() + + await _invoke(cog(factory), "session", interaction, session_id) + + assert "anna" in interaction.reply + assert "19 characters" in interaction.reply + assert "Copyright WDR" not in interaction.reply, "the transcript text must never be echoed" + + +async def test_session_reports_whether_the_audio_still_exists( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN, audio_deleted_at=T1)]) + interaction = _Interaction() + + await _invoke(cog(factory), "session", interaction, session_id) + + assert "audio: present" in interaction.reply + assert "audio: erased" in interaction.reply + + +async def test_session_from_another_guild_reads_as_not_existing( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Identical to the reply for an id that exists nowhere at all. + + A different answer would turn the command into a way to probe whether + a session id exists in some other server. + """ + session_id = await seed(factory, [Speaker(ANNA)], guild=OTHER_GUILD) + seen = _Interaction() + unseen = _Interaction() + + await _invoke(cog(factory), "session", seen, session_id) + await _invoke(cog(factory), "session", unseen, 999_999) + + assert seen.reply == NO_SUCH_SESSION + assert unseen.reply == NO_SUCH_SESSION + + +async def test_a_command_outside_a_guild_is_refused( + factory: async_sessionmaker[AsyncSession], +) -> None: + interaction = _Interaction(guild_id=None) + + await _invoke(cog(factory), "session", interaction, 1) + + assert "only be used in a server" in interaction.reply + + +# --------------------------------------------------------------------------- +# `/queue requeue` -- refusals, which never reach a confirmation at all. +# --------------------------------------------------------------------------- + + +async def test_requeue_defers_before_touching_the_database( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Reading and locking a session's jobs can exceed Discord's three seconds.""" + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert interaction.response.deferred is True + assert interaction.response.deferred_ephemeral is True + assert interaction.followup.messages, "the answer arrives as a followup" + assert interaction.ephemeral is True + + +async def test_requeue_refuses_a_session_with_a_running_job( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The worker holding it would overwrite the reset when it completes.""" + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN, status="running")]) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert "Refused" in interaction.reply + assert "ben" in interaction.reply + assert interaction.view is None, "a refusal must not offer a Confirm button" + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done", "nothing may change on a refusal" + + +async def test_requeue_refuses_when_every_speakers_audio_is_erased( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Making no change while reporting success is the failure to avoid.""" + session_id = await seed( + factory, + [Speaker(ANNA, audio_deleted_at=T1), Speaker(BEN, audio_deleted_at=T1)], + ) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert "Nothing to do" in interaction.reply + assert "erased" in interaction.reply + assert interaction.view is None + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].transcript == "old hallucinated text" + + +async def test_requeue_refuses_a_session_that_has_no_jobs( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, []) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert "Nothing to do" in interaction.reply + assert interaction.view is None + + +async def test_requeue_of_another_guilds_session_reads_as_not_existing( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [Speaker(ANNA)], guild=OTHER_GUILD) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert interaction.reply == NO_SUCH_SESSION + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done" + + +# --------------------------------------------------------------------------- +# `/queue requeue` -- the confirmation, and what it says before writing. +# --------------------------------------------------------------------------- + + +async def test_requeue_writes_nothing_until_the_confirmation_is_pressed( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The command itself is a question. Only the button is an action.""" + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert isinstance(interaction.view, RequeueConfirmView) + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done" + assert jobs[ANNA].transcript == "old hallucinated text" + assert (await read_session(factory, session_id)).status == "documented" + + +async def test_the_confirmation_names_every_consequence_an_admin_cannot_undo( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Three of them are irreversible or visible to people who are not admins. + + The stored transcripts are discarded, a second Outline document is + created and the first is orphaned, and a second link is posted into + the voice channel where non-admins will see it. + """ + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN, audio_deleted_at=T1)]) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + reply = interaction.reply + assert "1 speaker" in reply and "discarded" in reply + assert "ben" in reply and "erased" in reply and "carried" in reply + assert DOC_URL in reply and "not updated" in reply + assert f"<#{CHANNEL}>" in reply + + +async def test_only_the_invoker_may_press_confirm( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Discord components are clickable by anyone who can see the message.""" + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + intruder = _Interaction(user_id=ADMIN + 1) + allowed = await view.interaction_check(_as_interaction(intruder)) + + assert allowed is False + assert "Only" in intruder.reply + + +async def test_a_timeout_disables_the_buttons( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A stale prompt must not still be able to re-queue a session tomorrow.""" + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await view.on_timeout() + + assert all(item.disabled for item in view.children if isinstance(item, discord.ui.Button)) + assert interaction.message.edits, "the message carrying the buttons must be updated" + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done" + + +async def test_cancelling_changes_nothing( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + pressed = _Interaction() + await _press(view, "Cancel", pressed) + + assert "Cancelled" in pressed.reply + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done" + assert jobs[ANNA].transcript == "old hallucinated text" + assert (await read_session(factory, session_id)).announced_at == T1 + + +# --------------------------------------------------------------------------- +# `/queue requeue` -- the write itself. +# --------------------------------------------------------------------------- + + +async def test_confirming_resets_every_column_the_redo_depends_on( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`claim` selects only `pending`, so nothing less than this resurrects it. + + `attempts` and `error` describe the old run and `claimed_at` is a + lease that no longer means anything; leaving any of them would make + `/queue status` misreport the redo before it has even started. + """ + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + pressed = _Interaction() + await _press(view, "Confirm", pressed) + + job = (await read_jobs(factory, session_id))[ANNA] + assert job.status == "pending" + assert job.claimed_at is None + assert job.attempts == 0 + assert job.error is None + assert job.transcript is None + + +async def test_confirming_clears_the_transcript_so_a_half_done_redo_cannot_lie( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`assemble` reads every job of the session, not only the last to finish. + + A reset job that kept its old text would put the very hallucinations + the re-queue exists to remove into the new document if the session + were re-documented before the redo finished. Clearing it makes a + half-done redo visibly incomplete instead of plausibly wrong. + """ + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].transcript is None + assert jobs[BEN].transcript is None + + +async def test_confirming_returns_the_session_to_closed_and_not_to_open( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`closed` is exactly the post-`close_session`, pre-documentation state. + + `open` would make `find_open_session` believe this guild has a live + recording; `closed` is what makes `complete`'s last-job rule + (`remaining == 0 and session_status == "closed"`) fire a second time, + and it keeps `candidates_for_announcement` from selecting the session + while the redo is still running. + """ + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + row = await read_session(factory, session_id) + assert row.status == "closed" + assert row.ended_at == T1, "a re-closed session keeps the end it already had" + + +async def test_confirming_clears_announced_at_so_the_new_link_is_posted( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`mark_documented` never touches `announced_at`, and nothing else does. + + Leaving it set produces a session that transcribes, re-documents with + a fresh URL, and is then never announced -- nothing logged, nothing + raised, the channel simply never sees the new link. A corrected + transcript nobody is told about is indistinguishable from no + transcript, which is the defect being fixed. + """ + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + assert (await read_session(factory, session_id)).announced_at is None + + +async def test_confirming_leaves_the_old_document_on_the_session_row( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The next `mark_documented` overwrites them; clearing them early would + only stop `/queue session` showing which document is being superseded.""" + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + row = await read_session(factory, session_id) + assert row.document_url == DOC_URL + assert row.document_id == "doc-1" + assert row.document_provider == "outline" + + +async def test_confirming_skips_an_erased_speaker_and_keeps_their_transcript( + factory: async_sessionmaker[AsyncSession], +) -> None: + """Their `done` status keeps them terminal for `complete`'s count, and + their old text is what `assemble` puts in the new document for them.""" + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN, audio_deleted_at=T1)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + pressed = _Interaction() + await _press(view, "Confirm", pressed) + + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "pending" + assert jobs[BEN].status == "done" + assert jobs[BEN].transcript == "old hallucinated text" + assert jobs[BEN].audio_deleted_at == T1 + assert "ben" in pressed.reply and "carried" in pressed.reply + + +async def test_confirming_reports_how_many_of_how_many_speakers_were_requeued( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed( + factory, [Speaker(ANNA), Speaker(BEN), Speaker(CLARA, audio_deleted_at=T1)] + ) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + pressed = _Interaction() + await _press(view, "Confirm", pressed) + + assert "2 of 3" in pressed.reply + + +async def test_confirming_touches_no_other_session( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [Speaker(ANNA)]) + other = await seed(factory, [Speaker(BEN)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + assert (await read_jobs(factory, other))[BEN].status == "done" + assert (await read_session(factory, other)).announced_at == T1 + + +async def test_the_plan_is_recomputed_when_confirm_is_pressed( + factory: async_sessionmaker[AsyncSession], +) -> None: + """A worker can claim a sibling job between the question and the answer. + + The confirmation shows a plan built before the button existed; the + write must not act on it. Building the plan again inside the row lock + is what stops a re-queue racing a `complete()` that is already in + flight -- here simulated by a job going `running` while the prompt is + on screen. + """ + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await set_job_status(factory, session_id, BEN, "running") + pressed = _Interaction() + await _press(view, "Confirm", pressed) + + assert "Refused" in pressed.reply + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done", "the stale plan must not have been applied" + assert jobs[ANNA].transcript == "old hallucinated text" + assert (await read_session(factory, session_id)).status == "documented" + + +async def test_pressing_confirm_twice_cannot_reset_the_session_again( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The second press finds the jobs `pending` and refuses on that alone. + + Nothing depends on the buttons being disabled in time: after the first + write the session no longer qualifies, which is what stops a double + press turning into a second document and a second public post. + """ + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + second = _Interaction() + await _press(view, "Confirm", second) + + assert "Refused" in second.reply + assert (await read_session(factory, session_id)).status == "closed" + + +async def test_confirming_disables_the_buttons( + factory: async_sessionmaker[AsyncSession], +) -> None: + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + await _press(view, "Confirm", _Interaction()) + + assert all(item.disabled for item in view.children if isinstance(item, discord.ui.Button)) + + +# --------------------------------------------------------------------------- +# `_apply_requeue` on its own. Both of these cover a guarantee no path +# through the cog can reach, which is exactly why they exist: a mutation +# run showed the suite passing in full with either one removed from the +# implementation. +# --------------------------------------------------------------------------- + + +async def test_the_write_refuses_another_guilds_session_on_its_own( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The second lock on the door, and the one nothing else turns. + + `/queue requeue` already refuses a foreign session before a Confirm + button ever exists, so no test that goes through the cog can reach + this check -- which means dropping it from the write would look + entirely safe. It is not: the view holds a `guild_id` and a + `session_id` for the length of the prompt, and the guarantee that the + write applies to a session of *that* guild is the one this command + must never lose. + """ + session_id = await seed(factory, [Speaker(ANNA)], guild=OTHER_GUILD) + + assert await _apply_requeue(factory, GUILD, session_id) is None + assert (await read_jobs(factory, session_id))[ANNA].status == "done" + + +async def test_the_write_waits_for_a_worker_holding_the_sessions_jobs( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The plan must be built inside the row lock, not merely before the write. + + This drives the interleaving `JobQueue.complete` really produces: a + worker takes `SELECT ... FOR UPDATE` over every job of the session, + flips one of them, and only then commits. Under READ COMMITTED a + re-queue that read the job rows *without* first taking that same lock + would see the pre-commit snapshot -- every job still `done` -- decide + the whole session is resettable, and then block on the UPDATE instead; + when the worker's transaction landed, the reset would go through and + overwrite a job that is `running`, which is precisely the state the + refusal exists to protect. + + Taking the lock first turns that into a wait: the plan is not built + until the worker has committed, so it sees the `running` job and + refuses. The lock statement here is character-for-character the one + `JobQueue.complete` takes, `ORDER BY id` included, because matching + lock-acquisition order is what keeps the two from deadlocking. + """ + session_id = await seed(factory, [Speaker(ANNA), Speaker(BEN)]) + + async with factory() as worker: + await worker.execute( + select(TranscriptionJob.id) + .where(TranscriptionJob.session_id == session_id) + .order_by(TranscriptionJob.id) + .with_for_update() + ) + await worker.execute( + update(TranscriptionJob) + .where( + TranscriptionJob.session_id == session_id, + TranscriptionJob.discord_user_id == BEN, + ) + .values(status="running") + ) + task = asyncio.create_task(_apply_requeue(factory, GUILD, session_id)) + # Long enough for the task to reach the database and stop there. + await asyncio.sleep(0.3) + assert not task.done(), "the re-queue must wait on the lock, not race it" + await worker.commit() + + view = await asyncio.wait_for(task, timeout=10) + + assert view is not None + assert view.plan.is_blocked, "the plan was built from a snapshot taken before the commit" + jobs = await read_jobs(factory, session_id) + assert jobs[BEN].status == "running" + assert jobs[ANNA].status == "done" + assert jobs[ANNA].transcript == "old hallucinated text" + assert (await read_session(factory, session_id)).status == "documented" + + +# --------------------------------------------------------------------------- +# Reply rendering, without an `Interaction` at all. +# --------------------------------------------------------------------------- + + +def _summary(**overrides: Any) -> SessionSummary: + defaults: dict[str, Any] = { + "id": 4, + "channel_id": CHANNEL, + "status": "documented", + "ended_at": T1, + "end_reason": "empty", + "document_url": DOC_URL, + "announced_at": T1, + } + return SessionSummary(**{**defaults, **overrides}) + + +def test_a_confirmation_for_a_session_that_was_never_documented_says_so() -> None: + """There is no first document to orphan, so promising one would be wrong.""" + text = render_requeue_confirmation( + _summary(status="closed", document_url=None, announced_at=None), + RequeuePlan(resettable_job_ids=(1,), erased_user_ids=(), active_user_ids=()), + names={}, + ) + + assert "no document" in text + assert "not updated" not in text From 88aa4888622e46031ae5ad9f23d2455ad5ff8322 Mon Sep 17 00:00:00 2001 From: TheMeinerLP Date: Fri, 21 Aug 2026 00:24:13 +0200 Subject: [PATCH 2/3] docs(operations): document /queue and the silent-hallucination failure Section 5 previously had no way to name a job that succeeded end-to-end and was still wrong: `status` done, `transcript` non-NULL, nothing logged. That was every session transcribed before this branch's VAD fix. Add it as a third failure category alongside the existing transcription/document split, with the concrete tell (/queue session's transcript length, e.g. the 19-character " Copyright WDR 2021" fixture) and the fix (/queue requeue), and document the three /queue subcommands themselves, which had no writeup anywhere. Also point the stale "no admin command yet to list dead jobs" line at /queue status and /queue session. No settings table changes: the two commits on this branch touch no Settings/WorkerSettings/LinkSettings field, verified by diffing against the branch's merge-base. first-deployment.md needs no change for the same reason -- no first-deployment step touches any of this. --- docs/operations.md | 95 +++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 93 insertions(+), 2 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index 23961e3..2809206 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -496,14 +496,49 @@ stored is actually what is running. ## 5. Troubleshooting +**`/queue`, the admin command for most of what follows.** `QueueCog` +(`src/sturnus/infrastructure/discord/queue_cog.py`) adds three subcommands, +all admin-gated the same way `/setup` and `/config` are (`require_admin`, +section 3.1) and all replying `ephemeral=True`: + +- **`/queue status`** — a guild-wide, counts-only overview: jobs by status + (`pending` / `running` / `done` / `dead`), how many `running` jobs are + past the default lease, the age of the oldest `pending` job, and how many + `closed` sessions have every job finished but no document yet. Read-only. +- **`/queue session `** — one session in detail: its status, + end time and reason, its document URL and announcement time, and one + line per speaker giving job status, attempts, whether the audio is still + present, the last error, and the *length* of the stored transcript — + never its text; a slash command is deliberately not a way to read + meeting content. Read-only. +- **`/queue requeue `** — the only one that writes. It resets + the session's finished jobs back to `pending` so the worker transcribes + them again from the still-stored audio, discarding whatever was there + before. Nothing is written until an explicit Confirm press on a message + that names, in full, what cannot be undone: the discarded transcripts, a + second Outline document (the old one is left in place, not deleted or + updated — `DocumentSink` has no update path), and a second announcement + posted to the recording channel. It refuses outright if any job of the + session is still `pending` or `running` (there is nothing to redo yet, + and resetting a job a worker is about to finish would just let that + worker overwrite the reset), and it skips — rather than resets — any + speaker whose audio has already been erased, carrying their existing + transcript into the new document unchanged. + +Every `/queue` query is scoped to the guild the command was run in; a +session id from another guild gets the same reply as one that does not +exist, so it cannot be used to probe another guild's sessions. + **A job is `dead`.** `transcription_job.status` becomes `dead` once `attempts` reaches the worker's configured retry limit (`JobQueue.fail`). A dead job is deliberately excluded from the remaining-jobs count that decides whether a session is finished (`JobQueue.complete`) — so one unreadable recording does not block the rest of that session's document from being produced; it just means that -one speaker's portion is permanently missing from it. There is no admin -command yet to list dead jobs; query them directly: +one speaker's portion is permanently missing from it. `/queue status` +gives the guild-wide count and `/queue session ` names which speaker +and shows the last error for one session; for anything that cuts across +sessions or guilds, or that needs the raw row, query directly: ```sql SELECT id, session_id, discord_user_id, attempts, error FROM transcription_job WHERE status = 'dead'; @@ -531,6 +566,62 @@ These happen at different granularities and leave different traces: problem is fixed; anything else raised there is a transient failure worth retrying. +**A third kind, distinct from both of the above: a job that is `done`, +raised no error, and is still wrong.** Every session transcribed by a +worker running Silero VAD (`vad_filter=True`, before this fix landed — +`sturnus.infrastructure.whisper.WhisperEngine._transcribe`, commit +`12d4299`) produced an empty or hallucinated transcript while reporting +complete success: `status` is `done`, `transcript` is non-`NULL`, nothing +is logged, and `/queue status` shows nothing wrong, because nothing failed +from the code's point of view. The mechanism (full reasoning in +`sturnus.infrastructure.speech_gate`'s module docstring) was Silero's +recurrent state collapsing on the bit-exact digital-zero padding +`SpeakerWriter` writes into every gap between packets — on a real +100-minute recording it reported about one second of speech in two +minutes, and the transcript for that speaker came back +`" Copyright WDR 2021"`, a stock Whisper hallucination on near-silence +that has nothing to do with anything anyone said. `"Thank you."` is +another shape of the same failure: a short, generic, plausible-sounding +sentence standing in for a much longer real recording. + +*How to recognise it, without reading a transcript's content at all.* +`/queue session ` reports `transcript: N characters` per +speaker. A session that ran for an hour with a transcript of a few dozen +characters is the tell — the test fixture for this exact failure +(`tests/infrastructure/discord/test_queue_cog.py`) uses the real observed +hallucination `" Copyright WDR 2021"`, all of 19 characters, precisely +because a number that small against a long session is unambiguous at a +glance, and the command deliberately never echoes the transcript text +itself (only its length), so this check does not require reading meeting +content to make the call. + +*This affects only sessions transcribed before the fix.* A session +transcribed by a worker running the amplitude gate does not exhibit this — +the gate carries no state across frames, so there is no history for +padding to corrupt (again, see the module docstring for why). Do not chase +this section for a session transcribed after the fix shipped; a short +transcript there has some other cause. + +*What to do.* Check `/queue session ` for `audio: present` vs. +`audio: erased` per speaker — erased audio cannot be re-transcribed, only +carried forward unchanged (section 6 covers when audio is erased) — and +then run `/queue requeue `. Read what its confirmation says +before pressing Confirm: it names the old (bad) document and states that +it stays and is not deleted or updated, and it says a new link will be +posted publicly in the recording channel once the redo finishes. The +worker's ordinary pipeline carries the redo the rest of the way on its own +— nothing else needs to be run by hand. + +One more thing worth expecting rather than being surprised by: a redone +job actually transcribes the speech instead of ~1% of the file, so it +takes far longer than the original (garbage) run did — the design that +shipped this fix estimated roughly 20-25x, turning a couple of minutes +into tens of minutes for a long session. If `/queue status` starts +reporting `running` jobs past the default lease after a batch of +re-queues, that is very plausibly this, not a stuck worker — check whether +`STURNUS_JOB_LEASE_SECONDS` (section 1.2) has been raised to match before +assuming something is broken. + **The bot is sitting out of the channel while people are in it.** A session that ended with `capture_failure` or `decode_failure` means the bot could not hear, not that nobody spoke, and the guild is then held out of From 4873ef742096daa6017d19bd9cee339696f945fa Mon Sep 17 00:00:00 2001 From: TheMeinerLP Date: Fri, 21 Aug 2026 08:58:06 +0200 Subject: [PATCH 3/3] fix(discord): refuse the re-queues that would break the session they fix Three defects found reviewing /queue, all of them in what the command does when the pipeline has not actually finished with a session, plus the three holes in its tests that let them through. **A re-queue may only touch a `documented` session.** Neither `_load_requeue_view` nor `_apply_requeue` looked at `session.status`, and `_apply_requeue` writes `status="closed"` unconditionally -- so a session still inside `RecordingService.close` could be forced into exactly the state `JobQueue.complete`'s Defect 5 guard exists to refuse. That is not an exotic race: `close` uploads and enqueues one speaker at a time, each `enqueue` committing on its own, and calls `close_session` only after the last upload, so a long multi-speaker session stays `open` for minutes with early speakers already `done`; the first job to finish after the forced `closed` would be taken for the session's last, and the document would be built from part of the meeting. A `closed` session is refused for a different reason: `retry_pending_documents` still owns it and may be between its read and its `mark_documented` write, so a re-queue landing in that sweep gets documented from the transcripts it has just discarded, after which neither `complete`'s last-job rule nor the sweep itself ever looks at the session again. `SessionView.is_settled` is the rule, `is_refused` composes it with the two job rules from `sturnus.application.requeue`, and the check is taken inside the row lock as well as in front of the prompt -- the prompt holds nothing but ids while a human reads it, and the session can move in that time. The refusal names the status the session is in and says what to wait for; an administrator who is told only "no" tries again immediately. **Every reply is bounded to Discord's 2000 characters.** `render_session` emitted one ~85-character line per speaker plus `str(exc)` out of `job.error`, unbounded on both counts, and the resulting `HTTPException` from `followup.send` after a `thinking=True` defer is not a truncated answer but no answer at all -- the diagnostic command going silent exactly when the queue is in the state it exists to diagnose. Errors are collapsed to one line and cut at `MAX_ERROR_CHARS`, so one exception cannot swallow the other speakers; the speaker list is filled to the remaining budget and ends in a line stating how many speakers are not shown. Truncation rather than an attachment, because the readout is scanned for one thing -- a transcript length absurd for the length of the session -- and the first speakers sample that as well as any, while a file arrives as a download to open on a command whose value is that the answer is on screen in a second; `docs/operations.md` section 5 already documents the SQL for the rest. `_capped` is the backstop under that for lengths nobody budgeted, and every render function ends inside it, not only `/queue session`. **`mark_announced` is now a compare-and-set.** `announce_ready_sessions` awaits `announcer.post` -- seconds, under rate limiting -- and stamps `announced_at` only afterwards; a re-queue landing in that window clears the column precisely so the redo's link will be posted, and the late unconditional stamp put it straight back, leaving the corrected transcript documented and never announced, with nothing logged and nothing raised. The UPDATE now matches only the state the selection was made on (still `documented`, still unannounced), so the stamp lands only if the session is still the one the post was about. The cost is one duplicate post of the superseded link, which is the side `announce_ready_sessions` already documents itself as erring towards. The tests had three holes of their own, and every one of them is a mutation the old suite passed with. `_press` called `Button.callback` directly, which is not how discord.py dispatches: `View._scheduled_task` runs `item._run_checks(interaction) and self.interaction_check(interaction)` first and drops the press if either is falsy, so the confirm view's author check could be made to deny every press -- locking administrators out of their own Confirm -- with the file still green. `_press` goes through `_scheduled_task` now, with `on_error` replaced by one that re-raises, because the default logs and swallows exactly the assertions these fakes exist to make. `_Followup.send` accepted a followup on an interaction Discord had never been told about, so the `defer` in `RequeueConfirmView.confirm` was deletable with the suite green; it now refuses one, which is the contract the module docstring already claimed to enforce. And the four reply paths that never asserted `ephemeral is True` now do, ephemerality being a contract this cog states in its own docstring because these replies name who was recorded. docs/operations.md section 5 gains the refusal rule with both states and what to wait for, the fact that `/queue session` truncates and where to read the untruncated rows, and the duplicate announcement as something to expect rather than to be alarmed by. --- docs/operations.md | 53 ++- src/sturnus/infrastructure/db/repositories.py | 55 ++- .../infrastructure/discord/queue_cog.py | 283 +++++++++++++-- .../infrastructure/discord/test_queue_cog.py | 326 +++++++++++++++++- tests/infrastructure/test_repositories.py | 52 +++ 5 files changed, 727 insertions(+), 42 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index 2809206..1f37bb8 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -510,7 +510,14 @@ section 3.1) and all replying `ephemeral=True`: line per speaker giving job status, attempts, whether the audio is still present, the last error, and the *length* of the stored transcript — never its text; a slash command is deliberately not a way to read - meeting content. Read-only. + meeting content. Read-only. A Discord message holds 2000 characters, and + this reply is bounded to fit inside one: each speaker's `error` is shown + up to 160 characters (whitespace collapsed, an `…` marking the cut), and + a session with more speakers than fit ends in a line saying how many are + not shown. That is a real limit for a large channel — roughly the first + twenty speakers fit — so for a full picture of a big session, or for an + error too long to display, read the rows directly with the query at the + end of "A job is `dead`" below. - **`/queue requeue `** — the only one that writes. It resets the session's finished jobs back to `pending` so the worker transcribes them again from the still-stored audio, discarding whatever was there @@ -525,6 +532,31 @@ section 3.1) and all replying `ephemeral=True`: speaker whose audio has already been erased, carrying their existing transcript into the new document unchanged. + **It also refuses any session that is not `documented` yet**, and says + which state it is in instead. `documented` is the only status in which + nothing else in the pipeline still has a claim on the session, and the + two other states are refused for two different reasons: + + - `open` — the recording has not finished, or the bot is still uploading + the speakers it recorded. Sessions are enqueued one speaker at a time + and only marked `closed` after the last upload, so re-queueing here + would close the session while speakers are still being added to it, + and the next job to finish would be taken for the session's last: the + document would then be built from part of the meeting. This is the + likely mistake right after a long meeting ends — the recording looks + over in Discord well before the bot has finished uploading it. + - `closed` — transcription finished but no document exists yet. The + worker's retry sweep still owns the session and creates that document + on its next pass; a re-queue landing in the middle of the sweep can + leave the session documented from the transcripts the re-queue just + discarded, with nothing revisiting it afterwards. + + In both cases the remedy is to wait and re-run `/queue session + ` until it reports `documented`. If a session never gets + there, that is a separate fault — `/queue status` counts the sessions + stuck in it, and "Telling a transcription failure from a document + failure apart" below is where to start on it. + Every `/queue` query is scoped to the guild the command was run in; a session id from another guild gets the same reply as one that does not exist, so it cannot be used to probe another guild's sessions. @@ -602,8 +634,10 @@ padding to corrupt (again, see the module docstring for why). Do not chase this section for a session transcribed after the fix shipped; a short transcript there has some other cause. -*What to do.* Check `/queue session ` for `audio: present` vs. -`audio: erased` per speaker — erased audio cannot be re-transcribed, only +*What to do.* Check that `/queue session ` reports the session +as `documented` — a re-queue of an `open` or `closed` session is refused, +for the reasons listed under `/queue requeue` above — then check `audio: +present` vs. `audio: erased` per speaker — erased audio cannot be re-transcribed, only carried forward unchanged (section 6 covers when audio is erased) — and then run `/queue requeue `. Read what its confirmation says before pressing Confirm: it names the old (bad) document and states that @@ -622,6 +656,19 @@ re-queues, that is very plausibly this, not a stuck worker — check whether `STURNUS_JOB_LEASE_SECONDS` (section 1.2) has been raised to match before assuming something is broken. +A second, harmless surprise: a re-queue run in the same seconds as the +bot's announcement poll can put the *old* link into the channel one last +time before the redo starts. The poll posts the link and only afterwards +stamps `session.announced_at`; a re-queue that lands in between clears +that column on purpose, and the late stamp is then rejected +(`SessionRepository.mark_announced` only stamps a session that is still +`documented` and still unannounced) so that the redo's new link is +announced when it is ready. The bot logs `Session N changed while its +announcement was being posted` when that happens. The alternative — +letting the late stamp through — would mean the corrected transcript is +documented and its link never posted at all, so the duplicate is the +deliberate choice. + **The bot is sitting out of the channel while people are in it.** A session that ended with `capture_failure` or `decode_failure` means the bot could not hear, not that nobody spoke, and the guild is then held out of diff --git a/src/sturnus/infrastructure/db/repositories.py b/src/sturnus/infrastructure/db/repositories.py index 241e922..11a53e6 100644 --- a/src/sturnus/infrastructure/db/repositories.py +++ b/src/sturnus/infrastructure/db/repositories.py @@ -7,6 +7,7 @@ from __future__ import annotations +import logging from datetime import UTC, datetime from sqlalchemy import CursorResult, delete, select, update @@ -25,6 +26,8 @@ TranscriptionJob, ) +log = logging.getLogger(__name__) + class ConsentRepository: def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: @@ -301,11 +304,59 @@ async def candidates_for_announcement(self) -> list[dict[str, object]]: ] async def mark_announced(self, session_id: int, now: datetime) -> None: + """Stamps `announced_at`, but only on the session that was announced. + + A compare-and-set, not a plain write, and the condition is the + whole point. `sturnus.application.publishing. + announce_ready_sessions` selects a `documented` session whose + `announced_at` is null, awaits `announcer.post` -- a Discord HTTP + call that takes seconds under rate limiting -- and calls this only + afterwards. `/queue requeue` can land inside that window: it puts + the session back to `closed` and clears `announced_at` precisely + so the redo's new link will be posted. An unconditional stamp + arriving late would put a timestamp back on a session that has not + been announced since, and `sessions_to_announce` selects only + sessions whose `announced_at` is null -- so the corrected + transcript would be documented and then never posted, with nothing + logged and nothing raised. That is the exact failure clearing the + column exists to prevent, reintroduced by the sweep that was + racing it. + + Restricting the UPDATE to the state the selection was made on -- + still `documented`, still unannounced -- makes the stamp land only + if the session is still the one the post was about. When it is + not, no row matches, the re-queued session keeps its null + `announced_at`, and the sweep after the redo announces the new + link. The cost is one duplicate post of the superseded link, which + is the side `announce_ready_sessions` already documents itself as + erring towards: losing an announcement entirely is the worse half + of that trade. + """ async with self._session_factory() as session: - await session.execute( - update(Session).where(Session.id == session_id).values(announced_at=now) + result = await session.execute( + update(Session) + .where( + Session.id == session_id, + Session.status == DOCUMENTED_STATUS, + Session.announced_at.is_(None), + ) + .values(announced_at=now) ) await session.commit() + # Same narrowing `AccountLinkRepository.delete` uses: `execute` on + # a Core UPDATE always yields a `CursorResult` at runtime, and the + # assertion makes `.rowcount` available without an unchecked cast. + assert isinstance(result, CursorResult) + if result.rowcount == 0: + # Not an error: the announcement went out, and the session it + # went out for no longer exists in that form. Worth a line + # anyway -- it is the only trace connecting a link posted in a + # channel to a session row that does not say it was announced. + log.info( + "Session %d changed while its announcement was being posted; " + "announced_at left unset so the next sweep can announce it again", + session_id, + ) async def closed_undocumented_sessions(self) -> list[int]: """Closed sessions whose jobs are all terminal but which never got documented. diff --git a/src/sturnus/infrastructure/discord/queue_cog.py b/src/sturnus/infrastructure/discord/queue_cog.py index 7d37635..292a248 100644 --- a/src/sturnus/infrastructure/discord/queue_cog.py +++ b/src/sturnus/infrastructure/discord/queue_cog.py @@ -15,8 +15,13 @@ second time on their own. Nothing here orchestrates the redo. Which jobs may be reset is decided by `sturnus.application.requeue.plan_requeue`, a pure function tested without a database; read its module docstring first, -because the two rules that keep this command from being destructive live -there rather than here. +because two of the three rules that keep this command from being +destructive live there rather than here. The third is `SessionView. +is_settled`: only a `documented` session may be re-queued at all, because +that is the one status in which nothing else in the pipeline is still +working on the session -- a rule about the session row rather than about +its jobs, which is why it lives here beside the write and not in that +pure function over job rows. **Why the SQL is inline instead of in a repository.** This selection is specific to these three commands and used nowhere else, so @@ -65,6 +70,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sturnus.application.ports import Clock +from sturnus.application.publishing import DOCUMENTED_STATUS from sturnus.application.requeue import TERMINAL_STATUSES, RequeuePlan, plan_requeue from sturnus.infrastructure.db.models import Session, SessionParticipant, TranscriptionJob from sturnus.infrastructure.db.queue import DEFAULT_LEASE_SECONDS @@ -85,6 +91,23 @@ #: this timeout only keeps a forgotten prompt from lingering. CONFIRM_TIMEOUT_SECONDS = 60.0 +#: The longest message body Discord accepts. Not advisory: `followup.send` +#: raises `HTTPException` on anything longer, and every reply this cog +#: sends goes out *after* a `thinking=True` defer -- so an over-long reply +#: is not a truncated answer, it is no answer at all, on commands whose +#: whole job is to explain a queue that has gone wrong. Every render +#: function here therefore ends inside `_capped`. +DISCORD_MESSAGE_LIMIT = 2000 + +#: How much of one job's stored `error` a `/queue session` line may carry. +#: `transcription_job.error` is `str(exc)` -- arbitrary text of arbitrary +#: length, never truncated on the way in -- and one such string is easily +#: longer than the whole message budget. Enough to recognise a failure +#: ("An error occurred (AccessDenied) when calling the GetObject +#: operation..."), and `docs/operations.md` section 5 says where to read +#: the untruncated row. +MAX_ERROR_CHARS = 160 + #: Job statuses reported by `/queue status`, in lifecycle order rather than #: alphabetically: an administrator reads this line to see where work is #: piling up, and `pending -> running -> done | dead` is the order that @@ -125,6 +148,55 @@ class SessionView: #: `discord_user_id` -> display name, from `session_participant`. names: dict[int, str] + @property + def is_settled(self) -> bool: + """Whether the pipeline has finished with this session and let go of it. + + `documented` is the only status from which a re-queue is safe, and + each of the other two is unsafe for its own reason. + + An `open` session is the window `JobQueue.complete`'s "Defect 5" + guard exists to refuse. `RecordingService.close` uploads and + enqueues one speaker at a time, every `enqueue` committing on its + own, and calls `close_session` only after the last upload -- so a + long multi-speaker session spends a long time `open` with early + speakers already enqueued (and possibly already `done`) while + later ones do not exist as rows yet. `_apply_requeue` writes + `status="closed"` unconditionally, so re-queueing in that window + hands `complete` exactly the state its guard is there to prevent: + no outstanding jobs plus a `closed` session, therefore "this was + the session's last job", therefore a document assembled from the + speakers that happened to exist at that moment. That is this + command causing the failure it exists to repair. + + A `closed` session is still owned by + `sturnus.application.worker.retry_pending_documents`, which + documents any closed session whose jobs are all terminal and may + be between its read and its `mark_documented` write right now. + Re-queueing into that sweep gives it a session whose transcripts + this command has just cleared: it publishes that near-empty + document and flips the session to `documented`, after which + `complete`'s last-job rule (which requires `closed`) never fires + again and the sweep never selects it again either -- the redo runs + to completion and no document is ever made from it. + + Waiting costs nothing: a session that is genuinely finished + reaches `documented` on its own, and `/queue session` shows when. + """ + return self.summary.status == DOCUMENTED_STATUS + + @property + def is_refused(self) -> bool: + """Whether this session must be refused instead of re-queued. + + The three reasons in the order `render_requeue_refusal` reports + them, which is the order of what an administrator can do about it: + a blocked session only needs the queue to go idle, an unsettled + one needs the pipeline to finish, and an empty one will never + change. + """ + return self.plan.is_blocked or not self.is_settled or self.plan.is_empty + @dataclass(frozen=True) class JobLine: @@ -351,10 +423,18 @@ async def _apply_requeue( that was actually applied, or the one that caused a refusal. `None` means the session does not belong to this guild (or does not exist), which the caller renders as `NO_SUCH_SESSION`. Whether the write - happened is derivable: it did exactly when the returned plan is neither - blocked nor empty. + happened is derivable: it did exactly when the returned view is not + `is_refused`. - Two properties matter and neither is incidental. + Three properties matter and none is incidental. + + **The session's own status is checked here too, not only in front of + the prompt.** `SessionView.is_settled` spells out why only a + `documented` session may be reset; what matters at *this* point is + that the check is made against the row this transaction locked. The + prompt holds nothing but ids while an administrator reads it, and the + session can move in that time -- so a status read before the lock + would be exactly the stale snapshot the lock exists to rule out. **The lock comes first.** `SELECT TranscriptionJob.id WHERE session_id = ... ORDER BY id FOR UPDATE` is the same statement, with the same @@ -395,7 +475,7 @@ async def _apply_requeue( plan=plan_requeue(await _job_dicts(db, session_id)), names=await _participant_names(db, session_id), ) - if view.plan.is_blocked or view.plan.is_empty: + if view.is_refused: return view await db.execute( @@ -446,6 +526,15 @@ async def _apply_requeue( # Clearing it is the deliberate choice to post again, # exactly once, the same not-null guard preventing any # further repeats. + # + # Clearing the column is not by itself enough to make that + # second post happen: an announcement sweep can already be + # inside `announcer.post` for this session right now, and + # its `mark_announced` afterwards would stamp the column + # we have just cleared. `SessionRepository.mark_announced` + # is a compare-and-set on `status = 'documented' AND + # announced_at IS NULL` for that reason -- the `closed` + # written above is what makes the late stamp miss. announced_at=None, # `document_provider`/`document_id`/`document_url` are # deliberately untouched: the next `mark_documented` @@ -501,6 +590,45 @@ def _speakers(count: int) -> str: return "1 speaker" if count == 1 else f"{count} speakers" +def _one_line(text: str) -> str: + """Collapses every run of whitespace, so one value stays one line. + + `job.error` can carry newlines -- a wrapped traceback, an XML error + body -- and the `/queue session` readout is a bullet per speaker, + read as such. Left alone, one error would break its speaker into a + dozen lines that look like speakers of their own, and each of those + newlines would also spend budget that belongs to a speaker who then + does not get shown. + """ + return " ".join(text.split()) + + +def _shortened(text: str, limit: int) -> str: + """`text` cut to `limit` characters, ending in an ellipsis when it was cut.""" + if len(text) <= limit: + return text + return text[: limit - 1].rstrip() + "…" + + +def _capped(text: str) -> str: + """The last line of defence: never hand Discord a body it will reject. + + The render functions bound themselves where the length actually comes + from -- errors, speaker lists -- and that structured bound is what + keeps a truncated reply *readable*. This is the crude backstop under + it, for the lengths nobody budgeted: a session document URL a + self-hosted Outline made 900 characters long, a display name that is + all combining marks. A reply cut off mid-sentence is a poor answer; + an `HTTPException` behind a `thinking=True` defer is no answer at all, + and this cog is what an administrator reaches for when they already + cannot see what is happening. + """ + if len(text) <= DISCORD_MESSAGE_LIMIT: + return text + marker = f"\n… (cut off at Discord's {DISCORD_MESSAGE_LIMIT}-character limit)" + return text[: DISCORD_MESSAGE_LIMIT - len(marker)] + marker + + def _running_jobs(count: int) -> str: return "1 running job" if count == 1 else f"{count} running jobs" @@ -537,11 +665,65 @@ def render_status(status: QueueStatus, now: datetime, lease_seconds: float) -> s f"{_closed_sessions(status.closed_undocumented)} with every job finished but " "no document yet; the worker retries those on its own sweep." ) - return "\n".join(lines) + return _capped("\n".join(lines)) + + +def _speaker_line(job: JobLine, names: dict[int, str]) -> str: + """One speaker's bullet, bounded so no single job can eat the reply.""" + name = names.get(job.discord_user_id, f"user {job.discord_user_id}") + length = ( + "transcript: none stored" + if job.transcript_length is None + else f"transcript: {job.transcript_length} characters" + ) + audio = "audio: present" if job.audio_present else "audio: erased" + line = f"- {name} — status: `{job.status}`, attempts: {job.attempts}, {audio}, {length}" + if job.error: + line += f", last error: {_shortened(_one_line(job.error), MAX_ERROR_CHARS)}" + return line + + +def _omitted(count: int) -> str: + noun = "speaker" if count == 1 else "speakers" + return ( + f"…and {count} more {noun} not shown — the full readout is longer than the " + f"{DISCORD_MESSAGE_LIMIT} characters Discord allows in one message. Query " + "`transcription_job` directly for the rest (docs/operations.md, section 5)." + ) + + +def _fitted(header: list[str], speakers: list[str]) -> list[str]: + """`header` plus as many speaker lines as fit, then a line saying how many did not. + + Truncation rather than an attached file, and the choice is about what + a truncated answer is *for*. This readout is scanned for one thing -- + a speaker whose transcript length is absurd for the length of the + session -- and the first speakers are as good a sample of that as any; + an attachment would answer completely but reaches the administrator as + a download to open, on a command whose value is that the answer is on + screen in a second. Nothing is lost silently either way: the omitted + count is stated, and section 5 of `docs/operations.md` already + documents the SQL for the rows behind this command. + + Dropping the *tail* rather than the middle keeps the list in job-id + order, which is the order every other `/queue` reply and the document + itself use. + """ + kept: list[str] = [] + # `+ 1` per line for the newline `join` will add. One more than it + # really needs, which is the safe direction to be wrong in. + used = sum(len(line) + 1 for line in header) + for index, line in enumerate(speakers): + omitted = _omitted(len(speakers) - index) + if used + len(line) + 1 + len(omitted) + 1 > DISCORD_MESSAGE_LIMIT: + return [*header, *kept, omitted] + used += len(line) + 1 + kept.append(line) + return [*header, *kept] def render_session(summary: SessionSummary, jobs: list[JobLine], names: dict[int, str]) -> str: - lines = [ + header = [ f"**Session {summary.id}** in <#{summary.channel_id}>", f"Status: `{summary.status}` — ended {_stamp(summary.ended_at)}" f" ({summary.end_reason or 'no reason recorded'})", @@ -549,22 +731,11 @@ def render_session(summary: SessionSummary, jobs: list[JobLine], names: dict[int f"Announced: {_stamp(summary.announced_at)}", ] if not jobs: - lines.append("No transcription jobs — nobody spoke in this session.") - return "\n".join(lines) - lines.append(f"{_speakers(len(jobs))}:") - for job in jobs: - name = names.get(job.discord_user_id, f"user {job.discord_user_id}") - length = ( - "transcript: none stored" - if job.transcript_length is None - else f"transcript: {job.transcript_length} characters" - ) - audio = "audio: present" if job.audio_present else "audio: erased" - line = f"- {name} — status: `{job.status}`, attempts: {job.attempts}, {audio}, {length}" - if job.error: - line += f", last error: {job.error}" - lines.append(line) - return "\n".join(lines) + header.append("No transcription jobs — nobody spoke in this session.") + return _capped("\n".join(header)) + header.append(f"{_speakers(len(jobs))}:") + speakers = [_speaker_line(job, names) for job in jobs] + return _capped("\n".join(_fitted(header, speakers))) def _document_line(summary: SessionSummary, *, future: bool) -> str: @@ -614,7 +785,7 @@ def render_requeue_confirmation( f"- A new link is posted in <#{summary.channel_id}> once it is ready. Everyone " "who can see that channel sees the post, not only administrators." ) - return "\n".join(lines) + return _capped("\n".join(lines)) def render_requeue_applied( @@ -634,7 +805,46 @@ def render_requeue_applied( ) lines.append(_document_line(summary, future=False)) lines.append(f"- A new link is posted in <#{summary.channel_id}> when it is ready.") - return "\n".join(lines) + return _capped("\n".join(lines)) + + +def _unsettled_refusal(summary: SessionSummary) -> str: + """Why a session that is not `documented` cannot be re-queued yet. + + One sentence per status about what is still holding the session, then + the same instruction in both cases: wait for `documented`. The + reasoning behind each is on `SessionView.is_settled`; what an + administrator needs from the reply is that the pipeline is not + finished with this session and that waiting is the whole remedy -- + said plainly enough that the obvious next move is not to try again + immediately. + """ + if summary.status == "open": + return ( + f"Refused: session {summary.id} is still open — the recording has not " + "finished, or the bot is still uploading the speakers it recorded. A " + "re-queue would close the session while speakers are still being added to " + "it, and the first job to finish afterwards would be taken for the " + "session's last: the document would then be built from part of the " + "meeting. Stop the recording and wait until `/queue session " + f"{summary.id}` reports `documented`." + ) + if summary.status == "closed": + return ( + f"Refused: session {summary.id} has finished transcribing but has no " + "document yet. The worker's own retry sweep still owns it and creates " + "that document on its next pass; a re-queue landing in the middle of the " + "sweep can leave the session documented from the transcripts this command " + "has just discarded, and nothing revisits it afterwards. Wait until " + f"`/queue session {summary.id}` reports `documented` and re-queue then — " + "if it never gets there, that is a different fault, and `/queue status` " + "counts the sessions stuck in it." + ) + return ( + f"Refused: session {summary.id} is `{summary.status}`, and only a " + "`documented` session can be re-queued — that is the one state in which " + "nothing else in the pipeline is still working on it." + ) def render_requeue_refusal( @@ -645,29 +855,40 @@ def render_requeue_refusal( Split from the other two because a refusal is not a smaller success: `config_cog` names four honest outcomes rather than one cheerful confirmation, and this follows that. + + The branches are checked in `SessionView.is_refused`'s order and must + stay in step with it, or the reply would explain a reason other than + the one the write actually refused on. """ prefix = ( "The session changed while the confirmation was on screen, so nothing was written. " if rechecked else "" ) + return _capped(f"{prefix}{_refusal(summary, plan, names)}") + + +def _refusal(summary: SessionSummary, plan: RequeuePlan, names: dict[int, str]) -> str: + """The reason itself, in `SessionView.is_refused`'s order of checks.""" if plan.is_blocked: return ( - f"{prefix}Refused: {_speakers(len(plan.active_user_ids))} in session " + f"Refused: {_speakers(len(plan.active_user_ids))} in session " f"{summary.id} still have a job pending or running " f"({_named(plan.active_user_ids, names)}). Those recordings are already " "going to be transcribed, and resetting a job a worker is holding would be " "undone the moment that worker finishes. Try again once the queue is idle." ) + if summary.status != DOCUMENTED_STATUS: + return _unsettled_refusal(summary) if plan.erased_user_ids: return ( - f"{prefix}Nothing to do: every speaker in session {summary.id} has had " + f"Nothing to do: every speaker in session {summary.id} has had " f"their audio erased ({_named(plan.erased_user_ids, names)}). Re-queueing " "would only hand a worker a key it cannot download. Their existing " "transcripts are untouched." ) return ( - f"{prefix}Nothing to do: session {summary.id} has no transcription jobs at all " + f"Nothing to do: session {summary.id} has no transcription jobs at all " "— nobody spoke, so there is nothing to transcribe again." ) @@ -756,7 +977,7 @@ async def confirm( if view is None: await interaction.followup.send(NO_SUCH_SESSION, ephemeral=True) return - if view.plan.is_blocked or view.plan.is_empty: + if view.is_refused: await interaction.followup.send( render_requeue_refusal(view.summary, view.plan, view.names, rechecked=True), ephemeral=True, @@ -872,7 +1093,7 @@ async def requeue(self, interaction: discord.Interaction, session_id: int) -> No if view is None: await interaction.followup.send(NO_SUCH_SESSION, ephemeral=True) return - if view.plan.is_blocked or view.plan.is_empty: + if view.is_refused: # No buttons at all: there is nothing to confirm, and offering # a Confirm that would only be refused invites the # administrator to press it and learn nothing new. diff --git a/tests/infrastructure/discord/test_queue_cog.py b/tests/infrastructure/discord/test_queue_cog.py index bb9e8cd..4fc33cc 100644 --- a/tests/infrastructure/discord/test_queue_cog.py +++ b/tests/infrastructure/discord/test_queue_cog.py @@ -5,8 +5,18 @@ the cog defines, so calling it exercises the decision without a gateway. The hand-rolled `_Response`/`_Followup`/`_Interaction` fakes are the same shape as `test_config_commands`'s, and enforce the same contract Discord -really does -- an interaction is answered once, and a deferred one must -answer through `followup`. +really does -- an interaction is answered once, a deferred one must answer +through `followup`, and a followup is refused outright until the +interaction has been acknowledged, which is what makes the `defer` calls +in this cog impossible to delete unnoticed. + +*Buttons*, unlike commands, are not invoked directly: `_press` goes +through `discord.ui.View._scheduled_task`, the coroutine the gateway +actually schedules, because that is where `interaction_check` runs. A test +calling `Button.callback` itself never passes through the confirm view's +author check at all, so its allow branch would be untested and could be +broken -- locking every administrator out of their own Confirm -- with +this file still green. The writes run against a real ephemeral PostgreSQL through the `clean_database` fixture, because everything worth pinning about a @@ -39,12 +49,16 @@ from sturnus.infrastructure.db.repositories import JobRepository, SessionRepository from sturnus.infrastructure.discord.permissions import _has_admin_access from sturnus.infrastructure.discord.queue_cog import ( + DISCORD_MESSAGE_LIMIT, NO_SUCH_SESSION, + JobLine, QueueCog, RequeueConfirmView, SessionSummary, _apply_requeue, render_requeue_confirmation, + render_requeue_refusal, + render_session, ) T0 = datetime(2026, 8, 19, 20, 0, 0, tzinfo=UTC) @@ -106,13 +120,30 @@ async def send_message( class _Followup: - def __init__(self) -> None: + """Discord's followup webhook, which only exists once the interaction is acknowledged. + + The acknowledgement check is the contract, not decoration: a followup + sent for an interaction that was never deferred and never answered is + a 404 (`Unknown Webhook`) from Discord, so the administrator sees "The + application did not respond" and then nothing -- on a command that may + already have written. Without this assertion the `await + interaction.response.defer(...)` in `RequeueConfirmView.confirm` can + be deleted with the whole suite still green, which is exactly what a + fake claiming to enforce Discord's contract must not allow. + """ + + def __init__(self, response: _Response) -> None: + self._response = response self.messages: list[tuple[str, bool]] = [] self.views: list[discord.ui.View | None] = [] async def send( self, content: str, ephemeral: bool = False, view: discord.ui.View | None = None ) -> None: + assert self._response.deferred or self._response.messages, ( + "followup on an interaction Discord has not been told about yet; " + "defer or send_message first" + ) self.messages.append((content, ephemeral)) self.views.append(view) @@ -129,8 +160,13 @@ def __init__(self, guild_id: int | None = GUILD, user_id: int = ADMIN) -> None: self.guild_id = guild_id self.user = _User(user_id) self.response = _Response() - self.followup = _Followup() + self.followup = _Followup(self.response) self.message = _Message() + #: `discord.ui.View._scheduled_task` hands this to + #: `Item._refresh_state` before any check runs, so `_press` cannot + #: dispatch the way Discord does without it. Empty is faithful: + #: a button's component data carries nothing the callback reads. + self.data: dict[str, Any] = {} async def original_response(self) -> _Message: return self.message @@ -173,14 +209,51 @@ async def _invoke(cog: QueueCog, command: str, interaction: _Interaction, *args: async def _press(view: discord.ui.View, label: str, interaction: _Interaction) -> None: - """Presses one of the view's buttons the way Discord dispatches it.""" + """Presses one of the view's buttons through Discord's own dispatch path. + + Not `item.callback(...)`: discord.py never dispatches a component that + way. `View._scheduled_task` runs + `await item._run_checks(interaction) and await + self.interaction_check(interaction)` first and returns *without* + calling the callback if either is falsy -- so calling the callback + directly walks straight past `RequeueConfirmView.interaction_check`. + The deny direction can still be asserted by calling that method + itself, but the allow direction then never runs anywhere in this file: + `interaction_check` could be made to return `False` unconditionally, + locking every administrator out of their own Confirm button, and every + test here would keep passing. Going through `_scheduled_task` is what + makes the gate load-bearing in this suite. + + `_scheduled_task` also funnels any exception into `View.on_error`, + whose default logs it and returns. That would swallow every assertion + the fakes above make -- `_Followup.send`'s acknowledgement check most + of all -- and turn a broken button into a passing test, so the handler + is replaced for the duration of the press by one that re-raises. + """ for item in view.children: if isinstance(item, discord.ui.Button) and item.label == label: - await item.callback(_as_interaction(interaction)) + await _dispatch(view, item, interaction) return raise AssertionError(f"no button labelled {label!r} in {view.children}") +async def _dispatch( + view: discord.ui.View, item: discord.ui.Item[Any], interaction: _Interaction +) -> None: + """Runs `View._scheduled_task`, surfacing what `on_error` would hide.""" + failures: list[BaseException] = [] + + async def _capture( + _interaction: discord.Interaction, error: Exception, _item: discord.ui.Item[Any] + ) -> None: + failures.append(error) + + view.on_error = _capture # type: ignore[method-assign] + await view._scheduled_task(item, _as_interaction(interaction)) + if failures: + raise failures[0] + + # --------------------------------------------------------------------------- # Database fixtures -- copied from `tests/infrastructure/test_queue.py`. # --------------------------------------------------------------------------- @@ -422,6 +495,10 @@ async def test_session_reports_the_transcript_length_and_never_its_text( assert "anna" in interaction.reply assert "19 characters" in interaction.reply assert "Copyright WDR" not in interaction.reply, "the transcript text must never be echoed" + # Ephemerality is a contract this cog states in its own docstring: the + # reply names who was recorded and how much they said, which is not a + # fact for the channel to see. + assert interaction.ephemeral is True async def test_session_reports_whether_the_audio_still_exists( @@ -436,6 +513,30 @@ async def test_session_reports_whether_the_audio_still_exists( assert "audio: erased" in interaction.reply +async def test_the_session_reply_an_admin_receives_is_short_enough_to_send( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The bound has to hold for the values that really come out of the database. + + `transcription_job.error` is whatever `str(exc)` produced when the job + last failed, stored unbounded and never truncated on the way in. The + reply carrying it goes out through `followup.send` after a + `thinking=True` defer, and a body over Discord's limit is an + `HTTPException` there -- so an unbounded error would turn the one + command that can explain a broken queue into no answer whatsoever. + """ + session_id = await seed( + factory, [Speaker(ANNA, error="x" * 5_000), Speaker(BEN), Speaker(CLARA)] + ) + interaction = _Interaction() + + await _invoke(cog(factory), "session", interaction, session_id) + + assert len(interaction.reply) <= DISCORD_MESSAGE_LIMIT + reply = interaction.reply + assert "anna" in reply and "ben" in reply and "clara" in reply + + async def test_session_from_another_guild_reads_as_not_existing( factory: async_sessionmaker[AsyncSession], ) -> None: @@ -501,6 +602,72 @@ async def test_requeue_refuses_a_session_with_a_running_job( assert jobs[ANNA].status == "done", "nothing may change on a refusal" +async def test_requeue_refuses_a_session_that_is_still_open( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The window `JobQueue.complete`'s Defect 5 guard exists to close. + + `RecordingService.close` uploads and enqueues one speaker at a time, + each `enqueue` committing on its own, and only calls `close_session` + after the last upload -- so for a long multi-speaker session there is + a wide window in which the session is still `open` while an early + speaker's job is already `done` and later speakers have not been + enqueued at all. A re-queue writes `status="closed"` unconditionally, + which is precisely the state Defect 5's guard refuses to accept from + anyone else: the next `complete()` would see no outstanding jobs and a + `closed` session, call it the session's last job, and build the + document out of one speaker. An impatient administrator re-queueing + "the session that just ended" lands in exactly this window. + """ + session_id = await seed( + factory, + [Speaker(ANNA)], + session_status="open", + document_url=None, + announced_at=None, + ) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert "Refused" in interaction.reply + assert "still open" in interaction.reply, "the reply has to say why, not just refuse" + assert interaction.view is None, "a refusal must not offer a Confirm button" + assert (await read_jobs(factory, session_id))[ANNA].status == "done" + assert (await read_session(factory, session_id)).status == "open", "nothing may change" + + +async def test_requeue_refuses_a_closed_session_that_has_no_document_yet( + factory: async_sessionmaker[AsyncSession], +) -> None: + """`retry_pending_documents` still owns a `closed` session; racing it strands. + + That sweep documents any closed session whose jobs are all terminal, + and it can be between its `closed_undocumented_sessions` read and its + `mark_documented` write right now. A re-queue landing in between hands + it a session whose transcripts have just been cleared: it publishes + that empty document and flips the session to `documented`, after which + `complete`'s last-job rule (`status == "closed"`) never fires again + and the sweep never looks at it again either. The redo would finish + into a database nobody ever reads a document out of. + """ + session_id = await seed( + factory, + [Speaker(ANNA)], + session_status="closed", + document_url=None, + announced_at=None, + ) + interaction = _Interaction() + + await _invoke(cog(factory), "requeue", interaction, session_id) + + assert "Refused" in interaction.reply + assert "no document yet" in interaction.reply + assert interaction.view is None + assert (await read_jobs(factory, session_id))[ANNA].transcript == "old hallucinated text" + + async def test_requeue_refuses_when_every_speakers_audio_is_erased( factory: async_sessionmaker[AsyncSession], ) -> None: @@ -602,6 +769,37 @@ async def test_only_the_invoker_may_press_confirm( assert allowed is False assert "Only" in intruder.reply + assert intruder.ephemeral is True + + +async def test_a_press_by_someone_else_never_reaches_the_write( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The gate has to hold on the path Discord really dispatches through. + + The test above calls `interaction_check` itself, which proves only + what that method returns; this presses the button the way the gateway + does (`View._scheduled_task` -- see `_press`), and then asserts on the + database rather than on the reply, because "somebody else's press + changed nothing" is the property the check exists for. Between the two + of them both directions of the gate are load-bearing: a check that + always allowed would fail here, and a check that always denied would + fail every `test_confirming_*` above. + """ + session_id = await seed(factory, [Speaker(ANNA)]) + interaction = _Interaction() + await _invoke(cog(factory), "requeue", interaction, session_id) + view = interaction.view + assert isinstance(view, RequeueConfirmView) + + intruder = _Interaction(user_id=ADMIN + 1) + await _press(view, "Confirm", intruder) + + assert "Only" in intruder.reply + jobs = await read_jobs(factory, session_id) + assert jobs[ANNA].status == "done", "a press that failed the check must not write" + assert jobs[ANNA].transcript == "old hallucinated text" + assert (await read_session(factory, session_id)).status == "documented" async def test_a_timeout_disables_the_buttons( @@ -635,6 +833,7 @@ async def test_cancelling_changes_nothing( await _press(view, "Cancel", pressed) assert "Cancelled" in pressed.reply + assert pressed.ephemeral is True, "the reply names the session an admin was looking at" jobs = await read_jobs(factory, session_id) assert jobs[ANNA].status == "done" assert jobs[ANNA].transcript == "old hallucinated text" @@ -797,6 +996,7 @@ async def test_confirming_reports_how_many_of_how_many_speakers_were_requeued( await _press(view, "Confirm", pressed) assert "2 of 3" in pressed.reply + assert pressed.ephemeral is True, "the result names every speaker whose audio was erased" async def test_confirming_touches_no_other_session( @@ -907,6 +1107,30 @@ async def test_the_write_refuses_another_guilds_session_on_its_own( assert (await read_jobs(factory, session_id))[ANNA].status == "done" +async def test_the_write_refuses_a_session_that_is_not_documented_on_its_own( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The status check has to be inside the lock, not only in front of the prompt. + + `/queue requeue` refuses an `open` or `closed` session before a + Confirm button ever exists, so no test that goes through the cog can + reach this one -- dropping it from the write would look safe. It is + not: the prompt holds nothing but ids for as long as an administrator + reads it, and in that time the session can move (a second + administrator's re-queue of the same session leaves it `closed`; a + session that has just closed can be documented). The write must decide + on the row it locked, not on the row the prompt was rendered from. + """ + session_id = await seed(factory, [Speaker(ANNA)], session_status="open") + + view = await _apply_requeue(factory, GUILD, session_id) + + assert view is not None + assert view.is_refused is True + assert (await read_jobs(factory, session_id))[ANNA].status == "done" + assert (await read_session(factory, session_id)).status == "open" + + async def test_the_write_waits_for_a_worker_holding_the_sessions_jobs( factory: async_sessionmaker[AsyncSession], ) -> None: @@ -980,6 +1204,96 @@ def _summary(**overrides: Any) -> SessionSummary: return SessionSummary(**{**defaults, **overrides}) +def _job_line(user_id: int, error: str | None = None) -> JobLine: + return JobLine( + discord_user_id=user_id, + status="dead", + attempts=3, + audio_present=True, + error=error, + transcript_length=None, + ) + + +def test_a_session_readout_of_many_speakers_stays_inside_discords_limit() -> None: + """The reply that has to survive is the one from a session full of failures. + + `followup.send` raises `HTTPException` on a body over + `DISCORD_MESSAGE_LIMIT`, and by then the interaction has been deferred + with `thinking=True`, so the administrator gets no answer at all -- + the diagnostic command going silent exactly when the queue is in the + state it exists to diagnose. Every speaker line is worth ~85 + characters, so a 40-person voice channel passes the limit on speaker + count alone, with no long error involved. + """ + jobs = [ + _job_line(user_id, error="boto3 timed out talking to the object store") + for user_id in range(40) + ] + + text = render_session(_summary(), jobs, {}) + + assert len(text) <= DISCORD_MESSAGE_LIMIT + assert "**Session 4**" in text, "the header is what identifies the session being read" + assert "user 0" in text, "as many speakers as fit, from the first" + assert "more speakers not shown" in text, "silently dropping speakers would be worse" + + +def test_one_enormous_error_cannot_push_the_other_speakers_out_of_the_reply() -> None: + """`job.error` is `str(exc)` -- arbitrary text of arbitrary length. + + A single unbounded exception string (a boto3 error carrying a whole + request context, say) would otherwise consume the entire budget and + leave the readout to be truncated after one speaker, which is the + least useful place to cut a list of speakers. Bounding each error + first keeps the shape of the answer -- one line per speaker -- intact, + and the length of a transcript, which is the reason this command + exists, is on those lines rather than in the error text. + """ + jobs = [_job_line(ANNA), _job_line(BEN, error="x" * 5_000), _job_line(CLARA)] + + text = render_session(_summary(), jobs, NAMES) + + assert len(text) <= DISCORD_MESSAGE_LIMIT + assert "anna" in text and "ben" in text and "clara" in text + assert "more speakers not shown" not in text, "three speakers must all fit" + + +def test_an_error_full_of_newlines_still_renders_as_one_line_per_speaker() -> None: + """A multi-line exception string would otherwise break the list apart. + + The readout is one line per speaker and is read as such; an error + carrying newlines (a wrapped traceback, an XML error body) would turn + one speaker into a dozen lines that look like speakers of their own. + """ + jobs = [_job_line(ANNA, error="failed:\n line two\n line three"), _job_line(BEN)] + + text = render_session(_summary(), jobs, NAMES) + + assert text.count("\n- ") == 2, "exactly one bullet per speaker" + assert "failed: line two line three" in text + + +def test_a_refusal_naming_a_whole_channel_of_erased_speakers_is_still_sendable() -> None: + """The same limit applies to every reply, not only to `/queue session`. + + A refusal lists names, and a Discord display name is up to 32 + characters: a large enough session pushes even this text past the + limit, and a refusal that cannot be sent reads to the administrator + exactly like a command that did nothing. + """ + erased = tuple(range(80)) + names = {user_id: f"a-rather-long-display-name-{user_id}" for user_id in erased} + + text = render_requeue_refusal( + _summary(), + RequeuePlan(resettable_job_ids=(), erased_user_ids=erased, active_user_ids=()), + names, + ) + + assert len(text) <= DISCORD_MESSAGE_LIMIT + + def test_a_confirmation_for_a_session_that_was_never_documented_says_so() -> None: """There is no first document to orphan, so promising one would be wrong.""" text = render_requeue_confirmation( diff --git a/tests/infrastructure/test_repositories.py b/tests/infrastructure/test_repositories.py index adabc7b..1431de4 100644 --- a/tests/infrastructure/test_repositories.py +++ b/tests/infrastructure/test_repositories.py @@ -1,6 +1,7 @@ from datetime import UTC, datetime, timedelta import pytest +from sqlalchemy import update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sturnus.application.assembly import serialize_transcript @@ -494,6 +495,57 @@ async def test_mark_announced_stamps_the_session( assert candidates[0]["announced_at"] == T0 + timedelta(minutes=5) +async def test_mark_announced_does_not_stamp_a_session_that_was_requeued_meanwhile( + factory: async_sessionmaker[AsyncSession], +) -> None: + """The stamp has to land on the session the post was actually about. + + `announce_ready_sessions` selects a `documented`, unannounced session, + awaits `announcer.post` -- a Discord HTTP call that takes seconds + under rate limiting -- and only then calls `mark_announced`. A + `/queue requeue` can land inside that window: it puts the session back + to `closed` and clears `announced_at` precisely so the redo's fresh + link gets posted. An unconditional stamp arriving afterwards would put + a timestamp back on a session that has *not* been announced since, and + `sessions_to_announce` would then never select it again: the corrected + transcript would be documented and silently never posted, which is the + exact failure clearing the column exists to prevent. + + A duplicate post of the superseded link is the accepted cost here, and + the one `announce_ready_sessions` already documents itself as erring + towards -- losing an announcement is the worse half of that trade. + """ + repo = SessionRepository(factory) + session_id = await repo.open_session(GUILD, CHANNEL, "meeting-raum", T0) + await repo.close_session(session_id, T0 + timedelta(hours=1), "empty") + await _mark_documented(factory, session_id) + # The re-queue, exactly as `queue_cog._apply_requeue` writes it, while + # the sweep is somewhere inside `announcer.post`. + await _requeue(factory, session_id) + + await repo.mark_announced(session_id, T0 + timedelta(minutes=5)) + + async with factory() as session: + row = await session.get(Session, session_id) + assert row is not None + assert row.announced_at is None, "the late stamp belongs to a run that is superseded" + # And the consequence that matters: once the redo is documented, the + # session is a candidate again and the new link does get posted. + await _mark_documented(factory, session_id) + assert [c["id"] for c in await repo.candidates_for_announcement()] == [session_id] + + +async def _requeue(factory: async_sessionmaker[AsyncSession], session_id: int) -> None: + """The session-row half of a `/queue requeue`, without the cog.""" + async with factory() as session: + await session.execute( + update(Session) + .where(Session.id == session_id) + .values(status="closed", announced_at=None) + ) + await session.commit() + + async def test_closed_undocumented_sessions_finds_a_session_whose_jobs_are_all_terminal( factory: async_sessionmaker[AsyncSession], ) -> None: