Follow the data: detect the advertising, not the intro

The property that broke intro detection is the property that makes this work.
Eight episodes mirrored within one minute are maximally dependent, because the
ad server inserts whatever campaign is running into all of them identically.
For intro detection that destroys the independence the confirmation threshold
needs -- which is how it came to confirm an advertisement on 8 of 8 episodes.
For ad detection it is the signal.

And unlike intro detection it is checkable against a number the detector does
not produce: the feed declares the ad-free length, the file served is longer,
and the difference is the advertising.

**Measured on the live deployment, Radiolab's eight mirrored episodes: the scan
accounted for 53:29 of the 59:39 the length gap implies, with every individual
episode between 71% and 105%.** That is the first time detection in this
project has been checked against external ground truth and agreed with it.

The old sliding search could never have done this -- one full-length
correlation per window position, 199 seconds for a 240-second clip. BestLags
computes the cross-correlation over every lag in a single FFT pair, which says
where two recordings align; MatchedRuns then walks the extent of the agreement
directly, window by window, with no transform. Sized against measurements
rather than guesses: 2^22 is 1.69s, so a 70-minute episode decimated to 250Hz
is about five seconds a pair. Silence is explicitly not agreement, or separate
advertisements weld together across the pauses between them.

Breaks publish as ranges into episode_midrolls, which MidrollSkip and
PlayheadPolicy already read -- no playback changes and no migration. Intro and
outro offsets are still computed and logged but no longer published, and the
ad-derived ones already in the server's state were cleared.

Two guards publish nothing rather than something wrong: an episode whose
matches cover over 35% of it, and any single match over four minutes. A
re-aired episode or a two-parter would otherwise be eaten. Both record the
refusal, because "found nothing" and "refused what it found" are different
states and conflating them is the readout mistake over again.

Also three data-loss bugs, found by reading the journal after a restart rather
than by any test. The subscription list -- the one thing the server cannot
rediscover -- was persisted only when a whole poll cycle finished, which for 61
feeds is hours. A cycle re-applied its stale snapshot at the end, which would
undo a push that arrived during it and then delete the audio of every show the
snapshot omitted. And detection ignored cancellation, so systemd's 90-second
stop SIGKILLed the process before the save-on-exit could run. All three fixed.

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

1 files changed, 120 insertions, 23 deletions

diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go
index 588af0d..04c933f 100644
--- a/server/cmd/listend/main.go
+++ b/server/cmd/listend/main.go
@@ -72,9 +72,17 @@ func main() {
 	server := api.New(state, worker, api.Options{
 		Token:    token,
 		AudioDir: audioDir,
-		// Act on a new subscription list straight away: the phone sends it on refresh, and
-		// waiting for the next interval would leave a new show with nothing for hours.
-		OnSubscriptions: func([]string) { worker.wake() },
+		// Persist the list the moment it arrives, then poll.
+		//
+		// Both halves were learned the hard way. The list is the one thing the server cannot
+		// rediscover, and it used to live only in memory until a whole poll cycle completed --
+		// which for 61 feeds is hours -- so a restart in the meantime lost every subscription
+		// the phone had sent. And acting on it at once matters because waiting for the next
+		// interval leaves a newly-subscribed show with nothing for hours.
+		OnSubscriptions: func([]string) {
+			worker.save()
+			worker.wake()
+		},
 	})
 
 	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
@@ -192,16 +200,31 @@ func (w *worker) pollAll(ctx context.Context) {
 			return
 		}
 		w.pollOne(ctx, feedURL)
-	}
-	// A feed no longer subscribed stops being polled; reclaim its audio too.
-	for _, url := range w.state.SetFeeds(feeds) {
-		if show := w.state.Forget(url); show != nil {
-			for _, e := range show.Episodes {
+		// After every feed rather than at the end of the cycle: a cycle over 61 feeds takes
+		// hours, and a restart partway through used to discard all of it.
+		w.save()
+	}
+	// Reclaim the audio of shows no longer subscribed to.
+	//
+	// Against the list as it stands *now*, not the snapshot this cycle began with. A cycle over
+	// 61 feeds takes hours; re-applying a stale snapshot would silently undo a subscription the
+	// phone sent during it, and then delete the audio of every show the stale list omitted.
+	current := make(map[string]bool)
+	for _, url := range w.state.Feeds() {
+		current[url] = true
+	}
+	for _, show := range w.state.Shows() {
+		if current[show.FeedURL] {
+			continue
+		}
+		if dropped := w.state.Forget(show.FeedURL); dropped != nil {
+			for _, e := range dropped.Episodes {
 				_ = w.mirror.Prune(e)
 			}
-			log.Printf("%s: unsubscribed, reclaimed %d episodes", url, len(show.Episodes))
+			log.Printf("%s: unsubscribed, reclaimed %d episodes", show.FeedURL, len(dropped.Episodes))
 		}
 	}
+	w.save()
 }
 
 func (w *worker) pollOne(ctx context.Context, feedURL string) {
@@ -261,12 +284,13 @@ func (w *worker) detect(ctx context.Context, show *store.Show, candidates []*sto
 			continue
 		}
 		episodes = append(episodes, skips.Episode{
-			ID:   e.ID,
-			Path: filepath.Join(w.mirror.AudioDir, e.Path),
+			ID:                 e.ID,
+			Path:               filepath.Join(w.mirror.AudioDir, e.Path),
+			DeclaredDurationMs: e.DeclaredDurationMs,
 		})
 		byID[e.ID] = e
 	}
-	if len(episodes) < skips.DefaultConfirmThreshold {
+	if len(episodes) < skips.DefaultConfirmThreshold || ctx.Err() != nil {
 		return
 	}
 
@@ -278,24 +302,97 @@ func (w *worker) detect(ctx context.Context, show *store.Show, candidates []*sto
 	}
 
 	now := time.Now().Unix()
-	applied := 0
 	for _, o := range result.Offsets {
-		e := byID[o.EpisodeID]
-		if e == nil {
-			continue
-		}
-		e.DetectedAtUnix = now
-		e.IntroMs, e.OutroMs = o.IntroMs, o.OutroMs
-		if o.IntroMs != nil || o.OutroMs != nil {
-			applied++
+		if e := byID[o.EpisodeID]; e != nil {
+			e.DetectedAtUnix = now
 		}
 	}
 
-	log.Printf("%s: intro on %d/%d%s, outro on %d/%d%s, %d episode(s) carry offsets, took %s",
+	// Reported, deliberately not published.
+	//
+	// Intro detection assumes the agreeing episodes were fetched independently. Mirroring eight
+	// of them within one minute means they were not, and it showed: the segment confirmed on 8
+	// of 8 Radiolab episodes was searched for in an ad-free copy of the same episode fetched
+	// hours earlier and scored r = 0.0621. It was an advertisement. Until the agreeing episodes
+	// are required to span more than one fetch occasion these numbers cannot be trusted, and
+	// the advertising they keep finding is better served by the scan that measures itself
+	// against the declared-versus-served length gap.
+	log.Printf("%s: intro on %d/%d%s, outro on %d/%d%s (not published -- see detectAds), took %s",
 		show.Title,
 		result.IntroConfirmedOn, len(episodes), checkNote(result.IntroCheck),
 		result.OutroConfirmedOn, len(episodes), checkNote(result.OutroCheck),
-		applied, time.Since(started).Round(time.Second))
+		time.Since(started).Round(time.Second))
+
+	w.detectAds(ctx, show, episodes, byID)
+}
+
+// detectAds finds the advertising a batch of episodes shares, which is what simultaneous
+// mirroring actually makes detectable.
+//
+// Separate from intro detection, and trusted differently. Intro detection assumes the episodes
+// were fetched independently; mirroring eight of them in one minute means they were not, which
+// is how it came to confirm an advertisement as an intro. This uses that dependence rather than
+// being defeated by it -- and reports what it found against the declared-versus-served length
+// gap, a figure it did not produce and which can therefore contradict it.
+func (w *worker) detectAds(
+	ctx context.Context,
+	show *store.Show,
+	episodes []skips.Episode,
+	byID map[string]*store.Episode,
+) {
+	if ctx.Err() != nil {
+		return
+	}
+	started := time.Now()
+	ads, err := skips.DetectAds(ctx, w.decoder, episodes, skips.DefaultAdConfig())
+	if err != nil {
+		log.Printf("%s: ad scan failed: %v", show.Title, err)
+		return
+	}
+
+	var withBreaks, refused int
+	var foundTotal, insertedTotal int64
+	for _, ep := range episodes {
+		e := byID[ep.ID]
+		if e == nil {
+			continue
+		}
+		acc := ads.Accounting[ep.ID]
+		e.FoundAdMs, e.InsertedAdMs = acc.FoundMs, acc.InsertedMs
+		e.AdsRefused = ads.Refused[ep.ID]
+		e.Breaks = nil
+		for _, br := range ads.Breaks[ep.ID] {
+			e.Breaks = append(e.Breaks, store.Break{
+				StartMs:  br.StartMs,
+				EndMs:    br.EndMs,
+				Partners: br.Partners,
+			})
+		}
+		if len(e.Breaks) > 0 {
+			withBreaks++
+		}
+		if e.AdsRefused {
+			refused++
+		}
+		foundTotal += acc.FoundMs
+		insertedTotal += acc.InsertedMs
+	}
+
+	log.Printf("%s: ads on %d/%d episode(s)%s, accounted %s of %s inserted, took %s",
+		show.Title, withBreaks, len(episodes), refusedNote(refused),
+		clock(foundTotal), clock(insertedTotal), time.Since(started).Round(time.Second))
+}
+
+func refusedNote(refused int) string {
+	if refused == 0 {
+		return ""
+	}
+	return fmt.Sprintf(" (%d refused as too extensive to be advertising)", refused)
+}
+
+func clock(ms int64) string {
+	total := ms / 1000
+	return fmt.Sprintf("%d:%02d", total/60, total%60)
 }
 
 // checkNote reports a withheld result loudly. A confirmation that could not verify itself is