feat(playback): server-observed stream monitoring (async, no client trust)

Introduce a first-class, authoritative view of what is actually streaming,
observed server-side and never trusting client progress reports. This is the
base observation layer the kill switch and async over-cap enforcer build on.

- internal/streammonitor: live-stream snapshot model plus pluggable Sources
  (local func source, Redis source, multi-source fan-in) so a single node and a
  multi-node deployment expose the same picture.
- internal/nodesessions/tracker: serve-activity attribution — LastServedAt and
  served-byte counters advance from real serving, not client pings, giving an
  authoritative liveness signal.
- Client identity as monitoring attribution: Origin ("native" | "jellycompat")
  and ClientName ride the server-signed stream token (streamtoken.Claims) and
  the transcode-start request so an edge/transcode node — which never sees the
  originating API path — can stamp them onto its live-session record. These are
  attribution only: not byte-affecting and not a trust assertion.
- Serve-activity marks on the transcode node (MarkServed) so a node's own record
  reflects real serving instead of a LastServedAt frozen at start time.
- Admin observation surfaces: node/session listing carries owner + client
  identity and dedupes multi-record sessions.

Part of the stream monitoring & kill-switch epic.

internal/api/handlers/nodes.go

@@ -15,10 +15,45 @@ import (
 
 	"github.com/Silo-Server/silo-server/internal/cache"
 	"github.com/Silo-Server/silo-server/internal/nodepool"
+	"github.com/Silo-Server/silo-server/internal/nodesessions"
+	"github.com/Silo-Server/silo-server/internal/playback"
+	"github.com/Silo-Server/silo-server/internal/streammonitor"
 	"github.com/go-chi/chi/v5"
 	"github.com/redis/go-redis/v9"
 )
 
+// LiveLocalSessions maps the in-process playback sessions into monitoring
+// records (the same SessionInfo shape edge nodes write to Redis), so integrated
+// single-node streams — which never touch Redis — are still visible to the
+// monitor and the admin active-streams view. Shared by the enforcer's FuncSource
+// and the admin session list so there is exactly one Session→record mapping.
+func LiveLocalSessions(sm *playback.SessionManager, nodeName string) []nodesessions.SessionInfo {
+	if sm == nil {
+		return nil
+	}
+	live := sm.AllSessions()
+	out := make([]nodesessions.SessionInfo, 0, len(live))
+	for _, s := range live {
+		out = append(out, nodesessions.SessionInfo{
+			SessionID:    s.ID,
+			NodeName:     nodeName,
+			AuthUserID:   s.UserID,
+			ProfileID:    s.ProfileID,
+			Type:         string(s.PlayMethod),
+			Route:        s.Origin(),
+			MediaFileID:  s.MediaFileID,
+			ClientIP:     s.ClientIP,
+			ClientName:   s.ClientName,
+			Position:     s.Position,
+			Resolution:   s.TargetResolution,
+			HWAccel:      s.TranscodeHWAccel,
+			StartedAt:    s.StartedAt.UTC().Format(time.RFC3339),
+			LastServedAt: s.LastActivityAt.UTC().Format(time.RFC3339),
+		})
+	}
+	return out
+}
+
 // NodeRepository defines the operations the NodeHandler needs on the node store.
 type NodeRepository interface {
 	List(ctx context.Context) ([]*nodepool.Node, error)
@@ -43,6 +78,18 @@ type NodeHandler struct {
 	eventBus      cache.EventBus
 	redisClient   *redis.Client // for reading session keys
 	jwtSecret     string        // for bearer auth when calling force-reload on nodes
+
+	// sessionMgr + localNodeName let the session list include integrated
+	// single-node streams (which never write Redis). Optional; nil in edge modes.
+	sessionMgr    *playback.SessionManager
+	localNodeName string
+}
+
+// SetLocalSessionSource wires the in-process session manager so HandleListSessions
+// can union integrated streams with the Redis-backed edge records.
+func (h *NodeHandler) SetLocalSessionSource(sm *playback.SessionManager, nodeName string) {
+	h.sessionMgr = sm
+	h.localNodeName = nodeName
 }
 
 // NewNodeHandler creates a new NodeHandler.
@@ -301,51 +348,69 @@ func (h *NodeHandler) HandleForceReloadNode(w http.ResponseWriter, r *http.Reque
 }
 
 // HandleListSessions handles GET /admin/nodes/sessions — lists active playback
-// sessions from Redis, optionally filtered by node_id query parameter.
+// sessions, unioning the Redis-backed edge records with the in-process
+// integrated sessions (which never write Redis) so a single-node deployment is
+// not blind. The union is deduped by session id (the same stream is tracked by
+// the central manager AND by the edge serving it — one row per stream, like the
+// enforcer's monitoring picture, so an operator never sees a single stream
+// counted twice against a cap). Optionally filtered by node_id: a filter
+// targets a specific edge node, so integrated sessions are only included in the
+// unfiltered listing.
 func (h *NodeHandler) HandleListSessions(w http.ResponseWriter, r *http.Request) {
-	if h.redisClient == nil {
-		writeError(w, http.StatusServiceUnavailable, "redis_unavailable", "Redis not configured")
-		return
-	}
-
 	ctx := r.Context()
-	pattern := "silo:sessions:*"
+	nodeFilter := r.URL.Query().Get("node_id")
 
-	if nodeIDStr := r.URL.Query().Get("node_id"); nodeIDStr != "" {
-		nodeID, err := strconv.Atoi(nodeIDStr)
-		if err == nil {
-			node, err := h.repo.GetByID(ctx, nodeID)
-			if err == nil {
-				hashBytes := sha256.Sum256([]byte(node.URL))
-				nodeHash := hex.EncodeToString(hashBytes[:4])
-				pattern = "silo:sessions:" + nodeHash + ":*"
+	var infos []nodesessions.SessionInfo
+
+	if h.redisClient != nil {
+		pattern := nodesessions.KeyPrefix + "*"
+		if nodeFilter != "" {
+			if nodeID, err := strconv.Atoi(nodeFilter); err == nil {
+				if node, err := h.repo.GetByID(ctx, nodeID); err == nil {
+					hashBytes := sha256.Sum256([]byte(node.URL))
+					nodeHash := hex.EncodeToString(hashBytes[:4])
+					pattern = nodesessions.KeyPrefix + nodeHash + ":*"
+				}
 			}
 		}
-	}
 
-	var sessions []json.RawMessage
-	var cursor uint64
-	for {
-		keys, next, err := h.redisClient.Scan(ctx, cursor, pattern, 100).Result()
-		if err != nil {
-			writeError(w, http.StatusInternalServerError, "redis_error", err.Error())
-			return
-		}
-		for _, key := range keys {
-			val, err := h.redisClient.Get(ctx, key).Result()
+		var cursor uint64
+		for {
+			keys, next, err := h.redisClient.Scan(ctx, cursor, pattern, 100).Result()
 			if err != nil {
-				continue
+				writeError(w, http.StatusInternalServerError, "redis_error", err.Error())
+				return
+			}
+			for _, key := range keys {
+				val, err := h.redisClient.Get(ctx, key).Result()
+				if err != nil {
+					continue
+				}
+				var info nodesessions.SessionInfo
+				if err := json.Unmarshal([]byte(val), &info); err != nil {
+					slog.Debug("skip malformed session record", "key", key, "error", err)
+					continue
+				}
+				infos = append(infos, info)
+			}
+			cursor = next
+			if cursor == 0 {
+				break
 			}
-			sessions = append(sessions, json.RawMessage(val))
-		}
-		cursor = next
-		if cursor == 0 {
-			break
 		}
 	}
 
-	if sessions == nil {
-		sessions = []json.RawMessage{}
+	// Integrated single-node streams live only in the in-process session manager.
+	// Include them in the unfiltered listing (a node_id filter targets an edge).
+	if h.sessionMgr != nil && nodeFilter == "" {
+		infos = append(infos, LiveLocalSessions(h.sessionMgr, h.localNodeName)...)
+	}
+
+	sessions := []json.RawMessage{}
+	for _, info := range streammonitor.DedupeSessionInfos(infos) {
+		if data, err := json.Marshal(info); err == nil {
+			sessions = append(sessions, json.RawMessage(data))
+		}
 	}
 
 	type sessionsResponse struct {

internal/api/handlers/playback.go

@@ -612,7 +612,9 @@ func (h *PlaybackHandler) signSessionToken(card playback.RecipeCard) string {
 	if h.JWTSecret == "" {
 		return ""
 	}
-	token, err := streamtoken.Sign(card.ToClaims(), h.JWTSecret, playback.MaxTokenTTL)
+	claims := card.ToClaims()
+	claims.Origin = playback.OriginNative // native-only mint site
+	token, err := streamtoken.Sign(claims, h.JWTSecret, playback.MaxTokenTTL)
 	if err != nil {
 		slog.Warn("sign stream token failed", "error", err, "session", card.SessionID, "playback_session_id", card.SessionID)
 		return ""
@@ -1952,6 +1954,8 @@ func (h *PlaybackHandler) handleStartPlaybackLegacy(w http.ResponseWriter, r *ht
 				UserID:      session.UserID,
 				ProfileID:   session.ProfileID,
 				MediaFileID: session.MediaFileID,
+				Origin:      playback.OriginNative,
+				ClientName:  session.ClientName,
 			}
 
 			// Resolve media path if possible.
@@ -2603,6 +2607,11 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http.
 					SubtitleCodec:          subtitleCodec,
 					TotalDuration:          float64(file.Duration),
 					RequireReady:           atomicLegacyReplacement,
+					AuthUserID:             updatedSession.UserID,
+					ProfileID:              updatedSession.ProfileID,
+					MediaFileID:            updatedSession.MediaFileID,
+					Route:                  updatedSession.Origin(),
+					ClientName:             updatedSession.ClientName,
 				}
 				if strings.TrimSpace(nodeReq.HWAccel) == "" {
 					nodeReq.HWAccel = h.playbackConfig().HWAccel
@@ -2728,6 +2737,8 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http.
 					// mode; a re-minted token must not silently shed either.
 					TranscodeTransportID: updatedSession.TranscodeTransportID,
 					RemuxDVMode:          string(updatedSession.RemuxDVMode),
+					Origin:               updatedSession.Origin(),
+					ClientName:           updatedSession.ClientName,
 				}
 				if plan.TranscodeNode != nil {
 					tokenClaims.TranscodeNode = plan.TranscodeNode.URL
@@ -3273,6 +3284,11 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re
 			SubtitleCodec:          subtitleCodec,
 			TotalDuration:          float64(file.Duration),
 			RequireReady:           atomicLegacyReplacement,
+			AuthUserID:             session.UserID,
+			ProfileID:              session.ProfileID,
+			MediaFileID:            session.MediaFileID,
+			Route:                  session.Origin(),
+			ClientName:             session.ClientName,
 		}
 
 		nodeResp, status, err := h.startRemotePlaybackTransport(context.Background(), tcNode.URL, nodeReq)
@@ -3755,6 +3771,19 @@ func (h *PlaybackHandler) HandleGetTranscodeSegment(w http.ResponseWriter, r *ht
 
 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
+	// Hold a transport marker for the whole segment pour so a slow client draining
+	// a single segment past the liveness grace cannot be reaped mid-serve and go
+	// invisible to the monitor — matching the direct-play/remux handlers. This is
+	// server-observed liveness: it keeps the session visible independent of any
+	// client progress report (the hidden-stream defense).
+	if err := h.sessionMgr.BeginTransport(sessionID); err == nil {
+		defer func() { _ = h.sessionMgr.EndTransport(sessionID) }()
+	} else {
+		// The marker is what keeps a slow single-segment drain visible; the
+		// segment is still served on failure, so log rather than fail — but log so
+		// the hidden-stream window is observable.
+		slog.Warn("begin transport marker failed", "session", sessionID, "segment", segmentPath, "error", err)
+	}
 	http.ServeFile(w, r, segmentPath)
 }
 

internal/nodesessions/tracker.go

@@ -13,7 +13,12 @@ import (
 )
 
 const (
-	keyPrefix  = "silo:sessions:"
+	// KeyPrefix is the Redis key namespace for node session records
+	// (silo:sessions:{nodeHash}:{sessionID}). Exported so readers of the
+	// monitoring picture (streammonitor, the admin session list) SCAN the same
+	// namespace this tracker writes instead of duplicating the literal.
+	KeyPrefix = "silo:sessions:"
+
 	sessionTTL = 60 * time.Second
 	refreshInt = 30 * time.Second
 )
@@ -33,6 +38,14 @@ type SessionInfo struct {
 	HWAccel     string `json:"hw_accel,omitempty"`
 	StartedAt   string `json:"started_at"`
 
+	// LastServedAt / BytesServed are the authoritative, server-observed liveness
+	// signals: they are refreshed only when the node actually serves bytes for this
+	// session (a segment written, a direct-play/remux pour advancing), never from a
+	// client progress report. A "quiet" stream that keeps pulling stays fresh here;
+	// one that stops pulling ages out. This is the anti-abuse monitoring surface.
+	LastServedAt string `json:"last_served_at,omitempty"`
+	BytesServed  int64  `json:"bytes_served,omitempty"`
+
 	// AuthUserID / ProfileID / MediaFileID are the numeric ownership keys the
 	// node copies from the verified stream token. They enrich the live admin
 	// "active streams" view (served by SCANning these records) so it can answer
@@ -41,6 +54,17 @@ type SessionInfo struct {
 	AuthUserID  int    `json:"auth_user_id,omitempty"`
 	ProfileID   string `json:"profile_id,omitempty"`
 	MediaFileID int    `json:"media_file_id,omitempty"`
+
+	// Route is the origin protocol ("native" | "jellycompat"). Type is the play
+	// method (direct_play/remux/transcode); Route is orthogonal — a jellycompat
+	// stream and a native one share a Type but differ in Route. Client* are
+	// best-effort viewer identity for oversight and abuse heuristics (e.g. one
+	// re-streaming client fanning out to many viewers). Position is the last
+	// known playback position in seconds (secondary timing).
+	Route      string  `json:"route,omitempty"`
+	ClientIP   string  `json:"client_ip,omitempty"`
+	ClientName string  `json:"client_name,omitempty"`
+	Position   float64 `json:"position,omitempty"`
 }
 
 // Tracker manages session lifecycle in Redis for a single node.
@@ -52,8 +76,10 @@ type Tracker struct {
 	nodeHash string // first 8 chars of SHA-256 of nodeURL
 
 	mu       sync.Mutex
-	sessions map[string]struct{}  // set of active session IDs
-	touched  map[string]time.Time // ephemeral sessions by last-activity time
+	sessions map[string]struct{}    // set of active (explicitly tracked) session IDs
+	touched  map[string]time.Time   // ephemeral sessions by last-activity time
+	records  map[string]SessionInfo // last-written record per session, for enriched refresh
+	bytes    map[string]int64       // cumulative bytes served per session (monitoring only)
 }
 
 // NewTracker creates a session tracker for the given node.
@@ -68,12 +94,14 @@ func NewTracker(rdb *redis.Client, nodeURL, nodeName, nodeType string) *Tracker
 		nodeHash: hex.EncodeToString(h[:4]), // 8 hex chars
 		sessions: make(map[string]struct{}),
 		touched:  make(map[string]time.Time),
+		records:  make(map[string]SessionInfo),
+		bytes:    make(map[string]int64),
 	}
 }
 
 // redisKey returns the full Redis key for a session.
 func (tr *Tracker) redisKey(sessionID string) string {
-	return keyPrefix + tr.nodeHash + ":" + sessionID
+	return KeyPrefix + tr.nodeHash + ":" + sessionID
 }
 
 // NodeHash returns the node's hash prefix used in Redis keys.
@@ -109,12 +137,64 @@ func (tr *Tracker) ActiveCount() int {
 	return count
 }
 
+// Snapshot returns a copy of the currently-known session records on this node
+// (including ephemeral ones still within their TTL), stamped with the latest
+// LastServedAt/BytesServed. Used for the node /status view; central aggregates
+// the equivalent picture by reading Redis.
+func (tr *Tracker) Snapshot() []SessionInfo {
+	tr.mu.Lock()
+	defer tr.mu.Unlock()
+	now := time.Now()
+	out := make([]SessionInfo, 0, len(tr.records))
+	for id, rec := range tr.records {
+		// Session-backed entries are always live until Remove; only ephemeral
+		// (non-session) entries age out by idle timeout.
+		if _, isSession := tr.sessions[id]; !isSession {
+			if last, ok := tr.touched[id]; ok && now.Sub(last) > sessionTTL {
+				continue
+			}
+		}
+		out = append(out, tr.enrichLocked(id, rec))
+	}
+	return out
+}
+
+// enrichLocked stamps a record with the latest observed liveness/bytes. Caller
+// holds tr.mu.
+func (tr *Tracker) enrichLocked(id string, rec SessionInfo) SessionInfo {
+	if last, ok := tr.touched[id]; ok {
+		rec.LastServedAt = last.UTC().Format(time.RFC3339)
+	}
+	if b, ok := tr.bytes[id]; ok {
+		rec.BytesServed = b
+	}
+	return rec
+}
+
 // Track registers an active session in Redis with a TTL.
 func (tr *Tracker) Track(ctx context.Context, info SessionInfo) {
 	if tr.rdb == nil {
 		return
 	}
-	data, err := json.Marshal(info)
+	now := time.Now()
+	if info.LastServedAt == "" {
+		info.LastServedAt = now.UTC().Format(time.RFC3339)
+	}
+
+	// Explicitly-tracked sessions (direct play / remux) live in tr.sessions for
+	// the whole connection. touched holds their last-served time so LastServedAt
+	// reflects real activity; refreshAll only idle-prunes ephemeral (non-session)
+	// touched entries, so a long quiet pour is never expired while its connection
+	// is open.
+	tr.mu.Lock()
+	tr.preserveStartedAtLocked(&info)
+	tr.sessions[info.SessionID] = struct{}{}
+	tr.records[info.SessionID] = info
+	tr.touched[info.SessionID] = now
+	enriched := tr.enrichLocked(info.SessionID, info)
+	tr.mu.Unlock()
+
+	data, err := json.Marshal(enriched)
 	if err != nil {
 		slog.DebugContext(ctx, "session track marshal failed", "component", "nodesessions", "error", err)
 		return
@@ -124,29 +204,31 @@ func (tr *Tracker) Track(ctx context.Context, info SessionInfo) {
 		slog.DebugContext(ctx, "session track set failed", "component", "nodesessions", "error", err, "session", info.SessionID)
 		return
 	}
-
-	tr.mu.Lock()
-	tr.sessions[info.SessionID] = struct{}{}
-	tr.mu.Unlock()
 }
 
 // Touch registers or refreshes an ephemeral session that has no explicit end,
 // such as HLS manifest/segment fetches flowing through a proxy. The session is
 // written to Redis on first touch and drops out of the active count after
-// sessionTTL without further touches (pruned by the refresh loop).
+// sessionTTL without further touches (pruned by the refresh loop). Subsequent
+// touches update LastServedAt in memory; the body is re-flushed on the refresh
+// tick so the monitoring record reflects real serve activity.
 func (tr *Tracker) Touch(ctx context.Context, info SessionInfo) {
 	if tr.rdb == nil {
 		return
 	}
+	now := time.Now()
 	tr.mu.Lock()
+	tr.preserveStartedAtLocked(&info)
 	_, known := tr.touched[info.SessionID]
-	tr.touched[info.SessionID] = time.Now()
+	tr.touched[info.SessionID] = now
+	tr.records[info.SessionID] = info
+	enriched := tr.enrichLocked(info.SessionID, info)
 	tr.mu.Unlock()
 	if known {
 		return
 	}
 
-	data, err := json.Marshal(info)
+	data, err := json.Marshal(enriched)
 	if err != nil {
 		slog.DebugContext(ctx, "session touch marshal failed", "component", "nodesessions", "error", err)
 		return
@@ -156,6 +238,57 @@ func (tr *Tracker) Touch(ctx context.Context, info SessionInfo) {
 	}
 }
 
+// preserveStartedAtLocked keeps the first-seen StartedAt when a record is
+// re-written for a session we already track. Track re-fires on every range
+// reconnect and Touch stamps a fresh sessionInfo per segment fetch; without
+// this, StartedAt degrades to "time of last request", corrupting the admin
+// view and the enforcer's StartedAt tie-break in victim selection. Caller
+// holds tr.mu.
+func (tr *Tracker) preserveStartedAtLocked(info *SessionInfo) {
+	if prev, ok := tr.records[info.SessionID]; ok && prev.StartedAt != "" {
+		info.StartedAt = prev.StartedAt
+	}
+}
+
+// AddBytes records that n bytes were served for a session and marks it as active
+// now. Cheap and lock-guarded; the accumulated value is flushed to Redis on the
+// refresh tick. Bytes attribution is best-effort monitoring and never gates the
+// hot path. Bytes for a session with no live record are dropped: a late tally
+// arriving after the record was idle-pruned (or removed) must not recreate a
+// bytes entry nothing will ever clean up — that was a slow permanent leak on
+// busy edges.
+func (tr *Tracker) AddBytes(sessionID string, n int64) {
+	if tr.rdb == nil || n <= 0 {
+		return
+	}
+	tr.mu.Lock()
+	if _, known := tr.records[sessionID]; known {
+		tr.bytes[sessionID] += n
+		// Mark real serve activity so LastServedAt advances for every session
+		// type, including long direct-play/remux pours.
+		tr.touched[sessionID] = time.Now()
+	}
+	tr.mu.Unlock()
+}
+
+// MarkServed records real serve activity for a session without byte
+// attribution, advancing LastServedAt (flushed on the refresh tick). Used by
+// the transcode node's own serve paths, where wrapping ServeFile for byte
+// counting would cost the sendfile fast path for a signal the fronting proxy
+// already measures — without it the node record's LastServedAt stays frozen at
+// start time, which starves the enforcer's freshness ordering and the admin
+// view of the node-side truth.
+func (tr *Tracker) MarkServed(sessionID string) {
+	if tr == nil || tr.rdb == nil {
+		return
+	}
+	tr.mu.Lock()
+	if _, known := tr.records[sessionID]; known {
+		tr.touched[sessionID] = time.Now()
+	}
+	tr.mu.Unlock()
+}
+
 // Remove deletes a session from Redis and the in-memory set.
 func (tr *Tracker) Remove(ctx context.Context, sessionID string) {
 	if tr.rdb == nil {
@@ -164,6 +297,8 @@ func (tr *Tracker) Remove(ctx context.Context, sessionID string) {
 	tr.mu.Lock()
 	delete(tr.sessions, sessionID)
 	delete(tr.touched, sessionID)
+	delete(tr.records, sessionID)
+	delete(tr.bytes, sessionID)
 	tr.mu.Unlock()
 
 	if err := tr.rdb.Del(ctx, tr.redisKey(sessionID)).Err(); err != nil {
@@ -188,6 +323,8 @@ func (tr *Tracker) Cleanup(ctx context.Context) {
 	}
 	tr.sessions = make(map[string]struct{})
 	tr.touched = make(map[string]time.Time)
+	tr.records = make(map[string]SessionInfo)
+	tr.bytes = make(map[string]int64)
 	tr.mu.Unlock()
 
 	if len(ids) == 0 {
@@ -225,31 +362,53 @@ func (tr *Tracker) StartRefresh(ctx context.Context) {
 
 func (tr *Tracker) refreshAll(ctx context.Context) {
 	now := time.Now()
+	type flush struct {
+		id   string
+		data []byte
+	}
 	tr.mu.Lock()
-	ids := make([]string, 0, len(tr.sessions)+len(tr.touched))
+	ids := make([]string, 0, len(tr.sessions))
 	for id := range tr.sessions {
 		ids = append(ids, id)
 	}
 	for id, last := range tr.touched {
-		if now.Sub(last) > sessionTTL {
-			// Idle ephemeral session: stop refreshing and let the Redis
-			// key expire on its own.
+		_, isSession := tr.sessions[id]
+		if !isSession && now.Sub(last) > sessionTTL {
+			// Idle ephemeral session: stop refreshing and let the Redis key
+			// expire on its own. Session-backed entries (direct/remux) are pruned
+			// on Remove, never by idle timeout, so a quiet-but-open pour stays live.
 			delete(tr.touched, id)
+			delete(tr.records, id)
+			delete(tr.bytes, id)
 			continue
 		}
-		if _, dup := tr.sessions[id]; !dup {
+		if !isSession {
 			ids = append(ids, id)
 		}
 	}
+	flushes := make([]flush, 0, len(ids))
+	for _, id := range ids {
+		rec, ok := tr.records[id]
+		if !ok {
+			continue
+		}
+		data, err := json.Marshal(tr.enrichLocked(id, rec))
+		if err != nil {
+			continue
+		}
+		flushes = append(flushes, flush{id: id, data: data})
+	}
 	tr.mu.Unlock()
 
-	if len(ids) == 0 {
+	if len(flushes) == 0 {
 		return
 	}
 
+	// Re-SET (not just EXPIRE) so LastServedAt/BytesServed stay current in the
+	// monitoring record. 30s cadence keeps this cheap.
 	pipe := tr.rdb.Pipeline()
-	for _, id := range ids {
-		pipe.Expire(ctx, tr.redisKey(id), sessionTTL)
+	for _, f := range flushes {
+		pipe.Set(ctx, tr.redisKey(f.id), f.data, sessionTTL)
 	}
 	if _, err := pipe.Exec(ctx); err != nil {
 		slog.DebugContext(ctx, "session refresh pipeline failed", "component", "nodesessions", "error", err)

internal/playback/session.go

@@ -135,6 +135,14 @@ var ErrSessionReplacementSuperseded = errors.New("session replacement was supers
 
 type clientInfoContextKey struct{}
 
+// Origin identifies the protocol a playback session was started through, for
+// first-class monitoring attribution (a jellycompat session shares this manager
+// with native ones and is otherwise indistinguishable in the live snapshot).
+const (
+	OriginNative   = "native"
+	OriginJellyfin = "jellycompat"
+)
+
 // ClientInfo carries best-effort client metadata from request handling into
 // the playback session manager.
 type ClientInfo struct {
@@ -294,6 +302,17 @@ func normalizeClientMetadataValue(value string, maxLen int) string {
 	return value
 }
 
+// Origin reports the protocol this session was started through, as the label
+// monitoring and stream tokens attribute by. It derives from the stored
+// IsJellyfinCompat identity so the two can never disagree; native is the
+// historical default for anything not minted through the compat layer.
+func (s *Session) Origin() string {
+	if s == nil || !s.IsJellyfinCompat {
+		return OriginNative
+	}
+	return OriginJellyfin
+}
+
 // StartSession creates a new playback session using the same file as both the
 // requested and effective source.
 func (m *SessionManager) StartSession(userID int, profileID string, fileID int, method PlayMethod, transcodeAudio bool) (*Session, error) {

internal/playback/session_test.go

@@ -864,6 +864,35 @@ func TestStartSessionWithFiles(t *testing.T) {
 	}
 }
 
+func TestSessionOriginFromContext(t *testing.T) {
+	mgr := playback.NewSessionManager(0, 0)
+
+	// A jellycompat-originated session records its origin for monitor attribution.
+	ctx := playback.WithClientInfo(context.Background(), playback.ClientInfo{
+		Name:     "Infuse",
+		IsCompat: true,
+	})
+	s, err := mgr.StartSessionWithContext(ctx, 1, "p1", 100, playback.PlayTranscode, false)
+	if err != nil {
+		t.Fatalf("StartSessionWithContext: %v", err)
+	}
+	if s.Origin() != playback.OriginJellyfin {
+		t.Errorf("Origin = %q, want %q", s.Origin(), playback.OriginJellyfin)
+	}
+	if s.ClientName != "Infuse" {
+		t.Errorf("ClientName = %q, want Infuse", s.ClientName)
+	}
+
+	// An unset origin defaults to native (historical behavior), never empty.
+	s2, err := mgr.StartSession(2, "p2", 101, playback.PlayDirect, false)
+	if err != nil {
+		t.Fatalf("StartSession: %v", err)
+	}
+	if s2.Origin() != playback.OriginNative {
+		t.Errorf("default Origin = %q, want %q", s2.Origin(), playback.OriginNative)
+	}
+}
+
 func TestUpdateStreamState(t *testing.T) {
 	mgr := playback.NewSessionManager(0, 0)
 	session, err := mgr.StartSession(1, "profile-1", 100, playback.PlayRemux, false)

internal/streammonitor/monitor.go

@@ -0,0 +1,386 @@
+// Package streammonitor produces a normalized "live streams" snapshot grouped by
+// user from the authoritative server-side monitoring records written by
+// internal/nodesessions. It backs an async enforcement loop and admin views.
+//
+// EXISTENCE is server-observed on every path and never gated by a client report,
+// so a "hidden stream" (a disguised client that pulls bytes but withholds
+// progress) is always counted:
+//   - Edge (multi-node): a record exists in Redis for the whole connection
+//     (direct/remux Track..Remove) or while segments are pulled (transcode Touch),
+//     and BytesServed/LastServedAt advance only on real bytes.
+//   - Integrated: the FuncSource reflects SessionManager.AllSessions(), and a
+//     session is unreapable while it holds an in-flight transport marker
+//     (BeginTransport/EndTransport around every byte pour) — no client progress
+//     required to stay visible.
+//
+// TIMING is a secondary signal. On the edge, LastServedAt is purely byte-observed.
+// In integrated mode LastServedAt is mapped from SessionManager.LastActivityAt,
+// which client progress reports also advance; that is acceptable because it is
+// used only to order over-cap victims (selectVictims), never to decide whether a
+// stream exists or is counted. See internal/nodesessions for record production.
+package streammonitor
+
+import (
+	"context"
+	"encoding/json"
+	"log/slog"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/Silo-Server/silo-server/internal/nodesessions"
+)
+
+// sessionKeyPrefix is the Redis key prefix under which nodesessions.Tracker
+// stores its SessionInfo records — the tracker's own exported constant, so the
+// write and read sides can never drift.
+const sessionKeyPrefix = nodesessions.KeyPrefix
+
+// scanCount is the COUNT hint passed to SCAN. It bounds the amount of work per
+// round trip while keeping the number of round trips reasonable.
+const scanCount = 256
+
+// LiveStream is a normalized view of a single active streaming session.
+type LiveStream struct {
+	SessionID    string
+	UserID       int // from SessionInfo.AuthUserID
+	ProfileID    string
+	NodeName     string
+	NodeURL      string
+	Type         string // play method: direct_play | remux | transcode
+	Route        string // origin protocol: native | jellycompat
+	MediaFileID  int
+	ClientIP     string
+	ClientName   string
+	Position     float64 // last known playback position (seconds); secondary timing
+	HWAccel      string
+	LastServedAt time.Time // parsed from SessionInfo.LastServedAt (zero if absent)
+	BytesServed  int64
+	StartedAt    time.Time // parsed from SessionInfo.StartedAt (zero if unparseable)
+}
+
+// Snapshot is a point-in-time picture of the live streams.
+type Snapshot struct {
+	Streams []LiveStream
+}
+
+// CountByUser returns the number of live streams owned by userID.
+func (s Snapshot) CountByUser(userID int) int {
+	n := 0
+	for _, st := range s.Streams {
+		if st.UserID == userID {
+			n++
+		}
+	}
+	return n
+}
+
+// StreamsForUser returns the live streams owned by userID.
+func (s Snapshot) StreamsForUser(userID int) []LiveStream {
+	out := make([]LiveStream, 0)
+	for _, st := range s.Streams {
+		if st.UserID == userID {
+			out = append(out, st)
+		}
+	}
+	return out
+}
+
+// ByUser groups the live streams by owning user id.
+func (s Snapshot) ByUser() map[int][]LiveStream {
+	out := make(map[int][]LiveStream)
+	for _, st := range s.Streams {
+		out[st.UserID] = append(out[st.UserID], st)
+	}
+	return out
+}
+
+// Source yields the current live picture.
+type Source interface {
+	Snapshot(ctx context.Context) (Snapshot, error)
+}
+
+// MultiSource unions several sources into one snapshot, de-duplicating sessions
+// that appear in more than one backend (e.g. a session held both in the central
+// session manager and mirrored by an edge's Redis record) by keeping the most-
+// recently-served copy. A source that errors is skipped (logged), so one
+// unavailable backend never blinds the enforcer to the others. This is what lets
+// the enforcer see BOTH locally-served (integrated) and edge-served (multi-node)
+// streams regardless of whether Redis is configured.
+type MultiSource struct {
+	sources []Source
+}
+
+// NewMultiSource builds a union source over the given sources (nil entries are
+// ignored).
+func NewMultiSource(sources ...Source) *MultiSource {
+	return &MultiSource{sources: sources}
+}
+
+// Snapshot merges every sub-source's snapshot, de-duplicated by session id.
+func (m *MultiSource) Snapshot(ctx context.Context) (Snapshot, error) {
+	var all []LiveStream
+	for _, src := range m.sources {
+		if src == nil {
+			continue
+		}
+		snap, err := src.Snapshot(ctx)
+		if err != nil {
+			slog.Warn("streammonitor: source snapshot failed; skipping", "error", err)
+			continue
+		}
+		all = append(all, snap.Streams...)
+	}
+	return Snapshot{Streams: mergeStreams(all)}, nil
+}
+
+// toLiveStream converts a nodesessions.SessionInfo into a LiveStream, parsing
+// the RFC3339 timestamps and tolerating empty/unparseable values (which map to
+// the zero time).
+func toLiveStream(info nodesessions.SessionInfo) LiveStream {
+	return LiveStream{
+		SessionID:    info.SessionID,
+		UserID:       info.AuthUserID,
+		ProfileID:    info.ProfileID,
+		NodeName:     info.NodeName,
+		NodeURL:      info.NodeURL,
+		Type:         info.Type,
+		Route:        info.Route,
+		MediaFileID:  info.MediaFileID,
+		ClientIP:     info.ClientIP,
+		ClientName:   info.ClientName,
+		Position:     info.Position,
+		HWAccel:      info.HWAccel,
+		LastServedAt: parseTime(info.LastServedAt),
+		BytesServed:  info.BytesServed,
+		StartedAt:    parseTime(info.StartedAt),
+	}
+}
+
+// parseTime parses an RFC3339 timestamp, returning the zero time for empty or
+// unparseable input.
+func parseTime(s string) time.Time {
+	if s == "" {
+		return time.Time{}
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		slog.Debug("streammonitor: unparseable timestamp", "value", s, "error", err)
+		return time.Time{}
+	}
+	return t
+}
+
+// mergeStreams collapses records that share a SessionID (the same session can be
+// tracked by more than one node — e.g. a proxy record and a transcode-node
+// record), keeping the one with the most recent LastServedAt. Records with a
+// distinct SessionID are all retained. The relative order of the kept records is
+// not guaranteed.
+//
+// Ownership and attribution are carried forward independently of the freshness
+// pick: the transcode node's own start record has no resolved owner (UserID 0)
+// and thinner attribution, while the proxy record fronting it does. If the
+// ownerless node record happened to be the freshest copy, taking it wholesale
+// would bucket the session under user 0 (which the enforcer skips — silently
+// exempting it from the concurrency cap) and drop route/client detail from the
+// monitor view. So a merged record adopts a resolved owner from either candidate
+// and backfills any empty attribution/display field from the other copy.
+func mergeStreams(streams []LiveStream) []LiveStream {
+	bySession := make(map[string]LiveStream, len(streams))
+	for _, st := range streams {
+		existing, ok := bySession[st.SessionID]
+		if !ok {
+			bySession[st.SessionID] = st
+			continue
+		}
+		winner := existing
+		other := st
+		if st.LastServedAt.After(existing.LastServedAt) {
+			winner, other = st, existing
+		}
+		// If the freshest copy is ownerless, adopt a resolved owner from either
+		// candidate so the session is still attributed (and enforced) correctly.
+		if winner.UserID <= 0 {
+			for _, cand := range []LiveStream{existing, st} {
+				if cand.UserID > 0 {
+					winner.UserID = cand.UserID
+					winner.ProfileID = cand.ProfileID
+					winner.MediaFileID = cand.MediaFileID
+					break
+				}
+			}
+		}
+		// Backfill display/attribution fields the freshest copy lacks so the
+		// merged record is as complete as possible for the monitor view.
+		if winner.Route == "" {
+			winner.Route = other.Route
+		}
+		if winner.ClientIP == "" {
+			winner.ClientIP = other.ClientIP
+		}
+		if winner.ClientName == "" {
+			winner.ClientName = other.ClientName
+		}
+		if winner.HWAccel == "" {
+			winner.HWAccel = other.HWAccel
+		}
+		if winner.Position == 0 {
+			winner.Position = other.Position
+		}
+		bySession[st.SessionID] = winner
+	}
+	out := make([]LiveStream, 0, len(bySession))
+	for _, st := range bySession {
+		out = append(out, st)
+	}
+	return out
+}
+
+// DedupeSessionInfos collapses raw monitoring records that share a SessionID,
+// applying the same rules as mergeStreams — keep the most-recently-served copy,
+// carry a resolved owner forward, backfill missing attribution — but preserving
+// the SessionInfo shape for surfaces whose wire format IS the raw record (the
+// admin session list, which unions Redis edge records with the in-process
+// integrated sessions and would otherwise show the same stream twice). Kept as
+// a sibling of mergeStreams rather than a shared generic because mergeStreams
+// operates on the parsed LiveStream form; keep the two rule sets in sync.
+func DedupeSessionInfos(infos []nodesessions.SessionInfo) []nodesessions.SessionInfo {
+	bySession := make(map[string]nodesessions.SessionInfo, len(infos))
+	order := make([]string, 0, len(infos))
+	for _, in := range infos {
+		existing, ok := bySession[in.SessionID]
+		if !ok {
+			bySession[in.SessionID] = in
+			order = append(order, in.SessionID)
+			continue
+		}
+		winner, other := existing, in
+		if parseTime(in.LastServedAt).After(parseTime(existing.LastServedAt)) {
+			winner, other = in, existing
+		}
+		if winner.AuthUserID <= 0 && other.AuthUserID > 0 {
+			winner.AuthUserID = other.AuthUserID
+			winner.ProfileID = other.ProfileID
+			winner.MediaFileID = other.MediaFileID
+		}
+		if winner.Route == "" {
+			winner.Route = other.Route
+		}
+		if winner.ClientIP == "" {
+			winner.ClientIP = other.ClientIP
+		}
+		if winner.ClientName == "" {
+			winner.ClientName = other.ClientName
+		}
+		if winner.HWAccel == "" {
+			winner.HWAccel = other.HWAccel
+		}
+		if winner.Position == 0 {
+			winner.Position = other.Position
+		}
+		bySession[in.SessionID] = winner
+	}
+	out := make([]nodesessions.SessionInfo, 0, len(bySession))
+	for _, id := range order {
+		out = append(out, bySession[id])
+	}
+	return out
+}
+
+// RedisSource reads silo:sessions:* records, producing the multi-node
+// authoritative picture.
+type RedisSource struct {
+	rdb *redis.Client
+}
+
+// NewRedisSource creates a RedisSource backed by rdb.
+func NewRedisSource(rdb *redis.Client) *RedisSource {
+	return &RedisSource{rdb: rdb}
+}
+
+// Snapshot SCANs every silo:sessions:* record, decodes it, and returns the
+// deduped live picture. The same session appearing on multiple nodes is
+// collapsed to the record with the most recent LastServedAt.
+func (r *RedisSource) Snapshot(ctx context.Context) (Snapshot, error) {
+	if r.rdb == nil {
+		return Snapshot{Streams: []LiveStream{}}, nil
+	}
+
+	// Fully iterate the cursor to collect matching keys. SCAN (not KEYS) keeps
+	// this non-blocking against a large keyspace.
+	var keys []string
+	var cursor uint64
+	for {
+		batch, next, err := r.rdb.Scan(ctx, cursor, sessionKeyPrefix+"*", scanCount).Result()
+		if err != nil {
+			return Snapshot{}, err
+		}
+		keys = append(keys, batch...)
+		cursor = next
+		if cursor == 0 {
+			break
+		}
+	}
+
+	if len(keys) == 0 {
+		return Snapshot{Streams: []LiveStream{}}, nil
+	}
+
+	vals, err := r.rdb.MGet(ctx, keys...).Result()
+	if err != nil {
+		return Snapshot{}, err
+	}
+
+	streams := make([]LiveStream, 0, len(vals))
+	for i, v := range vals {
+		if v == nil {
+			// Key expired between SCAN and MGET; skip.
+			continue
+		}
+		var raw string
+		switch val := v.(type) {
+		case string:
+			raw = val
+		case []byte:
+			raw = string(val)
+		default:
+			slog.Debug("streammonitor: unexpected redis value type", "key", keys[i])
+			continue
+		}
+		var info nodesessions.SessionInfo
+		if err := json.Unmarshal([]byte(raw), &info); err != nil {
+			slog.Debug("streammonitor: unmarshal session record failed", "key", keys[i], "error", err)
+			continue
+		}
+		streams = append(streams, toLiveStream(info))
+	}
+
+	return Snapshot{Streams: mergeStreams(streams)}, nil
+}
+
+// FuncSource adapts an in-process provider (integrated single-node) that returns
+// the local tracker records.
+type FuncSource struct {
+	fn func(ctx context.Context) ([]nodesessions.SessionInfo, error)
+}
+
+// NewFuncSource creates a FuncSource backed by fn.
+func NewFuncSource(fn func(ctx context.Context) ([]nodesessions.SessionInfo, error)) *FuncSource {
+	return &FuncSource{fn: fn}
+}
+
+// Snapshot invokes the wrapped provider and returns the deduped live picture.
+func (f *FuncSource) Snapshot(ctx context.Context) (Snapshot, error) {
+	if f.fn == nil {
+		return Snapshot{Streams: []LiveStream{}}, nil
+	}
+	infos, err := f.fn(ctx)
+	if err != nil {
+		return Snapshot{}, err
+	}
+	streams := make([]LiveStream, 0, len(infos))
+	for _, info := range infos {
+		streams = append(streams, toLiveStream(info))
+	}
+	return Snapshot{Streams: mergeStreams(streams)}, nil
+}

internal/streammonitor/monitor_test.go

@@ -0,0 +1,276 @@
+package streammonitor
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/Silo-Server/silo-server/internal/nodesessions"
+)
+
+func fakeFn(infos []nodesessions.SessionInfo) func(ctx context.Context) ([]nodesessions.SessionInfo, error) {
+	return func(ctx context.Context) ([]nodesessions.SessionInfo, error) {
+		return infos, nil
+	}
+}
+
+func TestFuncSourceGrouping(t *testing.T) {
+	infos := []nodesessions.SessionInfo{
+		{SessionID: "s1", AuthUserID: 1, Type: "direct_play", StartedAt: "2026-07-04T10:00:00Z", LastServedAt: "2026-07-04T10:05:00Z"},
+		{SessionID: "s2", AuthUserID: 1, Type: "transcode", StartedAt: "2026-07-04T10:01:00Z", LastServedAt: "2026-07-04T10:06:00Z"},
+		{SessionID: "s3", AuthUserID: 2, Type: "remux", StartedAt: "2026-07-04T10:02:00Z", LastServedAt: "2026-07-04T10:07:00Z"},
+		{SessionID: "s4", AuthUserID: 0, Type: "direct_play"}, // no user, no timestamps
+	}
+
+	snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background())
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+
+	if got := len(snap.Streams); got != 4 {
+		t.Fatalf("Streams len = %d, want 4", got)
+	}
+
+	if got := snap.CountByUser(1); got != 2 {
+		t.Errorf("CountByUser(1) = %d, want 2", got)
+	}
+	if got := snap.CountByUser(2); got != 1 {
+		t.Errorf("CountByUser(2) = %d, want 1", got)
+	}
+	if got := snap.CountByUser(0); got != 1 {
+		t.Errorf("CountByUser(0) = %d, want 1", got)
+	}
+	if got := snap.CountByUser(99); got != 0 {
+		t.Errorf("CountByUser(99) = %d, want 0", got)
+	}
+
+	u1 := snap.StreamsForUser(1)
+	if len(u1) != 2 {
+		t.Fatalf("StreamsForUser(1) len = %d, want 2", len(u1))
+	}
+	for _, st := range u1 {
+		if st.UserID != 1 {
+			t.Errorf("StreamsForUser(1) returned stream with UserID %d", st.UserID)
+		}
+	}
+
+	byUser := snap.ByUser()
+	if len(byUser) != 3 {
+		t.Fatalf("ByUser len = %d, want 3 (users 0,1,2)", len(byUser))
+	}
+	if len(byUser[1]) != 2 || len(byUser[2]) != 1 || len(byUser[0]) != 1 {
+		t.Errorf("ByUser grouping wrong: %#v", map[int]int{0: len(byUser[0]), 1: len(byUser[1]), 2: len(byUser[2])})
+	}
+}
+
+func TestTimestampParsing(t *testing.T) {
+	infos := []nodesessions.SessionInfo{
+		{SessionID: "s1", AuthUserID: 1, StartedAt: "2026-07-04T10:00:00Z", LastServedAt: "2026-07-04T10:05:00Z"},
+		{SessionID: "s2", AuthUserID: 1, StartedAt: "not-a-time", LastServedAt: ""},
+	}
+	snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background())
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+	bySession := map[string]LiveStream{}
+	for _, st := range snap.Streams {
+		bySession[st.SessionID] = st
+	}
+
+	want := time.Date(2026, 7, 4, 10, 5, 0, 0, time.UTC)
+	if !bySession["s1"].LastServedAt.Equal(want) {
+		t.Errorf("s1 LastServedAt = %v, want %v", bySession["s1"].LastServedAt, want)
+	}
+	if !bySession["s2"].StartedAt.IsZero() {
+		t.Errorf("s2 StartedAt = %v, want zero (unparseable)", bySession["s2"].StartedAt)
+	}
+	if !bySession["s2"].LastServedAt.IsZero() {
+		t.Errorf("s2 LastServedAt = %v, want zero (empty)", bySession["s2"].LastServedAt)
+	}
+}
+
+func TestDedupeKeepsNewest(t *testing.T) {
+	// Same SessionID observed on two nodes (proxy vs transcode node). Keep the
+	// record with the most recent LastServedAt.
+	infos := []nodesessions.SessionInfo{
+		{SessionID: "dup", AuthUserID: 5, NodeName: "proxy", LastServedAt: "2026-07-04T10:00:00Z", BytesServed: 100},
+		{SessionID: "dup", AuthUserID: 5, NodeName: "transcode", LastServedAt: "2026-07-04T10:09:00Z", BytesServed: 900},
+		{SessionID: "other", AuthUserID: 5, NodeName: "proxy", LastServedAt: "2026-07-04T10:03:00Z"},
+	}
+	snap, err := NewFuncSource(fakeFn(infos)).Snapshot(context.Background())
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+
+	if got := len(snap.Streams); got != 2 {
+		t.Fatalf("Streams len = %d, want 2 (deduped)", got)
+	}
+	if got := snap.CountByUser(5); got != 2 {
+		t.Errorf("CountByUser(5) = %d, want 2", got)
+	}
+
+	var dup LiveStream
+	found := false
+	for _, st := range snap.Streams {
+		if st.SessionID == "dup" {
+			dup = st
+			found = true
+		}
+	}
+	if !found {
+		t.Fatal("deduped session 'dup' not present")
+	}
+	if dup.NodeName != "transcode" {
+		t.Errorf("dedupe kept NodeName %q, want transcode (newest LastServedAt)", dup.NodeName)
+	}
+	if dup.BytesServed != 900 {
+		t.Errorf("dedupe kept BytesServed %d, want 900", dup.BytesServed)
+	}
+}
+
+func TestMergeStreamsDirect(t *testing.T) {
+	// Direct unit test of the merge helper, mirroring RedisSource dedupe.
+	in := []LiveStream{
+		{SessionID: "a", NodeName: "n1", LastServedAt: time.Unix(100, 0)},
+		{SessionID: "a", NodeName: "n2", LastServedAt: time.Unix(200, 0)},
+		{SessionID: "b", NodeName: "n1", LastServedAt: time.Unix(150, 0)},
+	}
+	out := mergeStreams(in)
+	if len(out) != 2 {
+		t.Fatalf("mergeStreams len = %d, want 2", len(out))
+	}
+	for _, st := range out {
+		if st.SessionID == "a" && st.NodeName != "n2" {
+			t.Errorf("merge kept %q for session a, want n2 (newest)", st.NodeName)
+		}
+	}
+}
+
+func TestMergeStreamsCarriesOwnershipForward(t *testing.T) {
+	// The transcode node's own start record has no resolved owner (UserID 0) and
+	// can be the freshest copy of a session. Taking it wholesale would bucket the
+	// session under user 0, which the enforcer skips — exempting it from the cap.
+	// The merge must recover the owner from the proxy's (staler) owned record.
+	in := []LiveStream{
+		{SessionID: "s", NodeName: "proxy", UserID: 42, ProfileID: "p1", MediaFileID: 7, LastServedAt: time.Unix(100, 0)},
+		{SessionID: "s", NodeName: "transcode", UserID: 0, LastServedAt: time.Unix(200, 0)},
+	}
+	out := mergeStreams(in)
+	if len(out) != 1 {
+		t.Fatalf("mergeStreams len = %d, want 1", len(out))
+	}
+	got := out[0]
+	if got.NodeName != "transcode" {
+		t.Errorf("kept NodeName %q, want transcode (freshest)", got.NodeName)
+	}
+	if got.UserID != 42 {
+		t.Errorf("UserID = %d, want 42 (recovered from the proxy record)", got.UserID)
+	}
+	if got.ProfileID != "p1" || got.MediaFileID != 7 {
+		t.Errorf("ownership not fully recovered: ProfileID=%q MediaFileID=%d", got.ProfileID, got.MediaFileID)
+	}
+}
+
+func TestToLiveStreamCarriesRouteAndClient(t *testing.T) {
+	// The normalized view must surface route (native vs jellycompat) and client
+	// identity so the monitor is first-class, not just session id + method.
+	info := nodesessions.SessionInfo{
+		SessionID:  "s1",
+		AuthUserID: 9,
+		Type:       "transcode",
+		Route:      "jellycompat",
+		ClientIP:   "203.0.113.7",
+		ClientName: "Infuse",
+		Position:   61.5,
+		HWAccel:    "vaapi",
+	}
+	got := toLiveStream(info)
+	if got.Route != "jellycompat" || got.ClientIP != "203.0.113.7" || got.ClientName != "Infuse" {
+		t.Fatalf("route/client not carried: %+v", got)
+	}
+	if got.Position != 61.5 || got.HWAccel != "vaapi" {
+		t.Fatalf("position/hwaccel not carried: %+v", got)
+	}
+}
+
+func TestMergeStreamsBackfillsAttribution(t *testing.T) {
+	// A fresher-but-thinner record (e.g. an ownerless transcode-node start record)
+	// must not drop route/client that a staler record for the same session carries.
+	in := []LiveStream{
+		{SessionID: "s", NodeName: "proxy", UserID: 5, Route: "native", ClientName: "SiloTV", ClientIP: "10.0.0.9", LastServedAt: time.Unix(100, 0)},
+		{SessionID: "s", NodeName: "transcode", UserID: 0, LastServedAt: time.Unix(200, 0)},
+	}
+	out := mergeStreams(in)
+	if len(out) != 1 {
+		t.Fatalf("merge len = %d, want 1", len(out))
+	}
+	got := out[0]
+	if got.NodeName != "transcode" {
+		t.Errorf("kept NodeName %q, want transcode (freshest)", got.NodeName)
+	}
+	if got.UserID != 5 || got.Route != "native" || got.ClientName != "SiloTV" || got.ClientIP != "10.0.0.9" {
+		t.Errorf("attribution not backfilled from staler record: %+v", got)
+	}
+}
+
+func TestFuncSourceNilFn(t *testing.T) {
+	snap, err := NewFuncSource(nil).Snapshot(context.Background())
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+	if len(snap.Streams) != 0 {
+		t.Errorf("nil fn Streams len = %d, want 0", len(snap.Streams))
+	}
+}
+
+func TestRedisSourceNilClient(t *testing.T) {
+	snap, err := NewRedisSource(nil).Snapshot(context.Background())
+	if err != nil {
+		t.Fatalf("Snapshot: %v", err)
+	}
+	if len(snap.Streams) != 0 {
+		t.Errorf("nil client Streams len = %d, want 0", len(snap.Streams))
+	}
+}
+
+// TestDedupeSessionInfos mirrors the mergeStreams rules on the raw SessionInfo
+// shape used by the admin session list: one row per session id, freshest copy
+// wins, a resolved owner and missing attribution are carried across the merge
+// so the operator view matches the enforcer's picture (no double-counted
+// streams, no user-0 rows when any copy knows the owner).
+func TestDedupeSessionInfos(t *testing.T) {
+	newer := time.Now().UTC().Format(time.RFC3339)
+	older := time.Now().Add(-time.Minute).UTC().Format(time.RFC3339)
+	infos := []nodesessions.SessionInfo{
+		{SessionID: "s1", NodeName: "central", AuthUserID: 7, ProfileID: "p1", MediaFileID: 42,
+			Route: "native", ClientName: "SiloTV", Position: 130, LastServedAt: older},
+		// The same stream, seen from the edge serving it: freshest but ownerless.
+		{SessionID: "s1", NodeName: "edge-1", LastServedAt: newer, BytesServed: 9000},
+		{SessionID: "s2", NodeName: "edge-1", AuthUserID: 8, LastServedAt: newer},
+	}
+
+	out := DedupeSessionInfos(infos)
+	if len(out) != 2 {
+		t.Fatalf("dedupe len = %d, want 2", len(out))
+	}
+	byID := make(map[string]nodesessions.SessionInfo, len(out))
+	for _, info := range out {
+		byID[info.SessionID] = info
+	}
+	s1, ok := byID["s1"]
+	if !ok {
+		t.Fatalf("s1 missing from dedupe output")
+	}
+	if s1.NodeName != "edge-1" || s1.BytesServed != 9000 {
+		t.Errorf("s1 freshest copy not kept: %+v", s1)
+	}
+	if s1.AuthUserID != 7 || s1.ProfileID != "p1" || s1.MediaFileID != 42 {
+		t.Errorf("s1 owner not carried across merge: %+v", s1)
+	}
+	if s1.Route != "native" || s1.ClientName != "SiloTV" || s1.Position != 130 {
+		t.Errorf("s1 attribution not backfilled: %+v", s1)
+	}
+	if s2 := byID["s2"]; s2.AuthUserID != 8 {
+		t.Errorf("s2 mangled by dedupe: %+v", s2)
+	}
+}

internal/streamtoken/token.go

@@ -43,6 +43,13 @@ type Claims struct {
 	ProfileID   string `json:"pid,omitempty"`
 	MediaFileID int    `json:"mfid,omitempty"`
 
+	// Monitoring attribution (not byte-affecting, not a trust assertion): the
+	// origin protocol ("native" | "jellycompat") and the reported client/app
+	// name, carried so an edge node — which never sees the originating API path —
+	// can stamp them onto its live-session record for the first-class monitor view.
+	Origin     string `json:"org,omitempty"`
+	ClientName string `json:"cn,omitempty"`
+
 	// Reconstruction recipe — the byte-affecting encode parameters, mirroring the
 	// former playback.RecipeCard. Zero for direct/remux tokens, which reconstruct
 	// from identity alone plus the client-supplied position.

internal/transcodenode/server.go

@@ -48,6 +48,15 @@ type TranscodeStartRequest struct {
 	SubtitleCodec          string  `json:"subtitle_codec,omitempty"`
 	TotalDuration          float64 `json:"total_duration"`
 	RequireReady           bool    `json:"require_ready,omitempty"`
+
+	// Monitoring attribution copied from the originating session/token so the
+	// node's live-session record is owner-attributed and route-tagged even when
+	// no proxy record fronts it (closes the ownerless-record window).
+	AuthUserID  int    `json:"auth_user_id,omitempty"`
+	ProfileID   string `json:"profile_id,omitempty"`
+	MediaFileID int    `json:"media_file_id,omitempty"`
+	Route       string `json:"route,omitempty"`
+	ClientName  string `json:"client_name,omitempty"`
 }
 
 // TranscodeStartResponse is the JSON response for POST /transcode/start.
@@ -567,15 +576,20 @@ func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) {
 	effectiveHWAccel := session.Opts().HWAccel
 	trackCtx := context.WithoutCancel(r.Context())
 	go s.tracker.Track(trackCtx, nodesessions.SessionInfo{
-		SessionID:  req.SessionID,
-		NodeURL:    s.tracker.NodeURL(),
-		NodeName:   s.tracker.NodeName(),
-		Type:       "transcode",
-		CodecVideo: req.TargetCodecVideo,
-		CodecAudio: req.TargetCodecAudio,
-		Resolution: req.TargetResolution,
-		HWAccel:    effectiveHWAccel,
-		StartedAt:  time.Now().UTC().Format(time.RFC3339),
+		SessionID:   req.SessionID,
+		NodeURL:     s.tracker.NodeURL(),
+		NodeName:    s.tracker.NodeName(),
+		Type:        "transcode",
+		CodecVideo:  req.TargetCodecVideo,
+		CodecAudio:  req.TargetCodecAudio,
+		Resolution:  req.TargetResolution,
+		HWAccel:     effectiveHWAccel,
+		StartedAt:   time.Now().UTC().Format(time.RFC3339),
+		AuthUserID:  req.AuthUserID,
+		ProfileID:   req.ProfileID,
+		MediaFileID: req.MediaFileID,
+		Route:       req.Route,
+		ClientName:  req.ClientName,
 	})
 
 	w.WriteHeader(http.StatusAccepted)
@@ -651,7 +665,7 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request
 			}
 			resolved = *fetched
 		}
-		return s.spawnReconstruct(r, sessionID, requestedSegment, resolved), nil
+		return s.spawnReconstruct(r, sessionID, requestedSegment, resolved, claims.Origin, claims.ClientName), nil
 	})
 	if session, _ := v.(*playback.TranscodeSession); session != nil {
 		return session
@@ -663,7 +677,7 @@ func (s *Server) reconstructFromToken(r *http.Request, sessionID string, request
 // registers it in the live map. It is only ever called inside the per-session
 // single-flight in reconstructFromToken, so it is the sole writer racing to
 // register sessionID. Returns nil if the spawn fails or the slot wait is canceled.
-func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSegment int, card playback.RecipeCard) *playback.TranscodeSession {
+func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSegment int, card playback.RecipeCard, route, clientName string) *playback.TranscodeSession {
 	// Pace the cold-start burst so a node restart that loses many sessions does not
 	// launch every ffmpeg at once. A client that disconnects while waiting releases
 	// its slot rather than queueing dead work.
@@ -749,6 +763,8 @@ func (s *Server) spawnReconstruct(r *http.Request, sessionID string, requestedSe
 		AuthUserID:  card.UserID,
 		ProfileID:   card.ProfileID,
 		MediaFileID: card.MediaFileID,
+		Route:       route,
+		ClientName:  clientName,
 	})
 
 	slog.InfoContext(r.Context(), "transcode node session reconstructed from token", "component", "transcodenode",
@@ -854,6 +870,10 @@ func (s *Server) handleManifest(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
 	w.Write(manifest)
+	// Serve-activity mark so the node's own record reflects real serving instead
+	// of a LastServedAt frozen at start time (bytes are measured by the fronting
+	// proxy; wrapping ServeFile here would cost sendfile for a duplicate count).
+	s.tracker.MarkServed(sessionID)
 }
 
 func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
@@ -983,6 +1003,8 @@ func (s *Server) handleSegment(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Cache-Control", "no-store, max-age=0")
 	w.Header().Set("Pragma", "no-cache")
 	http.ServeFile(w, r, segPath)
+	// Serve-activity mark: see handleManifest.
+	s.tracker.MarkServed(sessionID)
 }
 
 func (s *Server) handleForceReload(w http.ResponseWriter, r *http.Request) {