feat(i18n): the server's closed English vocabularies named in the reader's language (E30 S2)

- lib/vocab + the `vocab` locale area: TMDB crew jobs and departments, auto-tags (D1 A: the data
  stays English, the catalog is named at its one source, useTags), built-in download profiles,
  YouTube topics; held key-for-key to the backend's own lists by backendIds.contract.
- Countries by ISO code: TMDB parse stores {code, name}; rows stored before are coded from TMDB's
  own country list in one request (run_enrich); Intl names them, historic codes (YU, SU, XI…)
  from the catalog.
- Player tracks named from the language tag, forced/SDH flags and codec tail, not Plex's
  server-language displayTitle; no invented "Audio N" on the server.
- Download lifecycle notices (shared / expiring / expired / evicted) named from data; the share
  notice carries the title.
- Scheduler: the job result rides along as data and is named per key (dashboard + run notice).
- Invite status named; a render crash's notice no longer shows the engine's English message.
- Fix: YouTube links …/wiki/Humour, the tagger looked for "Humor" — Comedy never came from a topic.

backend/app/plex/tmdb.py

@@ -12,6 +12,7 @@ the row. The enrich batch + scheduler entry live here too so the whole TMDB inte
 
 import time
 from datetime import datetime, timezone
+from typing import cast
 
 import httpx
 from sqlalchemy import func, select
@@ -125,6 +126,11 @@ class TmdbClient:
         """Resolve an external id (imdb_id / tvdb_id) to TMDB items via /find."""
         return self._get(f"/find/{external_id}", {"external_source": source})
 
+    def countries(self) -> list[dict] | None:
+        """TMDB's country list: `[{iso_3166_1, english_name, native_name}]` (a JSON array, which
+        `_get`'s dict annotation does not describe)."""
+        return cast("list[dict] | None", self._get("/configuration/countries"))
+
     def fetch_person(self, person_id: int) -> dict | None:
         """One person's bio + combined credits (E5 S6), in a single append_to_response request."""
         return self._get(f"/person/{person_id}", {"append_to_response": "combined_credits"})
@@ -255,8 +261,12 @@ def parse_tmdb(payload: dict, media_type: str) -> dict:
             c.get("name") for c in (payload.get("production_companies") or []) if c.get("name")
         ][:8]
         or None,
+        # The ISO code beside TMDB's English name (E30): the client names the country in the reader's
+        # language from the code; the name is the fallback for a code it cannot name.
         "production_countries": [
-            c.get("name") for c in (payload.get("production_countries") or []) if c.get("name")
+            {"code": c.get("iso_3166_1") or None, "name": c["name"]}
+            for c in (payload.get("production_countries") or [])
+            if c.get("name")
         ]
         or None,
         "recommendations": recs or None,
@@ -357,13 +367,49 @@ def enrich_batch(db: Session, tmdb: TmdbClient, limit: int) -> int:
     return processed
 
 
+def _country(c: dict | str) -> dict:
+    """One stored production country as served: `{code, name}`. A row enriched before E30 stored the
+    bare English name; until `backfill_country_codes` reaches it, it serves with no code (the client
+    shows the name)."""
+    return c if isinstance(c, dict) else {"code": None, "name": c}
+
+
+def backfill_country_codes(db: Session, tmdb: TmdbClient) -> int:
+    """Give the countries stored before E30 (bare English names) their ISO codes, from TMDB's own
+    country list — the names came from that list, so they match it exactly. One request, and only
+    while such a row is left: re-fetching each title instead would take the enrichment batch days.
+    A name the list does not hold keeps no code. Returns the number of rows rewritten."""
+    stale = [
+        row
+        for model in (PlexItem, PlexShow)
+        for row in db.execute(
+            select(model).where(func.jsonb_typeof(model.production_countries[0]) == "string")
+        ).scalars()
+    ]
+    if not stale:
+        return 0
+    codes = {
+        c["english_name"]: c["iso_3166_1"]
+        for c in tmdb.countries() or []
+        if c.get("english_name") and c.get("iso_3166_1")
+    }
+    for row in stale:
+        row.production_countries = [
+            {"code": codes.get(c), "name": c} if isinstance(c, str) else c
+            for c in row.production_countries or []
+        ]
+    db.commit()
+    return len(stale)
+
+
 def run_enrich(db: Session) -> dict:
     """Scheduler entry point. No-op (never an error) when the TMDB key isn't configured."""
     if not sysconfig.get_str(db, "tmdb_api_key"):
         return {"enriched": 0, "skipped": "tmdb_api_key not set"}
     with TmdbClient(db) as tmdb:
+        coded = backfill_country_codes(db, tmdb)
         n = enrich_batch(db, tmdb, settings.tmdb_enrich_batch)
-    return {"enriched": n}
+    return {"enriched": n, **({"countries_coded": coded} if coded else {})}
 
 
 def resolve_library_guids(
@@ -445,7 +491,7 @@ def serve_block(db: Session, row, *, lang: str) -> dict | None:
         "keywords": row.keywords or [],
         "crew": row.crew or [],
         "studios": row.studios or [],
-        "production_countries": row.production_countries or [],
+        "production_countries": [_country(c) for c in row.production_countries or []],
         "recommendations": resolve_recs(db, row.recommendations, lang=lang),
         "backdrops": row.backdrops or [],
     }

backend/app/routes/downloads.py

@@ -917,7 +917,14 @@ def share_download(
             user_id=recipient.id,
             type="download_shared",
             title=(job.display_name or "A download") + " was shared with you",
-            data={"kind": "download_shared", "from": user.email, "job_id": job_id},
+            # The client names the notice in the reader's language from `data` (E30); the title is the
+            # English fallback only.
+            data={
+                "kind": "download_shared",
+                "from": user.email,
+                "job_id": job_id,
+                "title": job.display_name,
+            },
             commit=False,
         )
         db.commit()

backend/app/routes/plex/detail.py

@@ -1,5 +1,6 @@
 """Plex show/item detail: the show drill-down + the per-item player payload (markers, cast, nav)."""
 
+import re
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
@@ -342,6 +343,26 @@ def _episode_nav(db: Session, it: PlexItem) -> tuple[str | None, str | None]:
     return (keys[i - 1] if i > 0 else None), (keys[i + 1] if i < len(keys) - 1 else None)
 
 
+# The technical tail of Plex's track title — "English (AC3 5.1)" → "AC3 5.1".
+_TRACK_TAIL = re.compile(r"\(([^()]+)\)\s*$")
+
+
+def _track_naming(s: dict) -> dict:
+    """What the player names a track from (E30). Plex's `displayTitle` is in the SERVER's language
+    ("English (AC3 5.1)", "magyar Forced", "中文 (EAC3 Stereo)"), so it is sent only as the fallback
+    `label`; the client builds the name in the reader's language from the language tag, the
+    forced / SDH flags and the codec tail. No title and no language → no label (the client numbers it)."""
+    title = s.get("displayTitle") or s.get("language") or None
+    tail = _TRACK_TAIL.search(title) if title else None
+    return {
+        "label": title,
+        "lang_tag": s.get("languageTag") or None,
+        "forced": bool(s.get("forced")),
+        "hearing_impaired": bool(s.get("hearingImpaired")),
+        "detail": tail.group(1).strip() if tail else None,
+    }
+
+
 @router.get("/item/{rating_key}")
 def item_detail(
     rating_key: str,
@@ -387,9 +408,7 @@ def item_detail(
                 audio_streams.append(
                     {
                         "ord": len(audio_streams),
-                        "label": s.get("displayTitle")
-                        or s.get("language")
-                        or f"Audio {len(audio_streams) + 1}",
+                        **_track_naming(s),
                         "language": s.get("languageTag") or s.get("language"),
                         "default": bool(s.get("selected") or s.get("default")),
                     }
@@ -399,9 +418,7 @@ def item_detail(
                 subtitle_streams.append(
                     {
                         "ord": len(subtitle_streams),
-                        "label": s.get("displayTitle")
-                        or s.get("language")
-                        or f"Subtitle {len(subtitle_streams) + 1}",
+                        **_track_naming(s),
                         "language": s.get("languageTag") or s.get("language"),
                         "codec": codec,
                         # Text subs (srt/ass/…) can be served as WebVTT tracks; image subs (PGS/VobSub)

backend/app/scheduler.py

@@ -115,7 +115,9 @@ def _record(name: str, **fields) -> None:
         _activity.setdefault(name, {}).update(fields)
 
 
-def _notify_done(db, actor_id: int, job_id: str, status: str, summary: str | None) -> None:
+def _notify_done(
+    db, actor_id: int, job_id: str, status: str, summary: str | None, result: dict | None = None
+) -> None:
     """Post a completion notice to the triggering admin's inbox (manual runs only). Best
     effort — a notification failure must never mask the job's own outcome."""
     try:
@@ -125,7 +127,7 @@ def _notify_done(db, actor_id: int, job_id: str, status: str, summary: str | Non
             type="scheduler",
             title=f"{job_id}: {status}",  # English fallback; UI renders translated from data
             body=summary,
-            data={"job_id": job_id, "status": status, "summary": summary},
+            data={"job_id": job_id, "status": status, "summary": summary, "result": result},
         )
     except Exception:
         db.rollback()
@@ -192,25 +194,41 @@ def _job(name: str, fn) -> None:
                 status="skipped",
                 last_finished=_now_iso(),
                 last_result="paused",
+                result=_PAUSED,
                 progress=None,
             )
             if actor_id is not None:
-                _notify_done(db, actor_id, name, "skipped", "sync paused")
+                _notify_done(db, actor_id, name, "skipped", "sync paused", _PAUSED)
                 _audit_run(db, actor_id, name, "skipped", "sync paused", None)
             return
         with progress.bind(sink):
             result = fn(db)
         summary = _summarize(result)
+        data = _result_data(result)
         log.info("job %s -> %s", name, result)
-        _record(name, status="ok", last_finished=_now_iso(), last_result=summary, progress=None)
+        _record(
+            name,
+            status="ok",
+            last_finished=_now_iso(),
+            last_result=summary,
+            result=data,
+            progress=None,
+        )
         if actor_id is not None:
-            _notify_done(db, actor_id, name, "ok", summary)
+            _notify_done(db, actor_id, name, "ok", summary, data)
         _audit_run(db, actor_id, name, "ok", summary, result)
     except Exception as exc:
         db.rollback()
         log.exception("job %s failed", name, extra=logs.event("scheduler.job.failed", job=name))
         err = str(exc) or exc.__class__.__name__
-        _record(name, status="error", last_finished=_now_iso(), last_error=err, progress=None)
+        _record(
+            name,
+            status="error",
+            last_finished=_now_iso(),
+            last_error=err,
+            result=None,
+            progress=None,
+        )
         if actor_id is not None:
             _notify_done(db, actor_id, name, "error", err)
         _audit_run(db, actor_id, name, "error", err, None)
@@ -221,6 +239,21 @@ def _job(name: str, fn) -> None:
         db.close()
 
 
+# A skipped run's result, in the same shape as a job's own `{"skipped": …}`.
+_PAUSED = {"skipped": "sync paused"}
+
+
+def _result_data(result) -> dict | None:
+    """A job's result as the client names it in the reader's language (E30): the dict itself (counts,
+    a `skipped` / `reason` word, a nested dict per phase), `enrich`'s bare count as one. The one-line
+    `_summarize` is the English fallback — and what the audit log keeps."""
+    if isinstance(result, dict):
+        return result or None
+    if isinstance(result, int) and not isinstance(result, bool):
+        return {"enriched": result}
+    return None
+
+
 def _summarize(result) -> str | None:
     """Compact one-line rendering of a job's return value for the dashboard."""
     if result is None:
@@ -259,6 +292,7 @@ def scheduler_snapshot() -> dict:
                 "last_started": a.get("last_started"),
                 "last_finished": a.get("last_finished"),
                 "last_result": a.get("last_result"),
+                "result": a.get("result"),
                 "last_error": a.get("last_error"),
                 "progress": a.get("progress"),
             }

backend/app/sync/autotag.py

@@ -115,7 +115,9 @@ _TOPIC_SLUGS = {
     "Tourism": "Travel",
     "Fashion": "Fashion & Beauty",
     "Physical_attractiveness": "Fashion & Beauty",
-    "Humor": "Comedy",
+    # YouTube links the British page (…/wiki/Humour); the key was "Humor", so no channel ever got Comedy
+    # from its topic (E30).
+    "Humour": "Comedy",
     "Entertainment": "Entertainment",
     "Film": "Film",
     "Television_program": "TV",

backend/tests/test_autotag_topics.py

@@ -0,0 +1,24 @@
+"""A YouTube topic URL's slug → the auto-tag it gives (pure). The slugs are the ones YouTube actually
+links (measured on the owner's channels, 2026-09-29): a key spelled differently from YouTube's page
+never matches, silently — "Humor" sat in the map for months while YouTube links …/wiki/Humour."""
+
+import pytest
+
+from app.sync.autotag import map_topic_slug
+
+
+@pytest.mark.parametrize(
+    ("slug", "tag"),
+    [
+        ("Humour", "Comedy"),
+        ("Video_game_culture", "Gaming"),
+        ("Electronic_music", "Music"),
+        ("Association_football", "Sports"),
+        ("Knowledge", "Education"),
+        ("Physical_fitness", "Health & Fitness"),
+        ("Television_program", "TV"),
+        ("Lifestyle_(sociology)", None),  # YouTube's catch-all: deliberately no tag
+    ],
+)
+def test_a_youtube_topic_slug_gives_its_tag(slug, tag):
+    assert map_topic_slug(slug) == tag

backend/tests/test_download_share_notice.py

@@ -0,0 +1,41 @@
+"""E30: the "shared with you" notice carries the download's title in its data, so the client can name
+the notice in the reader's language. Its stored title is an English sentence ("Cats was shared with
+you") — the only place the title lived, so a Hungarian reader got the English. Needs the DB."""
+
+from sqlalchemy import select
+
+from app.models import DownloadJob, MediaAsset, Notification
+
+
+def test_a_share_notice_carries_the_title_and_the_sharer(db, make_user, as_user):
+    owner = make_user("owner@example.com")
+    friend = make_user("friend@example.com")
+    asset = MediaAsset(
+        source_kind="url", source_ref="share-e30", format_sig="sig", status="ready", ref_count=1
+    )
+    db.add(asset)
+    db.flush()
+    job = DownloadJob(
+        user_id=owner.id,
+        asset_id=asset.id,
+        source_kind="url",
+        source_ref="share-e30",
+        status="done",
+        queue_pos=1,
+        profile_snapshot={},
+        display_name="Cats",
+    )
+    db.add(job)
+    db.commit()
+
+    r = as_user(owner).post(f"/api/downloads/{job.id}/share", json={"email": "friend@example.com"})
+    assert r.status_code == 200, r.text
+
+    notice = db.execute(select(Notification).where(Notification.user_id == friend.id)).scalar_one()
+    assert notice.type == "download_shared"
+    assert notice.data == {
+        "kind": "download_shared",
+        "from": "owner@example.com",
+        "job_id": job.id,
+        "title": "Cats",
+    }

backend/tests/test_plex_detail.py

@@ -352,3 +352,52 @@ def test_show_detail_lazy_fills_when_unenriched(db, monkeypatch, override_dep):
     assert show.imdb_id == "tt0068646"  # persisted for next time
     assert show.people_enriched_at is not None
     assert [c["name"] for c in show.full_cast] == ["Marlon Brando", "Al Pacino"]
+
+
+def test_item_detail_names_tracks_from_their_language_not_plexs_title(
+    db, monkeypatch, override_dep
+):
+    """E30: Plex titles a track in the Plex SERVER's language ("English (AC3 5.1)", "magyar Forced",
+    "中文 (EAC3 Stereo)"), so a Hungarian reader got a mix. The route sends what the client names it
+    from instead — the language tag, the forced / SDH flags, the codec tail — keeping Plex's title
+    only as the fallback, and no invented English "Audio 2" for a track with neither."""
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    db.add(PlexItem(rating_key="m1", library_id=lib.id, kind="movie", title="T"))
+    db.commit()
+    user = _user(db, "tracks@example.com")
+    db.commit()
+    streams = [
+        {"streamType": 2, "displayTitle": "English (AC3 5.1)", "languageTag": "en-US"},
+        {"streamType": 2, "displayTitle": "中文 (EAC3 Stereo)", "languageTag": "zh"},
+        {"streamType": 2},
+        {
+            "streamType": 3,
+            "displayTitle": "magyar Forced",
+            "languageTag": "hu",
+            "forced": True,
+            "codec": "srt",
+        },
+        {
+            "streamType": 3,
+            "displayTitle": "English SDH",
+            "languageTag": "en",
+            "hearingImpaired": True,
+            "codec": "srt",
+        },
+    ]
+    meta = {**RICH_META, "Media": [{"Part": [{"Stream": streams}]}]}
+    monkeypatch.setattr(detail_routes, "PlexClient", lambda db: _StubClient(meta))
+    body = _get(db, user, monkeypatch, override_dep, "/api/plex/item/m1").json()
+
+    naming = ("label", "lang_tag", "forced", "hearing_impaired", "detail")
+    assert [tuple(a[k] for k in naming) for a in body["audio_streams"]] == [
+        ("English (AC3 5.1)", "en-US", False, False, "AC3 5.1"),
+        ("中文 (EAC3 Stereo)", "zh", False, False, "EAC3 Stereo"),
+        (None, None, False, False, None),
+    ]
+    assert [tuple(s[k] for k in naming) for s in body["subtitle_streams"]] == [
+        ("magyar Forced", "hu", True, False, None),
+        ("English SDH", "en", False, True, None),
+    ]

backend/tests/test_plex_tmdb.py

@@ -49,7 +49,7 @@ MOVIE_PAYLOAD = {
         ],
     },
     "production_companies": [{"name": "Paramount"}, {"name": "Alfran"}],
-    "production_countries": [{"name": "United States of America"}],
+    "production_countries": [{"iso_3166_1": "US", "name": "United States of America"}],
     "recommendations": {
         "results": [
             {
@@ -126,7 +126,8 @@ def test_parse_movie_pulls_all_fields():
         },
     ]  # the Gaffer is filtered; the two Coppola jobs survive (dedup is by name+job)
     assert d["studios"] == ["Paramount", "Alfran"]
-    assert d["production_countries"] == ["United States of America"]
+    # E30: the code beside the English name, so the client can name the country in the reader's language.
+    assert d["production_countries"] == [{"code": "US", "name": "United States of America"}]
     assert [r["tmdb_id"] for r in d["recommendations"]] == [240, 999]
     assert d["backdrops"] == ["/bd1.jpg", "/bd2.jpg"]
     # cast + crew names → tmdb person ids for the plex_people foundation
@@ -429,3 +430,69 @@ def test_resolve_recs_matches_by_media_type_not_bare_id(db):
     }
     assert hit[900]["in_library"] is True and hit[900]["rating_key"] == "m900"
     assert hit[500]["in_library"] is True and hit[500]["rating_key"] == "s500"
+
+
+# ---- E30: countries by ISO code -----------------------------------------------------------------
+
+
+class FakeCountries(FakeTmdb):
+    """TMDB's /configuration/countries, counting the requests."""
+
+    def __init__(self):
+        super().__init__()
+        self.calls = 0
+
+    def countries(self):
+        self.calls += 1
+        return [
+            {"iso_3166_1": "US", "english_name": "United States of America"},
+            {"iso_3166_1": "YU", "english_name": "Yugoslavia"},
+        ]
+
+
+def test_countries_stored_before_codes_get_them_from_tmdbs_list(db):
+    lib = _lib(db)
+    old = PlexItem(
+        rating_key="m1",
+        library_id=lib.id,
+        kind="movie",
+        title="Old",
+        production_countries=["Yugoslavia", "Atlantis"],
+        tmdb_enriched_at=datetime.now(timezone.utc),
+    )
+    show_lib = _lib(db, key="2", kind="show")
+    show = PlexShow(
+        rating_key="s1",
+        library_id=show_lib.id,
+        title="Show",
+        production_countries=["United States of America"],
+    )
+    new = PlexItem(
+        rating_key="m2",
+        library_id=lib.id,
+        kind="movie",
+        title="New",
+        production_countries=[{"code": "US", "name": "United States of America"}],
+    )
+    db.add_all([old, show, new])
+    db.commit()
+
+    # Served before the backfill: the bare name, with no code — never a crash on the old shape.
+    assert tmdb.serve_block(db, old, lang="hu")["production_countries"] == [
+        {"code": None, "name": "Yugoslavia"},
+        {"code": None, "name": "Atlantis"},
+    ]
+
+    fake = FakeCountries()
+    assert tmdb.backfill_country_codes(db, fake) == 2
+    db.refresh(old)
+    db.refresh(show)
+    assert old.production_countries == [
+        {"code": "YU", "name": "Yugoslavia"},
+        {"code": None, "name": "Atlantis"},  # not in TMDB's list → no code, the name kept
+    ]
+    assert show.production_countries == [{"code": "US", "name": "United States of America"}]
+
+    # Nothing left to code: no request at all.
+    assert tmdb.backfill_country_codes(db, fake) == 0
+    assert fake.calls == 1

backend/tests/test_scheduler.py

@@ -2,7 +2,10 @@
 through (at the top of _job): holding _activity_lock across the check-and-set means exactly one
 caller can hold a job's "running" claim at a time, and it's claimable again once released."""
 
+from sqlalchemy import select
+
 from app import scheduler
+from app.models import Notification
 
 _JOB = "unit_test_claim"  # not a real job id — exercises the primitive without touching live jobs
 
@@ -23,3 +26,44 @@ def test_claim_is_exclusive_until_released():
     # ...and now it can be claimed again.
     assert scheduler._claim_running(_JOB) is True
     _release()
+
+
+# ---- E30: the result as data -------------------------------------------------------------------
+
+
+def _snapshot_job(job_id):
+    return next(j for j in scheduler.scheduler_snapshot()["jobs"] if j["id"] == job_id)
+
+
+def test_a_run_keeps_its_result_as_data_for_the_readers_language(db, make_user):
+    """The dashboard and the manual-run notice showed `_summarize`'s English one-liner ("new=3,
+    failed=0", a Python repr for a nested phase). The result rides along as data, so the client names
+    it; the one-liner stays as the fallback and the audit text."""
+    admin = make_user("sched@example.com", role="admin")
+    token = scheduler._manual_actor.set(admin.id)
+    try:
+        scheduler._job("rss_poll", lambda _db: {"new": 3, "failed": 0, "fallback": 1})
+        scheduler._job("enrich", lambda _db: 5)  # enrich returns a bare count
+    finally:
+        scheduler._manual_actor.reset(token)
+
+    rss = _snapshot_job("rss_poll")
+    assert rss["result"] == {"new": 3, "failed": 0, "fallback": 1}
+    assert rss["last_result"] == "new=3, failed=0, fallback=1"
+    assert _snapshot_job("enrich")["result"] == {"enriched": 5}
+
+    notices = db.execute(
+        select(Notification).where(Notification.user_id == admin.id).order_by(Notification.id)
+    ).scalars()
+    assert [n.data["result"] for n in notices] == [
+        {"new": 3, "failed": 0, "fallback": 1},
+        {"enriched": 5},
+    ]
+
+
+def test_a_failed_run_drops_the_previous_result(db):
+    scheduler._job("rss_poll", lambda _db: {"new": 1})
+    scheduler._job("rss_poll", lambda _db: (_ for _ in ()).throw(RuntimeError("boom")))
+    rss = _snapshot_job("rss_poll")
+    assert rss["last_error"] == "boom"
+    assert rss["result"] is None