Add reliable Jellycompat watch scrobbling

- Forward start, pause, resume, and stop events with stable media identities
- Persist and retry terminal scrobbles across teardown and restart paths
- Reject ambiguous playback-report route matches

cmd/silo/main.go

@@ -2451,6 +2451,7 @@ func main() {
 			compatDeps.SeasonRepo = seasonRepo
 			compatDeps.EpisodeRepo = episodeRepo
 			compatDeps.ProviderIDRepo = providerIDRepo
+			compatDeps.StableIdentityResolver = watchstate.NewStableIdentityResolver(itemRepo, episodeRepo, providerIDRepo)
 			compatDeps.DetailSvc = detailSvc
 			compatDeps.FolderRepo = folderRepo
 			compatDeps.SessionMgr = sessionMgr
@@ -2458,6 +2459,9 @@ func main() {
 			compatDeps.WatchCompletionObserver = deps.WatchCompletionObserver
 			compatDeps.SettingsRepo = settingsRepo
 			compatDeps.PersonRepo = personRepo
+			if watchProviderService != nil {
+				compatDeps.WatchScrobbler = watchProviderService
+			}
 			compatSearchService := catalog.NewCatalogSearchService(
 				appCtx,
 				settingsRepo,

internal/api/handlers/playback.go

@@ -1348,28 +1348,7 @@ func (h *PlaybackHandler) scrobbleEventForSession(ctx context.Context, session *
 		DurationSeconds:   duration,
 		OccurredAt:        time.Now().UTC(),
 	}
-	if h.StableIdentityResolver == nil {
-		event.Kind = "movie"
-		return event
-	}
-	identity := h.StableIdentityResolver.ResolveHistoryIdentity(ctx, mediaItemID)
-	event.Kind = identity.StableType
-	if event.Kind == "" {
-		event.Kind = "movie"
-	}
-	event.SeasonNumber = intPtrValue(identity.Season)
-	event.EpisodeNumber = intPtrValue(identity.Episode)
-	if identity.ProviderIDs != nil {
-		event.IMDbID = identity.ProviderIDs["imdb"]
-		event.TMDBID = identity.ProviderIDs["tmdb"]
-		event.TVDBID = identity.ProviderIDs["tvdb"]
-	}
-	if identity.SeriesProviderIDs != nil {
-		event.SeriesIMDbID = identity.SeriesProviderIDs["imdb"]
-		event.SeriesTMDBID = identity.SeriesProviderIDs["tmdb"]
-		event.SeriesTVDBID = identity.SeriesProviderIDs["tvdb"]
-	}
-	return event
+	return watchsync.ResolveScrobbleIdentity(ctx, h.StableIdentityResolver, event)
 }
 
 func (h *PlaybackHandler) scrobbleEventForStoppedSession(
@@ -1406,13 +1385,6 @@ func (h *PlaybackHandler) scrobbleEventForStoppedSession(
 	return event, true
 }
 
-func intPtrValue(value *int) int {
-	if value == nil {
-		return 0
-	}
-	return *value
-}
-
 func (h *PlaybackHandler) buildAdminHistoryEntry(
 	ctx context.Context,
 	session *playback.Session,

internal/jellycompat/audio_selection_test.go

@@ -66,9 +66,12 @@ func (m *testCompatSessionManager) UpdateProgress(sessionID string, position flo
 		isPaused:  isPaused,
 	})
 	if m.sessions != nil {
-		if _, ok := m.sessions[sessionID]; !ok {
+		session, ok := m.sessions[sessionID]
+		if !ok {
 			return playback.ErrSessionNotFound
 		}
+		session.Position = position
+		session.IsPaused = isPaused
 	}
 	return nil
 }
@@ -111,6 +114,7 @@ func (m *testCompatSessionManager) UpdateAudioTrack(sessionID string, audioTrack
 
 func (m *testCompatSessionManager) StopSession(sessionID string) error {
 	m.stopCalls = append(m.stopCalls, sessionID)
+	delete(m.sessions, sessionID)
 	return nil
 }
 

internal/jellycompat/handlers_playback.go

@@ -29,6 +29,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/subtitles"
 	"github.com/Silo-Server/silo-server/internal/transcodenode"
 	"github.com/Silo-Server/silo-server/internal/userstore"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
 )
 
 type playbackInfoRequest struct {
@@ -151,6 +152,14 @@ type PlaybackSessionSyncer interface {
 	SyncNow(ctx context.Context) error
 }
 
+// PlaybackWatchScrobbler forwards a playback lifecycle to connected watch
+// providers. The watchsync service implements this interface.
+type PlaybackWatchScrobbler interface {
+	ScrobbleStart(ctx context.Context, event watchsync.ScrobbleEvent) error
+	ScrobblePause(ctx context.Context, event watchsync.ScrobbleEvent) error
+	ScrobbleStop(ctx context.Context, event watchsync.ScrobbleEvent) error
+}
+
 // PlaybackHandler serves Jellyfin playback negotiation endpoints.
 type PlaybackHandler struct {
 	cfg                     *config.Config
@@ -173,12 +182,15 @@ type PlaybackHandler struct {
 	// and node-affinity rule for free. The reconstruction recipe is carried in the
 	// compat playback store (PlaybackSession.Recipe), since Jellyfin clients cannot
 	// round-trip a native stream token.
-	tm            *playback.TranscodeManager
-	SubtitleRepo  subtitles.Repository  // optional; enables downloaded subtitles
-	S3Client      subtitles.S3Client    // optional; for serving S3 subtitles
-	S3Bucket      string                // bucket for subtitle storage
-	SettingsRepo  SettingsReader        // optional; reads watched threshold setting
-	SessionSyncer PlaybackSessionSyncer // optional; enables immediate session sync to shared admin view
+	tm                     *playback.TranscodeManager
+	SubtitleRepo           subtitles.Repository  // optional; enables downloaded subtitles
+	S3Client               subtitles.S3Client    // optional; for serving S3 subtitles
+	S3Bucket               string                // bucket for subtitle storage
+	SettingsRepo           SettingsReader        // optional; reads watched threshold setting
+	SessionSyncer          PlaybackSessionSyncer // optional; enables immediate session sync to shared admin view
+	WatchScrobbler         PlaybackWatchScrobbler
+	StableIdentityResolver watchsync.ScrobbleIdentityResolver
+	terminalFallbackDelay  time.Duration
 	// RecipeNodeStore hands a remote transcode's reconstruction recipe to the
 	// control-plane recipe store (Redis) so a dedicated transcode node that
 	// restarts can rebuild ffmpeg from it. The node-hop token is server-minted and
@@ -292,8 +304,10 @@ func NewPlaybackHandler(
 		// ffmpeg crash: drop the dead transcode and stop the upstream native
 		// session. The recipe stays in the compat store so a resume reconstructs.
 		nodeURL := ""
+		var upstreamSession *playback.Session
 		if h.sessionMgr != nil {
 			if up, err := h.sessionMgr.GetSession(sessionID); err == nil && up != nil {
+				upstreamSession = up
 				nodeURL = up.TranscodeNodeURL
 			}
 		}
@@ -305,6 +319,11 @@ func NewPlaybackHandler(
 		// the dead transcode. The recipe stays in the compat store either way so a
 		// resume reconstructs.
 		if h.sessionMgr != nil && h.tm.CloseTranscodeSessionIf(sessionID, dead, nodeURL) {
+			if h.playbackStore != nil {
+				if playSession, ok := h.playbackStore.FindByUpstreamSessionID(sessionID); ok {
+					h.dispatchCompatScrobble(ctx, compatScrobblePause, playSession, upstreamSession, nil)
+				}
+			}
 			_ = h.sessionMgr.StopSession(sessionID)
 		}
 	}

internal/jellycompat/playback_report_liveness_test.go

@@ -38,14 +38,15 @@ func newReportLivenessHandler(upstreamID string, registerUpstream bool) (*Playba
 
 	playbackStore := NewPlaybackSessionStore(time.Hour, nil)
 	playbackStore.Put(PlaybackSession{
-		ID:                 "play-1",
-		CompatToken:        "token-1",
-		ItemID:             "movie-1",
-		RouteItemID:        encodedItemID,
-		UserID:             "user-1",
-		UpstreamSessionID:  upstreamID,
-		UpstreamPlayMethod: "direct",
-		MediaSources:       []PlaybackMediaSource{source},
+		ID:                       "play-1",
+		CompatToken:              "token-1",
+		ItemID:                   "movie-1",
+		RouteItemID:              encodedItemID,
+		UserID:                   "user-1",
+		UpstreamSessionID:        upstreamID,
+		UpstreamPlayMethod:       "direct",
+		ProgressPersistenceKnown: true,
+		MediaSources:             []PlaybackMediaSource{source},
 	})
 
 	sessions := map[string]*playback.Session{}
@@ -336,11 +337,10 @@ func TestHandlePlaybackReport_StopViaAliasTearsDown(t *testing.T) {
 	}
 }
 
-// TestHandlePlaybackReport_StopViaRouteMatchDoesNotTearDown proves a Stopped
-// report that only matched by item/source route (an ambiguous match when the
-// same item plays twice under one token) must not tear down the session it
-// happened to hit; stale cleanup owns that session's end of life.
-func TestHandlePlaybackReport_StopViaRouteMatchDoesNotTearDown(t *testing.T) {
+// TestHandlePlaybackReport_StopViaUniqueRouteTearsDown proves a client that
+// replaces its PlaySessionId can still finalize the single caller-owned play
+// identified by its item/source route.
+func TestHandlePlaybackReport_StopViaUniqueRouteTearsDown(t *testing.T) {
 	handler, mgr, encodedItemID, sourceID := newReportLivenessHandler("upstream-1", true)
 
 	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
@@ -353,18 +353,89 @@ func TestHandlePlaybackReport_StopViaRouteMatchDoesNotTearDown(t *testing.T) {
 	if rec.Code != http.StatusNoContent {
 		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
 	}
-	if len(mgr.stopCalls) != 0 {
-		t.Fatalf("StopSession calls = %v, want none for an ambiguous route-only match", mgr.stopCalls)
+	if len(mgr.stopCalls) != 1 || mgr.stopCalls[0] != "upstream-1" {
+		t.Fatalf("StopSession calls = %v, want upstream-1", mgr.stopCalls)
 	}
-	if _, ok := handler.playbackStore.Get("play-1"); !ok {
-		t.Fatal("expected route-only Stopped report to leave the play session in place")
+	if _, ok := handler.playbackStore.Get("play-1"); ok {
+		t.Fatal("unique route-only Stopped report left the play session in place")
 	}
-	// The final position still lands on the upstream session.
 	if len(mgr.progressUpdates) != 1 || mgr.progressUpdates[0].sessionID != "upstream-1" {
 		t.Fatalf("progress updates = %+v, want one update on upstream-1", mgr.progressUpdates)
 	}
 }
 
+// TestHandlePlaybackReport_StopViaAmbiguousRouteDoesNotTearDown proves route
+// fallback refuses to choose when the same item/source is playing twice under
+// one token.
+func TestHandlePlaybackReport_StopViaAmbiguousRouteDoesNotTearDown(t *testing.T) {
+	handler, mgr, encodedItemID, sourceID := newReportLivenessHandler("upstream-1", true)
+	original, ok := handler.playbackStore.Get("play-1")
+	if !ok {
+		t.Fatal("original play session missing")
+	}
+	sibling := *original
+	sibling.ID = "play-2"
+	sibling.UpstreamSessionID = "upstream-2"
+	handler.playbackStore.Put(sibling)
+	mgr.sessions["upstream-2"] = &playback.Session{ID: "upstream-2", PlayMethod: playback.PlayDirect}
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"never-seen-psid","ItemId":"`+encodedItemID+`","MediaSourceId":"`+sourceID+`","PositionTicks":9000000000}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 1, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+	if len(mgr.stopCalls) != 0 || len(mgr.progressUpdates) != 0 {
+		t.Fatalf("ambiguous route mutated playback: stops=%v progress=%+v", mgr.stopCalls, mgr.progressUpdates)
+	}
+	if _, ok := handler.playbackStore.Get("play-1"); !ok {
+		t.Fatal("ambiguous route removed play-1")
+	}
+	if _, ok := handler.playbackStore.Get("play-2"); !ok {
+		t.Fatal("ambiguous route removed play-2")
+	}
+}
+
+func TestHandlePlaybackReport_StopRejectsMixedRouteIdentifiers(t *testing.T) {
+	handler, mgr, encodedItemID, _ := newReportLivenessHandler("upstream-1", true)
+	original, ok := handler.playbackStore.Get("play-1")
+	if !ok {
+		t.Fatal("original play session missing")
+	}
+	sibling := *original
+	sibling.ID = "play-2"
+	sibling.ItemID = "movie-2"
+	sibling.RouteItemID = handler.codec.EncodeStringID(EncodedIDItem, "movie-2")
+	sibling.UpstreamSessionID = "upstream-2"
+	sibling.MediaSources = []PlaybackMediaSource{{ID: "source-2", FileID: 43}}
+	handler.playbackStore.Put(sibling)
+	mgr.sessions["upstream-2"] = &playback.Session{ID: "upstream-2", PlayMethod: playback.PlayDirect}
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"stale-play-id","ItemId":"`+encodedItemID+`","MediaSourceId":"source-2","PositionTicks":9000000000}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 1, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+	if len(mgr.stopCalls) != 0 || len(mgr.progressUpdates) != 0 {
+		t.Fatalf("mixed route identifiers mutated playback: stops=%v progress=%+v", mgr.stopCalls, mgr.progressUpdates)
+	}
+	if _, ok := handler.playbackStore.Get("play-1"); !ok {
+		t.Fatal("mixed route identifiers removed play-1")
+	}
+	if _, ok := handler.playbackStore.Get("play-2"); !ok {
+		t.Fatal("mixed route identifiers removed play-2")
+	}
+}
+
 // TestEnsureUpstreamPlayback_ReviveClosesStaleTranscode proves that when a
 // reaped upstream session is recreated under the same play method, any
 // transcode still keyed to the stale upstream id is closed first — otherwise

internal/jellycompat/playback_scrobble.go

@@ -0,0 +1,171 @@
+package jellycompat
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/userstore"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
+)
+
+type compatScrobbleAction string
+
+const (
+	compatScrobbleStart compatScrobbleAction = "start"
+	compatScrobblePause compatScrobbleAction = "pause"
+	compatScrobbleStop  compatScrobbleAction = "stop"
+)
+
+func (h *PlaybackHandler) dispatchCompatScrobble(
+	ctx context.Context,
+	action compatScrobbleAction,
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	preferredSource *PlaybackMediaSource,
+) error {
+	return h.dispatchCompatScrobbleAt(ctx, action, playSession, upstreamSession, preferredSource, nil)
+}
+
+func (h *PlaybackHandler) dispatchCompatScrobbleAt(
+	ctx context.Context,
+	action compatScrobbleAction,
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	preferredSource *PlaybackMediaSource,
+	positionOverride *float64,
+) error {
+	event, ok := h.compatScrobbleEvent(
+		ctx, action, playSession, upstreamSession, preferredSource, positionOverride,
+	)
+	if !ok {
+		return nil
+	}
+	return h.dispatchCompatScrobbleEvent(ctx, action, event)
+}
+
+func (h *PlaybackHandler) compatScrobbleEvent(
+	ctx context.Context,
+	action compatScrobbleAction,
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	preferredSource *PlaybackMediaSource,
+	positionOverride *float64,
+) (watchsync.ScrobbleEvent, bool) {
+	if h == nil || h.WatchScrobbler == nil || playSession == nil || upstreamSession == nil ||
+		upstreamSession.DisableProgressPersistence || playSession.ItemID == "" {
+		return watchsync.ScrobbleEvent{}, false
+	}
+	scrobbleCtx, cancel := compatDetachedContext(ctx)
+	defer cancel()
+
+	source := compatScrobbleSource(playSession, upstreamSession, preferredSource)
+	duration := 0.0
+	if source != nil {
+		duration = float64(source.Version.Duration)
+	}
+	position := upstreamSession.Position
+	if positionOverride != nil {
+		position = *positionOverride
+	} else if position <= 0 && playSession.InitialSeekSeconds > 0 {
+		position = playSession.InitialSeekSeconds
+	}
+	completed := false
+	if action == compatScrobbleStop && duration > 0 {
+		_, completed, _ = userstore.ResolveProgressState(position, duration, h.playbackThresholds(scrobbleCtx))
+	}
+	event := watchsync.ResolveScrobbleIdentity(scrobbleCtx, h.StableIdentityResolver, watchsync.ScrobbleEvent{
+		PlaybackSessionID: upstreamSession.ID,
+		UserID:            upstreamSession.UserID,
+		ProfileID:         upstreamSession.ProfileID,
+		MediaItemID:       playSession.ItemID,
+		PositionSeconds:   position,
+		DurationSeconds:   duration,
+		Completed:         completed,
+		OccurredAt:        time.Now().UTC(),
+	})
+	return event, true
+}
+
+func (h *PlaybackHandler) dispatchCompatScrobbleEvent(
+	ctx context.Context,
+	action compatScrobbleAction,
+	event watchsync.ScrobbleEvent,
+) error {
+	if h == nil || h.WatchScrobbler == nil {
+		return nil
+	}
+	scrobbleCtx, cancel := compatDetachedContext(ctx)
+	defer cancel()
+	var err error
+	switch action {
+	case compatScrobblePause:
+		err = h.WatchScrobbler.ScrobblePause(scrobbleCtx, event)
+	case compatScrobbleStop:
+		err = h.WatchScrobbler.ScrobbleStop(scrobbleCtx, event)
+	default:
+		err = h.WatchScrobbler.ScrobbleStart(scrobbleCtx, event)
+	}
+	if err != nil {
+		slog.WarnContext(scrobbleCtx, "failed to queue jellycompat watch provider scrobble",
+			"component", "jellycompat",
+			"action", action,
+			"playback_session_id", event.PlaybackSessionID,
+			"error", err,
+		)
+	}
+	return err
+}
+
+func compatScrobbleSource(
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	preferredSource *PlaybackMediaSource,
+) *PlaybackMediaSource {
+	if preferredSource != nil {
+		return preferredSource
+	}
+	if playSession == nil {
+		return nil
+	}
+	if upstreamSession != nil {
+		for _, source := range playSession.MediaSources {
+			if source.FileID == upstreamSession.MediaFileID {
+				copy := source
+				return &copy
+			}
+		}
+	}
+	return firstMediaSource(playSession)
+}
+
+// compatScrobbleFallbackSession preserves enough authenticated report state to
+// emit a terminal event after the in-memory upstream session has already been
+// reaped. The compat play session remains the source of media identity.
+func compatScrobbleFallbackSession(
+	compatSession *Session,
+	playSession *PlaybackSession,
+	preferredSource *PlaybackMediaSource,
+	position float64,
+	positionKnown bool,
+	isPaused bool,
+) *playback.Session {
+	if compatSession == nil || playSession == nil || playSession.UpstreamSessionID == "" || !positionKnown {
+		return nil
+	}
+	source := compatScrobbleSource(playSession, nil, preferredSource)
+	fileID := 0
+	if source != nil {
+		fileID = source.FileID
+	}
+	return &playback.Session{
+		ID:                         playSession.UpstreamSessionID,
+		UserID:                     compatSession.StreamAppUserID,
+		ProfileID:                  compatSession.ProfileID,
+		MediaFileID:                fileID,
+		Position:                   position,
+		IsPaused:                   isPaused,
+		DisableProgressPersistence: !playSession.ProgressPersistenceKnown || playSession.DisableProgressPersistence,
+	}
+}

internal/jellycompat/playback_scrobble_test.go

@@ -0,0 +1,793 @@
+package jellycompat
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
+)
+
+type compatScrobbleCall struct {
+	action string
+	event  watchsync.ScrobbleEvent
+}
+
+type recordingCompatWatchScrobbler struct {
+	calls []compatScrobbleCall
+}
+
+type channelCompatWatchScrobbler struct {
+	stopEvents chan watchsync.ScrobbleEvent
+	failStops  int
+}
+
+type failingCompatWatchScrobbler struct {
+	stopCalls atomic.Int32
+}
+
+type poisonBatchCompatWatchScrobbler struct {
+	deliverableSessionID string
+	stopCalls            atomic.Int32
+}
+
+func (*failingCompatWatchScrobbler) ScrobbleStart(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (*failingCompatWatchScrobbler) ScrobblePause(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (s *failingCompatWatchScrobbler) ScrobbleStop(context.Context, watchsync.ScrobbleEvent) error {
+	s.stopCalls.Add(1)
+	return errors.New("queue unavailable")
+}
+
+func (*poisonBatchCompatWatchScrobbler) ScrobbleStart(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (*poisonBatchCompatWatchScrobbler) ScrobblePause(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (s *poisonBatchCompatWatchScrobbler) ScrobbleStop(_ context.Context, event watchsync.ScrobbleEvent) error {
+	s.stopCalls.Add(1)
+	if event.PlaybackSessionID != s.deliverableSessionID {
+		return errors.New("poison terminal event")
+	}
+	return nil
+}
+
+type flakyTerminalPlaybackStore struct {
+	*PlaybackSessionStore
+	mu         sync.Mutex
+	failStages int
+	stageCalls int
+}
+
+func (s *flakyTerminalPlaybackStore) StageTerminal(
+	id string,
+	compatToken string,
+	event watchsync.ScrobbleEvent,
+	authoritative bool,
+) (*PlaybackSession, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.stageCalls++
+	if s.failStages > 0 {
+		s.failStages--
+		return nil, errors.New("terminal store unavailable")
+	}
+	return s.PlaybackSessionStore.StageTerminal(id, compatToken, event, authoritative)
+}
+
+func (s *flakyTerminalPlaybackStore) calls() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.stageCalls
+}
+
+func (s *channelCompatWatchScrobbler) ScrobbleStart(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (s *channelCompatWatchScrobbler) ScrobblePause(context.Context, watchsync.ScrobbleEvent) error {
+	return nil
+}
+
+func (s *channelCompatWatchScrobbler) ScrobbleStop(_ context.Context, event watchsync.ScrobbleEvent) error {
+	if s.failStops > 0 {
+		s.failStops--
+		return errors.New("queue unavailable")
+	}
+	s.stopEvents <- event
+	return nil
+}
+
+func (s *recordingCompatWatchScrobbler) ScrobbleStart(_ context.Context, event watchsync.ScrobbleEvent) error {
+	s.calls = append(s.calls, compatScrobbleCall{action: "start", event: event})
+	return nil
+}
+
+func (s *recordingCompatWatchScrobbler) ScrobblePause(_ context.Context, event watchsync.ScrobbleEvent) error {
+	s.calls = append(s.calls, compatScrobbleCall{action: "pause", event: event})
+	return nil
+}
+
+func (s *recordingCompatWatchScrobbler) ScrobbleStop(_ context.Context, event watchsync.ScrobbleEvent) error {
+	s.calls = append(s.calls, compatScrobbleCall{action: "stop", event: event})
+	return nil
+}
+
+func TestEnsureUpstreamPlaybackStartsWatchProviderScrobble(t *testing.T) {
+	mgr := &testCompatSessionManager{}
+	h, store := newActiveEncodingsHandler(mgr)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	h.WatchScrobbler = scrobbler
+	source := PlaybackMediaSource{ID: "source-1", FileID: 42, Version: testCompatVersion()}
+	store.Put(PlaybackSession{
+		ID:                 "play-1",
+		CompatToken:        "token-1",
+		ItemID:             "movie-1",
+		InitialSeekSeconds: 125,
+		MediaSources:       []PlaybackMediaSource{source},
+	})
+	compatSession := &Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}
+
+	if _, err := h.ensureUpstreamPlayback(context.Background(), compatSession, "play-1", source, "direct"); err != nil {
+		t.Fatalf("ensureUpstreamPlayback: %v", err)
+	}
+	if len(scrobbler.calls) != 1 {
+		t.Fatalf("scrobble calls = %d, want 1", len(scrobbler.calls))
+	}
+	call := scrobbler.calls[0]
+	if call.action != "start" || call.event.PlaybackSessionID != "upstream-started" {
+		t.Fatalf("start call = %+v", call)
+	}
+	if call.event.UserID != 7 || call.event.ProfileID != "profile-1" || call.event.MediaItemID != "movie-1" {
+		t.Fatalf("start scope = %+v", call.event)
+	}
+	if call.event.PositionSeconds != 125 || call.event.DurationSeconds != 3600 {
+		t.Fatalf("start progress = %v/%v, want 125/3600", call.event.PositionSeconds, call.event.DurationSeconds)
+	}
+
+	if _, err := h.ensureUpstreamPlayback(context.Background(), compatSession, "play-1", source, "direct"); err != nil {
+		t.Fatalf("ensureUpstreamPlayback reuse: %v", err)
+	}
+	if len(scrobbler.calls) != 1 {
+		t.Fatalf("reuse emitted %d scrobbles, want the original start only", len(scrobbler.calls))
+	}
+}
+
+func TestHandlePlaybackReportScrobblesPauseAndResumeTransitions(t *testing.T) {
+	handler, mgr, _, sourceID := newReportLivenessHandler("upstream-1", true)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	handler.WatchScrobbler = scrobbler
+	mgr.sessions["upstream-1"].UserID = 7
+	mgr.sessions["upstream-1"].ProfileID = "profile-1"
+	mgr.sessions["upstream-1"].MediaFileID = 42
+
+	post := func(paused bool, ticks int64) {
+		body := strings.NewReader(`{"PlaySessionId":"play-1","MediaSourceId":"` + sourceID +
+			`","PositionTicks":` + strconv.FormatInt(ticks, 10) + `,"IsPaused":` + strconv.FormatBool(paused) + `}`)
+		req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Progress", body)
+		req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+			&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+		rec := httptest.NewRecorder()
+		handler.HandleSessionPlayingProgress(rec, req)
+		if rec.Code != http.StatusNoContent {
+			t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+		}
+	}
+
+	post(true, 600_000_000)
+	post(true, 700_000_000)
+	post(false, 800_000_000)
+
+	if len(scrobbler.calls) != 2 {
+		t.Fatalf("scrobble calls = %+v, want pause and resume only", scrobbler.calls)
+	}
+	if scrobbler.calls[0].action != "pause" || scrobbler.calls[0].event.PositionSeconds != 60 {
+		t.Fatalf("pause call = %+v", scrobbler.calls[0])
+	}
+	if scrobbler.calls[1].action != "start" || scrobbler.calls[1].event.PositionSeconds != 80 {
+		t.Fatalf("resume call = %+v", scrobbler.calls[1])
+	}
+}
+
+func TestHandlePlaybackReportPreservesExplicitZeroOnPause(t *testing.T) {
+	handler, mgr, _, sourceID := newReportLivenessHandler("upstream-1", true)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	handler.WatchScrobbler = scrobbler
+	mgr.sessions["upstream-1"].UserID = 7
+	mgr.sessions["upstream-1"].ProfileID = "profile-1"
+	mgr.sessions["upstream-1"].MediaFileID = 42
+	if err := handler.playbackStore.Update("play-1", func(session *PlaybackSession) error {
+		session.InitialSeekSeconds = 125
+		return nil
+	}); err != nil {
+		t.Fatalf("set initial seek: %v", err)
+	}
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Progress", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":0,"IsPaused":true}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingProgress(rec, req)
+
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+	if len(scrobbler.calls) != 1 || scrobbler.calls[0].action != "pause" ||
+		scrobbler.calls[0].event.PositionSeconds != 0 {
+		t.Fatalf("explicit-zero pause scrobble = %+v", scrobbler.calls)
+	}
+}
+
+func TestCompatTeardownScrobblesAuthoritativeStopExactlyOnce(t *testing.T) {
+	tests := []struct {
+		name          string
+		stoppedFirst  bool
+		wantPositions []float64
+	}{
+		{name: "stopped report first", stoppedFirst: true, wantPositions: []float64{90}},
+		{name: "active encodings first", stoppedFirst: false, wantPositions: []float64{90}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{
+				"upstream-1": {
+					ID:          "upstream-1",
+					UserID:      7,
+					ProfileID:   "profile-1",
+					MediaFileID: 42,
+					Position:    45,
+				},
+			}}
+			h, store := newActiveEncodingsHandler(mgr)
+			scrobbler := &recordingCompatWatchScrobbler{}
+			h.WatchScrobbler = scrobbler
+			source := PlaybackMediaSource{ID: "source-1", FileID: 42, Version: testCompatVersion()}
+			store.Put(PlaybackSession{
+				ID:                       "play-1",
+				CompatToken:              "token-1",
+				ItemID:                   "movie-1",
+				UpstreamSessionID:        "upstream-1",
+				UpstreamPlayMethod:       "direct",
+				ProgressPersistenceKnown: true,
+				MediaSources:             []PlaybackMediaSource{source},
+			})
+
+			stopped := func() {
+				req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped",
+					strings.NewReader(`{"PlaySessionId":"play-1","MediaSourceId":"source-1","PositionTicks":900000000}`))
+				req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+					&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+				rec := httptest.NewRecorder()
+				h.HandleSessionPlayingStopped(rec, req)
+				if rec.Code != http.StatusNoContent {
+					t.Fatalf("stopped status = %d", rec.Code)
+				}
+			}
+			activeEncodings := func() {
+				req := withCompatSession(httptest.NewRequest(http.MethodDelete,
+					"/Videos/ActiveEncodings?PlaySessionId=play-1", nil), "token-1")
+				rec := httptest.NewRecorder()
+				h.HandleDeleteActiveEncodings(rec, req)
+				if rec.Code != http.StatusNoContent {
+					t.Fatalf("active encodings status = %d", rec.Code)
+				}
+			}
+
+			if tt.stoppedFirst {
+				stopped()
+				activeEncodings()
+			} else {
+				activeEncodings()
+				stopped()
+			}
+
+			if len(scrobbler.calls) != len(tt.wantPositions) {
+				t.Fatalf("scrobble calls = %+v, want positions %v", scrobbler.calls, tt.wantPositions)
+			}
+			for i, wantPosition := range tt.wantPositions {
+				if scrobbler.calls[i].action != "stop" || scrobbler.calls[i].event.PositionSeconds != wantPosition {
+					t.Fatalf("stop call %d = %+v, want position %v", i, scrobbler.calls[i], wantPosition)
+				}
+			}
+		})
+	}
+}
+
+func TestActiveEncodingsOnNonOwnerDefersScrobbleToStoppedReport(t *testing.T) {
+	mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{}}
+	h, store := newActiveEncodingsHandler(mgr)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	h.WatchScrobbler = scrobbler
+	h.terminalFallbackDelay = 10 * time.Millisecond
+	source := PlaybackMediaSource{ID: "source-1", FileID: 42, Version: testCompatVersion()}
+	store.Put(PlaybackSession{
+		ID:                       "play-1",
+		CompatToken:              "token-1",
+		ItemID:                   "movie-1",
+		UpstreamSessionID:        "upstream-1",
+		ProgressPersistenceKnown: true,
+		MediaSources:             []PlaybackMediaSource{source},
+	})
+
+	activeReq := withCompatSession(httptest.NewRequest(http.MethodDelete,
+		"/Videos/ActiveEncodings?PlaySessionId=play-1", nil), "token-1")
+	activeRec := httptest.NewRecorder()
+	h.HandleDeleteActiveEncodings(activeRec, activeReq)
+	if activeRec.Code != http.StatusNoContent {
+		t.Fatalf("active encodings status = %d", activeRec.Code)
+	}
+	time.Sleep(3 * h.terminalFallbackDelay)
+	if len(scrobbler.calls) != 0 {
+		t.Fatalf("non-owner cleanup emitted stale scrobble: %+v", scrobbler.calls)
+	}
+	if _, ok := store.Get("play-1"); ok {
+		t.Fatal("terminal session remained routable after encoder cleanup")
+	}
+	if _, ok := store.GetFinalizable("play-1", "token-1"); !ok {
+		t.Fatal("terminal session was not retained for the final report")
+	}
+
+	stoppedReq := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"client-replaced-play-id","MediaSourceId":"source-1","PositionTicks":900000000}`))
+	stoppedReq = stoppedReq.WithContext(context.WithValue(stoppedReq.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	stoppedRec := httptest.NewRecorder()
+	h.HandleSessionPlayingStopped(stoppedRec, stoppedReq)
+
+	if stoppedRec.Code != http.StatusNoContent {
+		t.Fatalf("stopped status = %d, body = %s", stoppedRec.Code, stoppedRec.Body.String())
+	}
+	if len(scrobbler.calls) != 1 || scrobbler.calls[0].action != "stop" ||
+		scrobbler.calls[0].event.PositionSeconds != 90 {
+		t.Fatalf("authoritative stopped scrobble = %+v, want one stop at 90s", scrobbler.calls)
+	}
+}
+
+func TestActiveEncodingsFallbackAllowsLaterAuthoritativeStop(t *testing.T) {
+	mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{
+		"upstream-1": {
+			ID:          "upstream-1",
+			UserID:      7,
+			ProfileID:   "profile-1",
+			MediaFileID: 42,
+			Position:    45,
+		},
+	}}
+	h, store := newActiveEncodingsHandler(mgr)
+	scrobbler := &channelCompatWatchScrobbler{stopEvents: make(chan watchsync.ScrobbleEvent, 2)}
+	h.WatchScrobbler = scrobbler
+	h.terminalFallbackDelay = 10 * time.Millisecond
+	store.Put(PlaybackSession{
+		ID:                       "play-1",
+		CompatToken:              "token-1",
+		ItemID:                   "movie-1",
+		UpstreamSessionID:        "upstream-1",
+		ProgressPersistenceKnown: true,
+		MediaSources: []PlaybackMediaSource{{
+			ID: "source-1", FileID: 42, Version: testCompatVersion(),
+		}},
+	})
+
+	req := withCompatSession(httptest.NewRequest(http.MethodDelete,
+		"/Videos/ActiveEncodings?PlaySessionId=play-1", nil), "token-1")
+	rec := httptest.NewRecorder()
+	h.HandleDeleteActiveEncodings(rec, req)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d", rec.Code)
+	}
+
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 45 || event.PlaybackSessionID != "upstream-1" {
+			t.Fatalf("fallback stop = %+v, want upstream-1 at 45s", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for ActiveEncodings terminal fallback")
+	}
+	terminal, ok := store.GetFinalizable("play-1", "token-1")
+	if !ok || !terminal.TerminalFallbackSent || terminal.TerminalAuthoritative {
+		t.Fatalf("fallback terminal state = ok=%v session=%+v", ok, terminal)
+	}
+
+	stoppedReq := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"source-1","PositionTicks":900000000}`))
+	stoppedReq = stoppedReq.WithContext(context.WithValue(stoppedReq.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	stoppedRec := httptest.NewRecorder()
+	h.HandleSessionPlayingStopped(stoppedRec, stoppedReq)
+	if stoppedRec.Code != http.StatusNoContent {
+		t.Fatalf("stopped status = %d", stoppedRec.Code)
+	}
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 90 {
+			t.Fatalf("authoritative stop = %+v, want 90s", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for authoritative stop after fallback")
+	}
+	if _, ok := store.GetFinalizable("play-1", "token-1"); ok {
+		t.Fatal("authoritative terminal event remained after delivery")
+	}
+}
+
+func TestPositionlessLateStopPreservesAndDeliversPendingFallback(t *testing.T) {
+	mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{
+		"upstream-1": {
+			ID:          "upstream-1",
+			UserID:      7,
+			ProfileID:   "profile-1",
+			MediaFileID: 42,
+			Position:    45,
+		},
+	}}
+	h, store := newActiveEncodingsHandler(mgr)
+	scrobbler := &channelCompatWatchScrobbler{stopEvents: make(chan watchsync.ScrobbleEvent, 1)}
+	h.WatchScrobbler = scrobbler
+	h.terminalFallbackDelay = time.Hour
+	store.Put(PlaybackSession{
+		ID:                       "play-1",
+		CompatToken:              "token-1",
+		ItemID:                   "movie-1",
+		UpstreamSessionID:        "upstream-1",
+		ProgressPersistenceKnown: true,
+		MediaSources: []PlaybackMediaSource{{
+			ID: "source-1", FileID: 42, Version: testCompatVersion(),
+		}},
+	})
+
+	activeReq := withCompatSession(httptest.NewRequest(
+		http.MethodDelete, "/Videos/ActiveEncodings?PlaySessionId=play-1", nil,
+	), "token-1")
+	h.HandleDeleteActiveEncodings(httptest.NewRecorder(), activeReq)
+	terminal, ok := store.GetFinalizable("play-1", "token-1")
+	if !ok || terminal.TerminalScrobbleEvent == nil || terminal.TerminalFallbackSent {
+		t.Fatalf("pending fallback = ok=%v session=%+v", ok, terminal)
+	}
+
+	stoppedReq := httptest.NewRequest(
+		http.MethodPost,
+		"/Sessions/Playing/Stopped",
+		strings.NewReader(`{"PlaySessionId":"play-1","MediaSourceId":"source-1"}`),
+	)
+	stoppedReq = stoppedReq.WithContext(context.WithValue(
+		stoppedReq.Context(),
+		compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"},
+	))
+	h.HandleSessionPlayingStopped(httptest.NewRecorder(), stoppedReq)
+
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 45 {
+			t.Fatalf("preserved fallback = %+v, want 45s", event)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for preserved terminal fallback")
+	}
+	terminal, ok = store.GetFinalizable("play-1", "token-1")
+	if !ok || !terminal.TerminalFallbackSent || terminal.TerminalAuthoritative {
+		t.Fatalf("delivered fallback state = ok=%v session=%+v", ok, terminal)
+	}
+}
+
+func TestStoppedScrobbleQueueFailureRetainsAndRetriesTerminalEvent(t *testing.T) {
+	handler, mgr, _, sourceID := newReportLivenessHandler("upstream-1", true)
+	scrobbler := &channelCompatWatchScrobbler{
+		stopEvents: make(chan watchsync.ScrobbleEvent, 1),
+		failStops:  1,
+	}
+	handler.WatchScrobbler = scrobbler
+	mgr.sessions["upstream-1"].UserID = 7
+	mgr.sessions["upstream-1"].ProfileID = "profile-1"
+	mgr.sessions["upstream-1"].MediaFileID = 42
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":900000000}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	if _, ok := handler.playbackStore.GetFinalizable("play-1", "token-1"); !ok {
+		t.Fatal("terminal event was deleted after queue failure")
+	}
+
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 90 {
+			t.Fatalf("retried stop = %+v, want 90s", event)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for terminal queue retry")
+	}
+	if _, ok := handler.playbackStore.GetFinalizable("play-1", "token-1"); ok {
+		t.Fatal("authoritative terminal event remained after successful retry")
+	}
+}
+
+func TestStoppedScrobbleRestagesAfterTerminalPersistenceFailure(t *testing.T) {
+	handler, mgr, _, sourceID := newReportLivenessHandler("upstream-1", true)
+	baseStore := handler.playbackStore.(*PlaybackSessionStore)
+	flakyStore := &flakyTerminalPlaybackStore{PlaybackSessionStore: baseStore, failStages: 1}
+	handler.playbackStore = flakyStore
+	scrobbler := &channelCompatWatchScrobbler{stopEvents: make(chan watchsync.ScrobbleEvent, 1)}
+	handler.WatchScrobbler = scrobbler
+	mgr.sessions["upstream-1"].UserID = 7
+	mgr.sessions["upstream-1"].ProfileID = "profile-1"
+	mgr.sessions["upstream-1"].MediaFileID = 42
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":900000000}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	if calls := flakyStore.calls(); calls != 1 {
+		t.Fatalf("synchronous stage calls = %d, want 1", calls)
+	}
+	if _, ok := flakyStore.Get("play-1"); ok {
+		t.Fatal("failed durable terminal stage left the stopped session routable")
+	}
+	if len(mgr.stopCalls) != 1 || mgr.stopCalls[0] != "upstream-1" {
+		t.Fatalf("cleanup after failed stage = %v, want upstream-1 stopped immediately", mgr.stopCalls)
+	}
+
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 90 {
+			t.Fatalf("restaged stop = %+v, want 90s", event)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for terminal restage retry")
+	}
+	if calls := flakyStore.calls(); calls < 2 {
+		t.Fatalf("stage calls = %d, want persistence retry", calls)
+	}
+	if _, ok := flakyStore.GetFinalizable("play-1", "token-1"); ok {
+		t.Fatal("restaged authoritative event remained after delivery")
+	}
+}
+
+func TestStoppedScrobblePreservesExplicitZeroPosition(t *testing.T) {
+	handler, mgr, _, sourceID := newReportLivenessHandler("upstream-1", true)
+	scrobbler := &channelCompatWatchScrobbler{stopEvents: make(chan watchsync.ScrobbleEvent, 1)}
+	handler.WatchScrobbler = scrobbler
+	mgr.sessions["upstream-1"].Position = 45
+	mgr.sessions["upstream-1"].UserID = 7
+	mgr.sessions["upstream-1"].ProfileID = "profile-1"
+	mgr.sessions["upstream-1"].MediaFileID = 42
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":0}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+
+	select {
+	case event := <-scrobbler.stopEvents:
+		if event.PositionSeconds != 0 {
+			t.Fatalf("stop position = %v, want explicit zero", event.PositionSeconds)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for zero-position stop")
+	}
+}
+
+func TestTerminalScrobbleRecoveryDeliversPersistedEventAfterRestart(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "token-1"})
+	event := watchsync.ScrobbleEvent{
+		PlaybackSessionID: "upstream-1",
+		UserID:            7,
+		ProfileID:         "profile-1",
+		MediaItemID:       "movie-1",
+		PositionSeconds:   90,
+	}
+	if _, err := store.StageTerminal("play-1", "token-1", event, true); err != nil {
+		t.Fatalf("stage terminal event: %v", err)
+	}
+	scrobbler := &channelCompatWatchScrobbler{stopEvents: make(chan watchsync.ScrobbleEvent, 1)}
+	handler := &PlaybackHandler{playbackStore: store, WatchScrobbler: scrobbler}
+
+	if err := recoverPendingTerminalScrobbles(context.Background(), handler); err != nil {
+		t.Fatalf("recover terminal events: %v", err)
+	}
+	select {
+	case got := <-scrobbler.stopEvents:
+		if got.PlaybackSessionID != "upstream-1" || got.PositionSeconds != 90 {
+			t.Fatalf("recovered event = %+v", got)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for recovered terminal event")
+	}
+	if _, ok := store.GetFinalizable("play-1", "token-1"); ok {
+		t.Fatal("recovered authoritative event remained pending")
+	}
+}
+
+func TestTerminalScrobbleRecoveryLeavesRetryToNextScan(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "token-1"})
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+	if _, err := store.StageTerminal("play-1", "token-1", event, true); err != nil {
+		t.Fatalf("stage terminal event: %v", err)
+	}
+	scrobbler := &failingCompatWatchScrobbler{}
+	handler := &PlaybackHandler{playbackStore: store, WatchScrobbler: scrobbler}
+
+	if err := recoverPendingTerminalScrobbles(context.Background(), handler); err != nil {
+		t.Fatalf("recover terminal events: %v", err)
+	}
+	time.Sleep(compatTerminalInitialRetryDelay + 100*time.Millisecond)
+	if calls := scrobbler.stopCalls.Load(); calls != 1 {
+		t.Fatalf("recovery stop attempts = %d, want one attempt per scan", calls)
+	}
+	if _, ok := store.GetFinalizable("play-1", "token-1"); !ok {
+		t.Fatal("failed recovery did not retain the terminal event for the next scan")
+	}
+}
+
+func TestTerminalScrobbleRecoveryRotatesPastPoisonBatch(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	for i := 0; i <= compatTerminalRecoveryBatchSize; i++ {
+		id := fmt.Sprintf("play-%03d", i)
+		store.Put(PlaybackSession{ID: id, CompatToken: "token-1"})
+		if _, err := store.StageTerminal(
+			id,
+			"token-1",
+			watchsync.ScrobbleEvent{PlaybackSessionID: fmt.Sprintf("upstream-%03d", i)},
+			true,
+		); err != nil {
+			t.Fatalf("stage terminal event %s: %v", id, err)
+		}
+	}
+	scrobbler := &poisonBatchCompatWatchScrobbler{deliverableSessionID: "upstream-100"}
+	handler := &PlaybackHandler{playbackStore: store, WatchScrobbler: scrobbler}
+
+	if err := recoverPendingTerminalScrobbles(context.Background(), handler); err != nil {
+		t.Fatalf("recover poison batch: %v", err)
+	}
+	if _, ok := store.GetFinalizable("play-100", "token-1"); !ok {
+		t.Fatal("first bounded scan unexpectedly reached the event after the poison batch")
+	}
+	if err := recoverPendingTerminalScrobbles(context.Background(), handler); err != nil {
+		t.Fatalf("recover after poison batch: %v", err)
+	}
+	if _, ok := store.GetFinalizable("play-100", "token-1"); ok {
+		t.Fatal("event after poison batch remained starved on the next scan")
+	}
+	if calls := scrobbler.stopCalls.Load(); calls != compatTerminalRecoveryBatchSize+1 {
+		t.Fatalf("recovery attempts = %d, want %d", calls, compatTerminalRecoveryBatchSize+1)
+	}
+}
+
+func TestReapedSessionFallbackHonorsProgressPersistencePolicy(t *testing.T) {
+	tests := []struct {
+		name     string
+		known    bool
+		disabled bool
+	}{
+		{name: "disabled", known: true, disabled: true},
+		{name: "legacy row with unknown policy", known: false, disabled: false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler, _, _, sourceID := newReportLivenessHandler("upstream-reaped", false)
+			scrobbler := &recordingCompatWatchScrobbler{}
+			handler.WatchScrobbler = scrobbler
+			if err := handler.playbackStore.Update("play-1", func(session *PlaybackSession) error {
+				session.ProgressPersistenceKnown = tt.known
+				session.DisableProgressPersistence = tt.disabled
+				return nil
+			}); err != nil {
+				t.Fatalf("set progress policy: %v", err)
+			}
+
+			req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+				`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":900000000}`))
+			req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+				&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+			rec := httptest.NewRecorder()
+			handler.HandleSessionPlayingStopped(rec, req)
+
+			if rec.Code != http.StatusNoContent {
+				t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+			}
+			if len(scrobbler.calls) != 0 {
+				t.Fatalf("privacy-suppressed fallback emitted scrobble: %+v", scrobbler.calls)
+			}
+		})
+	}
+}
+
+func TestHandleSessionStoppedScrobblesAfterUpstreamSessionWasReaped(t *testing.T) {
+	handler, _, _, sourceID := newReportLivenessHandler("upstream-reaped", false)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	handler.WatchScrobbler = scrobbler
+
+	req := httptest.NewRequest(http.MethodPost, "/Sessions/Playing/Stopped", strings.NewReader(
+		`{"PlaySessionId":"play-1","MediaSourceId":"`+sourceID+`","PositionTicks":900000000}`))
+	req = req.WithContext(context.WithValue(req.Context(), compatSessionKey,
+		&Session{Token: "token-1", StreamAppUserID: 7, ProfileID: "profile-1"}))
+	rec := httptest.NewRecorder()
+	handler.HandleSessionPlayingStopped(rec, req)
+
+	if rec.Code != http.StatusNoContent {
+		t.Fatalf("status = %d, body = %s", rec.Code, rec.Body.String())
+	}
+	if len(scrobbler.calls) != 1 || scrobbler.calls[0].action != "stop" {
+		t.Fatalf("scrobble calls = %+v, want one stop", scrobbler.calls)
+	}
+	event := scrobbler.calls[0].event
+	if event.PlaybackSessionID != "upstream-reaped" || event.UserID != 7 || event.ProfileID != "profile-1" {
+		t.Fatalf("stop scope = %+v", event)
+	}
+	if event.MediaItemID != "movie-1" || event.PositionSeconds != 90 || event.DurationSeconds != 3600 {
+		t.Fatalf("stop progress = %+v", event)
+	}
+	if _, ok := handler.playbackStore.GetFinalizable("play-1", "token-1"); ok {
+		t.Fatal("stopped compat session should be consumed")
+	}
+}
+
+func TestTeardownStillCleansLocalPlaybackAfterAnotherCallerClaimsStop(t *testing.T) {
+	mgr := &testCompatSessionManager{sessions: map[string]*playback.Session{
+		"upstream-1": {ID: "upstream-1", UserID: 7, ProfileID: "profile-1", MediaFileID: 42},
+	}}
+	h, store := newActiveEncodingsHandler(mgr)
+	scrobbler := &recordingCompatWatchScrobbler{}
+	h.WatchScrobbler = scrobbler
+	store.Put(PlaybackSession{
+		ID:                "play-1",
+		CompatToken:       "token-1",
+		ItemID:            "movie-1",
+		UpstreamSessionID: "upstream-1",
+		MediaSources:      []PlaybackMediaSource{{ID: "source-1", FileID: 42, Version: testCompatVersion()}},
+	})
+	candidate, ok := store.Get("play-1")
+	if !ok {
+		t.Fatal("playback session missing")
+	}
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1", UserID: 7, ProfileID: "profile-1"}
+	if _, err := store.StageTerminal("play-1", "token-1", event, true); err != nil {
+		t.Fatal("failed to stage competing terminal event")
+	}
+	if _, err := store.ClaimTerminal("play-1", "token-1", time.Now().Add(compatTerminalClaimLease)); err != nil {
+		t.Fatal("failed to simulate a competing terminal delivery claim")
+	}
+
+	h.teardownPlaySession(context.Background(), candidate, nil, nil)
+
+	if _, err := mgr.GetSession("upstream-1"); err == nil {
+		t.Fatal("local upstream session was not cleaned up after losing the terminal claim")
+	}
+	if len(scrobbler.calls) != 0 {
+		t.Fatalf("losing teardown emitted provider event: %+v", scrobbler.calls)
+	}
+}

internal/jellycompat/playback_sessions.go

@@ -1,13 +1,21 @@
 package jellycompat
 
 import (
+	"context"
+	"errors"
+	"sort"
 	"sync"
 	"time"
 
 	"github.com/Silo-Server/silo-server/internal/catalog"
 	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
 )
 
+// ErrTerminalClaimUnavailable means a staged terminal event still exists but
+// another process owns its delivery lease (or already sent its fallback).
+var ErrTerminalClaimUnavailable = errors.New("compat terminal event claim unavailable")
+
 // PlaybackSession stores compat-owned playback negotiation state before the
 // native Silo playback session starts.
 type PlaybackSession struct {
@@ -19,13 +27,25 @@ type PlaybackSession struct {
 	// when it differs from ours (Static=true direct play skips PlaybackInfo,
 	// so the client never learns the server id). Playback reports carrying
 	// that id resolve to this session directly instead of by ambiguous route.
-	ClientPlaySessionID string
-	UserID              string
-	InitialSeekSeconds  float64
-	MediaSources        []PlaybackMediaSource
-	UpstreamSessionID   string
-	UpstreamPlayMethod  string
-	TranscodeStarted    bool
+	ClientPlaySessionID        string
+	UserID                     string
+	InitialSeekSeconds         float64
+	MediaSources               []PlaybackMediaSource
+	UpstreamSessionID          string
+	UpstreamPlayMethod         string
+	TranscodeStarted           bool
+	ProgressPersistenceKnown   bool
+	DisableProgressPersistence bool
+	// Terminal hides a play session from stream and progress routing after
+	// ActiveEncodings cleanup while retaining the authenticated mapping long
+	// enough for a later Stopped report to publish its authoritative position.
+	Terminal              bool
+	TerminalAuthoritative bool
+	TerminalFallbackSent  bool
+	TerminalClaimUntil    time.Time
+	TerminalEventVersion  int64
+	TerminalClaimVersion  int64
+	TerminalScrobbleEvent *watchsync.ScrobbleEvent
 	// Recipe is the transcode reconstruction descriptor for this session. Jellyfin
 	// clients cannot round-trip a native stream token, so jellycompat carries the
 	// recipe in its own durable compat store (this struct, persisted as JSONB)
@@ -65,6 +85,27 @@ type CompatPlaybackStore interface {
 	Get(id string) (*PlaybackSession, bool)
 	// Delete removes a session.
 	Delete(id string)
+	// HideFromRouting immediately makes a caller-owned session unavailable to
+	// stream/progress routing before slower durable terminal staging begins.
+	HideFromRouting(id, compatToken string) error
+	// StageTerminal hides a session from playback routing and durably records
+	// the provider event. An authoritative Stopped event replaces a fallback;
+	// a later fallback can never replace an authoritative event.
+	StageTerminal(id, compatToken string, event watchsync.ScrobbleEvent, authoritative bool) (*PlaybackSession, error)
+	// ClaimTerminal leases one staged event for delivery across server processes.
+	ClaimTerminal(id, compatToken string, claimUntil time.Time) (*PlaybackSession, error)
+	// ReleaseTerminalClaim releases an exact lease after delivery failure, or
+	// records a delivered fallback while retaining the row for a later Stopped.
+	ReleaseTerminalClaim(id, compatToken string, claimUntil time.Time, claimVersion int64, fallbackSent bool)
+	// CompleteTerminal deletes an authoritatively delivered terminal row only
+	// when the caller still owns the exact lease.
+	CompleteTerminal(id, compatToken string, claimUntil time.Time, claimVersion int64)
+	// ListPendingTerminals returns retryable authoritative events and unsent
+	// fallbacks for startup/periodic delivery recovery.
+	ListPendingTerminals(ctx context.Context, limit int) ([]PlaybackSession, error)
+	// GetFinalizable reads an active or terminal caller-owned session for report
+	// validation before an atomic Take.
+	GetFinalizable(id, compatToken string) (*PlaybackSession, bool)
 	// Update modifies a session in place under the store's lock.
 	Update(id string, fn func(*PlaybackSession) error) error
 	// FindByRoute resolves a route item / media-source id to a session.
@@ -73,6 +114,18 @@ type CompatPlaybackStore interface {
 	// alias recorded for plays that skipped PlaybackInfo. The alias must
 	// identify exactly one live session; ambiguity returns not-found.
 	FindByClientPlaySessionID(compatToken, clientPlaySessionID string) (*PlaybackSession, bool)
+	// FindFinalizableByClientPlaySessionID is the stop-report variant of alias
+	// lookup and includes terminal sessions retained by Deactivate. The report
+	// identifiers disambiguate clients that reuse an alias across plays.
+	FindFinalizableByClientPlaySessionID(
+		compatToken, clientPlaySessionID, routeItemID, mediaSourceID string,
+	) (*PlaybackSession, bool)
+	// FindFinalizableByRoute resolves exactly one active or terminal session for
+	// a caller-owned route. Ambiguous matches return not-found.
+	FindFinalizableByRoute(compatToken, routeID string) (*PlaybackSession, *PlaybackMediaSource, bool)
+	// FindByUpstreamSessionID resolves the local upstream session that owns a
+	// compat play. It is used for process-local failure lifecycle handling.
+	FindByUpstreamSessionID(upstreamSessionID string) (*PlaybackSession, bool)
 }
 
 // PlaybackSessionStore keeps compat playback sessions in memory. It is the
@@ -82,6 +135,9 @@ type PlaybackSessionStore struct {
 	sessions map[string]PlaybackSession
 	ttl      time.Duration
 	now      func() time.Time
+	// pendingCursor rotates bounded recovery scans through the full queue so a
+	// permanently failing first batch cannot starve later terminal events.
+	pendingCursor string
 }
 
 // NewPlaybackSessionStore creates a new playback session store.
@@ -140,6 +196,9 @@ func (s *PlaybackSessionStore) Get(id string) (*PlaybackSession, bool) {
 		s.Delete(id)
 		return nil, false
 	}
+	if session.Terminal {
+		return nil, false
+	}
 	cp := session
 	return &cp, true
 }
@@ -151,13 +210,254 @@ func (s *PlaybackSessionStore) Delete(id string) {
 	delete(s.sessions, id)
 }
 
+func (s *PlaybackSessionStore) compatTokenForID(id string) string {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.sessions[id].CompatToken
+}
+
+// HideFromRouting marks a session terminal without requiring its provider event
+// to have been staged yet. Final-report lookups remain available for retries.
+func (s *PlaybackSessionStore) HideFromRouting(id, compatToken string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken || !session.ExpiresAt.After(s.now()) {
+		return ErrSessionNotFound
+	}
+	session.Terminal = true
+	session.UpdatedAt = s.now()
+	s.sessions[id] = session
+	return nil
+}
+
+func (s *PlaybackSessionStore) terminalIDsByCompatToken(compatToken string) map[string]struct{} {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	result := make(map[string]struct{})
+	now := s.now()
+	for id, session := range s.sessions {
+		if !session.ExpiresAt.After(now) {
+			delete(s.sessions, id)
+			continue
+		}
+		if session.CompatToken == compatToken && session.Terminal {
+			result[id] = struct{}{}
+		}
+	}
+	return result
+}
+
+func (s *PlaybackSessionStore) replaceByCompatToken(
+	compatToken string,
+	replacements []PlaybackSession,
+	preserveIDs map[string]struct{},
+) []string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	affected := make(map[string]struct{}, len(replacements))
+	for id, session := range s.sessions {
+		_, preserve := preserveIDs[id]
+		if session.CompatToken == compatToken && !preserve {
+			delete(s.sessions, id)
+			affected[id] = struct{}{}
+		}
+	}
+	for _, session := range replacements {
+		if _, preserve := preserveIDs[session.ID]; preserve {
+			continue
+		}
+		if session.CreatedAt.IsZero() {
+			session.CreatedAt = s.now()
+		}
+		if session.ExpiresAt.IsZero() {
+			session.ExpiresAt = session.CreatedAt.Add(s.ttl)
+		}
+		s.sessions[session.ID] = session
+		affected[session.ID] = struct{}{}
+	}
+	result := make([]string, 0, len(affected))
+	for id := range affected {
+		result = append(result, id)
+	}
+	return result
+}
+
+// StageTerminal hides a playback session and records the event that must reach
+// watch providers. Authoritative final reports replace provisional cleanup
+// events; provisional events never overwrite an authoritative report.
+func (s *PlaybackSessionStore) StageTerminal(
+	id string,
+	compatToken string,
+	event watchsync.ScrobbleEvent,
+	authoritative bool,
+) (*PlaybackSession, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken {
+		return nil, ErrSessionNotFound
+	}
+	if !session.ExpiresAt.After(s.now()) {
+		delete(s.sessions, id)
+		return nil, ErrSessionNotFound
+	}
+	if !session.TerminalAuthoritative || authoritative {
+		eventCopy := event
+		session.TerminalScrobbleEvent = &eventCopy
+		session.TerminalAuthoritative = authoritative
+		session.TerminalEventVersion++
+	}
+	session.Terminal = true
+	session.UpdatedAt = s.now()
+	s.sessions[id] = session
+	return &session, nil
+}
+
+// ClaimTerminal leases one staged event. Expired leases may be reclaimed after
+// a process dies; a delivered fallback is skipped unless Stopped subsequently
+// staged an authoritative replacement.
+func (s *PlaybackSessionStore) ClaimTerminal(id, compatToken string, claimUntil time.Time) (*PlaybackSession, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken || !session.Terminal || session.TerminalScrobbleEvent == nil {
+		return nil, ErrSessionNotFound
+	}
+	if !session.ExpiresAt.After(s.now()) {
+		delete(s.sessions, id)
+		return nil, ErrSessionNotFound
+	}
+	if session.TerminalClaimUntil.After(s.now()) || (session.TerminalFallbackSent && !session.TerminalAuthoritative) {
+		return nil, ErrTerminalClaimUnavailable
+	}
+	session.TerminalClaimUntil = claimUntil
+	session.TerminalClaimVersion = session.TerminalEventVersion
+	session.UpdatedAt = s.now()
+	s.sessions[id] = session
+	return &session, nil
+}
+
+// ReleaseTerminalClaim releases an exact delivery lease. A stale caller cannot
+// clear a successor's lease after its own lease expires.
+func (s *PlaybackSessionStore) ReleaseTerminalClaim(
+	id string,
+	compatToken string,
+	claimUntil time.Time,
+	claimVersion int64,
+	fallbackSent bool,
+) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken || !session.TerminalClaimUntil.Equal(claimUntil) ||
+		session.TerminalClaimVersion != claimVersion {
+		return
+	}
+	session.TerminalClaimUntil = time.Time{}
+	session.TerminalClaimVersion = 0
+	if fallbackSent {
+		session.TerminalFallbackSent = true
+	}
+	session.UpdatedAt = s.now()
+	s.sessions[id] = session
+}
+
+// CompleteTerminal removes an authoritatively delivered event while protecting
+// a newer lease from a stale completion.
+func (s *PlaybackSessionStore) CompleteTerminal(id, compatToken string, claimUntil time.Time, claimVersion int64) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken || !session.TerminalAuthoritative ||
+		!session.TerminalClaimUntil.Equal(claimUntil) || session.TerminalClaimVersion != claimVersion ||
+		session.TerminalEventVersion != claimVersion {
+		return
+	}
+	delete(s.sessions, id)
+}
+
+// ListPendingTerminals returns staged events that still need delivery. A
+// successfully sent fallback remains retained for a possible authoritative
+// replacement but is not itself pending.
+func (s *PlaybackSessionStore) ListPendingTerminals(_ context.Context, limit int) ([]PlaybackSession, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if limit <= 0 {
+		limit = 100
+	}
+	now := s.now()
+	eligible := make([]PlaybackSession, 0, len(s.sessions))
+	for id, session := range s.sessions {
+		if !session.ExpiresAt.After(now) {
+			delete(s.sessions, id)
+			continue
+		}
+		if !session.Terminal || session.TerminalScrobbleEvent == nil ||
+			(session.TerminalFallbackSent && !session.TerminalAuthoritative) {
+			continue
+		}
+		eligible = append(eligible, session)
+	}
+	sort.Slice(eligible, func(i, j int) bool { return eligible[i].ID < eligible[j].ID })
+	start := sort.Search(len(eligible), func(i int) bool { return eligible[i].ID > s.pendingCursor })
+	if start == len(eligible) && s.pendingCursor != "" {
+		start = 0
+		s.pendingCursor = ""
+	}
+	end := min(start+limit, len(eligible))
+	result := append([]PlaybackSession(nil), eligible[start:end]...)
+	if len(result) > 0 {
+		s.pendingCursor = result[len(result)-1].ID
+	} else {
+		s.pendingCursor = ""
+	}
+	return result, nil
+}
+
+func (s *PlaybackSessionStore) deleteExpired() map[string]string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	now := s.now()
+	removed := make(map[string]string)
+	for id, session := range s.sessions {
+		if !session.ExpiresAt.After(now) {
+			removed[id] = session.CompatToken
+			delete(s.sessions, id)
+		}
+	}
+	return removed
+}
+
+// GetFinalizable returns an active or terminal caller-owned session so a stop
+// report can validate its media fields before atomically consuming it.
+func (s *PlaybackSessionStore) GetFinalizable(id, compatToken string) (*PlaybackSession, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	session, ok := s.sessions[id]
+	if !ok || session.CompatToken != compatToken {
+		return nil, false
+	}
+	if !session.ExpiresAt.After(s.now()) {
+		delete(s.sessions, id)
+		return nil, false
+	}
+	copy := session
+	return &copy, true
+}
+
 // Update modifies a playback session in place.
 func (s *PlaybackSessionStore) Update(id string, fn func(*PlaybackSession) error) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
 	session, ok := s.sessions[id]
-	if !ok {
+	if !ok || session.Terminal {
 		return ErrSessionNotFound
 	}
 	if !session.ExpiresAt.After(s.now()) {
@@ -178,6 +478,27 @@ func (s *PlaybackSessionStore) Update(id string, fn func(*PlaybackSession) error
 // PlaySessionId across plays makes the alias ambiguous, and the caller should
 // fall back to route matching instead of binding an arbitrary session.
 func (s *PlaybackSessionStore) FindByClientPlaySessionID(compatToken, clientPlaySessionID string) (*PlaybackSession, bool) {
+	return s.findByClientPlaySessionID(compatToken, clientPlaySessionID, "", "", false)
+}
+
+// FindFinalizableByClientPlaySessionID includes terminal sessions retained for
+// an authoritative final report. Route identifiers narrow reused client aliases
+// to the play described by that report before the uniqueness check runs.
+func (s *PlaybackSessionStore) FindFinalizableByClientPlaySessionID(
+	compatToken, clientPlaySessionID, routeItemID, mediaSourceID string,
+) (*PlaybackSession, bool) {
+	return s.findByClientPlaySessionID(
+		compatToken, clientPlaySessionID, routeItemID, mediaSourceID, true,
+	)
+}
+
+func (s *PlaybackSessionStore) findByClientPlaySessionID(
+	compatToken string,
+	clientPlaySessionID string,
+	routeItemID string,
+	mediaSourceID string,
+	includeTerminal bool,
+) (*PlaybackSession, bool) {
 	if clientPlaySessionID == "" {
 		return nil, false
 	}
@@ -187,12 +508,18 @@ func (s *PlaybackSessionStore) FindByClientPlaySessionID(compatToken, clientPlay
 	now := s.now()
 	var match *PlaybackSession
 	for _, session := range s.sessions {
-		if !session.ExpiresAt.After(now) {
+		if !session.ExpiresAt.After(now) || (!includeTerminal && session.Terminal) {
 			continue
 		}
 		if session.CompatToken != compatToken {
 			continue
 		}
+		if routeItemID != "" && !mediaSourceIDsEqual(session.RouteItemID, routeItemID) {
+			continue
+		}
+		if mediaSourceID != "" && findMediaSource(&session, mediaSourceID) == nil {
+			continue
+		}
 		if session.ClientPlaySessionID == clientPlaySessionID {
 			if match != nil {
 				return nil, false
@@ -204,14 +531,50 @@ func (s *PlaybackSessionStore) FindByClientPlaySessionID(compatToken, clientPlay
 	return match, match != nil
 }
 
-// FindByRoute resolves a route item/media-source identifier to a compat playback session.
-func (s *PlaybackSessionStore) FindByRoute(compatToken, routeID string) (*PlaybackSession, *PlaybackMediaSource, bool) {
+// FindByUpstreamSessionID resolves the unique compat play attached to a local
+// upstream session.
+func (s *PlaybackSessionStore) FindByUpstreamSessionID(upstreamSessionID string) (*PlaybackSession, bool) {
+	if upstreamSessionID == "" {
+		return nil, false
+	}
 	s.mu.RLock()
 	defer s.mu.RUnlock()
 
 	now := s.now()
 	for _, session := range s.sessions {
-		if !session.ExpiresAt.After(now) {
+		if session.ExpiresAt.After(now) && !session.Terminal && session.UpstreamSessionID == upstreamSessionID {
+			copy := session
+			return &copy, true
+		}
+	}
+	return nil, false
+}
+
+// FindByRoute resolves a route item/media-source identifier to a compat playback session.
+func (s *PlaybackSessionStore) FindByRoute(compatToken, routeID string) (*PlaybackSession, *PlaybackMediaSource, bool) {
+	return s.findByRoute(compatToken, routeID, false, false)
+}
+
+// FindFinalizableByRoute includes terminal sessions retained for a stopped
+// report, but only returns a unique token-scoped match.
+func (s *PlaybackSessionStore) FindFinalizableByRoute(
+	compatToken, routeID string,
+) (*PlaybackSession, *PlaybackMediaSource, bool) {
+	return s.findByRoute(compatToken, routeID, true, true)
+}
+
+func (s *PlaybackSessionStore) findByRoute(
+	compatToken, routeID string,
+	includeTerminal, requireUnique bool,
+) (*PlaybackSession, *PlaybackMediaSource, bool) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	now := s.now()
+	var matchedSession *PlaybackSession
+	var matchedSource *PlaybackMediaSource
+	for _, session := range s.sessions {
+		if !session.ExpiresAt.After(now) || (!includeTerminal && session.Terminal) {
 			continue
 		}
 		if compatToken != "" && session.CompatToken != compatToken {
@@ -221,17 +584,33 @@ func (s *PlaybackSessionStore) FindByRoute(compatToken, routeID string) (*Playba
 		// whatever casing/dash format the client model uses, which may differ
 		// from the raw route param captured at stream time.
 		if mediaSourceIDsEqual(session.RouteItemID, routeID) {
+			if requireUnique && matchedSession != nil {
+				return nil, nil, false
+			}
 			cp := session
-			return &cp, nil, true
+			matchedSession = &cp
+			matchedSource = nil
+			if !requireUnique {
+				return matchedSession, nil, true
+			}
+			continue
 		}
 		for _, source := range session.MediaSources {
 			if mediaSourceIDsEqual(source.ID, routeID) {
+				if requireUnique && matchedSession != nil {
+					return nil, nil, false
+				}
 				cp := session
 				sourceCopy := source
-				return &cp, &sourceCopy, true
+				matchedSession = &cp
+				matchedSource = &sourceCopy
+				if !requireUnique {
+					return matchedSession, matchedSource, true
+				}
+				break
 			}
 		}
 	}
 
-	return nil, nil, false
+	return matchedSession, matchedSource, matchedSession != nil
 }

internal/jellycompat/playback_sessions_postgres_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Silo-Server/silo-server/internal/watchsync"
 	"github.com/jackc/pgx/v5/pgxpool"
 )
 
@@ -216,3 +217,668 @@ func TestDurableCompatPlaybackStore_NilPoolInMemory(t *testing.T) {
 		t.Fatal("nil-pool Delete failed")
 	}
 }
+
+func TestPlaybackSessionStoreTerminalClaimIsAtomicAndOwnerScoped(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+
+	if _, err := store.StageTerminal("play-1", "other", event, true); err == nil {
+		t.Fatal("foreign token staged terminal event")
+	}
+	if _, err := store.StageTerminal("play-1", "owner", event, true); err != nil {
+		t.Fatal("owner failed to stage terminal event")
+	}
+	claimUntil := time.Now().Add(time.Minute)
+	if _, err := store.ClaimTerminal("play-1", "other", claimUntil); err == nil {
+		t.Fatal("foreign token claimed terminal event")
+	}
+	got, err := store.ClaimTerminal("play-1", "owner", claimUntil)
+	if err != nil || got.UpstreamSessionID != "upstream-1" {
+		t.Fatalf("owner claim failed: err=%v session=%+v", err, got)
+	}
+	if _, err := store.ClaimTerminal("play-1", "owner", claimUntil.Add(time.Minute)); err == nil {
+		t.Fatal("terminal event was claimed twice")
+	}
+}
+
+func TestPlaybackSessionStoreDeactivateRetainsOnlyFinalReportLookup(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{
+		ID:                  "play-1",
+		CompatToken:         "owner",
+		ClientPlaySessionID: "client-play-1",
+		UpstreamSessionID:   "upstream-1",
+	})
+
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+	terminal, err := store.StageTerminal("play-1", "owner", event, true)
+	if err != nil || !terminal.Terminal {
+		t.Fatalf("terminal stage failed: err=%v session=%+v", err, terminal)
+	}
+	if _, ok := store.Get("play-1"); ok {
+		t.Fatal("terminal session remained available to ordinary lookup")
+	}
+	if _, ok := store.FindByClientPlaySessionID("owner", "client-play-1"); ok {
+		t.Fatal("terminal session remained available to ordinary alias lookup")
+	}
+	if _, ok := store.GetFinalizable("play-1", "owner"); !ok {
+		t.Fatal("terminal session was unavailable to final report lookup")
+	}
+	if _, ok := store.FindFinalizableByClientPlaySessionID("owner", "client-play-1", "", ""); !ok {
+		t.Fatal("terminal session was unavailable to final alias lookup")
+	}
+	claimUntil := time.Now().Add(time.Minute)
+	claimed, err := store.ClaimTerminal("play-1", "owner", claimUntil)
+	if err != nil || !claimed.Terminal {
+		t.Fatalf("final report could not claim terminal session: err=%v session=%+v", err, claimed)
+	}
+	store.CompleteTerminal("play-1", "owner", claimUntil, claimed.TerminalClaimVersion)
+	if _, ok := store.GetFinalizable("play-1", "owner"); ok {
+		t.Fatal("authoritatively completed terminal session was retained")
+	}
+}
+
+func TestPlaybackSessionStoreFinalAliasLookupDisambiguatesReusedAlias(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{
+		ID:                  "old-play",
+		CompatToken:         "owner",
+		ClientPlaySessionID: "reused-client-play",
+		RouteItemID:         "old-item",
+		MediaSources:        []PlaybackMediaSource{{ID: "old-source"}},
+		UpstreamSessionID:   "old-upstream",
+	})
+	if _, err := store.StageTerminal(
+		"old-play",
+		"owner",
+		watchsync.ScrobbleEvent{PlaybackSessionID: "old-upstream"},
+		false,
+	); err != nil {
+		t.Fatalf("stage old terminal play: %v", err)
+	}
+	store.Put(PlaybackSession{
+		ID:                  "current-play",
+		CompatToken:         "owner",
+		ClientPlaySessionID: "reused-client-play",
+		RouteItemID:         "current-item",
+		MediaSources:        []PlaybackMediaSource{{ID: "current-source"}},
+		UpstreamSessionID:   "current-upstream",
+	})
+
+	if _, ok := store.FindFinalizableByClientPlaySessionID(
+		"owner", "reused-client-play", "", "",
+	); ok {
+		t.Fatal("unscoped reused alias unexpectedly selected an arbitrary play")
+	}
+	current, ok := store.FindFinalizableByClientPlaySessionID(
+		"owner", "reused-client-play", "current-item", "current-source",
+	)
+	if !ok || current.ID != "current-play" {
+		t.Fatalf("current report resolved to ok=%v session=%+v", ok, current)
+	}
+	old, ok := store.FindFinalizableByClientPlaySessionID(
+		"owner", "reused-client-play", "old-item", "old-source",
+	)
+	if !ok || old.ID != "old-play" {
+		t.Fatalf("late old report resolved to ok=%v session=%+v", ok, old)
+	}
+	oldByRoute, _, ok := store.FindFinalizableByRoute("owner", "old-source")
+	if !ok || oldByRoute.ID != "old-play" {
+		t.Fatalf("terminal route resolved to ok=%v session=%+v", ok, oldByRoute)
+	}
+}
+
+func TestPlaybackSessionStoreExpiredTerminalLeaseCanBeReclaimed(t *testing.T) {
+	now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC)
+	store := NewPlaybackSessionStore(time.Hour, func() time.Time { return now })
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "owner"})
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+	if _, err := store.StageTerminal("play-1", "owner", event, true); err != nil {
+		t.Fatalf("stage terminal: %v", err)
+	}
+	firstLease := now.Add(10 * time.Second)
+	firstClaim, err := store.ClaimTerminal("play-1", "owner", firstLease)
+	if err != nil {
+		t.Fatalf("first claim: %v", err)
+	}
+	now = firstLease.Add(time.Microsecond)
+	secondLease := now.Add(10 * time.Second)
+	secondClaim, err := store.ClaimTerminal("play-1", "owner", secondLease)
+	if err != nil {
+		t.Fatalf("reclaim expired lease: %v", err)
+	}
+	store.CompleteTerminal("play-1", "owner", firstLease, firstClaim.TerminalClaimVersion)
+	if _, ok := store.GetFinalizable("play-1", "owner"); !ok {
+		t.Fatal("stale first lease completed the successor's terminal row")
+	}
+	store.CompleteTerminal("play-1", "owner", secondLease, secondClaim.TerminalClaimVersion)
+	if _, ok := store.GetFinalizable("play-1", "owner"); ok {
+		t.Fatal("successor lease did not complete terminal row")
+	}
+}
+
+func TestPlaybackSessionStorePendingScanEvictsExpiredTerminal(t *testing.T) {
+	now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC)
+	store := NewPlaybackSessionStore(time.Minute, func() time.Time { return now })
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "owner"})
+	if _, err := store.StageTerminal(
+		"play-1", "owner", watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}, false,
+	); err != nil {
+		t.Fatalf("stage terminal: %v", err)
+	}
+	now = now.Add(2 * time.Minute)
+	pending, err := store.ListPendingTerminals(context.Background(), 100)
+	if err != nil {
+		t.Fatalf("list pending terminals: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("expired pending terminals = %d, want 0", len(pending))
+	}
+	store.mu.RLock()
+	remaining := len(store.sessions)
+	store.mu.RUnlock()
+	if remaining != 0 {
+		t.Fatalf("expired terminal entries retained in memory = %d", remaining)
+	}
+}
+
+func TestDurableCompatPlaybackStoreExpiryClearsFailureBookkeeping(t *testing.T) {
+	now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC)
+	store := NewDurableCompatPlaybackStore(nil, time.Minute, func() time.Time { return now })
+	store.mem.Put(PlaybackSession{ID: "play-1", CompatToken: "owner"})
+	store.markUnpersisted("play-1")
+	store.appendPendingUpdate("play-1", "owner", func(*PlaybackSession) error { return nil })
+	now = now.Add(2 * time.Minute)
+
+	if _, err := store.DeleteExpired(context.Background()); err != nil {
+		t.Fatalf("delete expired: %v", err)
+	}
+	if store.isUnpersisted("play-1") {
+		t.Fatal("expired session retained its unpersisted marker")
+	}
+	if store.hasPendingUpdates("play-1") {
+		t.Fatal("expired session retained pending update closures")
+	}
+}
+
+func TestDurableCompatPlaybackStoreBoundsGenerationTombstones(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	before := store.idGenerationSnapshot("old-session")
+	for i := 0; i <= compatValidationCacheLimit; i++ {
+		store.bumpCacheGenerations(fmt.Sprintf("missing-%d", i), "")
+	}
+	store.generationMu.Lock()
+	count := len(store.idGenerations)
+	epoch := store.generationEpoch
+	store.generationMu.Unlock()
+	if count > compatValidationCacheLimit {
+		t.Fatalf("generation tombstones = %d, limit = %d", count, compatValidationCacheLimit)
+	}
+	if epoch == 0 || before == store.idGenerationSnapshot("old-session") {
+		t.Fatal("generation eviction did not invalidate an older captured stamp")
+	}
+}
+
+func TestDurableCompatPlaybackStoreIDWriteDoesNotRefreshTokenSnapshot(t *testing.T) {
+	now := time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC)
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, func() time.Time { return now })
+	store.markTokenValidated("owner")
+	now = now.Add(4 * time.Second)
+	store.markIDValidated("play-2")
+	now = now.Add(2 * time.Second)
+	if !store.shouldRevalidateToken("owner") {
+		t.Fatal("single-row validation incorrectly extended the token-wide cache window")
+	}
+}
+
+func TestPlaybackSessionStoreNewAuthoritativeEventSurvivesStaleCompletion(t *testing.T) {
+	store := NewPlaybackSessionStore(time.Hour, nil)
+	store.Put(PlaybackSession{ID: "play-1", CompatToken: "owner"})
+	firstEvent := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1", PositionSeconds: 45}
+	if _, err := store.StageTerminal("play-1", "owner", firstEvent, true); err != nil {
+		t.Fatalf("stage first event: %v", err)
+	}
+	firstLease := time.Now().Add(time.Minute)
+	firstClaim, err := store.ClaimTerminal("play-1", "owner", firstLease)
+	if err != nil {
+		t.Fatalf("claim first event: %v", err)
+	}
+	secondEvent := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1", PositionSeconds: 90}
+	if _, err := store.StageTerminal("play-1", "owner", secondEvent, true); err != nil {
+		t.Fatalf("stage replacement event: %v", err)
+	}
+
+	store.CompleteTerminal("play-1", "owner", firstLease, firstClaim.TerminalClaimVersion)
+	store.ReleaseTerminalClaim("play-1", "owner", firstLease, firstClaim.TerminalClaimVersion, false)
+	pending, ok := store.GetFinalizable("play-1", "owner")
+	if !ok || pending.TerminalScrobbleEvent == nil || pending.TerminalScrobbleEvent.PositionSeconds != 90 {
+		t.Fatalf("replacement event was lost: ok=%v session=%+v", ok, pending)
+	}
+	secondLease := firstLease.Add(time.Minute)
+	secondClaim, err := store.ClaimTerminal("play-1", "owner", secondLease)
+	if err != nil {
+		t.Fatalf("claim replacement event: %v", err)
+	}
+	store.CompleteTerminal("play-1", "owner", secondLease, secondClaim.TerminalClaimVersion)
+	if _, ok := store.GetFinalizable("play-1", "owner"); ok {
+		t.Fatal("replacement event was not completed")
+	}
+}
+
+func TestDurableCompatPlaybackStoreTerminalClaimAcrossInstances(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-take-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+
+	seed := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	seed.Put(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+	skewedNow := time.Now().Add(-6 * time.Hour)
+	first := NewDurableCompatPlaybackStore(pool, time.Hour, func() time.Time { return skewedNow })
+	second := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	if _, ok := first.Get(id); !ok {
+		t.Fatal("first instance did not load session")
+	}
+	if _, ok := second.Get(id); !ok {
+		t.Fatal("second instance did not load session")
+	}
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+	if _, err := seed.StageTerminal(id, "owner", event, true); err != nil {
+		t.Fatal("failed to stage durable terminal event")
+	}
+
+	var dbBefore time.Time
+	if err := pool.QueryRow(ctx, `SELECT clock_timestamp()`).Scan(&dbBefore); err != nil {
+		t.Fatalf("read database clock: %v", err)
+	}
+	claimUntil := skewedNow.Add(time.Minute)
+	claimed, err := first.ClaimTerminal(id, "owner", claimUntil)
+	if err != nil {
+		t.Fatal("first instance did not claim session")
+	}
+	var dbAfter time.Time
+	if err := pool.QueryRow(ctx, `SELECT clock_timestamp()`).Scan(&dbAfter); err != nil {
+		t.Fatalf("read database clock after claim: %v", err)
+	}
+	if claimed.TerminalClaimUntil.Before(dbBefore.Add(55*time.Second)) ||
+		claimed.TerminalClaimUntil.After(dbAfter.Add(65*time.Second)) {
+		t.Fatalf(
+			"claim deadline %s was not anchored to database clock interval [%s, %s]",
+			claimed.TerminalClaimUntil,
+			dbBefore.Add(55*time.Second),
+			dbAfter.Add(65*time.Second),
+		)
+	}
+	if _, err := second.ClaimTerminal(id, "owner", time.Now().Add(time.Minute)); err == nil {
+		t.Fatal("second instance claimed an already leased durable session")
+	}
+}
+
+func TestDurableCompatPlaybackStoreTerminalStageSurvivesInstanceBoundary(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-deactivate-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	now := time.Now()
+	clock := func() time.Time { return now }
+
+	seed := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	seed.Put(PlaybackSession{
+		ID:                  id,
+		CompatToken:         "owner",
+		ClientPlaySessionID: "client-play-1",
+		UpstreamSessionID:   "upstream-1",
+	})
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	if _, ok := fresh.Get(id); !ok {
+		t.Fatal("fresh instance did not preload the active session")
+	}
+	if _, ok := fresh.FindByClientPlaySessionID("owner", "client-play-1"); !ok {
+		t.Fatal("fresh instance did not preload the active alias")
+	}
+	event := watchsync.ScrobbleEvent{PlaybackSessionID: "upstream-1"}
+	if terminal, err := seed.StageTerminal(id, "owner", event, false); err != nil || !terminal.Terminal {
+		t.Fatalf("durable terminal stage failed: err=%v session=%+v", err, terminal)
+	}
+	// Cross-process invalidation is deliberately bounded rather than putting a
+	// DB query on every segment request. Advance past that window before the
+	// other instance revalidates its preloaded cache.
+	now = now.Add(compatCacheRevalidationInterval)
+
+	if _, ok := fresh.Get(id); ok {
+		t.Fatal("instance routed a terminal session from its stale active cache")
+	}
+	if _, ok := fresh.FindByClientPlaySessionID("owner", "client-play-1"); ok {
+		t.Fatal("instance routed a terminal alias from its stale active cache")
+	}
+	if _, ok := fresh.GetFinalizable(id, "owner"); !ok {
+		t.Fatal("fresh instance could not resolve terminal session for final report")
+	}
+	if _, ok := fresh.FindFinalizableByClientPlaySessionID("owner", "client-play-1", "", ""); !ok {
+		t.Fatal("fresh instance could not resolve terminal alias for final report")
+	}
+	claimUntil := time.Now().UTC().Truncate(time.Microsecond).Add(time.Minute)
+	claimed, err := fresh.ClaimTerminal(id, "owner", claimUntil)
+	if err != nil || !claimed.Terminal {
+		t.Fatalf("fresh instance could not claim terminal session: err=%v session=%+v", err, claimed)
+	}
+	fresh.ReleaseTerminalClaim(id, "owner", claimed.TerminalClaimUntil, claimed.TerminalClaimVersion, true)
+	if claimed, err := fresh.ClaimTerminal(id, "owner", claimUntil.Add(time.Minute)); err == nil || claimed != nil {
+		t.Fatalf("delivered fallback was claimed again: err=%v session=%+v", err, claimed)
+	}
+}
+
+func TestDurableCompatPlaybackStorePreservesCachedSessionOnValidationFailure(t *testing.T) {
+	pool := newCompatTestPool(t)
+	now := time.Now()
+	clock := func() time.Time { return now }
+	store := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	pool.Close()
+	store.Put(PlaybackSession{ID: "cached-session", CompatToken: "owner"})
+
+	// A request inside the validation window must be served entirely from cache.
+	if _, ok := store.Get("cached-session"); !ok {
+		t.Fatal("hot cache lookup consulted the closed database")
+	}
+
+	// Once validation is due, the failed query must not evict a last-known-good
+	// active session and interrupt playback.
+	now = now.Add(compatCacheRevalidationInterval)
+	if _, ok := store.Get("cached-session"); !ok {
+		t.Fatal("database validation failure evicted the cached active session")
+	}
+}
+
+func TestDurableCompatPlaybackStoreRepairsFailedInitialPersistence(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-repair-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	now := time.Now()
+	clock := func() time.Time { return now }
+	store := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	stored := store.mem.putNormalized(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+	store.markIDValidated(id)
+	store.markUnpersisted(id)
+
+	now = now.Add(compatCacheRevalidationInterval)
+	if got, ok := store.Get(id); !ok || got.UpstreamSessionID != "upstream-1" {
+		t.Fatalf("unpersisted cache entry was not retained and repaired: ok=%v session=%+v", ok, got)
+	}
+	if store.isUnpersisted(id) {
+		t.Fatal("successfully repaired session remained marked unpersisted")
+	}
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	if got, ok := fresh.Get(stored.ID); !ok || got.UpstreamSessionID != "upstream-1" {
+		t.Fatalf("repaired session was not durable: ok=%v session=%+v", ok, got)
+	}
+}
+
+func TestDurableCompatPlaybackStorePreservesUnpersistedTerminalOnRevalidation(t *testing.T) {
+	pool := newCompatTestPool(t)
+	id := fmt.Sprintf("compat-unpersisted-terminal-%d", time.Now().UnixNano())
+	now := time.Now()
+	clock := func() time.Time { return now }
+	store := NewDurableCompatPlaybackStore(pool, time.Hour, clock)
+	store.mem.Put(PlaybackSession{ID: id, CompatToken: "owner"})
+	if err := store.mem.HideFromRouting(id, "owner"); err != nil {
+		t.Fatalf("hide terminal: %v", err)
+	}
+	store.markUnpersisted(id)
+	store.markIDValidated(id)
+	now = now.Add(compatCacheRevalidationInterval)
+
+	if _, ok := store.Get(id); ok {
+		t.Fatal("terminal session became routable during missing-row revalidation")
+	}
+	terminal, ok := store.GetFinalizable(id, "owner")
+	if !ok || !terminal.Terminal || !store.isUnpersisted(id) {
+		t.Fatalf("unpersisted terminal state was lost: ok=%v session=%+v", ok, terminal)
+	}
+}
+
+func TestDurableCompatPlaybackStoreDiscardsTokenSnapshotAfterLocalMutation(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	generation := store.tokenGenerationSnapshot("owner")
+	store.cacheMutationMu.Lock()
+	store.mem.Put(PlaybackSession{ID: "new-session", CompatToken: "owner", RouteItemID: "route-1"})
+	store.bumpCacheGenerations("new-session", "owner")
+	store.cacheMutationMu.Unlock()
+
+	if store.applyCompatTokenSnapshot("owner", nil, generation) {
+		t.Fatal("stale token snapshot applied after a concurrent local mutation")
+	}
+	if _, _, ok := store.mem.FindByRoute("owner", "route-1"); !ok {
+		t.Fatal("stale token snapshot deleted the concurrent local session")
+	}
+}
+
+func TestDurableCompatPlaybackStoreAppliesTokenSnapshotAfterUnrelatedMutation(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	store.mem.Put(PlaybackSession{ID: "stale-session", CompatToken: "owner", RouteItemID: "route-1"})
+	generation := store.tokenGenerationSnapshot("owner")
+	store.cacheMutationMu.Lock()
+	store.mem.Put(PlaybackSession{ID: "other-session", CompatToken: "other"})
+	store.bumpCacheGenerations("other-session", "other")
+	store.cacheMutationMu.Unlock()
+
+	if !store.applyCompatTokenSnapshot("owner", nil, generation) {
+		t.Fatal("unrelated local mutation incorrectly invalidated the token snapshot")
+	}
+	if _, _, ok := store.mem.FindByRoute("owner", "route-1"); ok {
+		t.Fatal("valid token snapshot was not applied after unrelated mutation")
+	}
+}
+
+func TestDurableCompatPlaybackStoreDoesNotReviveLocalTerminalFromActiveSnapshot(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	active := PlaybackSession{ID: "play-1", CompatToken: "owner", RouteItemID: "route-1"}
+	store.mem.Put(active)
+	if err := store.mem.HideFromRouting("play-1", "owner"); err != nil {
+		t.Fatalf("hide local terminal: %v", err)
+	}
+	generation := store.tokenGenerationSnapshot("owner")
+
+	if !store.applyCompatTokenSnapshot("owner", []PlaybackSession{active}, generation) {
+		t.Fatal("active durable snapshot was unexpectedly discarded")
+	}
+	if _, ok := store.mem.Get("play-1"); ok {
+		t.Fatal("active durable snapshot revived a locally terminal session")
+	}
+	if terminal, ok := store.mem.GetFinalizable("play-1", "owner"); !ok || !terminal.Terminal {
+		t.Fatalf("local terminal marker was lost: ok=%v session=%+v", ok, terminal)
+	}
+}
+
+func TestDurableCompatPlaybackStorePreservesPendingUpdateFromOlderSnapshot(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	local := PlaybackSession{ID: "play-1", CompatToken: "owner", UpstreamSessionID: "upstream-new"}
+	store.mem.Put(local)
+	store.appendPendingUpdate("play-1", "owner", func(session *PlaybackSession) error {
+		session.UpstreamSessionID = "upstream-new"
+		return nil
+	})
+	generation := store.tokenGenerationSnapshot("owner")
+	durable := local
+	durable.UpstreamSessionID = "upstream-old"
+
+	if !store.applyCompatTokenSnapshot("owner", []PlaybackSession{durable}, generation) {
+		t.Fatal("durable snapshot was unexpectedly discarded")
+	}
+	if got, ok := store.mem.Get("play-1"); !ok || got.UpstreamSessionID != "upstream-new" {
+		t.Fatalf("pending local update was overwritten: ok=%v session=%+v", ok, got)
+	}
+}
+
+func TestDurableCompatPlaybackStoreDropsPendingSessionMissingFromSnapshot(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	store.mem.Put(PlaybackSession{ID: "play-1", CompatToken: "owner", RouteItemID: "route-1"})
+	store.appendPendingUpdate("play-1", "owner", func(session *PlaybackSession) error {
+		session.UpstreamSessionID = "upstream-new"
+		return nil
+	})
+	generation := store.tokenGenerationSnapshot("owner")
+
+	if !store.applyCompatTokenSnapshot("owner", nil, generation) {
+		t.Fatal("durable deletion snapshot was unexpectedly discarded")
+	}
+	if _, _, ok := store.mem.FindByRoute("owner", "route-1"); ok {
+		t.Fatal("missing durable row remained routable due to a pending update")
+	}
+	if store.hasPendingUpdates("play-1") {
+		t.Fatal("pending update survived authoritative durable deletion")
+	}
+}
+
+func TestDurableCompatPlaybackStoreTokenSnapshotKeepsOtherTokenPendingUpdates(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	store.mem.Put(PlaybackSession{ID: "play-a", CompatToken: "owner-a"})
+	store.mem.Put(PlaybackSession{ID: "play-b", CompatToken: "owner-b"})
+	store.appendPendingUpdate("play-b", "owner-b", func(session *PlaybackSession) error {
+		session.UpstreamSessionID = "upstream-b"
+		return nil
+	})
+	generation := store.tokenGenerationSnapshot("owner-a")
+
+	if !store.applyCompatTokenSnapshot(
+		"owner-a", []PlaybackSession{{ID: "play-a", CompatToken: "owner-a"}}, generation,
+	) {
+		t.Fatal("token A snapshot was unexpectedly discarded")
+	}
+	if !store.hasPendingUpdates("play-b") {
+		t.Fatal("token A snapshot discarded token B's pending update")
+	}
+}
+
+func TestDurableCompatPlaybackStoreSerializesSameSessionMutations(t *testing.T) {
+	store := NewDurableCompatPlaybackStore(nil, time.Hour, nil)
+	unlock := store.lockSessionMutation("play-1")
+	acquired := make(chan struct{})
+	released := make(chan struct{})
+	go func() {
+		secondUnlock := store.lockSessionMutation("play-1")
+		close(acquired)
+		<-released
+		secondUnlock()
+	}()
+	select {
+	case <-acquired:
+		t.Fatal("same-session mutation lock was acquired concurrently")
+	case <-time.After(20 * time.Millisecond):
+	}
+	unlock()
+	select {
+	case <-acquired:
+		close(released)
+	case <-time.After(time.Second):
+		t.Fatal("same-session mutation did not resume after release")
+	}
+}
+
+func TestDurableCompatPlaybackStorePersistsTerminalShell(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-terminal-shell-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	store := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	store.Put(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+
+	if err := store.HideFromRouting(id, "owner"); err != nil {
+		t.Fatalf("persist terminal shell: %v", err)
+	}
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	if _, ok := fresh.Get(id); ok {
+		t.Fatal("fresh process routed a durably hidden terminal shell")
+	}
+	terminal, ok := fresh.GetFinalizable(id, "owner")
+	if !ok || !terminal.Terminal || terminal.TerminalScrobbleEvent != nil {
+		t.Fatalf("terminal shell = ok=%v session=%+v", ok, terminal)
+	}
+	pending, err := fresh.ListPendingTerminals(context.Background(), 100)
+	if err != nil {
+		t.Fatalf("list terminal shells: %v", err)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("terminal shell appeared as a pending event: %+v", pending)
+	}
+	claimUntil := time.Now().UTC().Truncate(time.Microsecond).Add(time.Minute)
+	if claimed, err := fresh.ClaimTerminal(id, "owner", claimUntil); err == nil || claimed != nil {
+		t.Fatalf("terminal shell without an event was claimable: err=%v session=%+v", err, claimed)
+	}
+}
+
+func TestDurableCompatPlaybackStoreTerminalizesConflictingUnpersistedShell(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-terminal-conflict-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	seed := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	seed.Put(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+	local := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	local.mem.Put(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-1"})
+	local.markUnpersisted(id)
+
+	if err := local.HideFromRouting(id, "owner"); err != nil {
+		t.Fatalf("hide conflicting terminal shell: %v", err)
+	}
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	if _, ok := fresh.Get(id); ok {
+		t.Fatal("insert conflict left the durable session routable")
+	}
+}
+
+func TestDurableCompatPlaybackStoreReplaysPendingUpdates(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-pending-update-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	store := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	store.Put(PlaybackSession{ID: id, CompatToken: "owner", UpstreamSessionID: "upstream-old"})
+	if err := store.mem.Update(id, func(session *PlaybackSession) error {
+		session.UpstreamSessionID = "upstream-new"
+		return nil
+	}); err != nil {
+		t.Fatalf("seed pending cache update: %v", err)
+	}
+	store.appendPendingUpdate(id, "owner", func(session *PlaybackSession) error {
+		session.UpstreamSessionID = "upstream-new"
+		return nil
+	})
+
+	if err := store.Update(id, func(session *PlaybackSession) error {
+		session.TranscodeStarted = true
+		return nil
+	}); err != nil {
+		t.Fatalf("update with pending replay: %v", err)
+	}
+	if store.hasPendingUpdates(id) {
+		t.Fatal("successfully replayed update remained pending")
+	}
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	got, ok := fresh.Get(id)
+	if !ok || got.UpstreamSessionID != "upstream-new" || !got.TranscodeStarted {
+		t.Fatalf("durable replay lost updates: ok=%v session=%+v", ok, got)
+	}
+}
+
+func TestDurableCompatPlaybackStoreColdLookupIgnoresReservedThrottle(t *testing.T) {
+	pool := newCompatTestPool(t)
+	ctx := context.Background()
+	id := fmt.Sprintf("compat-cold-load-%d", time.Now().UnixNano())
+	t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM jellycompat_playback_sessions WHERE id = $1`, id) })
+	seed := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	seed.Put(PlaybackSession{ID: id, CompatToken: "owner", RouteItemID: "route-1"})
+	fresh := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	_ = fresh.shouldRevalidateID(id)
+	_ = fresh.shouldRevalidateToken("owner")
+
+	if _, ok := fresh.Get(id); !ok {
+		t.Fatal("cold ID lookup treated an in-flight validation reservation as not-found")
+	}
+	other := NewDurableCompatPlaybackStore(pool, time.Hour, nil)
+	_ = other.shouldRevalidateToken("owner")
+	if _, _, ok := other.FindByRoute("owner", "route-1"); !ok {
+		t.Fatal("cold token lookup treated an in-flight validation reservation as not-found")
+	}
+}

internal/jellycompat/router.go

@@ -120,6 +120,8 @@ func NewRouter(deps Dependencies) chi.Router {
 	playbackHandler.SettingsRepo = deps.SettingsRepo
 	playbackHandler.RecipeNodeStore = deps.RecipeNodeStore
 	playbackHandler.SessionSyncer = deps.SessionSyncer
+	playbackHandler.WatchScrobbler = deps.WatchScrobbler
+	playbackHandler.StableIdentityResolver = deps.StableIdentityResolver
 	if subtitleRepo != nil {
 		playbackHandler.SubtitleRepo = subtitleRepo
 		playbackHandler.S3Client = deps.S3Client

internal/jellycompat/server.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/subtitles"
 	"github.com/Silo-Server/silo-server/internal/userstore"
 	"github.com/Silo-Server/silo-server/internal/watchstate"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
 )
 
 // Dependencies holds the pluggable pieces used by the compat server.
@@ -92,14 +93,16 @@ type Dependencies struct {
 	// admin live-session table right after compat playback starts/stops, so
 	// the activity dashboard doesn't wait for the periodic reconciler tick.
 	// Optional.
-	SessionSyncer     PlaybackSessionSyncer
-	FileResolver      FilePathResolver
-	UserStoreProvider userstore.UserStoreProvider
-	AccessFilterFn    AccessFilterResolver
-	NodePlanner       nodepool.SessionPlanner
-	JWTSecret         string
-	Recommender       recommendations.Recommender
-	RecWorker         *recommendations.Worker
+	SessionSyncer          PlaybackSessionSyncer
+	FileResolver           FilePathResolver
+	UserStoreProvider      userstore.UserStoreProvider
+	WatchScrobbler         PlaybackWatchScrobbler
+	StableIdentityResolver watchsync.ScrobbleIdentityResolver
+	AccessFilterFn         AccessFilterResolver
+	NodePlanner            nodepool.SessionPlanner
+	JWTSecret              string
+	Recommender            recommendations.Recommender
+	RecWorker              *recommendations.Worker
 
 	// Settings (optional; reads server_settings for watched threshold, etc.)
 	SettingsRepo SettingsReader
@@ -173,6 +176,7 @@ func (s *Server) StartBackgroundTasks(ctx context.Context) {
 		repo := NewSessionRepository(s.deps.DB, s.deps.SecretCipher)
 		StartSessionCleanupWithPlaybackStore(ctx, repo, s.deps.PlaybackStore, 1*time.Hour)
 	}
+	StartTerminalScrobbleRecovery(ctx, s.deps.PlaybackStore, s.deps.WatchScrobbler, 30*time.Second)
 }
 
 // NewDependencies fills in sensible defaults for optional compat dependencies.

internal/jellycompat/streams.go

@@ -25,6 +25,7 @@ import (
 	"github.com/Silo-Server/silo-server/internal/nodepool"
 	"github.com/Silo-Server/silo-server/internal/playback"
 	"github.com/Silo-Server/silo-server/internal/subtitles"
+	"github.com/Silo-Server/silo-server/internal/watchsync"
 )
 
 // Jellyfin Web is sensitive to startup latency. Use shorter compat segments
@@ -40,7 +41,7 @@ type sessionReportRequest struct {
 	ItemID              string          `json:"ItemId"`
 	MediaSourceID       string          `json:"MediaSourceId"`
 	PlaySessionID       string          `json:"PlaySessionId"`
-	PositionTicks       int64           `json:"PositionTicks"`
+	PositionTicks       *int64          `json:"PositionTicks,omitempty"`
 	IsPaused            bool            `json:"IsPaused"`
 	AudioStreamIndex    *compatIntValue `json:"AudioStreamIndex,omitempty"`
 	SubtitleStreamIndex *compatIntValue `json:"SubtitleStreamIndex,omitempty"`
@@ -778,22 +779,98 @@ func (h *PlaybackHandler) HandleDeleteActiveEncodings(w http.ResponseWriter, r *
 		return
 	}
 
-	h.teardownPlaySession(r.Context(), playSession)
+	fallback := compatScrobbleFallbackSession(session, playSession, nil, 0, false, false)
+	upstreamSession, transcodeNodeURL := h.compatStopSnapshot(playSession, fallback)
+	if event, ok := h.compatScrobbleEvent(
+		r.Context(), compatScrobbleStop, playSession, upstreamSession, nil, nil,
+	); ok {
+		h.stageCompatTerminal(r.Context(), playSession, upstreamSession, transcodeNodeURL, event, false, false, 0)
+	} else if upstreamSession == nil {
+		// With no native session and no reported position, publishing a zero-value
+		// fallback could move provider progress backwards. Keep only the terminal
+		// authenticated mapping for a possible later Stopped report.
+		if err := h.playbackStore.HideFromRouting(playSession.ID, playSession.CompatToken); err != nil &&
+			!errors.Is(err, ErrSessionNotFound) {
+			h.scheduleCompatTerminalHide(playSession.ID, playSession.CompatToken, playSession.ExpiresAt, 1)
+		}
+		h.cleanupPlaySession(r.Context(), playSession, nil, transcodeNodeURL)
+	} else {
+		h.playbackStore.Delete(playSession.ID)
+		h.cleanupPlaySession(r.Context(), playSession, upstreamSession, transcodeNodeURL)
+	}
 
 	w.WriteHeader(http.StatusNoContent)
 }
 
-// teardownPlaySession stops the upstream playback session and removes the compat
-// play session from the store. Every step is idempotent, so it is safe to call
-// from both the explicit ActiveEncodings teardown and a Stopped playback report
-// (which may race). It is a no-op-safe teardown for an already-gone session.
-func (h *PlaybackHandler) teardownPlaySession(ctx context.Context, playSession *PlaybackSession) {
+// teardownPlaySession stages the authoritative stop before resource cleanup,
+// then delivers it through a leased durable record. The record is removed only
+// after watch-sync accepts the event, so a provider-queue failure remains
+// retryable by the client or the delayed ActiveEncodings fallback.
+func (h *PlaybackHandler) teardownPlaySession(
+	ctx context.Context,
+	playSession *PlaybackSession,
+	fallbackSession *playback.Session,
+	positionOverride *float64,
+) {
+	upstreamSession, transcodeNodeURL := h.compatStopSnapshot(playSession, fallbackSession)
+	if event, ok := h.compatScrobbleEvent(
+		ctx, compatScrobbleStop, playSession, upstreamSession, nil, positionOverride,
+	); ok {
+		h.stageCompatTerminal(ctx, playSession, upstreamSession, transcodeNodeURL, event, true, false, 0)
+	} else if playSession.Terminal {
+		// A late Stopped report without PositionTicks cannot replace a staged
+		// fallback after ActiveEncodings already removed the native session. Keep
+		// that durable event (or terminal shell) and retry its delivery instead of
+		// deleting the only recoverable stop position.
+		h.cleanupPlaySession(ctx, playSession, upstreamSession, transcodeNodeURL)
+		if playSession.TerminalScrobbleEvent != nil {
+			h.deliverCompatTerminal(
+				ctx,
+				playSession.ID,
+				playSession.CompatToken,
+				playSession.TerminalAuthoritative,
+				playSession.ExpiresAt,
+				0,
+				true,
+			)
+		}
+	} else {
+		h.playbackStore.Delete(playSession.ID)
+		h.cleanupPlaySession(ctx, playSession, upstreamSession, transcodeNodeURL)
+	}
+}
+
+func (h *PlaybackHandler) compatStopSnapshot(
+	playSession *PlaybackSession,
+	fallbackSession *playback.Session,
+) (*playback.Session, string) {
 	transcodeNodeURL := ""
+	var upstreamSession *playback.Session
 	if h.sessionMgr != nil {
-		if upstreamSession, err := h.sessionMgr.GetSession(playSession.UpstreamSessionID); err == nil {
+		if current, err := h.sessionMgr.GetSession(playSession.UpstreamSessionID); err == nil {
+			upstreamSession = current
 			transcodeNodeURL = upstreamSession.TranscodeNodeURL
 		}
 	}
+	if upstreamSession == nil && fallbackSession != nil {
+		copy := *fallbackSession
+		copy.ID = playSession.UpstreamSessionID
+		if source := compatScrobbleSource(playSession, &copy, nil); source != nil {
+			copy.MediaFileID = source.FileID
+		}
+		upstreamSession = &copy
+	}
+	return upstreamSession, transcodeNodeURL
+}
+
+// cleanupPlaySession performs idempotent process/resource cleanup after the
+// terminal provider event has been staged (or intentionally omitted).
+func (h *PlaybackHandler) cleanupPlaySession(
+	ctx context.Context,
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	transcodeNodeURL string,
+) {
 	h.tm.CloseTranscodeSession(playSession.UpstreamSessionID, transcodeNodeURL)
 	if h.sessionMgr != nil {
 		_ = h.sessionMgr.StopSession(playSession.UpstreamSessionID)
@@ -809,17 +886,232 @@ func (h *PlaybackHandler) teardownPlaySession(ctx context.Context, playSession *
 				"playback_session_id", playSession.UpstreamSessionID)
 		}
 	}
-	h.playbackStore.Delete(playSession.ID)
 	// Clients often drop the connection right after reporting a stop, so detach
 	// the sync from request cancellation to keep the admin view accurate.
 	h.syncSessionsNow(context.WithoutCancel(ctx), "compat_stop")
 }
 
+const (
+	compatTerminalClaimLease           = 10 * time.Second
+	compatTerminalInitialRetryDelay    = 250 * time.Millisecond
+	compatTerminalMaxRetryDelay        = 30 * time.Second
+	defaultCompatTerminalFallbackDelay = 2 * time.Second
+)
+
+func (h *PlaybackHandler) compatTerminalFallbackDelay() time.Duration {
+	if h != nil && h.terminalFallbackDelay > 0 {
+		return h.terminalFallbackDelay
+	}
+	return defaultCompatTerminalFallbackDelay
+}
+
+func compatTerminalRetryDelay(attempt int) time.Duration {
+	delay := compatTerminalInitialRetryDelay
+	for i := 0; i < attempt && delay < compatTerminalMaxRetryDelay; i++ {
+		delay *= 2
+		if delay > compatTerminalMaxRetryDelay {
+			return compatTerminalMaxRetryDelay
+		}
+	}
+	return delay
+}
+
+func (h *PlaybackHandler) stageCompatTerminal(
+	ctx context.Context,
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	transcodeNodeURL string,
+	event watchsync.ScrobbleEvent,
+	authoritative bool,
+	cleanupDone bool,
+	attempt int,
+) {
+	staged, err := h.playbackStore.StageTerminal(playSession.ID, playSession.CompatToken, event, authoritative)
+	if err != nil {
+		// Production durable staging installs its local marker before I/O. Keep
+		// the interface invariant for alternate stores that fail before doing so.
+		_ = h.playbackStore.HideFromRouting(playSession.ID, playSession.CompatToken)
+		if errors.Is(err, ErrSessionNotFound) {
+			if !cleanupDone {
+				h.cleanupPlaySession(ctx, playSession, upstreamSession, transcodeNodeURL)
+			}
+			return
+		}
+		if !cleanupDone {
+			h.cleanupPlaySession(ctx, playSession, upstreamSession, transcodeNodeURL)
+			cleanupDone = true
+		}
+		if playSession.ExpiresAt.IsZero() || time.Now().Before(playSession.ExpiresAt) {
+			h.scheduleCompatTerminalStage(
+				playSession, upstreamSession, transcodeNodeURL, event, authoritative, cleanupDone, attempt+1,
+			)
+		} else if !cleanupDone {
+			h.cleanupPlaySession(ctx, playSession, upstreamSession, transcodeNodeURL)
+		}
+		return
+	}
+	if !cleanupDone {
+		h.cleanupPlaySession(ctx, staged, upstreamSession, transcodeNodeURL)
+	}
+	if authoritative {
+		h.deliverCompatTerminal(ctx, staged.ID, staged.CompatToken, true, staged.ExpiresAt, 0, true)
+		return
+	}
+	h.scheduleCompatTerminalDelivery(
+		staged.ID, staged.CompatToken, false, staged.ExpiresAt, h.compatTerminalFallbackDelay(), 0,
+	)
+}
+
+func (h *PlaybackHandler) scheduleCompatTerminalHide(
+	playSessionID string,
+	compatToken string,
+	expiresAt time.Time,
+	attempt int,
+) {
+	time.AfterFunc(compatTerminalRetryDelay(attempt), func() {
+		if !expiresAt.IsZero() && !time.Now().Before(expiresAt) {
+			return
+		}
+		err := h.playbackStore.HideFromRouting(playSessionID, compatToken)
+		if err != nil && !errors.Is(err, ErrSessionNotFound) {
+			h.scheduleCompatTerminalHide(playSessionID, compatToken, expiresAt, attempt+1)
+		}
+	})
+}
+
+func (h *PlaybackHandler) scheduleCompatTerminalStage(
+	playSession *PlaybackSession,
+	upstreamSession *playback.Session,
+	transcodeNodeURL string,
+	event watchsync.ScrobbleEvent,
+	authoritative bool,
+	cleanupDone bool,
+	attempt int,
+) {
+	playSessionCopy := *playSession
+	var upstreamCopy *playback.Session
+	if upstreamSession != nil {
+		copy := *upstreamSession
+		upstreamCopy = &copy
+	}
+	time.AfterFunc(compatTerminalRetryDelay(attempt), func() {
+		h.stageCompatTerminal(
+			context.Background(), &playSessionCopy, upstreamCopy, transcodeNodeURL,
+			event, authoritative, cleanupDone, attempt,
+		)
+	})
+}
+
+func (h *PlaybackHandler) scheduleCompatTerminalDelivery(
+	playSessionID string,
+	compatToken string,
+	requireAuthoritative bool,
+	expiresAt time.Time,
+	delay time.Duration,
+	attempt int,
+) {
+	time.AfterFunc(delay, func() {
+		h.deliverCompatTerminal(
+			context.Background(), playSessionID, compatToken, requireAuthoritative, expiresAt, attempt, true,
+		)
+	})
+}
+
+// deliverCompatTerminal leases the staged event, persists it into watch-sync's
+// durable queue, and only then completes the compat terminal record. A
+// provisional ActiveEncodings fallback remains available for a later
+// authoritative Stopped replacement.
+func (h *PlaybackHandler) deliverCompatTerminal(
+	ctx context.Context,
+	playSessionID string,
+	compatToken string,
+	requireAuthoritative bool,
+	expiresAt time.Time,
+	attempt int,
+	retry bool,
+) {
+	if h == nil || h.playbackStore == nil || h.WatchScrobbler == nil {
+		return
+	}
+	if !expiresAt.IsZero() && !time.Now().Before(expiresAt) {
+		return
+	}
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	claimUntil := now.Add(compatTerminalClaimLease)
+	playSession, err := h.playbackStore.ClaimTerminal(playSessionID, compatToken, claimUntil)
+	if err != nil {
+		if !requireAuthoritative && errors.Is(err, ErrTerminalClaimUnavailable) {
+			if pending, ok := h.playbackStore.GetFinalizable(playSessionID, compatToken); ok &&
+				pending.TerminalFallbackSent && !pending.TerminalAuthoritative {
+				return
+			}
+		}
+		if retry && !errors.Is(err, ErrSessionNotFound) {
+			h.scheduleCompatTerminalDelivery(
+				playSessionID, compatToken, requireAuthoritative, expiresAt,
+				compatTerminalRetryDelay(attempt), attempt+1,
+			)
+		}
+		return
+	}
+	ownedClaimUntil := playSession.TerminalClaimUntil
+	if playSession.TerminalScrobbleEvent == nil || (requireAuthoritative && !playSession.TerminalAuthoritative) {
+		h.playbackStore.ReleaseTerminalClaim(
+			playSessionID, compatToken, ownedClaimUntil, playSession.TerminalClaimVersion, false,
+		)
+		if retry {
+			h.scheduleCompatTerminalDelivery(
+				playSessionID, compatToken, requireAuthoritative, expiresAt,
+				compatTerminalRetryDelay(attempt), attempt+1,
+			)
+		}
+		return
+	}
+
+	err = h.dispatchCompatScrobbleEvent(ctx, compatScrobbleStop, *playSession.TerminalScrobbleEvent)
+	if err != nil {
+		h.playbackStore.ReleaseTerminalClaim(
+			playSessionID, compatToken, ownedClaimUntil, playSession.TerminalClaimVersion, false,
+		)
+		if retry {
+			h.scheduleCompatTerminalDelivery(
+				playSessionID, compatToken, requireAuthoritative, expiresAt,
+				compatTerminalRetryDelay(attempt), attempt+1,
+			)
+		}
+		return
+	}
+	if playSession.TerminalAuthoritative {
+		h.playbackStore.CompleteTerminal(
+			playSessionID, compatToken, ownedClaimUntil, playSession.TerminalClaimVersion,
+		)
+		// If a newer authoritative report replaced this event while it was in
+		// flight, completion intentionally failed. Release the old lease so the
+		// replacement can be claimed immediately instead of waiting for expiry.
+		h.playbackStore.ReleaseTerminalClaim(
+			playSessionID, compatToken, ownedClaimUntil, playSession.TerminalClaimVersion, false,
+		)
+		return
+	}
+	h.playbackStore.ReleaseTerminalClaim(
+		playSessionID, compatToken, ownedClaimUntil, playSession.TerminalClaimVersion, true,
+	)
+}
+
 // compatSessionSyncTimeout bounds the immediate session sync issued from
 // request paths, so a stalled database degrades to the periodic reconciler
 // tick instead of pinning request goroutines.
 const compatSessionSyncTimeout = 5 * time.Second
 
+func compatDetachedContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	if ctx == nil {
+		ctx = context.Background()
+	} else {
+		ctx = context.WithoutCancel(ctx)
+	}
+	return context.WithTimeout(ctx, compatSessionSyncTimeout)
+}
+
 // syncSessionsNow flushes the native-session snapshot to the shared admin
 // live-session table so compat start/stop events are visible immediately
 // instead of on the next reconciler tick.
@@ -851,11 +1143,16 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	playSession, ok := h.playbackStore.Get(req.PlaySessionID)
-	if ok && playSession.CompatToken != session.Token {
-		playSession, ok = nil, false
+	var playSession *PlaybackSession
+	var ok bool
+	if stop {
+		playSession, ok = h.playbackStore.GetFinalizable(req.PlaySessionID, session.Token)
+	} else {
+		playSession, ok = h.playbackStore.Get(req.PlaySessionID)
+		if ok && playSession.CompatToken != session.Token {
+			playSession, ok = nil, false
+		}
 	}
-	matchedByRouteOnly := false
 	if !ok {
 		// Static=true direct play (Infuse, SenPlayer) skips PlaybackInfo, so the
 		// client reports progress under its own generated PlaySessionId. The
@@ -864,7 +1161,13 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 		// route-scoped lookup the stream path uses (see resolvePlaybackRoute).
 		// Without either, these reports silently no-op, the admin activity view
 		// position freezes, and stale cleanup drops the still-active session.
-		playSession, ok = h.playbackStore.FindByClientPlaySessionID(session.Token, req.PlaySessionID)
+		if stop {
+			playSession, ok = h.playbackStore.FindFinalizableByClientPlaySessionID(
+				session.Token, req.PlaySessionID, req.ItemID, req.MediaSourceID,
+			)
+		} else {
+			playSession, ok = h.playbackStore.FindByClientPlaySessionID(session.Token, req.PlaySessionID)
+		}
 		if ok && !reportMatchesPlaySession(playSession, req) {
 			playSession, ok = nil, false
 		}
@@ -874,10 +1177,15 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 			if routeID == "" {
 				continue
 			}
-			if playSession, _, ok = h.playbackStore.FindByRoute(session.Token, routeID); ok {
-				matchedByRouteOnly = true
+			if stop {
+				playSession, _, ok = h.playbackStore.FindFinalizableByRoute(session.Token, routeID)
+			} else {
+				playSession, _, ok = h.playbackStore.FindByRoute(session.Token, routeID)
+			}
+			if ok && reportMatchesPlaySession(playSession, req) {
 				break
 			}
+			playSession, ok = nil, false
 		}
 	}
 	if !ok || playSession.UpstreamSessionID == "" {
@@ -885,7 +1193,14 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 		return
 	}
 
-	positionSeconds := float64(req.PositionTicks) / 10_000_000
+	positionSeconds := 0.0
+	positionReported := req.PositionTicks != nil
+	if positionReported {
+		positionSeconds = float64(*req.PositionTicks) / 10_000_000
+		if positionSeconds < 0 {
+			positionSeconds = 0
+		}
+	}
 	audioTrackIndex := 0
 	audioRestarted := false
 	// Jellyfin web/mobile clients send AudioStreamIndex on every progress
@@ -928,8 +1243,15 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 			)
 		}
 	}
-	if positionSeconds > 0 && h.sessionMgr != nil {
+	var previousSession *playback.Session
+	progressUpdated := false
+	if positionReported && h.sessionMgr != nil {
+		if current, err := h.sessionMgr.GetSession(playSession.UpstreamSessionID); err == nil && current != nil {
+			copy := *current
+			previousSession = &copy
+		}
 		err := h.sessionMgr.UpdateProgress(playSession.UpstreamSessionID, positionSeconds, req.IsPaused)
+		progressUpdated = err == nil
 		if errors.Is(err, playback.ErrSessionNotFound) && !stop {
 			// The upstream session was reaped as stale (e.g. the client buffered
 			// far ahead and went quiet between range requests). The report proves
@@ -937,10 +1259,24 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 			// dropping it from session tracking for the rest of playback.
 			if revived := h.reviveUpstreamForReport(r.Context(), session, playSession, req.MediaSourceID); revived != nil {
 				playSession = revived
-				_ = h.sessionMgr.UpdateProgress(playSession.UpstreamSessionID, positionSeconds, req.IsPaused)
+				progressUpdated = h.sessionMgr.UpdateProgress(playSession.UpstreamSessionID, positionSeconds, req.IsPaused) == nil
+				previousSession = nil
 			}
 		}
 	}
+	if progressUpdated && !stop && previousSession != nil && previousSession.IsPaused != req.IsPaused {
+		updatedSession := *previousSession
+		updatedSession.Position = positionSeconds
+		updatedSession.IsPaused = req.IsPaused
+		action := compatScrobbleStart
+		if req.IsPaused {
+			action = compatScrobblePause
+		}
+		h.dispatchCompatScrobbleAt(
+			r.Context(), action, playSession, &updatedSession,
+			findMediaSource(playSession, req.MediaSourceID), &positionSeconds,
+		)
+	}
 	// Persist progress to user store
 	if positionSeconds > 0 && h.storeProvider != nil && playSession.ItemID != "" {
 		if store, storeErr := h.storeProvider.ForUser(r.Context(), session.StreamAppUserID); storeErr == nil {
@@ -957,12 +1293,19 @@ func (h *PlaybackHandler) handlePlaybackReport(w http.ResponseWriter, r *http.Re
 			}
 		}
 	}
-	if stop && !matchedByRouteOnly {
-		// A bare item/source route match is ambiguous when the same item plays
-		// twice under one token, so never tear down a session the report may
-		// not own. A session that really stopped emits no further reports or
-		// transport, so stale cleanup reaps it shortly anyway.
-		h.teardownPlaySession(r.Context(), playSession)
+	if stop {
+		// Direct ids and aliases are caller-owned; route-only stopped reports are
+		// accepted only when FindFinalizableByRoute proves the token-scoped match
+		// is unique.
+		source := findMediaSource(playSession, req.MediaSourceID)
+		fallback := compatScrobbleFallbackSession(
+			session, playSession, source, positionSeconds, positionReported, req.IsPaused,
+		)
+		var positionOverride *float64
+		if positionReported {
+			positionOverride = &positionSeconds
+		}
+		h.teardownPlaySession(r.Context(), playSession, fallback, positionOverride)
 	}
 
 	w.WriteHeader(http.StatusNoContent)
@@ -1051,7 +1394,12 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess
 					card.ClientName, card.ClientVersion, card.ClientUserAgent = info.Name, info.Version, info.UserAgent
 				}
 				if reconstructed := h.tm.ReconstructSession(ctx, playSession.UpstreamSessionID, compatSession.StreamAppUserID, card); reconstructed != nil {
+					if !playSession.ProgressPersistenceKnown ||
+						playSession.DisableProgressPersistence != reconstructed.DisableProgressPersistence {
+						h.recordCompatProgressPersistence(playSession.ID, reconstructed.DisableProgressPersistence)
+					}
 					_ = h.syncUpstreamAudioSelection(playSession, source)
+					h.dispatchCompatScrobble(ctx, compatScrobbleStart, playSession, reconstructed, &source)
 					return playSession, nil
 				}
 			}
@@ -1067,6 +1415,11 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess
 			playSession.UpstreamPlayMethod = ""
 			playSession.TranscodeStarted = false
 		} else {
+			if current, currentErr := h.sessionMgr.GetSession(playSession.UpstreamSessionID); currentErr == nil &&
+				(!playSession.ProgressPersistenceKnown ||
+					playSession.DisableProgressPersistence != current.DisableProgressPersistence) {
+				h.recordCompatProgressPersistence(playSession.ID, current.DisableProgressPersistence)
+			}
 			_ = h.syncUpstreamAudioSelection(playSession, source)
 			return playSession, nil
 		}
@@ -1093,6 +1446,7 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess
 		transcodeNodeURL := ""
 		if current, err := h.sessionMgr.GetSession(oldUpstreamSessionID); err == nil {
 			transcodeNodeURL = current.TranscodeNodeURL
+			h.dispatchCompatScrobble(ctx, compatScrobblePause, playSession, current, nil)
 		}
 		_ = h.sessionMgr.StopSession(oldUpstreamSessionID)
 		h.tm.CloseTranscodeSession(oldUpstreamSessionID, transcodeNodeURL)
@@ -1134,6 +1488,8 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess
 		current.UpstreamSessionID = session.ID
 		current.UpstreamPlayMethod = method
 		current.TranscodeStarted = false
+		current.ProgressPersistenceKnown = true
+		current.DisableProgressPersistence = session.DisableProgressPersistence
 		return nil
 	}); updateErr != nil {
 		_ = h.sessionMgr.StopSession(session.ID)
@@ -1154,9 +1510,21 @@ func (h *PlaybackHandler) ensureUpstreamPlayback(ctx context.Context, compatSess
 		return nil, ErrSessionNotFound
 	}
 	h.syncSessionsNow(ctx, "compat_start")
+	h.dispatchCompatScrobble(ctx, compatScrobbleStart, updated, session, &source)
 	return updated, nil
 }
 
+func (h *PlaybackHandler) recordCompatProgressPersistence(playSessionID string, disabled bool) {
+	if h == nil || h.playbackStore == nil || playSessionID == "" {
+		return
+	}
+	_ = h.playbackStore.Update(playSessionID, func(session *PlaybackSession) error {
+		session.ProgressPersistenceKnown = true
+		session.DisableProgressPersistence = disabled
+		return nil
+	})
+}
+
 func (h *PlaybackHandler) ensureTranscodeManifest(ctx context.Context, compatSession *Session, playSessionID string, source PlaybackMediaSource) ([]byte, error) {
 	playSession, err := h.ensureUpstreamPlayback(ctx, compatSession, playSessionID, source, "transcode")
 	if err != nil {

internal/jellycompat/streams_test.go

@@ -198,7 +198,7 @@ func TestTeardownPlaySession_DeletesNodeRecipe(t *testing.T) {
 	if !ok {
 		t.Fatal("expected play session")
 	}
-	h.teardownPlaySession(context.Background(), playSession)
+	h.teardownPlaySession(context.Background(), playSession, nil, nil)
 
 	if _, ok := recipeStore.Get("upstream-1"); ok {
 		t.Fatal("node recipe should be deleted on deliberate teardown")

internal/jellycompat/terminal_scrobble_recovery.go

@@ -0,0 +1,65 @@
+package jellycompat
+
+import (
+	"context"
+	"log/slog"
+	"time"
+)
+
+const compatTerminalRecoveryBatchSize = 100
+
+// StartTerminalScrobbleRecovery resumes terminal provider events that survived
+// a server exit after staging but before delivery. The first scan runs at
+// startup; periodic scans also recover expired leases from crashed peers.
+func StartTerminalScrobbleRecovery(
+	ctx context.Context,
+	store CompatPlaybackStore,
+	scrobbler PlaybackWatchScrobbler,
+	interval time.Duration,
+) {
+	if store == nil || scrobbler == nil {
+		return
+	}
+	if interval <= 0 {
+		interval = 30 * time.Second
+	}
+	handler := &PlaybackHandler{playbackStore: store, WatchScrobbler: scrobbler}
+	run := func() {
+		if err := recoverPendingTerminalScrobbles(ctx, handler); err != nil {
+			slog.WarnContext(ctx, "recover jellycompat terminal scrobbles failed",
+				"component", "jellycompat", "error", err)
+		}
+	}
+	go func() {
+		run()
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				run()
+			}
+		}
+	}()
+}
+
+func recoverPendingTerminalScrobbles(ctx context.Context, handler *PlaybackHandler) error {
+	pending, err := handler.playbackStore.ListPendingTerminals(ctx, compatTerminalRecoveryBatchSize)
+	if err != nil {
+		return err
+	}
+	for _, session := range pending {
+		handler.deliverCompatTerminal(
+			ctx,
+			session.ID,
+			session.CompatToken,
+			session.TerminalAuthoritative,
+			session.ExpiresAt,
+			0,
+			false,
+		)
+	}
+	return nil
+}

internal/watchsync/scrobble_identity.go

@@ -0,0 +1,55 @@
+package watchsync
+
+import (
+	"context"
+
+	"github.com/Silo-Server/silo-server/internal/historyimport"
+	"github.com/Silo-Server/silo-server/internal/userstore"
+)
+
+// ScrobbleIdentityResolver resolves a local media item to the stable provider
+// identity required by external scrobble APIs.
+type ScrobbleIdentityResolver interface {
+	ResolveHistoryIdentity(ctx context.Context, mediaItemID string) userstore.WatchIdentity
+}
+
+// ResolveScrobbleIdentity enriches an event with stable movie or episode IDs.
+// Unknown identities retain the native playback fallback of treating the item
+// as a movie; providers then report their normal missing-identity error instead
+// of silently dropping the lifecycle event.
+func ResolveScrobbleIdentity(ctx context.Context, resolver ScrobbleIdentityResolver, event ScrobbleEvent) ScrobbleEvent {
+	if resolver == nil {
+		if event.Kind == "" {
+			event.Kind = historyimport.KindMovie
+		}
+		return event
+	}
+
+	identity := resolver.ResolveHistoryIdentity(ctx, event.MediaItemID)
+	if identity.StableType != "" {
+		event.Kind = identity.StableType
+	}
+	if event.Kind == "" {
+		event.Kind = historyimport.KindMovie
+	}
+	event.SeasonNumber = optionalIntValue(identity.Season)
+	event.EpisodeNumber = optionalIntValue(identity.Episode)
+	if identity.ProviderIDs != nil {
+		event.IMDbID = identity.ProviderIDs["imdb"]
+		event.TMDBID = identity.ProviderIDs["tmdb"]
+		event.TVDBID = identity.ProviderIDs["tvdb"]
+	}
+	if identity.SeriesProviderIDs != nil {
+		event.SeriesIMDbID = identity.SeriesProviderIDs["imdb"]
+		event.SeriesTMDBID = identity.SeriesProviderIDs["tmdb"]
+		event.SeriesTVDBID = identity.SeriesProviderIDs["tvdb"]
+	}
+	return event
+}
+
+func optionalIntValue(value *int) int {
+	if value == nil {
+		return 0
+	}
+	return *value
+}

internal/watchsync/scrobble_identity_test.go

@@ -0,0 +1,44 @@
+package watchsync
+
+import (
+	"context"
+	"testing"
+
+	"github.com/Silo-Server/silo-server/internal/userstore"
+)
+
+type fixedScrobbleIdentityResolver struct {
+	identity userstore.WatchIdentity
+}
+
+func (r fixedScrobbleIdentityResolver) ResolveHistoryIdentity(context.Context, string) userstore.WatchIdentity {
+	return r.identity
+}
+
+func TestResolveScrobbleIdentityEpisode(t *testing.T) {
+	season := 2
+	episode := 7
+	event := ResolveScrobbleIdentity(context.Background(), fixedScrobbleIdentityResolver{
+		identity: userstore.WatchIdentity{
+			StableType:        "episode",
+			ProviderIDs:       map[string]string{"tmdb": "episode-tmdb"},
+			SeriesProviderIDs: map[string]string{"tvdb": "series-tvdb"},
+			Season:            &season,
+			Episode:           &episode,
+		},
+	}, ScrobbleEvent{MediaItemID: "episode-1"})
+
+	if event.Kind != "episode" || event.TMDBID != "episode-tmdb" || event.SeriesTVDBID != "series-tvdb" {
+		t.Fatalf("resolved identity = %+v", event)
+	}
+	if event.SeasonNumber != season || event.EpisodeNumber != episode {
+		t.Fatalf("resolved episode numbers = S%dE%d, want S%dE%d", event.SeasonNumber, event.EpisodeNumber, season, episode)
+	}
+}
+
+func TestResolveScrobbleIdentityDefaultsUnknownItemToMovie(t *testing.T) {
+	event := ResolveScrobbleIdentity(context.Background(), nil, ScrobbleEvent{MediaItemID: "movie-1"})
+	if event.Kind != "movie" {
+		t.Fatalf("kind = %q, want movie", event.Kind)
+	}
+}