Recognise the words in episodes no publisher transcript can describe

whisper.cpp, installed on tungsten and shelled out to per episode. The other
304 of 342 mirrored episodes -- the ones whose publisher's transcript was
declined because this copy carries inserted advertising the master does not
describe, which is both why it is unusable and why having words would be worth
most.

**Measured, and an order of magnitude cheaper than I told Peter.** base.en on
six threads does 10.3x realtime on this box: a 63-minute Darknet Diaries episode
took 6m11s at 793MB peak. The earlier "roughly realtime, days of backfill" was
wrong. The real backlog is 298 episodes, 266 hours of audio, about 26 hours of
CPU -- three days at 45 minutes per two-hour cycle, which is the default budget.

Bounded per cycle rather than exhaustive, and only on the timer: a wakeup is the
phone pushing a subscription list and wanting that show polled now, and
following it with three quarters of an hour of recognition would make a new
subscription feel broken. Newest first, because an episode about to be played is
worth having words for before one from the back of a catalogue.

**The first real run found two bugs the tests could not have.** The working
directory was never created, so ffmpeg could not write and every episode failed
-- and each failure was recorded as "do not try this one again". A misconfigured
deploy would have marked the whole backlog attempted in seconds and nothing
would ever have been transcribed, which is precisely the shape of the midroll
scan marking episodes checked forever after an OutOfMemoryError nobody saw.
transcribe.Breaker now abandons a pass after three consecutive failures and
gives those attempts back, since a run of failures is the engine rather than the
episodes -- while one bad file among good ones is still marked, because the same
bytes will fail the same way next time.

NeedsRecognising is deliberately not the negation of TranscriptExamined. A
declined transcript means examined, no words, and exactly the episode
recognition exists for.

**First validation against ground truth the detector did not produce.** Up First
"Democrats Lead In NPR Poll": the feed declares 788s, the file served 890s, so
102s of advertising was inserted -- and reading the recognised words found 100s
of it, in two pods, every line unmistakable ad copy. The sibling episode found
21s of 131s: a whole pod at 54-130s was missed because its copy carries only one
kind of marker. Reported rather than tuned around, on two episodes.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B

1 files changed, 151 insertions, 4 deletions

diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go
index 495ed03..1b746f1 100644
--- a/server/cmd/listend/main.go
+++ b/server/cmd/listend/main.go
@@ -23,6 +23,7 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"sort"
 	"sync"
 	"syscall"
 	"time"
@@ -35,6 +36,7 @@ import (
 	"github.com/terst/listen/server/internal/signature"
 	"github.com/terst/listen/server/internal/skips"
 	"github.com/terst/listen/server/internal/store"
+	"github.com/terst/listen/server/internal/transcribe"
 	"github.com/terst/listen/server/internal/transcript"
 )
 
@@ -46,6 +48,12 @@ func main() {
 		pollEvery = flag.Duration("poll", 2*time.Hour, "how often to refresh every feed")
 		tokenFile = flag.String("token-file", "", "file holding the API token (defaults to <data>/token)")
 		once      = flag.Bool("once", false, "run one poll cycle and exit, for checking a deployment")
+
+		whisperBin       = flag.String("whisper", "", "whisper-cli binary; empty disables recognition")
+		whisperModel     = flag.String("whisper-model", "", "ggml model file for recognition")
+		whisperThreads   = flag.Int("whisper-threads", 6, "cores one recognition run may use")
+		transcribeBudget = flag.Duration("transcribe-budget", 45*time.Minute,
+			"how long each cycle may spend recognising episodes")
 	)
 	flag.Parse()
 
@@ -83,8 +91,17 @@ func main() {
 		decoder:       decoder,
 		transcripts:   transcripts,
 		harvester:     transcript.Harvester{Store: transcripts},
-		requests:      make(chan request, 64),
-		wakeups:       make(chan struct{}, 1),
+		recogniser: transcribe.Whisper{
+			Binary:  *whisperBin,
+			Model:   *whisperModel,
+			Threads: *whisperThreads,
+			// Beside the audio rather than in /tmp: a decoded hour is 115MB of PCM, and the
+			// data directory is the volume with room for it.
+			WorkDir: filepath.Join(*dataDir, "work"),
+		},
+		transcribeBudget: *transcribeBudget,
+		requests:         make(chan request, 64),
+		wakeups:          make(chan struct{}, 1),
 	}
 
 	server := api.New(state, worker, api.Options{
@@ -113,6 +130,9 @@ func main() {
 
 	if *once {
 		worker.pollAll(ctx)
+		// Recognition too, so a deployment check exercises the expensive path rather than
+		// discovering it is misconfigured hours later in a log nobody is reading.
+		worker.transcribeSome(ctx)
 		if err := state.Save(statePath); err != nil {
 			log.Fatalf("save: %v", err)
 		}
@@ -167,8 +187,14 @@ type worker struct {
 	// they describe the copy that was mirrored. See transcript.Fits for when they do not.
 	transcripts transcript.Store
 	harvester   transcript.Harvester
-	requests    chan request
-	wakeups     chan struct{}
+	// recogniser turns an episode's audio into its words where no usable published transcript
+	// exists. Zero when recognition is not configured, which is a working state.
+	recogniser transcribe.Whisper
+	// transcribeBudget is how long one cycle may spend recognising. Bounded so the backlog is
+	// worked through steadily rather than monopolising the machine -- see transcribeSome.
+	transcribeBudget time.Duration
+	requests         chan request
+	wakeups          chan struct{}
 }
 
 // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll.
@@ -192,6 +218,7 @@ func (w *worker) Request(feedURL, guid, enclosureURL, title string) {
 func (w *worker) run(ctx context.Context, every time.Duration) {
 	w.pollAll(ctx)
 	w.save()
+	w.transcribeSome(ctx)
 
 	ticker := time.NewTicker(every)
 	defer ticker.Stop()
@@ -202,6 +229,10 @@ func (w *worker) run(ctx context.Context, every time.Duration) {
 		case <-ticker.C:
 			w.pollAll(ctx)
 			w.save()
+			// Only on the timer. A wakeup is the phone pushing a subscription list and
+			// wanting that show polled now; following it with three quarters of an hour of
+			// recognition would make a new subscription feel broken.
+			w.transcribeSome(ctx)
 		case <-w.wakeups:
 			w.pollAll(ctx)
 			w.save()
@@ -387,6 +418,122 @@ func (w *worker) harvestTranscripts(ctx context.Context, show *store.Show, candi
 	}
 }
 
+// transcribeSome recognises the words in episodes that have none, for as long as the budget
+// allows.
+//
+// Bounded rather than exhaustive, and run after polling rather than inside it. Recognition is
+// the most expensive thing this server does -- measured at about 11x realtime with base.en on
+// six threads, so an hour-long episode costs about five minutes -- and the backlog at the time
+// this was built was 304 episodes. Spending a fixed slice of each cycle works through that in a
+// few days without ever making a poll wait hours or taking the machine over.
+//
+// Newest first. An episode Peter is likely to play soon is worth having words for before one
+// from the back of the catalogue.
+func (w *worker) transcribeSome(ctx context.Context) {
+	if !w.recogniser.Available() || w.transcribeBudget <= 0 {
+		return
+	}
+	deadline := time.Now().Add(w.transcribeBudget)
+	done, failed := 0, 0
+	// Attempts made during the current run of failures, so they can be given back if the run
+	// turns out to be the engine rather than the episodes. See transcribe.Breaker.
+	var breaker transcribe.Breaker
+	var failing []*store.Episode
+
+	for _, e := range w.pendingRecognition() {
+		if ctx.Err() != nil || time.Now().After(deadline) {
+			break
+		}
+		started := time.Now()
+		result, err := w.recogniser.Transcribe(ctx, filepath.Join(w.mirror.AudioDir, e.Path))
+		if ctx.Err() != nil {
+			// Interrupted rather than attempted. Recording it would quietly skip whatever was
+			// in flight across a redeploy, and this runs for days at a time.
+			return
+		}
+		e.TranscribeAttemptedAtUnix = time.Now().Unix()
+		if err != nil {
+			log.Printf("transcribe %q: %v", e.Title, err)
+			failed++
+			failing = append(failing, e)
+			if breaker.Failure() {
+				giveBackAttempts(failing)
+				log.Printf("recognition abandoned after %d failures in a row -- "+
+					"giving those attempts back, since this is the engine and not the episodes",
+					breaker.Run())
+				w.save()
+				return
+			}
+			w.save()
+			continue
+		}
+
+		stored := transcript.Transcript{
+			EpisodeID:    e.ID,
+			Source:       transcript.SourceASR,
+			LanguageCode: result.LanguageCode,
+			Cues:         result.Cues,
+			Ads:          transcript.Spot(result.Cues),
+		}
+		if err := w.transcripts.Put(stored); err != nil {
+			log.Printf("transcribe %q: %v", e.Title, err)
+			failed++
+			failing = append(failing, e)
+			if breaker.Failure() {
+				giveBackAttempts(failing)
+				log.Printf("recognition abandoned after %d failures in a row -- "+
+					"giving those attempts back, since this is the engine and not the episodes",
+					breaker.Run())
+				w.save()
+				return
+			}
+			w.save()
+			continue
+		}
+		e.TranscriptSource = string(transcript.SourceASR)
+		e.TranscriptCues = len(stored.Cues)
+		e.TranscriptAtUnix = time.Now().Unix()
+		// The note said why the publisher's transcript could not be used. There are words now,
+		// so it is answered rather than still true.
+		e.TranscriptNote = ""
+		done++
+		breaker.Success()
+		failing = nil
+		log.Printf("transcribed %q: %d lines, %d ad candidate(s), took %s",
+			e.Title, len(stored.Cues), len(stored.Ads), time.Since(started).Round(time.Second))
+		w.save()
+	}
+
+	if done > 0 || failed > 0 {
+		log.Printf("recognition -- %d done, %d failed, %d still waiting",
+			done, failed, len(w.pendingRecognition()))
+	}
+}
+
+// giveBackAttempts unmarks episodes that failed for a reason that was nothing to do with them,
+// so they are tried again rather than hidden for ever.
+func giveBackAttempts(episodes []*store.Episode) {
+	for _, e := range episodes {
+		e.TranscribeAttemptedAtUnix = 0
+	}
+}
+
+// pendingRecognition is every mirrored episode still without words, newest first.
+func (w *worker) pendingRecognition() []*store.Episode {
+	var pending []*store.Episode
+	for _, show := range w.state.Shows() {
+		for _, e := range show.Episodes {
+			if e.NeedsRecognising() {
+				pending = append(pending, e)
+			}
+		}
+	}
+	sort.Slice(pending, func(i, j int) bool {
+		return pending[i].PublishedUnix > pending[j].PublishedUnix
+	})
+	return pending
+}
+
 // detect runs the intro/outro search over a show's mirrored episodes and records the result.
 //
 // Every mirrored episode at once, which is the whole reason this is a server: the phone could