diff --git a/docs/operations.md b/docs/operations.md index fdd5e3c..96f577d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -496,6 +496,71 @@ 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. 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 + 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. + + **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. + **A speaker's audio arrives with no level.** During a meeting the bot may post, into the recording channel and naming the person: @@ -547,8 +612,10 @@ consented — a peak is a number about loudness, not about content. 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'; @@ -576,6 +643,77 @@ 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 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 +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. + +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/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/db/repositories.py b/src/sturnus/infrastructure/db/repositories.py index 7e69b87..f67fa0e 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: @@ -331,11 +334,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/client.py b/src/sturnus/infrastructure/discord/client.py index 43ece80..318ad3f 100644 --- a/src/sturnus/infrastructure/discord/client.py +++ b/src/sturnus/infrastructure/discord/client.py @@ -66,6 +66,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 @@ -228,6 +229,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..292a248 --- /dev/null +++ b/src/sturnus/infrastructure/discord/queue_cog.py @@ -0,0 +1,1115 @@ +"""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 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 +`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.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 +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 + +#: 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 +#: 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] + + @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: + """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 view is not + `is_refused`. + + 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 + 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.is_refused: + 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. + # + # 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` + # 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 _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" + + +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 _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: + 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'})", + f"Document: {summary.document_url or '*(none)*'}", + f"Announced: {_stamp(summary.announced_at)}", + ] + if not jobs: + 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: + """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 _capped("\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 _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( + 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. + + 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"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"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"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.is_refused: + 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.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. + 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..4fc33cc --- /dev/null +++ b/tests/infrastructure/discord/test_queue_cog.py @@ -0,0 +1,1306 @@ +"""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, 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 +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 ( + 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) +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: + """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) + + +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.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 + + @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 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 _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`. +# --------------------------------------------------------------------------- + + +@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" + # 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( + 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_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: + """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_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: + """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 + 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( + 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 + 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" + 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 + assert pressed.ephemeral is True, "the result names every speaker whose audio was erased" + + +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_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: + """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 _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( + _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 diff --git a/tests/infrastructure/test_repositories.py b/tests/infrastructure/test_repositories.py index 439b681..7533586 100644 --- a/tests/infrastructure/test_repositories.py +++ b/tests/infrastructure/test_repositories.py @@ -1,7 +1,7 @@ from datetime import UTC, datetime, timedelta import pytest -from sqlalchemy import select +from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sturnus.application.assembly import serialize_transcript @@ -569,6 +569,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: