Feat/agent harness

backend/app/collection/config.py

@@ -67,12 +67,6 @@ class PlaylistConfig:
 ]
 
 PLAYLISTS: list[PlaylistConfig] = [
-    PlaylistConfig(
-        name="Radio | Luyện nghe tiếng Trung",
-        playlist_id="PLniHrP5FDBnjRxYSs0j8_siiNPVs58pZ_",
-        default_difficulty="HSK 1-2",
-        default_topic="Daily Conversation",
-    ),
     PlaylistConfig(
         name="Mr.Chinese Channel",
         playlist_id="PLN7MEvFrgspUfsYuGJord_LVV9gp-L1bZ",

backend/app/collection/router.py

@@ -3,7 +3,12 @@
 
 from fastapi import APIRouter, HTTPException
 
-from app.collection.service import get_collection, get_playlist_videos, get_video_metadata
+from app.collection.service import (
+    get_collection,
+    get_playlist_videos,
+    get_video_metadata,
+    resolve_curated_video,
+)
 
 router = APIRouter(prefix="/api")
 
@@ -30,3 +35,14 @@ async def get_video_endpoint(video_id: str) -> dict:
     if result is None:
         raise HTTPException(status_code=404, detail="Video not found")
     return result
+
+
+@router.get("/collection/resolve/{video_id}")
+async def resolve_video_endpoint(video_id: str) -> dict:
+    """Resolve a recommended YouTube video to its internal tip route (playlist or standalone)."""
+    result = await asyncio.to_thread(resolve_curated_video, video_id)
+    if result["status"] in ("video", "playlist"):
+        return result
+    if result["status"] == "not_curated":
+        raise HTTPException(status_code=404, detail="Not a curated video")
+    raise HTTPException(status_code=503, detail="Resolution unavailable")

backend/app/collection/service.py

@@ -175,6 +175,40 @@ def get_cached_playlist(playlist_id: str) -> list[dict]:
     return entries
 
 
+def resolve_curated_video(video_id: str) -> dict:
+    """Resolve a YouTube video_id to its curated tip home.
+
+    The YouTube Data API has no reverse video→playlist lookup, so we enumerate the
+    curated tip playlists and check membership (reusing the cached playlistItems fetch).
+
+    Returns one of:
+    - {"status": "video", "video_id": ...}        standalone tip video
+    - {"status": "playlist", "playlist_id": ..., "video_id": ...}   tip-playlist member
+    - {"status": "not_curated"}   every tip playlist fetched, no match (definitive)
+    - {"status": "unresolved"}    a tip playlist fetch was empty/failed, no match (transient)
+    """
+    for video in STANDALONE_VIDEOS:
+        if video.video_id == video_id and video.content_type == "tip":
+            return {"status": "video", "video_id": video_id}
+
+    had_empty_fetch = False
+    for playlist in PLAYLISTS:
+        if playlist.default_content_type != "tip":
+            continue
+        entries = get_cached_playlist(playlist.playlist_id)
+        if not entries:
+            had_empty_fetch = True
+            continue
+        if any(entry["id"] == video_id for entry in entries):
+            return {
+                "status": "playlist",
+                "playlist_id": playlist.playlist_id,
+                "video_id": video_id,
+            }
+
+    return {"status": "unresolved"} if had_empty_fetch else {"status": "not_curated"}
+
+
 # Separate cache for playlist-level metadata (thumbnail, video_count)
 _meta_cache: dict[str, tuple[float, dict]] = {}
 

backend/app/lessons/router.py

@@ -25,12 +25,13 @@
     pick_manual_subtitle,
 )
 from app.lessons.services.romanization_provider import get_romanization_provider
+from app.lessons.services.segmentation_provider import get_segmentation_provider
 from app.lessons.services.chinese_normalizer import normalize_chinese
 from app.transcription.services.transcription_provider import STTProvider, TranscriptionKeys
 from app.translation.services.translation import translate_segments
 from app.lessons.services.validation import ValidationError, validate_upload_file, validate_youtube_url
 from app.shared.language_config import get_language_config
-from app.lessons.services.vocabulary import extract_vocabulary
+from app.lessons.services.vocabulary import enrich_vocabulary, extract_vocabulary
 from app.lessons.services.blog_scraper import scrape_article
 from app.tts.services.tts_provider import TTSProvider, TTSKeys
 
@@ -70,18 +71,29 @@ async def _shared_pipeline(
         enriched_segments.append({**seg, "romanization": romanizer.romanize_text(seg["text"])})
     logger.info("[pipeline] romanization: done in %.1fs (source_language=%s)", time.monotonic() - t0, source_language)
 
+    meaning_language = get_language_config(translation_languages[0])["language_name"]
+    segmenter = get_segmentation_provider(source_language)
+    if segmenter is not None:
+        for seg in enriched_segments:
+            seg["tokens"] = segmenter.segment(seg["text"])
+        vocab_coro = enrich_vocabulary(
+            enriched_segments, romanizer, api_key,
+            source_language=source_language, meaning_language=meaning_language,
+        )
+    else:
+        vocab_coro = extract_vocabulary(
+            enriched_segments, api_key,
+            source_language=source_language, meaning_language=meaning_language,
+        )
+
     jobs[job_id].step = "translation"
     t0 = time.monotonic()
     translated_segments, vocab_map = await asyncio.gather(
         translate_segments(
             enriched_segments, translation_languages, api_key,
             source_language=source_language,
         ),
-        extract_vocabulary(
-            enriched_segments, api_key,
-            source_language=source_language,
-            meaning_language=get_language_config(translation_languages[0])["language_name"],
-        ),
+        vocab_coro,
     )
     logger.info(
         "[pipeline] translation+vocabulary: done in %.1fs, %d segments, %d vocab entries",

backend/app/lessons/services/segmentation_provider.py

@@ -0,0 +1,29 @@
+"""Pluggable word-segmentation providers — deterministic, one per language family."""
+
+import re
+from typing import Protocol
+
+# A token is meaningful if it contains at least one word character. \w under
+# Python's default Unicode matching covers CJK; fullwidth/halfwidth punctuation
+# and whitespace do not — so this drops pure-punctuation tokens jieba emits.
+_HAS_WORD_CHAR = re.compile(r"\w", re.UNICODE)
+
+
+class SegmentationProvider(Protocol):
+    def segment(self, text: str) -> list[str]: ...
+
+
+class ChineseSegmentationProvider:
+    def segment(self, text: str) -> list[str]:
+        import jieba  # type: ignore[import]
+        return [
+            tok for tok in jieba.lcut(text)
+            if _HAS_WORD_CHAR.search(tok)
+        ]
+
+
+def get_segmentation_provider(source_language: str) -> SegmentationProvider | None:
+    """Return a deterministic segmenter for source_language, or None to fall back to LLM."""
+    if source_language.startswith("zh"):
+        return ChineseSegmentationProvider()
+    return None