Stop a second device from emptying the library, and recover what one did

I pointed the emulator at the live server to verify that published breaks
actually skip. It worked. It also pushed the emulator's own four-show
subscription list, which replaced the real one, and at 07:07:13 the server
reclaimed every show that device had never heard of: 26 shows and 290 mirrored
episodes, deleted in a second.

Three fixes, in the order they matter.

**The list can no longer be emptied by one push.** It is still replaced
wholesale -- the phone owns it, and reconciling two editable copies of something
changed this rarely costs more in sync bugs than it saves -- but a push that
keeps fewer than half of ten or more feeds is refused with a 409 and logged.
Losing most of a library in one push is a second device or a bug; it is never
somebody unsubscribing. Below ten feeds there is nothing to protect: a new
phone's first push legitimately replaces nothing with everything.

**Recognition is recovered rather than redone.** A transcript is keyed by the
episode id, which comes from the feed and the guid, so it outlives the record
and matches again when the episode is mirrored afresh. 332 transcript files
survived on disk. Reclaim restores them -- 18 immediately, the rest as
re-mirroring reaches them -- instead of spending twenty hours of CPU recognising
text the server is already holding. It refuses a transcript whose last cue runs
past the end of the audio, which is the same check made of a publisher's.

**The pipeline's Save was never wired**, which the security review caught before
it could panic in anger: the extraction moved the call and not the field. It is
now optional, because every test wants it nil, and assumed-non-nil is what
shipped a server that would have died on its first recognised episode.

Reclaim runs before each show is polled as well as at startup, or a re-mirrored
episode would wait for a restart to find the words already sitting next to it.

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

6 files changed, 315 insertions, 9 deletions

diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go
index e1329a9..2298db5 100644
--- a/server/cmd/listend/main.go
+++ b/server/cmd/listend/main.go
@@ -117,6 +117,10 @@ func main() {
 		requests: make(chan request, 64),
 		wakeups:  make(chan struct{}, 1),
 	}
+	// Wired after the worker exists, because it is the worker's own save. The pipeline runs
+	// for tens of minutes at a time and persists as each episode finishes, so a restart in
+	// the middle keeps the ones already done.
+	worker.words.Save = worker.save
 
 	server := api.New(state, worker, api.Options{
 		Token:         token,
@@ -143,6 +147,7 @@ func main() {
 	defer stop()
 
 	if *once {
+		worker.words.Reclaim()
 		worker.words.Rederive()
 		worker.pollAll(ctx)
 		// Recognition too, so a deployment check exercises the expensive path rather than
@@ -228,6 +233,7 @@ func (w *worker) Request(feedURL, guid, enclosureURL, title string) {
 }
 
 func (w *worker) run(ctx context.Context, every time.Duration) {
+	w.words.Reclaim()
 	w.words.Rederive()
 	w.save()
 	w.pollAll(ctx)
@@ -373,6 +379,10 @@ func (w *worker) pollOne(ctx context.Context, feedURL string) {
 	if fetched > 0 {
 		log.Printf("%s: mirrored %d new episode(s)", channel.Title, fetched)
 	}
+	// Before anything else looks at this show: an episode just re-mirrored may already have
+	// its words on disk from before its record was lost, and recognising it again would cost
+	// five minutes for text the server is holding.
+	w.words.Reclaim()
 	w.words.Harvest(ctx, show, kept)
 	w.words.Recurring(show)
 	w.detect(ctx, show, kept)
diff --git a/server/internal/api/api.go b/server/internal/api/api.go
index 4b7fb2d..6114c1b 100644
--- a/server/internal/api/api.go
+++ b/server/internal/api/api.go
@@ -3,6 +3,7 @@ package api
 
 import (
 	"encoding/json"
+	"log"
 	"net/http"
 	"net/url"
 	"os"
@@ -178,7 +179,15 @@ func (s *Server) subscriptions(w http.ResponseWriter, r *http.Request) {
 		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expected {\"feeds\":[...]}"})
 		return
 	}
-	removed := s.state.SetFeeds(body.Feeds)
+	removed, err := s.state.SetFeedsChecked(body.Feeds)
+	if err != nil {
+		// Refused rather than applied: losing most of a library in one push is another device
+		// or a bug. Said out loud, because a silent refusal looks exactly like a silent
+		// acceptance from the phone's end.
+		log.Printf("subscriptions: %v", err)
+		writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()})
+		return
+	}
 	// After the list, so a tier for a feed that was just removed is dropped with it.
 	s.state.SetTiers(body.Tiers)
 	if s.opts.OnSubscriptions != nil {
diff --git a/server/internal/store/store.go b/server/internal/store/store.go
index f69ff90..2e20f2e 100644
--- a/server/internal/store/store.go
+++ b/server/internal/store/store.go
@@ -360,6 +360,58 @@ func (s *State) Save(path string) error {
 	return nil
 }
 
+/*
+guardFloor is the library size below which a shrinking push is not questioned.
+
+A new phone's first push legitimately replaces nothing with everything, and a library of three
+can legitimately go to one. Ten is where "most of it vanished" stops being plausible as
+something somebody did on purpose.
+*/
+const guardFloor = 10
+
+/*
+SetFeedsChecked replaces the subscription list unless doing so would empty the library.
+
+This exists because it happened. A test device was pointed at the live server to check
+playback, its own short subscription list replaced the real one, and the server reclaimed every
+show that device had never heard of: 290 mirrored episodes and days of recognition, gone in one
+second.
+
+The list is still replaced wholesale -- the phone owns it, and reconciling two editable copies
+of something changed this rarely would cost more in sync bugs than it saves. What is refused is
+the magnitude. Losing most of a library in a single push is a second device or a bug; it is
+never somebody unsubscribing.
+
+Returns what was removed, or an error and nothing removed.
+*/
+func (s *State) SetFeedsChecked(feeds []string) (removed []string, err error) {
+	s.mu.RLock()
+	existing := len(s.feeds)
+	s.mu.RUnlock()
+
+	if existing >= guardFloor {
+		wanted := map[string]bool{}
+		for _, f := range feeds {
+			wanted[strings.TrimSpace(f)] = true
+		}
+		kept := 0
+		s.mu.RLock()
+		for _, f := range s.feeds {
+			if wanted[f] {
+				kept++
+			}
+		}
+		s.mu.RUnlock()
+		if kept*2 < existing {
+			return nil, fmt.Errorf(
+				"refusing a subscription push that keeps %d of %d feeds: losing most of a "+
+					"library in one push is another device or a bug, not an unsubscribe",
+				kept, existing)
+		}
+	}
+	return s.SetFeeds(feeds), nil
+}
+
 // SetFeeds replaces the subscription list and reports the feeds that are no longer wanted.
 //
 // Wholesale replacement because the phone owns this list; reconciling two editable copies of
diff --git a/server/internal/store/store_test.go b/server/internal/store/store_test.go
index f0c111b..a88fdd4 100644
--- a/server/internal/store/store_test.go
+++ b/server/internal/store/store_test.go
@@ -1,8 +1,10 @@
 package store
 
 import (
+	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/terst/listen/server/internal/feed"
@@ -640,3 +642,106 @@ func TestABreakBothDetectorsFoundSaysSo(t *testing.T) {
 		t.Errorf("origin = %q, want both named", got[0].Origin)
 	}
 }
+
+/*
+A subscription push that would empty the library is refused.
+
+This happened. A test device was pointed at the live server to check playback, its own short
+subscription list replaced the real one, and the server reclaimed every show the test device
+had never heard of -- 290 mirrored episodes and days of recognition, deleted in one second at
+07:07:13 on 2026-10-31.
+
+The list is still wholesale-replaced, because the phone owns it and reconciling two editable
+copies would cost more in sync bugs than it saves. What is refused is the magnitude: losing
+most of a library in one push is a second device or a bug, never somebody unsubscribing.
+
+Below [guardFloor] shows there is nothing to protect -- a new phone's first push legitimately
+replaces nothing with everything, and a small library can legitimately halve.
+*/
+
+func TestAPushThatWouldEmptyTheLibraryIsRefused(t *testing.T) {
+	s := New()
+	many := make([]string, 0, 20)
+	for i := 0; i < 20; i++ {
+		many = append(many, fmt.Sprintf("https://feed%d", i))
+	}
+	s.SetFeeds(many)
+
+	removed, err := s.SetFeedsChecked([]string{"https://feed0", "https://feed1"})
+
+	if err == nil {
+		t.Fatal("accepted a push that drops 18 of 20 shows")
+	}
+	if len(removed) != 0 {
+		t.Errorf("removed %d shows anyway", len(removed))
+	}
+	if len(s.Feeds()) != 20 {
+		t.Errorf("the list is %d long, want it untouched", len(s.Feeds()))
+	}
+}
+
+// Unsubscribing from a few shows is ordinary and must keep working.
+func TestAnOrdinaryUnsubscribeIsAccepted(t *testing.T) {
+	s := New()
+	many := make([]string, 0, 20)
+	for i := 0; i < 20; i++ {
+		many = append(many, fmt.Sprintf("https://feed%d", i))
+	}
+	s.SetFeeds(many)
+
+	// Merged so there are shows to reclaim; removed names shows with audio, not feed URLs.
+	for _, url := range many {
+		s.Merge(url, "A Show", nil)
+	}
+
+	removed, err := s.SetFeedsChecked(many[:17])
+
+	if err != nil {
+		t.Fatalf("refused an ordinary unsubscribe: %v", err)
+	}
+	if len(removed) != 3 {
+		t.Errorf("removed %d, want 3", len(removed))
+	}
+	if len(s.Feeds()) != 17 {
+		t.Errorf("the list is %d long, want 17", len(s.Feeds()))
+	}
+}
+
+// A first push replaces nothing with everything, which must not look like a catastrophe.
+func TestAFirstPushIsAccepted(t *testing.T) {
+	s := New()
+
+	if _, err := s.SetFeedsChecked([]string{"https://a", "https://b"}); err != nil {
+		t.Fatalf("refused a first push: %v", err)
+	}
+}
+
+// A small library can legitimately halve; there is nothing worth protecting yet.
+func TestASmallLibraryIsNotGuarded(t *testing.T) {
+	s := New()
+	s.SetFeeds([]string{"https://a", "https://b", "https://c"})
+
+	if _, err := s.SetFeedsChecked([]string{"https://a"}); err != nil {
+		t.Fatalf("refused a small library shrinking: %v", err)
+	}
+}
+
+// Clearing the list deliberately is still possible, because an empty push is unambiguous in a
+// way a short one is not -- but only from empty, so it cannot happen by accident here.
+func TestTheGuardNamesWhatItRefused(t *testing.T) {
+	s := New()
+	many := make([]string, 0, 12)
+	for i := 0; i < 12; i++ {
+		many = append(many, fmt.Sprintf("https://feed%d", i))
+	}
+	s.SetFeeds(many)
+
+	_, err := s.SetFeedsChecked(nil)
+
+	if err == nil {
+		t.Fatal("accepted an empty push over a full library")
+	}
+	if !strings.Contains(err.Error(), "12") {
+		t.Errorf("error %q does not say how much it would have lost", err)
+	}
+}
diff --git a/server/internal/words/pipeline.go b/server/internal/words/pipeline.go
index b37efa3..2ae2c4e 100644
--- a/server/internal/words/pipeline.go
+++ b/server/internal/words/pipeline.go
@@ -68,9 +68,20 @@ type Pipeline struct {
 	// Save persists the state. Called as each episode finishes rather than at the end,
 	// because these passes run for tens of minutes and a restart in the middle should not
 	// lose the ones already done.
+	//
+	// Optional: a nil Save means the caller is not persisting, which is what every test wants.
+	// It must be optional rather than assumed, because assuming it shipped a server that
+	// panicked on its first recognised episode.
 	Save func()
 }
 
+// save persists the state if the caller asked for persistence.
+func (p *Pipeline) save() {
+	if p.Save != nil {
+		p.Save()
+	}
+}
+
 /*
 Forget drops everything known about an episode's words.
 
@@ -221,10 +232,10 @@ func (p *Pipeline) Recognise(ctx context.Context) {
 				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())
-				p.Save()
+				p.save()
 				return
 			}
-			p.Save()
+			p.save()
 			continue
 		}
 
@@ -244,10 +255,10 @@ func (p *Pipeline) Recognise(ctx context.Context) {
 				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())
-				p.Save()
+				p.save()
 				return
 			}
-			p.Save()
+			p.save()
 			continue
 		}
 		e.TranscriptSource = string(transcript.SourceASR)
@@ -262,7 +273,7 @@ func (p *Pipeline) Recognise(ctx context.Context) {
 		log.Printf("transcribed %q: %d lines, %s, took %s",
 			e.Title, len(stored.Cues), adAccounting(e, stored.Ads),
 			time.Since(started).Round(time.Second))
-		p.Save()
+		p.save()
 	}
 
 	if done > 0 || failed > 0 {
@@ -569,10 +580,10 @@ func (p *Pipeline) SecondCopy(ctx context.Context) {
 				}
 				log.Printf("refetching abandoned after %d failures in a row -- giving those "+
 					"attempts back, since this is the machine and not the episodes", breaker.Run())
-				p.Save()
+				p.save()
 				return
 			}
-			p.Save()
+			p.save()
 			continue
 		}
 		breaker.Success()
@@ -583,7 +594,7 @@ func (p *Pipeline) SecondCopy(ctx context.Context) {
 		// then nothing for subtraction to see.
 		log.Printf("second copy of %q: %s against the mirrored %s, %d lines, %s",
 			e.Title, clock(secondMs), clock(e.DurationMs), e.MasterCues, adAccounting(e, found))
-		p.Save()
+		p.save()
 	}
 	if done > 0 || failed > 0 {
 		log.Printf("second copies -- %d done, %d failed, %d still waiting",
@@ -704,3 +715,53 @@ func (p *Pipeline) pendingRecognition() []*store.Episode {
 	})
 	return pending
 }
+
+/*
+Reclaim restores what the transcript files on disk say about episodes the state has forgotten.
+
+Recognition is the expensive thing this server does -- about five minutes of CPU an episode --
+and its output is a file keyed by the episode's id, which is derived from the feed and the
+guid. So a transcript survives an episode record being lost and matches again the moment that
+episode is mirrored afresh.
+
+That is not hypothetical. A test device pointed at the live server replaced the subscription
+list with its own, the server reclaimed every show the device had never heard of, and 290
+mirrored episodes went with it -- while their transcripts stayed on disk. Without this, every
+one of them would be recognised again from scratch.
+
+Only where the file genuinely matches the episode. A transcript whose last cue runs past the
+end of the audio describes a different recording, which is the same check transcript.Fits
+makes of a publisher's.
+*/
+func (p *Pipeline) Reclaim() {
+	restored := 0
+	for _, show := range p.State.Shows() {
+		for _, e := range show.Episodes {
+			if !e.Mirrored() || e.HasTranscript() || !p.Transcripts.Has(e.ID) {
+				continue
+			}
+			held, err := p.Transcripts.Get(e.ID)
+			if err != nil || len(held.Cues) == 0 {
+				continue
+			}
+			if e.DurationMs > 0 && held.LastEndMs() > e.DurationMs {
+				// Describes a longer recording than the one on disk, so it is not this copy.
+				continue
+			}
+			e.TranscriptSource = string(held.Source)
+			e.TranscriptCues = len(held.Cues)
+			e.TranscriptNote = ""
+			if p.Transcripts.HasMaster(e.ID) {
+				if master, err := p.Transcripts.Master(e.ID); err == nil && len(master.Cues) > 0 {
+					e.MasterCues = len(master.Cues)
+					e.MasterSource = string(master.Source)
+				}
+			}
+			p.publishWordBreaks(e, held.Ads)
+			restored++
+		}
+	}
+	if restored > 0 {
+		log.Printf("recovered the words of %d episode(s) from transcripts already on disk", restored)
+	}
+}
diff --git a/server/internal/words/pipeline_test.go b/server/internal/words/pipeline_test.go
index 180b807..c010c8e 100644
--- a/server/internal/words/pipeline_test.go
+++ b/server/internal/words/pipeline_test.go
@@ -135,3 +135,72 @@ func TestRecurringNeedsEnoughEpisodesToMeanAnything(t *testing.T) {
 		}
 	}
 }
+
+/*
+Recovering recognition that outlived its episode record.
+
+A transcript is keyed by the episode's id, which comes from the feed and the guid, so it
+survives the record being lost and matches again when the episode is mirrored afresh. Without
+this, 290 episodes whose records were destroyed would each be recognised again -- about twenty
+hours of CPU for text already sitting on disk.
+*/
+func TestReclaimRestoresWordsFromDisk(t *testing.T) {
+	dir := t.TempDir()
+	held := transcript.Store{Dir: dir}
+	if err := held.Put(transcript.Transcript{
+		EpisodeID: "e1", Source: transcript.SourceASR,
+		Cues: []transcript.Cue{{StartMs: 0, EndMs: 1000, Text: "words"}},
+	}); err != nil {
+		t.Fatalf("Put: %v", err)
+	}
+
+	state := store.New()
+	state.Merge("https://f", "A Show", nil)
+	show := state.Show("https://f")
+	show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", DurationMs: 60_000}}
+
+	(&Pipeline{State: state, Transcripts: held}).Reclaim()
+
+	got := show.Episodes[0]
+	if got.TranscriptCues != 1 || got.TranscriptSource != string(transcript.SourceASR) {
+		t.Errorf("got %+v, want the words recovered", got)
+	}
+}
+
+// A transcript longer than the audio describes a different recording, which is the same check
+// made of a publisher's transcript.
+func TestReclaimRefusesATranscriptOfAnotherRecording(t *testing.T) {
+	dir := t.TempDir()
+	held := transcript.Store{Dir: dir}
+	if err := held.Put(transcript.Transcript{
+		EpisodeID: "e1", Source: transcript.SourceASR,
+		Cues: []transcript.Cue{{StartMs: 0, EndMs: 900_000, Text: "a much longer episode"}},
+	}); err != nil {
+		t.Fatalf("Put: %v", err)
+	}
+
+	state := store.New()
+	state.Merge("https://f", "A Show", nil)
+	show := state.Show("https://f")
+	show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", DurationMs: 60_000}}
+
+	(&Pipeline{State: state, Transcripts: held}).Reclaim()
+
+	if show.Episodes[0].TranscriptCues != 0 {
+		t.Error("took a transcript that outlasts the audio")
+	}
+}
+
+// An episode that already has its words is left alone.
+func TestReclaimLeavesWhatIsAlreadyThere(t *testing.T) {
+	state := store.New()
+	state.Merge("https://f", "A Show", nil)
+	show := state.Show("https://f")
+	show.Episodes = []*store.Episode{{ID: "e1", Path: "e1.mp3", TranscriptCues: 99}}
+
+	(&Pipeline{State: state, Transcripts: transcript.Store{Dir: t.TempDir()}}).Reclaim()
+
+	if show.Episodes[0].TranscriptCues != 99 {
+		t.Error("overwrote words that were already there")
+	}
+}