feat(discord): add /queue and let an admin re-run a finished session

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 <session_id>`** — 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 <session_id>`** — 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
+  <session_id>` 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 <id>` 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 <session_id>` 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 <session_id>` 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 <session_id>`. 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

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),
+    )