Upload 52 files

docs/superpowers/plans/2026-07-21-feel-scorer.md

@@ -0,0 +1,1672 @@
+# Feel Scorer (Phase 4) Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Gemini re-watches every mechanically-passed clip, judges its feel against a taste rubric, and writes a score-ranked shortlist to a new `feel_scores` table with a UI to browse it.
+
+**Architecture:** Reuse the Phase-1 vision pipeline (download → VisionRouter → store). Add a `prompt`/`parser` override to `GenAIBackend` so the same router can run `FEEL_PROMPT`. New backend module (`feel_scorer.py`), runner (`feel_score_runner.py`), store methods (`feel_scores` table), UI tab + glue (`feel_score_app.py`), CLI flag.
+
+**Tech Stack:** Python 3, google-genai (Gemini/Gemma vision), Neon Postgres, Gradio 6.19, pytest, InMemoryCatalogStore for tests.
+
+## Global Constraints
+
+- **Two trees, kept in sync:** every non-test file written to BOTH `video_library/<f>` AND `hf_space/video_library/<f>`, byte-identical. Tests live only in `video_library/tests/`.
+- **Run tests from repo root** `C:\EXTRA PROJECTS\INPENIX\narration-shorts` with `python -m pytest video_library/tests/ -q`.
+- **No network in tests** — inject fakes; use `InMemoryCatalogStore`.
+- **Crash safety in UI:** video players `autoplay=False, loop=True`, page size 6, verdict/browse never triggers mass video reload, guarded render (catch → error header, no 500).
+- **No bytes persist:** clips downloaded to temp, deleted after scoring.
+- **`clips` table untouched.** `feel_scores` is a fresh, disposable table.
+- **FEEL_PROMPT is verbatim** from the user's rubric (score /100, KEEP≥80 / BORDERLINE 60-79 / REJECT<60, JSON keys: decision, score, camera_motion, subject_motion, visual_complexity, human_presence, mood, reason).
+- **No git repo** — skip commit steps; "commit" means "save both trees + run tests green".
+
+---
+
+### Task 1: `GenAIBackend` prompt/parser override (non-breaking)
+
+Make the existing vision backend accept an alternate prompt + parser so the feel pass reuses `VisionRouter` unchanged. Phase-1 callers keep current behavior via defaults.
+
+**Files:**
+- Modify: `video_library/vision_describe.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_vision_describe.py`
+
+**Interfaces:**
+- Produces: `GenAIBackend(api_key, model, *, strip_audio=False, prompt=VISION_PROMPT, parser=parse_vision_json, ...)`. `describe()` uses `self._prompt` and `self._parser`.
+
+- [ ] **Step 1: Write failing test** — append to `test_vision_describe.py`:
+
+```python
+def test_backend_uses_injected_prompt_and_parser():
+    captured = {}
+
+    class _FakeResp:
+        text = '{"ok": true}'
+
+    class _FakeModels:
+        def generate_content(self, model, contents, config):
+            captured["prompt"] = contents[1]
+            return _FakeResp()
+
+    class _FakeFiles:
+        def upload(self, file):
+            return object()
+
+    class _FakeClient:
+        files = _FakeFiles()
+        models = _FakeModels()
+
+    sentinel = object()
+    backend = GenAIBackend("k", "gemini-x", client=_FakeClient(),
+                           prompt="FEEL PROMPT HERE",
+                           parser=lambda text: sentinel)
+    result = backend.describe("dummy.mp4")
+    assert result is sentinel
+    assert captured["prompt"] == "FEEL PROMPT HERE"
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_vision_describe.py::test_backend_uses_injected_prompt_and_parser -q`
+Expected: FAIL (`GenAIBackend` got unexpected keyword `prompt`).
+
+- [ ] **Step 3: Implement** — in `GenAIBackend.__init__` add params and store them. Edit the signature:
+
+```python
+    def __init__(self, api_key: str, model: str, *, strip_audio: bool = False,
+                 client=None, audio_stripper=strip_audio_track,
+                 sleep=time.sleep, poll_interval: float = 5.0, max_poll: int = 60,
+                 gen_config=None, prompt: str = VISION_PROMPT,
+                 parser=parse_vision_json):
+        self._key = api_key
+        self._model = model
+        self._strip_audio = strip_audio
+        self._client = client
+        self._audio_stripper = audio_stripper
+        self._sleep = sleep
+        self._poll_interval = poll_interval
+        self._max_poll = max_poll
+        self._gen_config = gen_config
+        self._prompt = prompt
+        self._parser = parser
+```
+
+In `describe()`, replace the two hardcoded references:
+
+```python
+            resp = client.models.generate_content(
+                model=self._model,
+                contents=[uploaded, self._prompt],
+                config=self._gen_config_obj(),
+            )
+            desc = self._parser(self._text_of(resp))
+            desc.described_by = self._model
+            return desc
+```
+
+Note: `FeelResult` (Task 2) also carries a `described_by` attribute so this assignment stays valid for the feel parser.
+
+- [ ] **Step 4: Run tests** — verify new test + existing vision tests pass:
+
+Run: `python -m pytest video_library/tests/test_vision_describe.py -q`
+Expected: PASS (all).
+
+- [ ] **Step 5: Sync mirror** — copy `video_library/vision_describe.py` to `hf_space/video_library/vision_describe.py` (identical). Verify:
+
+Run: `diff video_library/vision_describe.py hf_space/video_library/vision_describe.py && echo SYNC`
+Expected: `SYNC`.
+
+---
+
+### Task 2: `feel_scorer.py` — FEEL_PROMPT, FeelResult, parse_feel_json
+
+The vision judge's data layer: the verbatim prompt, the result dataclass, and a strict-JSON parser with validation/clamping.
+
+**Files:**
+- Create: `video_library/feel_scorer.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_feel_scorer.py`
+
+**Interfaces:**
+- Produces:
+  - `FEEL_PROMPT: str`
+  - `FeelResult` dataclass: `decision: str, score: int, camera_motion: str, subject_motion: str, visual_complexity: str, human_presence: str, mood: str, reason: str, described_by: str = ""`. Property `scored_by` returns `described_by`.
+  - `FeelParseError(Exception)`
+  - `parse_feel_json(text: str) -> FeelResult`
+  - `FEEL_DECISIONS = ("KEEP", "BORDERLINE", "REJECT")`
+
+- [ ] **Step 1: Write failing tests** — create `video_library/tests/test_feel_scorer.py`:
+
+```python
+import pytest
+from video_library.feel_scorer import (
+    FEEL_PROMPT, FeelResult, parse_feel_json, FeelParseError, FEEL_DECISIONS,
+)
+
+GOOD = ('{"decision": "KEEP", "score": 88, "camera_motion": "slow drone glide", '
+        '"subject_motion": "still", "visual_complexity": "low", '
+        '"human_presence": "none", "mood": "peaceful", "reason": "calm open nature"}')
+
+
+def test_parses_good_json():
+    r = parse_feel_json(GOOD)
+    assert r.decision == "KEEP"
+    assert r.score == 88
+    assert r.visual_complexity == "low"
+    assert r.human_presence == "none"
+    assert r.reason == "calm open nature"
+
+
+def test_parses_fenced_json():
+    r = parse_feel_json("```json\n" + GOOD + "\n```")
+    assert r.decision == "KEEP"
+
+
+def test_score_clamped_to_0_100():
+    hi = parse_feel_json('{"decision":"KEEP","score":150,"reason":"x"}')
+    lo = parse_feel_json('{"decision":"REJECT","score":-5,"reason":"x"}')
+    assert hi.score == 100
+    assert lo.score == 0
+
+
+def test_score_coerced_from_string():
+    r = parse_feel_json('{"decision":"KEEP","score":"73","reason":"x"}')
+    assert r.score == 73
+
+
+def test_missing_optional_keys_default_empty():
+    r = parse_feel_json('{"decision":"BORDERLINE","score":65}')
+    assert r.reason == ""
+    assert r.mood == ""
+    assert r.camera_motion == ""
+
+
+def test_bad_decision_raises():
+    with pytest.raises(FeelParseError):
+        parse_feel_json('{"decision":"MAYBE","score":50}')
+
+
+def test_no_json_raises():
+    with pytest.raises(FeelParseError):
+        parse_feel_json("not json at all")
+
+
+def test_scored_by_mirrors_described_by():
+    r = FeelResult(decision="KEEP", score=90, described_by="gemini-x")
+    assert r.scored_by == "gemini-x"
+
+
+def test_prompt_mentions_core_rubric():
+    p = FEEL_PROMPT.lower()
+    assert "narration" in p
+    assert "camera" in p and "subject" in p
+    assert "keep" in p and "reject" in p
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_feel_scorer.py -q`
+Expected: FAIL (module `feel_scorer` not found).
+
+- [ ] **Step 3: Implement** — create `video_library/feel_scorer.py` exactly as below. `FEEL_PROMPT` is the user's rubric verbatim; keep it byte-for-byte:
+
+```python
+"""Phase-4 vision judge: watch a clip, score its FEEL for contemplative narration.
+
+Unlike the Phase-1 VISION_PROMPT (which forbids feel-judgment on purpose), this
+prompt asks the model to decide whether the clip can sit quietly *under* Osho-style
+narration without competing for attention. Output is strict JSON we parse into a
+FeelResult. The taste lives entirely in FEEL_PROMPT — no few-shot examples.
+"""
+from __future__ import annotations
+import json
+import re
+from dataclasses import dataclass
+
+FEEL_DECISIONS = ("KEEP", "BORDERLINE", "REJECT")
+
+FEEL_PROMPT = """# Role
+
+You are an expert curator of stock footage for a YouTube Shorts channel featuring slow, contemplative Osho narration.
+
+Your job is **not** to judge whether a clip is beautiful. Your job is to judge whether it quietly supports spoken wisdom without competing for the viewer's attention.
+
+The narration is always the hero.
+
+The footage exists only to create atmosphere.
+
+A perfect clip should almost disappear behind the words while gently deepening their emotional impact.
+
+---
+
+# Core Principle
+
+**Ask yourself one question above all else:**
+
+> **Would this clip make the viewer watch the visuals instead of listening to the narration?**
+
+If **yes**, reject it.
+
+If the visuals simply create a peaceful atmosphere while allowing the narration to remain the focus, keep it.
+
+---
+
+# The Most Important Distinction
+
+Evaluate **camera movement** and **subject movement** separately.
+
+### Good Camera Movement
+
+The camera may move if it slowly reveals the scene.
+
+Examples:
+
+* slow drone glide
+* slow dolly
+* gentle pan
+* slow tilt
+* slow push-in
+* smooth floating movement
+
+This creates immersion without distraction.
+
+### Bad Camera Movement
+
+Reject:
+
+* handheld shake
+* whip pans
+* rapid zooms
+* aggressive orbiting
+* jerky stabilization
+* fast tracking
+
+---
+
+### Good Subject Movement
+
+Minimal natural movement is ideal.
+
+Examples:
+
+* clouds drifting
+* mist moving
+* gentle waves
+* leaves lightly swaying
+* rain
+* smoke
+* calm flowing river
+
+The subject should feel almost still.
+
+### Bad Subject Movement
+
+Reject if the subject itself becomes the attraction.
+
+Examples:
+
+* people walking prominently
+* animals running
+* birds dominating the frame
+* vehicles
+* traffic
+* crowds
+* dancing
+* sports
+* waterfalls with violent motion
+* fireworks
+* explosions
+* anything energetic
+
+**Camera motion is good. Subject motion is usually bad.**
+
+---
+
+# Desired Visual Style
+
+Prefer clips with:
+
+* open landscapes
+* mountains
+* forests
+* rivers
+* lakes
+* ocean horizons
+* beaches
+* deserts
+* snow
+* fog
+* mist
+* clouds
+* sunrise
+* sunset
+* moonlight
+* empty temples
+* quiet interiors
+* empty roads
+* abandoned structures
+* fields
+* minimal architecture
+* cinematic nature
+
+The frame should evoke:
+
+* silence
+* spaciousness
+* solitude
+* timelessness
+* contemplation
+* stillness
+* peace
+* wonder
+* presence
+
+---
+
+# Visual Simplicity
+
+The composition should be clean and easy to absorb.
+
+Prefer:
+
+* few dominant elements
+* negative space
+* uncluttered backgrounds
+* natural lighting
+* cinematic framing
+* soft colors
+* slow visual rhythm
+
+Reject:
+
+* clutter
+* visual chaos
+* many competing objects
+* busy forests
+* dense city scenes
+* heavy visual texture
+* flashing lights
+* excessive detail that constantly pulls attention
+
+---
+
+# Human Presence
+
+People should rarely be the subject.
+
+KEEP:
+
+* tiny distant silhouette
+* lone person occupying very little of the frame
+* person integrated into the landscape
+* person remaining still
+
+REJECT:
+
+* close-up faces
+* portraits
+* talking people
+* emotional acting
+* influencers
+* vlog footage
+* people performing actions
+* crowds
+* dancing
+* weddings
+* celebrations
+
+If a viewer immediately notices the person before the environment, reject.
+
+---
+
+# Commercial or Stock Feel
+
+Reject clips that resemble:
+
+* advertisements
+* travel influencer footage
+* luxury lifestyle
+* product showcases
+* fashion videos
+* cinematic hero shots
+* corporate footage
+* generic motivational stock
+* obvious AI-generated visuals
+* overprocessed HDR
+* unrealistic colors
+
+The footage should feel authentic rather than promotional.
+
+---
+
+# Technical Rejections
+
+Automatically reject clips containing:
+
+* text
+* subtitles
+* logos
+* watermarks
+* UI elements
+* borders
+* split screens
+* multiple scenes
+* fast cuts
+* jump cuts
+* transitions
+* glitches
+* poor stabilization
+* distracting compression artifacts
+
+One uninterrupted shot is strongly preferred.
+
+---
+
+# Emotional Tone
+
+Desired emotions:
+
+* peaceful
+* reflective
+* meditative
+* calm
+* expansive
+* spiritual
+* quiet
+* gentle
+* grounded
+
+Avoid clips that feel:
+
+* exciting
+* urgent
+* dramatic
+* suspenseful
+* comedic
+* romantic
+* aggressive
+* adventurous
+* action-oriented
+* emotionally manipulative
+
+---
+
+# Scoring Rubric
+
+Score each clip out of 100.
+
+**Subject Motion (35 points)**
+
+* Minimal movement = high score
+* Active subjects = low score
+
+**Camera Movement (20 points)**
+
+* Smooth slow movement = high score
+* Fast or shaky movement = low score
+
+**Visual Simplicity (20 points)**
+
+* Clean composition = high score
+* Busy composition = low score
+
+**Mood & Atmosphere (15 points)**
+
+* Peaceful and contemplative = high score
+* Exciting or dramatic = low score
+
+**Human Presence (10 points)**
+
+* None or tiny silhouette = high score
+* Human as subject = low score
+
+---
+
+# Decision Rules
+
+**KEEP**
+
+* Score **80-100**
+* Strong contemplative background
+* Supports narration effortlessly
+
+**BORDERLINE**
+
+* Score **60-79**
+* Usable but contains mild distractions
+
+**REJECT**
+
+* Score **below 60**
+* Competes with narration or breaks the contemplative atmosphere
+
+---
+
+# Output Format
+
+Return only valid JSON.
+
+{
+  "decision": "KEEP | BORDERLINE | REJECT",
+  "score": 0,
+  "camera_motion": "",
+  "subject_motion": "",
+  "visual_complexity": "low | medium | high",
+  "human_presence": "none | distant | prominent",
+  "mood": "",
+  "reason": ""
+}
+
+Keep the "reason" concise (1-2 sentences), focusing on why the clip does or does not work as a quiet background for contemplative narration.
+
+Never judge the philosophical content. Judge only the visual suitability for an Osho-style spoken wisdom video."""
+
+
+class FeelParseError(Exception):
+    pass
+
+
+@dataclass
+class FeelResult:
+    decision: str
+    score: int
+    camera_motion: str = ""
+    subject_motion: str = ""
+    visual_complexity: str = ""
+    human_presence: str = ""
+    mood: str = ""
+    reason: str = ""
+    described_by: str = ""
+
+    @property
+    def scored_by(self) -> str:
+        return self.described_by
+
+
+def _coerce_score(v) -> int:
+    try:
+        n = int(float(v))
+    except (TypeError, ValueError):
+        raise FeelParseError(f"bad score: {v!r}")
+    return max(0, min(100, n))
+
+
+def _from_obj(obj: dict) -> FeelResult:
+    decision = str(obj.get("decision", "")).strip().upper()
+    if decision not in FEEL_DECISIONS:
+        raise FeelParseError(f"bad decision: {decision!r}")
+    return FeelResult(
+        decision=decision,
+        score=_coerce_score(obj.get("score", 0)),
+        camera_motion=str(obj.get("camera_motion", "") or ""),
+        subject_motion=str(obj.get("subject_motion", "") or ""),
+        visual_complexity=str(obj.get("visual_complexity", "") or ""),
+        human_presence=str(obj.get("human_presence", "") or ""),
+        mood=str(obj.get("mood", "") or ""),
+        reason=str(obj.get("reason", "") or ""),
+    )
+
+
+def parse_feel_json(text: str) -> FeelResult:
+    """Extract the first valid JSON object from model text -> FeelResult."""
+    if not text:
+        raise FeelParseError("empty response")
+    candidates = []
+    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+    if fence:
+        candidates.append(fence.group(1))
+    brace = re.search(r"\{.*\}", text, re.DOTALL)
+    if brace:
+        candidates.append(brace.group(0))
+    for c in candidates:
+        try:
+            return _from_obj(json.loads(c))
+        except (json.JSONDecodeError, TypeError):
+            continue
+    raise FeelParseError("no valid JSON object in response")
+```
+
+IMPORTANT: replace `<PASTE THE USER'S FULL RUBRIC PROMPT VERBATIM HERE>` with the exact rubric the user supplied (the `# Role ... Judge only the visual suitability` text). Do not paraphrase.
+
+- [ ] **Step 4: Run tests**
+
+Run: `python -m pytest video_library/tests/test_feel_scorer.py -q`
+Expected: PASS (all 9).
+
+- [ ] **Step 5: Sync mirror**
+
+Run: `cp video_library/feel_scorer.py hf_space/video_library/feel_scorer.py && diff video_library/feel_scorer.py hf_space/video_library/feel_scorer.py && echo SYNC`
+Expected: `SYNC`.
+
+---
+
+### Task 3: `feel_scores` table + store methods
+
+Add the fresh table and its CRUD to both `InMemoryCatalogStore` and `NeonCatalogStore`.
+
+**Files:**
+- Modify: `video_library/catalog_store.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_catalog_store_feel.py`
+
+**Interfaces:**
+- Consumes: `FeelResult` from Task 2.
+- Produces on both stores:
+  - `ensure_feel_schema() -> None`
+  - `load_for_feel_scoring(limit: int = 100) -> list[dict]` — mechanically-passed clip records NOT already in feel_scores.
+  - `set_feel_score(provider, provider_id, *, result: FeelResult, clip_url: str, now: str) -> None` — UPSERT.
+  - `load_feel_scores() -> list[dict]` — rows as dicts with keys: provider, provider_id, decision, score, camera_motion, subject_motion, visual_complexity, human_presence, mood, reason, clip_url, scored_at, scored_by.
+  - `feel_page(*, decision=None, min_score=0, limit=6, offset=0) -> list[dict]` — filtered, sorted score DESC.
+  - `feel_counts() -> dict` — `{"KEEP": n, "BORDERLINE": n, "REJECT": n, "total": n, "scored": n, "unscored": n}` where scored/unscored are over mechanically-passed clips.
+
+- [ ] **Step 1: Write failing tests** — create `video_library/tests/test_catalog_store_feel.py`:
+
+```python
+from video_library.catalog_store import InMemoryCatalogStore
+from video_library.models import ClipRecord, Candidate, FileOption, Description, CheckResult
+from video_library.feel_scorer import FeelResult
+
+
+def _clip(pid, *, passed=True):
+    cand = Candidate(provider="pexels", provider_id=pid, page_url="p",
+                     files=[FileOption("sd", 640, 360, 25.0, "http://x/" + pid)],
+                     duration=10.0, width=640, height=360, fps=25.0,
+                     creator_name="c", creator_url="u", license="pexels",
+                     original_query="q")
+    return ClipRecord(candidate=cand,
+                      description=Description(summary="s", environment="nature"),
+                      mechanical_check=CheckResult(passed=passed),
+                      analyzed_at="t" + pid).to_dict()
+
+
+def _store(*clips):
+    s = InMemoryCatalogStore()
+    for c in clips:
+        s._records[(c["provider"], c["provider_id"])] = c
+    return s
+
+
+def _res(decision="KEEP", score=88):
+    return FeelResult(decision=decision, score=score, reason="r",
+                      camera_motion="slow", subject_motion="still",
+                      visual_complexity="low", human_presence="none",
+                      mood="calm", described_by="gemini-x")
+
+
+def test_load_for_feel_scoring_only_passed_and_unscored():
+    s = _store(_clip("1"), _clip("2", passed=False))
+    got = s.load_for_feel_scoring()
+    assert [r["provider_id"] for r in got] == ["1"]
+
+
+def test_set_and_load_feel_score_roundtrip():
+    s = _store(_clip("1"))
+    s.set_feel_score("pexels", "1", result=_res(), clip_url="http://c/1", now="t1")
+    rows = s.load_feel_scores()
+    assert len(rows) == 1
+    r = rows[0]
+    assert r["decision"] == "KEEP" and r["score"] == 88
+    assert r["scored_by"] == "gemini-x" and r["clip_url"] == "http://c/1"
+    assert r["human_presence"] == "none"
+
+
+def test_scored_clip_excluded_from_next_load():
+    s = _store(_clip("1"), _clip("2"))
+    s.set_feel_score("pexels", "1", result=_res(), clip_url="u", now="t")
+    assert [r["provider_id"] for r in s.load_for_feel_scoring()] == ["2"]
+
+
+def test_set_feel_score_upsert_overwrites():
+    s = _store(_clip("1"))
+    s.set_feel_score("pexels", "1", result=_res(score=50), clip_url="u", now="t1")
+    s.set_feel_score("pexels", "1", result=_res(score=95), clip_url="u", now="t2")
+    rows = s.load_feel_scores()
+    assert len(rows) == 1 and rows[0]["score"] == 95
+
+
+def test_feel_page_sorts_desc_and_filters():
+    s = _store(_clip("1"), _clip("2"), _clip("3"))
+    s.set_feel_score("pexels", "1", result=_res("KEEP", 90), clip_url="u", now="t")
+    s.set_feel_score("pexels", "2", result=_res("REJECT", 30), clip_url="u", now="t")
+    s.set_feel_score("pexels", "3", result=_res("BORDERLINE", 70), clip_url="u", now="t")
+    ordered = [r["score"] for r in s.feel_page(limit=10)]
+    assert ordered == [90, 70, 30]
+    keeps = s.feel_page(decision="KEEP", limit=10)
+    assert [r["provider_id"] for r in keeps] == ["1"]
+    strong = [r["score"] for r in s.feel_page(min_score=60, limit=10)]
+    assert strong == [90, 70]
+
+
+def test_feel_counts():
+    s = _store(_clip("1"), _clip("2"), _clip("3"))
+    s.set_feel_score("pexels", "1", result=_res("KEEP", 90), clip_url="u", now="t")
+    s.set_feel_score("pexels", "2", result=_res("REJECT", 30), clip_url="u", now="t")
+    c = s.feel_counts()
+    assert c["KEEP"] == 1 and c["REJECT"] == 1 and c["BORDERLINE"] == 0
+    assert c["total"] == 2 and c["scored"] == 2 and c["unscored"] == 1
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_catalog_store_feel.py -q`
+Expected: FAIL (`load_for_feel_scoring` missing).
+
+- [ ] **Step 3a: Implement SQL constants** — in `catalog_store.py`, after the seed SQL block (near `SELECT_SEED`), add:
+
+```python
+CREATE_FEEL_SCORES = """
+CREATE TABLE IF NOT EXISTS feel_scores (
+    provider          TEXT NOT NULL,
+    provider_id       TEXT NOT NULL,
+    decision          TEXT NOT NULL,
+    score             INT  NOT NULL DEFAULT 0,
+    camera_motion     TEXT NOT NULL DEFAULT '',
+    subject_motion    TEXT NOT NULL DEFAULT '',
+    visual_complexity TEXT NOT NULL DEFAULT '',
+    human_presence    TEXT NOT NULL DEFAULT '',
+    mood              TEXT NOT NULL DEFAULT '',
+    reason            TEXT NOT NULL DEFAULT '',
+    clip_url          TEXT NOT NULL DEFAULT '',
+    scored_at         TEXT,
+    scored_by         TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (provider, provider_id)
+)
+"""
+
+UPSERT_FEEL = """
+INSERT INTO feel_scores (provider, provider_id, decision, score, camera_motion,
+    subject_motion, visual_complexity, human_presence, mood, reason, clip_url,
+    scored_at, scored_by)
+VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+ON CONFLICT (provider, provider_id) DO UPDATE SET
+    decision = EXCLUDED.decision, score = EXCLUDED.score,
+    camera_motion = EXCLUDED.camera_motion, subject_motion = EXCLUDED.subject_motion,
+    visual_complexity = EXCLUDED.visual_complexity,
+    human_presence = EXCLUDED.human_presence, mood = EXCLUDED.mood,
+    reason = EXCLUDED.reason, clip_url = EXCLUDED.clip_url,
+    scored_at = EXCLUDED.scored_at, scored_by = EXCLUDED.scored_by
+"""
+
+SELECT_FEEL = """
+SELECT provider, provider_id, decision, score, camera_motion, subject_motion,
+    visual_complexity, human_presence, mood, reason, clip_url, scored_at, scored_by
+FROM feel_scores
+ORDER BY score DESC, provider, provider_id
+"""
+
+# Mechanically-passed clips with no row yet in feel_scores.
+SELECT_FOR_FEEL = """
+SELECT c.record FROM clips c
+WHERE (c.record->'mechanical_check'->>'passed')::bool = true
+  AND NOT EXISTS (
+      SELECT 1 FROM feel_scores f
+      WHERE f.provider = c.provider AND f.provider_id = c.provider_id)
+ORDER BY c.analyzed_at
+LIMIT %s
+"""
+
+COUNT_FEEL_PASSED = """
+SELECT COUNT(*) FROM clips
+WHERE (record->'mechanical_check'->>'passed')::bool = true
+"""
+```
+
+- [ ] **Step 3b: Implement InMemory methods** — add to `InMemoryCatalogStore` (it already has `self._records`; add `self._feel = {}` in `__init__` alongside `self._seed`):
+
+```python
+    # --- Phase-4 feel scoring ---
+
+    def ensure_feel_schema(self) -> None:
+        if not hasattr(self, "_feel"):
+            self._feel = {}
+
+    def load_for_feel_scoring(self, limit: int = 100) -> list[dict]:
+        self.ensure_feel_schema()
+        out = []
+        for rec in self._records.values():
+            if not (rec.get("mechanical_check") or {}).get("passed"):
+                continue
+            if (rec.get("provider"), rec.get("provider_id")) in self._feel:
+                continue
+            out.append(rec)
+            if len(out) >= limit:
+                break
+        return out
+
+    def set_feel_score(self, provider, provider_id, *, result, clip_url, now) -> None:
+        self.ensure_feel_schema()
+        self._feel[(provider, provider_id)] = {
+            "provider": provider, "provider_id": provider_id,
+            "decision": result.decision, "score": int(result.score),
+            "camera_motion": result.camera_motion,
+            "subject_motion": result.subject_motion,
+            "visual_complexity": result.visual_complexity,
+            "human_presence": result.human_presence, "mood": result.mood,
+            "reason": result.reason, "clip_url": clip_url,
+            "scored_at": now, "scored_by": result.scored_by,
+        }
+
+    def load_feel_scores(self) -> list[dict]:
+        self.ensure_feel_schema()
+        rows = list(self._feel.values())
+        rows.sort(key=lambda r: (-int(r["score"]), r["provider"], r["provider_id"]))
+        return [dict(r) for r in rows]
+
+    def feel_page(self, *, decision=None, min_score=0, limit=6, offset=0) -> list[dict]:
+        rows = self.load_feel_scores()
+        if decision and decision != "all":
+            rows = [r for r in rows if r["decision"] == decision]
+        if min_score:
+            rows = [r for r in rows if int(r["score"]) >= int(min_score)]
+        return rows[offset:offset + limit]
+
+    def feel_counts(self) -> dict:
+        self.ensure_feel_schema()
+        c = {"KEEP": 0, "BORDERLINE": 0, "REJECT": 0, "total": 0,
+             "scored": 0, "unscored": 0}
+        for r in self._feel.values():
+            if r["decision"] in c:
+                c[r["decision"]] += 1
+            c["total"] += 1
+        passed = sum(1 for rec in self._records.values()
+                     if (rec.get("mechanical_check") or {}).get("passed"))
+        c["scored"] = len(self._feel)
+        c["unscored"] = max(0, passed - len(self._feel))
+        return c
+```
+
+Also add `self._feel = {}` to `InMemoryCatalogStore.__init__`.
+
+- [ ] **Step 3c: Implement Neon methods + abstract stubs** — add abstract method declarations on `CatalogStore` (mirroring the seed section) for the six methods, then add to `NeonCatalogStore`:
+
+```python
+    # --- Phase-4 feel scoring ---
+
+    def ensure_feel_schema(self) -> None:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(CREATE_FEEL_SCORES)
+            conn.commit()
+
+    def load_for_feel_scoring(self, limit: int = 100) -> list[dict]:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(SELECT_FOR_FEEL, (limit,))
+                rows = cur.fetchall()
+        out = []
+        for row in rows:
+            rec = row[0]
+            if isinstance(rec, str):
+                rec = json.loads(rec)
+            out.append(rec)
+        return out
+
+    def set_feel_score(self, provider, provider_id, *, result, clip_url, now) -> None:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(UPSERT_FEEL, (
+                    provider, provider_id, result.decision, int(result.score),
+                    result.camera_motion, result.subject_motion,
+                    result.visual_complexity, result.human_presence, result.mood,
+                    result.reason, clip_url, now, result.scored_by))
+            conn.commit()
+
+    def load_feel_scores(self) -> list[dict]:
+        cols = ("provider", "provider_id", "decision", "score", "camera_motion",
+                "subject_motion", "visual_complexity", "human_presence", "mood",
+                "reason", "clip_url", "scored_at", "scored_by")
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(SELECT_FEEL)
+                rows = cur.fetchall()
+        return [dict(zip(cols, row)) for row in rows]
+
+    def feel_page(self, *, decision=None, min_score=0, limit=6, offset=0) -> list[dict]:
+        rows = self.load_feel_scores()
+        if decision and decision != "all":
+            rows = [r for r in rows if r["decision"] == decision]
+        if min_score:
+            rows = [r for r in rows if int(r["score"]) >= int(min_score)]
+        return rows[offset:offset + limit]
+
+    def feel_counts(self) -> dict:
+        c = {"KEEP": 0, "BORDERLINE": 0, "REJECT": 0, "total": 0,
+             "scored": 0, "unscored": 0}
+        for r in self.load_feel_scores():
+            if r["decision"] in c:
+                c[r["decision"]] += 1
+            c["total"] += 1
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(COUNT_FEEL_PASSED)
+                passed = int((cur.fetchone() or [0])[0])
+        c["scored"] = c["total"]
+        c["unscored"] = max(0, passed - c["total"])
+        return c
+```
+
+- [ ] **Step 4: Run tests**
+
+Run: `python -m pytest video_library/tests/test_catalog_store_feel.py -q`
+Expected: PASS (all 6).
+
+- [ ] **Step 5: Sync mirror + full suite**
+
+Run: `cp video_library/catalog_store.py hf_space/video_library/catalog_store.py && python -m pytest video_library/tests/ -q`
+Expected: PASS (all).
+
+---
+
+### Task 4: `feel_score_runner.py` — the scoring loop
+
+Orchestrate download → router.describe (with FEEL_PROMPT) → set_feel_score. Resumable, quota-aware, per-clip error skip-list.
+
+**Files:**
+- Create: `video_library/feel_score_runner.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_feel_score_runner.py`
+
+**Interfaces:**
+- Consumes: store methods (Task 3), `pick_stream_url` (curation_app), `download_to_temp`/`delete_quietly` (downloader), `NoModelAvailableError` (vision_describe).
+- Produces: `run_feel_scoring(store, router, *, limit=10000, batch_size=100, tmp_dir, now=_now_iso, sleep=time.sleep, rpm_seconds=2.5, downloader=download_to_temp, deleter=delete_quietly, url_picker=pick_stream_url, quota_detector=is_quota_error) -> dict`. Summary keys: `processed, scored, skipped, errors, stopped_quota, by_decision`.
+
+- [ ] **Step 1: Write failing tests** — create `video_library/tests/test_feel_score_runner.py`:
+
+```python
+from video_library.catalog_store import InMemoryCatalogStore
+from video_library.models import ClipRecord, Candidate, FileOption, Description, CheckResult
+from video_library.feel_scorer import FeelResult
+from video_library.feel_score_runner import run_feel_scoring
+from video_library.vision_describe import NoModelAvailableError
+
+
+def _clip(pid, *, passed=True):
+    cand = Candidate(provider="pexels", provider_id=pid, page_url="p",
+                     files=[FileOption("sd", 640, 360, 25.0, "http://x/" + pid)],
+                     duration=10.0, width=640, height=360, fps=25.0,
+                     creator_name="c", creator_url="u", license="pexels",
+                     original_query="q")
+    return ClipRecord(candidate=cand,
+                      description=Description(summary="s", environment="nature"),
+                      mechanical_check=CheckResult(passed=passed),
+                      analyzed_at="t" + pid).to_dict()
+
+
+def _store(*clips):
+    s = InMemoryCatalogStore()
+    for c in clips:
+        s._records[(c["provider"], c["provider_id"])] = c
+    return s
+
+
+class _FakeRouter:
+    def __init__(self, *, by_url=None, raise_on=None):
+        self._by_url = by_url or {}
+        self._raise_on = raise_on or {}
+        self.calls = []
+
+    def describe(self, path):
+        self.calls.append(path)
+        if path in self._raise_on:
+            raise self._raise_on[path]
+        r = self._by_url.get(path)
+        return r or FeelResult(decision="KEEP", score=80, described_by="fake")
+
+
+def _noop_dl(url, tmp, pid):
+    return url            # use the url itself as the "path" so router keys on it
+
+
+def _noop_del(path):
+    pass
+
+
+def test_scores_all_passed_clips():
+    s = _store(_clip("1"), _clip("2"), _clip("3", passed=False))
+    router = _FakeRouter()
+    summary = run_feel_scoring(s, router, tmp_dir="/tmp", sleep=lambda x: None,
+                               downloader=_noop_dl, deleter=_noop_del)
+    assert summary["processed"] == 2
+    assert summary["scored"] == 2
+    assert summary["by_decision"]["KEEP"] == 2
+    assert len(s.load_feel_scores()) == 2
+
+
+def test_skips_already_scored():
+    s = _store(_clip("1"), _clip("2"))
+    s.set_feel_score("pexels", "1",
+                     result=FeelResult(decision="KEEP", score=90), clip_url="u", now="t")
+    router = _FakeRouter()
+    summary = run_feel_scoring(s, router, tmp_dir="/tmp", sleep=lambda x: None,
+                               downloader=_noop_dl, deleter=_noop_del)
+    assert summary["processed"] == 1
+    assert [r["provider_id"] for r in s.load_feel_scores() if r["score"] == 90]
+
+
+def test_quota_stops_cleanly():
+    s = _store(_clip("1"), _clip("2"))
+    router = _FakeRouter(raise_on={"http://x/1": NoModelAvailableError("over cap")})
+    summary = run_feel_scoring(s, router, tmp_dir="/tmp", sleep=lambda x: None,
+                               downloader=_noop_dl, deleter=_noop_del)
+    assert summary["stopped_quota"] is True
+    assert summary["scored"] == 0
+
+
+def test_per_clip_error_skips_and_continues():
+    s = _store(_clip("1"), _clip("2"))
+    router = _FakeRouter(raise_on={"http://x/1": RuntimeError("bad json")})
+    summary = run_feel_scoring(s, router, tmp_dir="/tmp", sleep=lambda x: None,
+                               downloader=_noop_dl, deleter=_noop_del)
+    assert summary["errors"] == 1
+    assert summary["scored"] == 1
+
+
+def test_limit_caps_processing():
+    s = _store(*[_clip(str(i)) for i in range(5)])
+    router = _FakeRouter()
+    summary = run_feel_scoring(s, router, tmp_dir="/tmp", limit=2,
+                               sleep=lambda x: None,
+                               downloader=_noop_dl, deleter=_noop_del)
+    assert summary["processed"] == 2
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_feel_score_runner.py -q`
+Expected: FAIL (module not found).
+
+- [ ] **Step 3: Implement** — create `video_library/feel_score_runner.py`:
+
+```python
+"""Phase-4 orchestrator: score every mechanically-passed clip's FEEL.
+
+Loop mirrors tag_runner / library_builder:
+  1. Load a batch of mechanically-passed, not-yet-scored clips.
+  2. Pick the smallest streamable file, download to temp.
+  3. router.describe(path) -> FeelResult (router carries FEEL_PROMPT).
+  4. store.set_feel_score(...); delete temp.
+  5. Sleep rpm_seconds.
+Resumable (each row written immediately) and quota-aware (NoModelAvailableError or
+a detected quota error stops the run cleanly; rerun to continue).
+"""
+from __future__ import annotations
+import datetime as _dt
+import time
+from typing import Callable
+
+from .downloader import download_to_temp, delete_quietly
+from .curation_app import pick_stream_url
+from .vision_describe import NoModelAvailableError, is_quota_error
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def run_feel_scoring(store, router, *,
+                     limit: int = 10_000,
+                     batch_size: int = 100,
+                     tmp_dir,
+                     now: Callable[[], str] = _now_iso,
+                     sleep: Callable[[float], None] = time.sleep,
+                     rpm_seconds: float = 2.5,
+                     downloader=download_to_temp,
+                     deleter=delete_quietly,
+                     url_picker=pick_stream_url,
+                     quota_detector: Callable[[Exception], bool] = is_quota_error,
+                     ) -> dict:
+    summary = {"processed": 0, "scored": 0, "skipped": 0, "errors": 0,
+               "stopped_quota": False, "by_decision": {}}
+    errored: set[tuple[str, str]] = set()
+    remaining = limit
+    while remaining > 0:
+        batch = store.load_for_feel_scoring(limit=min(batch_size, remaining))
+        batch = [r for r in batch
+                 if (r.get("provider"), r.get("provider_id")) not in errored]
+        if not batch:
+            break
+        for rec in batch:
+            if remaining <= 0:
+                break
+            provider = rec.get("provider")
+            provider_id = rec.get("provider_id")
+            summary["processed"] += 1
+            remaining -= 1
+
+            url = url_picker(rec)
+            if not url:
+                summary["skipped"] += 1
+                continue
+
+            temp_path = None
+            try:
+                temp_path = downloader(url, tmp_dir, provider_id)
+                result = router.describe(str(temp_path))
+            except NoModelAvailableError:
+                summary["stopped_quota"] = True
+                if temp_path is not None:
+                    deleter(temp_path)
+                return summary
+            except Exception as e:
+                if temp_path is not None:
+                    deleter(temp_path)
+                if quota_detector(e):
+                    summary["stopped_quota"] = True
+                    return summary
+                summary["errors"] += 1
+                errored.add((provider, provider_id))
+                continue
+            else:
+                deleter(temp_path)
+
+            store.set_feel_score(provider, provider_id, result=result,
+                                 clip_url=url, now=now())
+            summary["scored"] += 1
+            d = result.decision
+            summary["by_decision"][d] = summary["by_decision"].get(d, 0) + 1
+            sleep(rpm_seconds)
+    return summary
+```
+
+Note: `_noop_dl` in the test returns the URL as the path, so the fake router keys its `raise_on`/`by_url` on `http://x/<id>`.
+
+- [ ] **Step 4: Run tests**
+
+Run: `python -m pytest video_library/tests/test_feel_score_runner.py -q`
+Expected: PASS (all 5).
+
+- [ ] **Step 5: Sync mirror**
+
+Run: `cp video_library/feel_score_runner.py hf_space/video_library/feel_score_runner.py && echo SYNC`
+
+---
+
+### Task 5: `feel_score_app.py` — UI card helpers + trigger glue
+
+Split-out logic (testable without Gradio) + a background-thread trigger mirroring `seed_reason_app.py` + the Gradio results tab (built lazily).
+
+**Files:**
+- Create: `video_library/feel_score_app.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_feel_score_app.py`
+
+**Interfaces:**
+- Consumes: store methods (Task 3), `run_feel_scoring` (Task 4).
+- Produces:
+  - `feel_card(row: dict) -> dict` — `{video_url, score, decision, badge, caption}`.
+  - `feel_caption(row: dict) -> str`.
+  - `FEEL_SCORE_STATUS` dict + `run_feel_scorer(...)`, `format_feel_status(status)`, `start_feel_scorer_thread()`, `default_feel_components()`.
+  - `build_feel_tab(store_factory, *, now)` — returns `{"init_fn", "init_inputs", "init_outputs"}`.
+  - `FEEL_PAGE_SIZE = 6`, `DECISION_CHOICES = ["all", "KEEP", "BORDERLINE", "REJECT"]`.
+
+- [ ] **Step 1: Write failing tests** — create `video_library/tests/test_feel_score_app.py`:
+
+```python
+from video_library.feel_score_app import (
+    feel_card, feel_caption, format_feel_status, FEEL_PAGE_SIZE, DECISION_CHOICES,
+)
+
+
+def _row(**kw):
+    base = {"provider": "pexels", "provider_id": "1", "decision": "KEEP",
+            "score": 88, "camera_motion": "slow drone", "subject_motion": "still",
+            "visual_complexity": "low", "human_presence": "none", "mood": "calm",
+            "reason": "calm open nature", "clip_url": "http://c/1",
+            "scored_at": "t", "scored_by": "gemini-x"}
+    base.update(kw)
+    return base
+
+
+def test_feel_card_shape():
+    c = feel_card(_row())
+    assert c["video_url"] == "http://c/1"
+    assert c["score"] == 88
+    assert c["decision"] == "KEEP"
+    assert "88" in c["caption"]
+
+
+def test_feel_caption_includes_reason_and_decision():
+    cap = feel_caption(_row(reason="busy motion", decision="REJECT", score=30))
+    assert "busy motion" in cap
+    assert "REJECT" in cap
+    assert "30" in cap
+
+
+def test_page_size_is_crash_safe():
+    assert FEEL_PAGE_SIZE <= 6
+
+
+def test_decision_choices():
+    assert DECISION_CHOICES[0] == "all"
+    assert "KEEP" in DECISION_CHOICES
+
+
+def test_format_feel_status_running():
+    s = format_feel_status({"state": "running", "summary": None, "error": None})
+    assert "running" in s
+
+
+def test_format_feel_status_summary():
+    s = format_feel_status({"state": "done", "error": None,
+        "summary": {"processed": 10, "scored": 8, "skipped": 1, "errors": 1,
+                    "stopped_quota": False, "by_decision": {"KEEP": 5}}})
+    assert "scored 8" in s
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_feel_score_app.py -q`
+Expected: FAIL (module not found).
+
+- [ ] **Step 3: Implement** — create `video_library/feel_score_app.py`. The pure helpers + trigger glue first (Gradio import is lazy inside `build_feel_tab`):
+
+```python
+"""Phase-4 feel-scorer glue: pure card/status helpers, a background-thread trigger
+(mirrors seed_reason_app), and the Gradio 'Feel Score' results tab.
+
+Crash safety carried from the curation tab: video players autoplay=False, loop=True,
+page size 6, browsing repaints a small page, guarded render (no 500 on cold DB).
+"""
+from __future__ import annotations
+import threading
+import traceback
+
+from . import config
+from .feel_score_runner import run_feel_scoring
+from .__main__ import _now_iso
+
+FEEL_PAGE_SIZE = 6
+DECISION_CHOICES = ["all", "KEEP", "BORDERLINE", "REJECT"]
+
+_BADGE = {"KEEP": "✅ KEEP", "BORDERLINE": "🟡 BORDERLINE", "REJECT": "❌ REJECT"}
+
+
+def feel_caption(row: dict) -> str:
+    dec = row.get("decision", "")
+    badge = _BADGE.get(dec, dec)
+    parts = [f"**{badge} · {row.get('score', 0)}/100**"]
+    meta = []
+    if row.get("mood"):
+        meta.append(row["mood"])
+    if row.get("camera_motion"):
+        meta.append("cam: " + row["camera_motion"])
+    if row.get("subject_motion"):
+        meta.append("subj: " + row["subject_motion"])
+    if meta:
+        parts.append(" · ".join(meta))
+    if row.get("reason"):
+        parts.append("_" + row["reason"] + "_")
+    return "  \n".join(parts)
+
+
+def feel_card(row: dict) -> dict:
+    return {
+        "video_url": row.get("clip_url", ""),
+        "score": int(row.get("score", 0) or 0),
+        "decision": row.get("decision", ""),
+        "badge": _BADGE.get(row.get("decision", ""), row.get("decision", "")),
+        "caption": feel_caption(row),
+    }
+
+
+# --- background-thread trigger (mirrors seed_reason_app) ---
+
+FEEL_SCORE_STATUS = {"state": "idle", "summary": None, "error": None}
+
+
+def default_feel_components():
+    """Build the Neon store (feel schema ensured) + a FEEL_PROMPT vision router."""
+    from .catalog_store import NeonCatalogStore
+    from .vision_describe import GenAIBackend, VisionRouter
+    from .feel_scorer import FEEL_PROMPT, parse_feel_json
+    from .__main__ import _build_model_configs, _utc_date
+
+    store = NeonCatalogStore(config.NEON_DATABASE_URL)
+    store.ensure_schema()
+    store.ensure_feel_schema()
+    router = VisionRouter(
+        _build_model_configs(), store,
+        backend_factory=lambda cfg: GenAIBackend(
+            cfg.api_key or config.GEMINI_API_KEY, cfg.model_id,
+            strip_audio=cfg.strip_audio,
+            prompt=FEEL_PROMPT, parser=parse_feel_json),
+        today=_utc_date,
+    )
+    return store, router
+
+
+def run_feel_scorer(*, components_factory=default_feel_components,
+                    runner=run_feel_scoring, limit=10000):
+    if not config.NEON_DATABASE_URL:
+        FEEL_SCORE_STATUS.update(state="missing-key", error="set NEON_DATABASE_URL")
+        return FEEL_SCORE_STATUS
+    if not config.GEMINI_API_KEY:
+        FEEL_SCORE_STATUS.update(state="missing-key", error="set GEMINI_API_KEY")
+        return FEEL_SCORE_STATUS
+    FEEL_SCORE_STATUS.update(state="running", error=None, summary=None)
+    try:
+        store, router = components_factory()
+        summary = runner(store, router, limit=limit, tmp_dir=config.TMP_DIR,
+                         now=_now_iso)
+        FEEL_SCORE_STATUS.update(state="done", summary=summary)
+    except Exception as e:
+        FEEL_SCORE_STATUS.update(
+            state="error", error=repr(e) + "\n" + traceback.format_exc())
+    return FEEL_SCORE_STATUS
+
+
+def format_feel_status(status) -> str:
+    lines = ["Feel scorer: " + str(status.get("state"))]
+    summary = status.get("summary")
+    if summary:
+        lines.append(
+            "processed {processed} | scored {scored} | "
+            "skipped {skipped} | errors {errors}".format(**summary))
+        if summary.get("by_decision"):
+            lines.append("decisions: " + ", ".join(
+                f"{k} {v}" for k, v in summary["by_decision"].items()))
+        if summary.get("stopped_quota"):
+            lines.append("Stopped: Gemini quota hit; resume later.")
+    if status.get("error"):
+        lines.append("Error: " + str(status["error"]).splitlines()[0])
+    return "\n".join(lines)
+
+
+def start_feel_scorer_thread():
+    t = threading.Thread(target=run_feel_scorer, name="feel-scorer-loop", daemon=True)
+    t.start()
+    return t
+```
+
+Then append the Gradio tab builder in the same file (mirrors curation_app's snapshot/paging, but read-only — no verdict buttons):
+
+```python
+def build_feel_tab(store_factory, *, now):
+    """Construct the 'Feel Score' results tab. Imported lazily by app.py."""
+    import gradio as gr
+
+    session = {"rows": None, "decision": "all", "min_score": 0, "page": 1}
+
+    gr.Markdown(
+        "### Feel Score\nGemini watched every clip and scored how well it works as a "
+        "quiet background for narration. Sorted best-first. Filter by decision or "
+        "minimum score, then click a clip to play it.")
+
+    with gr.Row():
+        dec_dd = gr.Dropdown(DECISION_CHOICES, value="all", label="Decision")
+        min_slider = gr.Slider(0, 100, value=0, step=5, label="Min score")
+    header = gr.Markdown("")
+    page_label = gr.Markdown("")
+
+    slots = []
+    with gr.Column():
+        for _ in range(FEEL_PAGE_SIZE):
+            with gr.Row(visible=False) as row:
+                player = gr.Video(interactive=False, height=260, show_label=False,
+                                  autoplay=False, loop=True)
+                caption = gr.Markdown("")
+            slots.append({"row": row, "player": player, "caption": caption})
+
+    with gr.Row():
+        prev_btn = gr.Button("← Prev")
+        next_btn = gr.Button("Next →")
+
+    def _load_rows():
+        if session["rows"] is None:
+            session["rows"] = store_factory().load_feel_scores()
+        return session["rows"]
+
+    def _filtered():
+        rows = _load_rows()
+        dec, ms = session["decision"], session["min_score"]
+        if dec and dec != "all":
+            rows = [r for r in rows if r["decision"] == dec]
+        if ms:
+            rows = [r for r in rows if int(r["score"]) >= int(ms)]
+        return rows
+
+    def _outputs():
+        out = [header, page_label]
+        for slot in slots:
+            out += [slot["row"], slot["player"], slot["caption"]]
+        return out
+
+    def _render():
+        view = _filtered()
+        pages = max(1, (len(view) + FEEL_PAGE_SIZE - 1) // FEEL_PAGE_SIZE)
+        session["page"] = max(1, min(session["page"], pages))
+        off = (session["page"] - 1) * FEEL_PAGE_SIZE
+        clips = view[off:off + FEEL_PAGE_SIZE]
+        counts = store_factory().feel_counts() if session["rows"] is None else None
+        hdr = (f"{len(view)} clips shown · {len(_load_rows())} scored total")
+        updates = [gr.update(value=hdr),
+                   gr.update(value=f"Page {session['page']} / {pages}")]
+        for i, slot in enumerate(slots):
+            if i < len(clips):
+                card = feel_card(clips[i])
+                updates += [gr.update(visible=True),
+                            gr.update(value=card["video_url"]),
+                            gr.update(value=card["caption"])]
+            else:
+                updates += [gr.update(visible=False), gr.update(value=None),
+                            gr.update(value="")]
+        return updates
+
+    def _error_render(exc):
+        updates = [gr.update(value=f"⚠️ Could not load scores: {exc}"),
+                   gr.update(value="")]
+        for _ in slots:
+            updates += [gr.update(visible=False), gr.update(value=None),
+                        gr.update(value="")]
+        return updates
+
+    def _on_filter(dec, ms):
+        session["decision"], session["min_score"], session["page"] = dec, ms, 1
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    def _on_page(delta):
+        session["page"] = max(1, session["page"] + delta)
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    def _on_init():
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    full_out = _outputs()
+    dec_dd.change(_on_filter, [dec_dd, min_slider], full_out)
+    min_slider.release(_on_filter, [dec_dd, min_slider], full_out)
+    prev_btn.click(lambda: _on_page(-1), None, full_out)
+    next_btn.click(lambda: _on_page(1), None, full_out)
+
+    return {"init_fn": _on_init, "init_inputs": None, "init_outputs": _outputs()}
+```
+
+Note: the unused `counts` line is dead — remove it; keep `hdr` simple as shown. (Self-review fix applied: drop the `counts = ...` line.)
+
+- [ ] **Step 4: Run tests**
+
+Run: `python -m pytest video_library/tests/test_feel_score_app.py -q`
+Expected: PASS (all 6).
+
+- [ ] **Step 5: Sync mirror**
+
+Run: `cp video_library/feel_score_app.py hf_space/video_library/feel_score_app.py && echo SYNC`
+
+---
+
+### Task 6: Wire the tab into `app.py` + CLI flag
+
+Add the "Feel Score" tab (button + status + results grid) to the Gradio app, and a `--feel-score` CLI entry.
+
+**Files:**
+- Modify: `video_library/app.py` (+ hf_space mirror)
+- Modify: `video_library/__main__.py` (+ hf_space mirror)
+- Test: `video_library/tests/test_cli.py` (append)
+
+**Interfaces:**
+- Consumes: `build_feel_tab`, `FEEL_SCORE_STATUS`, `run_feel_scorer`, `format_feel_status`, `start_feel_scorer_thread` from Task 5.
+
+- [ ] **Step 1: Write failing CLI test** — append to `video_library/tests/test_cli.py`:
+
+```python
+def test_feel_score_flag_dispatches(monkeypatch):
+    import video_library.__main__ as m
+    called = {}
+
+    def _fake_run(limit):
+        called["limit"] = limit
+        return 0
+
+    monkeypatch.setattr(m, "_run_feel_score", _fake_run)
+    rc = m.main(["--feel-score", "--limit", "7"])
+    assert rc == 0
+    assert called["limit"] == 7
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `python -m pytest video_library/tests/test_cli.py::test_feel_score_flag_dispatches -q`
+Expected: FAIL (`_run_feel_score` missing / flag unknown).
+
+- [ ] **Step 3a: Implement CLI** — in `__main__.py` add the arg in `build_arg_parser`:
+
+```python
+    p.add_argument("--feel-score", action="store_true",
+                   help="Phase-4: Gemini re-watches clips and scores their feel, "
+                        "then exit")
+```
+
+Add the entry function:
+
+```python
+def _run_feel_score(limit):
+    """Phase-4 entry: score clip feel via Gemini. Returns exit code."""
+    if not config.NEON_DATABASE_URL:
+        print("ERROR: set NEON_DATABASE_URL", file=sys.stderr)
+        return 2
+    if not config.GEMINI_API_KEY:
+        print("ERROR: set GEMINI_API_KEY", file=sys.stderr)
+        return 2
+    from .feel_score_app import default_feel_components
+    from .feel_score_runner import run_feel_scoring
+    store, router = default_feel_components()
+    summary = run_feel_scoring(store, router, limit=limit, tmp_dir=config.TMP_DIR,
+                               now=_now_iso)
+    print("Feel-scored:", summary)
+    return 0
+```
+
+Wire dispatch in `main()` after the `--seed-reason` block:
+
+```python
+    if args.feel_score:
+        return _run_feel_score(args.limit)
+```
+
+- [ ] **Step 3b: Implement app.py tab** — add import near the other tab imports:
+
+```python
+from .feel_score_app import (
+    FEEL_SCORE_STATUS, run_feel_scorer, format_feel_status,
+    start_feel_scorer_thread, build_feel_tab,
+)
+```
+
+In `build_ui`, after the Seed Reasons tab block, add a new tab (reuse the cached curation store factory `_curation_store` / `_seed_store` pattern already in the file — add a `_feel_store` that ensures the feel schema once):
+
+```python
+        with gr.Tab("Feel Score"):
+            _feel_state = {"store": None}
+
+            def _feel_store():
+                if _feel_state["store"] is None:
+                    store = _curation_store()
+                    store.ensure_feel_schema()
+                    _feel_state["store"] = store
+                return _feel_state["store"]
+
+            feel_status_box = gr.Textbox(
+                label="Gemini feel scorer",
+                value=format_feel_status(FEEL_SCORE_STATUS),
+                lines=4, interactive=False)
+            with gr.Row():
+                def _start_feel():
+                    if FEEL_SCORE_STATUS.get("state") != "running":
+                        start_feel_scorer_thread()
+                    return format_feel_status(FEEL_SCORE_STATUS)
+
+                gr.Button("Score clips (Gemini)").click(
+                    _start_feel, outputs=feel_status_box)
+                gr.Button("Refresh scorer").click(
+                    lambda: format_feel_status(FEEL_SCORE_STATUS),
+                    outputs=feel_status_box)
+
+            feel_init = build_feel_tab(_feel_store, now=_now_iso)
+            demo.load(feel_init["init_fn"], feel_init["init_inputs"],
+                      feel_init["init_outputs"])
+```
+
+If `_curation_store` isn't the exact name in the file, use whatever cached store factory the curation/seed tabs already share (verify by reading `app.py:build_ui` first).
+
+- [ ] **Step 4: Run tests + import-sanity**
+
+Run: `python -m pytest video_library/tests/test_cli.py -q`
+Expected: PASS.
+
+Run: `python -c "import video_library.app"` (from repo root)
+Expected: no error (gradio import ok) — if gradio unavailable in dev, skip; the tab body is lazy.
+
+- [ ] **Step 5: Sync mirrors + full suite**
+
+Run: `cp video_library/__main__.py hf_space/video_library/__main__.py && cp video_library/app.py hf_space/video_library/app.py && python -m pytest video_library/tests/ -q`
+Expected: PASS (all).
+
+---
+
+### Task 7: End-to-end sanity + tree-sync verification
+
+Final guard: confirm both trees are byte-identical for every touched file and the whole suite is green.
+
+**Files:** none (verification only).
+
+- [ ] **Step 1: Verify tree sync**
+
+Run:
+```bash
+for f in vision_describe.py feel_scorer.py catalog_store.py feel_score_runner.py feel_score_app.py __main__.py app.py; do
+  diff -q "video_library/$f" "hf_space/video_library/$f" && echo "SYNC $f" || echo "DIFF $f";
+done
+```
+Expected: `SYNC` for all 7.
+
+- [ ] **Step 2: Full test suite**
+
+Run: `python -m pytest video_library/tests/ -q`
+Expected: PASS (prior count + ~26 new tests).
+
+- [ ] **Step 3: Confirm FEEL_PROMPT is verbatim**
+
+Manually open `video_library/feel_scorer.py` and confirm `FEEL_PROMPT` is the user's exact rubric (role, core principle, camera-vs-subject distinction, desired style, scoring rubric out of 100, decision bands, JSON output format). No paraphrasing.
+
+- [ ] **Step 4: Deploy note**
+
+The HF Space runs `python -m video_library.app`. On next Space restart the "Feel Score" tab appears; clicking "Score clips (Gemini)" starts the background pass (resumable, quota-aware). `feel_scores` table is created on first tab load. User runs the actual scoring on the Space (fast internet, Gemini quota).
+
+---
+
+## Self-Review
+
+**Spec coverage:**
+- Output score/100 + decision + full feel schema → Task 2 (`FeelResult`, `parse_feel_json`) + verbatim `FEEL_PROMPT`. ✓
+- Which clips = mechanically-passed, ignore hand-decisions → Task 3 `load_for_feel_scoring` (filters on `mechanical_check.passed`, not curation). ✓
+- Fresh `feel_scores` table, clips untouched, store all incl REJECT → Task 3. ✓
+- Resumable + quota-aware → Task 4 (`stopped_quota`, skip already-scored). ✓
+- UI: trigger + results grid, crash-safe → Tasks 5-6 (autoplay off, loop on, page 6, guarded render). ✓
+- No few-shot seed injection → confirmed; prompt is sole taste source (Task 2). ✓
+- Two-tree sync → every non-test task step 5 + Task 7. ✓
+
+**Placeholder scan:** `FEEL_PROMPT` in Task 2 Step 3 now holds the user's full rubric verbatim (no placeholder). No TBDs remain.
+
+**Type consistency:** `FeelResult` fields identical across Tasks 2/3/4/5. `run_feel_scoring` signature matches its test and the CLI/app callers. `set_feel_score(provider, provider_id, *, result, clip_url, now)` consistent Task 3 ↔ 4. `build_feel_tab(store_factory, *, now) -> {"init_fn","init_inputs","init_outputs"}` matches curation_app convention used in app.py wiring.

docs/superpowers/specs/2026-07-20-feel-scorer-design.md

@@ -0,0 +1,192 @@
+# Feel Scorer (Phase 4) — Design
+
+**Date:** 2026-07-20
+**Status:** Approved for planning
+
+## Goal
+
+Gemini re-watches every mechanically-passed clip and judges its *feel* — whether
+it can sit quietly under contemplative Osho narration without competing for
+attention — producing a fresh, ready-to-use, score-ranked shortlist of good clips.
+
+This replaces hand-curating ~1200 clips. The problem it fixes: the Phase-1 vision
+pass (`VISION_PROMPT`) deliberately forbids feel-judgment ("Do NOT guess a mood,
+emotion, theme, or meaning"), so the catalog is *feel-blind*. This pass gives
+Gemini eyes on the pixels plus a precise taste spec, and writes the verdict to a
+new table.
+
+## Scope decisions (locked)
+
+1. **Output per clip:** score /100 + KEEP/BORDERLINE/REJECT + full feel schema
+   (`camera_motion`, `subject_motion`, `visual_complexity`, `human_presence`,
+   `mood`, `reason`). Exactly the user-supplied rubric prompt (verbatim, below).
+2. **Which clips:** all mechanically-passed clips (Phase-1 `mechanical_check.passed`
+   == true). Ignore whether the user hand-decided them in Phase 3 — Gemini's
+   judgment with this prompt supersedes the rushed hand-tags. Resumable + quota-aware.
+3. **Storage:** fresh dedicated `feel_scores` table. `clips` table untouched.
+   Store *every* scored clip (including REJECTs); filter at query time.
+4. **UI:** new "Feel Score" tab — trigger button with live progress + a results
+   grid sorted best-first. Carries all crash-safety guards from the curation tab.
+5. **No few-shot seed injection.** The taste lives entirely in the prompt. The
+   Phase-3.5 seed set is NOT used as calibration examples (hand-tags are noisy).
+
+## The taste prompt (authoritative — used verbatim)
+
+The full rubric prompt supplied by the user is the single source of taste truth.
+It is stored as `FEEL_PROMPT` in the new backend module. Key contract points the
+rest of the design depends on:
+
+- Judges *visual suitability as a quiet background*, never philosophical content.
+- Camera motion good (slow reveal) vs subject motion bad (in-frame distraction) —
+  judged separately.
+- Scoring rubric out of **100**: Subject Motion 35, Camera Movement 20, Visual
+  Simplicity 20, Mood & Atmosphere 15, Human Presence 10.
+- Decision bands: **KEEP 80–100**, **BORDERLINE 60–79**, **REJECT <60**.
+- Output is strict JSON:
+  ```json
+  {
+    "decision": "KEEP | BORDERLINE | REJECT",
+    "score": 0,
+    "camera_motion": "",
+    "subject_motion": "",
+    "visual_complexity": "low | medium | high",
+    "human_presence": "none | distant | prominent",
+    "mood": "",
+    "reason": ""
+  }
+  ```
+
+The exact prompt text is pasted into the code as a module constant; this doc does
+not re-transcribe it to avoid drift — the code is canonical.
+
+## Architecture
+
+Mirrors the existing Phase-1 vision pipeline. Reuses, does not reinvent:
+
+- **`downloader.download_to_temp` / `delete_quietly`** — stream clip to temp, delete
+  after. No bytes persist.
+- **`VisionRouter`** — model selection, Gemini→Gemma fallback, RPD/RPM accounting,
+  quota exhaustion. **Reused as-is.**
+- **Buffer/snapshot/paging patterns** from `curation_app.py` for the results grid.
+
+New components:
+
+### 1. `feel_scorer.py` — the vision judge (backend)
+- `FEEL_PROMPT` — the user's rubric, verbatim.
+- `FeelResult` dataclass: `decision, score, camera_motion, subject_motion,
+  visual_complexity, human_presence, mood, reason, scored_by`.
+- `parse_feel_json(text) -> FeelResult` — extract first valid JSON object (same
+  fence/brace strategy as `parse_vision_json`), validate `decision` ∈
+  {KEEP,BORDERLINE,REJECT} and coerce `score` to int 0–100.
+- A thin describe-with-feel-prompt path. **Design choice:** rather than duplicate
+  `GenAIBackend`, add an optional `prompt` override so the router can be pointed at
+  `FEEL_PROMPT` instead of `VISION_PROMPT`. The router/backend stay generic; the
+  prompt is injected. (Backend already stamps `described_by` → reused as `scored_by`.)
+
+  Concretely: `GenAIBackend.__init__` gains `prompt: str = VISION_PROMPT` and
+  `parser=parse_vision_json`. Phase-1 callers keep the defaults (no behavior change);
+  the feel pass constructs backends with `prompt=FEEL_PROMPT, parser=parse_feel_json`.
+  This keeps `VisionRouter` untouched.
+
+### 2. `feel_score_runner.py` — the orchestrator loop
+`run_feel_scoring(store, router, *, limit, tmp_dir, now, sleep, rpm_seconds,
+downloader, file_picker, overwrite=False) -> summary`
+
+Per clip (loop mirrors `tag_runner.run_tagging` + `library_builder.analyze_candidate`):
+1. Load a batch of mechanically-passed clips from the store via a new
+   `load_for_feel_scoring(limit)` — clips with `mechanical_check.passed == true`,
+   skipping those already in `feel_scores` (unless `overwrite`).
+2. Pick the smallest streamable file ≥360p via `pick_stream_url` (reused from
+   curation_app) — same clip URL the UI will show.
+3. `download_to_temp` → `router.describe(path)` (router now carries FEEL_PROMPT) →
+   `delete_quietly`.
+4. `store.set_feel_score(provider, provider_id, result=..., clip_url=..., now=...)`.
+5. Sleep `rpm_seconds`.
+6. On `NoModelAvailableError` / quota → stop gracefully, `stopped_quota=True`.
+   On other errors → count, skip clip, continue (per-run skip list, no infinite retry).
+
+Summary: `{processed, scored, skipped, errors, stopped_quota, by_decision}`.
+
+### 3. `catalog_store.py` additions (both InMem + Neon)
+- `CREATE_FEEL_SCORES` DDL: `feel_scores(provider TEXT, provider_id TEXT, decision
+  TEXT, score INT, camera_motion TEXT, subject_motion TEXT, visual_complexity TEXT,
+  human_presence TEXT, mood TEXT, reason TEXT, clip_url TEXT, scored_at TEXT,
+  scored_by TEXT, PRIMARY KEY (provider, provider_id))`.
+- `ensure_feel_schema()` — create table if not exists (called once via cached store).
+- `load_for_feel_scoring(limit)` — mechanically-passed clips not yet in feel_scores.
+- `set_feel_score(provider, provider_id, *, result, clip_url, now)` — UPSERT one row.
+- `feel_page(page, page_size, *, decision_filter, min_score)` — paged results
+  sorted by score DESC for the grid.
+- `feel_counts()` — `{KEEP, BORDERLINE, REJECT, total}` for the header.
+
+### 4. `feel_score_app.py` — HF Space UI tab
+Two parts in one tab, mirroring existing tabs:
+- **Trigger:** "Score clips" button → background thread runs `run_feel_scoring`,
+  status line shows live progress (mirrors `tag_app.py` / `seed_reason_app.py`).
+- **Results grid:** snapshot-backed, paged (PAGE_SIZE = 6). Each card: video player
+  (`autoplay=False`, `loop=True`, `height≈260`), score, decision, and reason.
+  Sorted best-first. Filter control: KEEP only (default) / KEEP+BORDERLINE / all.
+
+  **Crash-safety (carried from curation tab, non-negotiable):** autoplay OFF, small
+  page size, no violent re-render, snapshot loaded once into RAM, guarded `_render`
+  that catches exceptions and surfaces them in the header instead of 500-ing.
+
+### 5. `app.py` — register the tab
+Add "Feel Score" (Phase 4) tab. Store cached via existing closure dict; call
+`ensure_feel_schema()` once at build.
+
+### 6. `__main__.py` — CLI
+Add `--feel-score` flag (+ `--overwrite` to re-score existing rows) mirroring
+`--tag` / `--seed-reason`.
+
+## Data flow
+
+```
+mechanically-passed clip
+  -> load_for_feel_scoring (skip already-scored)
+  -> pick_stream_url (smallest >=360p)
+  -> download_to_temp
+  -> VisionRouter.describe  [FEEL_PROMPT, Gemini->Gemma fallback, RPD/RPM]
+  -> parse_feel_json -> FeelResult
+  -> set_feel_score  [UPSERT into feel_scores]
+  -> delete_quietly
+  (sleep rpm_seconds; stop cleanly on quota; resume next run)
+
+query time:
+  feel_page(decision_filter, min_score) sorted score DESC -> UI grid / shortlist
+```
+
+## Error handling
+
+- **Quota exhaustion:** `NoModelAvailableError` bubbles up → runner stops with
+  `stopped_quota=True`; already-written rows persist; next run resumes (skips scored).
+- **Per-clip failure** (download/parse/network): counted in `errors`, clip skipped,
+  added to per-run skip list so the loop never spins on it. Left unscored → retried
+  on a future run.
+- **Malformed model JSON:** `parse_feel_json` raises → treated as a per-clip error.
+- **UI cold-DB hiccup:** guarded `_render` catches, shows error in header, no 500.
+- **Idempotent:** UPSERT keyed on (provider, provider_id); re-running is safe.
+
+## Testing
+
+New test files (mirror existing suite; all against `InMemoryCatalogStore`, no network):
+- `test_feel_scorer.py` — `parse_feel_json` (valid, fenced, braces, bad decision,
+  score coercion/clamp, missing keys); `FeelResult`; backend prompt-injection path
+  with a fake client.
+- `test_feel_score_runner.py` — happy path, skip-already-scored, quota stop,
+  per-clip error skip-list, overwrite flag, summary counts.
+- `test_catalog_store_feel.py` — schema, `set_feel_score` UPSERT, `load_for_
+  feel_scoring` filtering, `feel_page` sort/filter, `feel_counts`.
+- `test_feel_score_app.py` — caption/card rendering, filter, guarded render,
+  trigger glue.
+
+Run full suite (`pytest`) after each component; keep both `video_library/` and
+`hf_space/video_library/` trees in sync (every new file written to both).
+
+## Explicitly out of scope (YAGNI)
+
+- No re-scoring of mechanically-rejected clips.
+- No seed/few-shot injection into the prompt.
+- No changes to `clips` table shape, Phase-1/2/3/3.5 behavior.
+- No auto-selection of clips into montages — this pass only *scores*; using the
+  shortlist is a later, separate concern.

video_library/__main__.py

@@ -78,8 +78,17 @@ def build_arg_parser():
                    help="print resolved (mood, query) pairs and exit")
     p.add_argument("--tag", action="store_true",
                    help="Phase-2: tag untagged clips with moods, then exit")
+    p.add_argument("--seed-reason", action="store_true",
+                   help="Phase-3.5: draft AI calibration reasons for decided clips, "
+                        "then exit")
+    p.add_argument("--overwrite-ai", action="store_true",
+                   help="with --seed-reason, redraft clips that already have an AI "
+                        "draft (never touches human-confirmed reasons)")
+    p.add_argument("--feel-score", action="store_true",
+                   help="Phase-4: Gemini re-watches clips and scores their feel, "
+                        "then exit")
     p.add_argument("--limit", type=int, default=10000,
-                   help="max clips to tag in this --tag run")
+                   help="max clips to process in this --tag / --seed-reason run")
     return p
 
 
@@ -104,6 +113,40 @@ def _run_tag(limit):
     return 0
 
 
+def _run_seed_reason(limit, overwrite_ai):
+    """Phase-3.5 entry: draft AI reasons for decided clips. Returns exit code."""
+    if not config.MISTRAL_API_KEY:
+        print("ERROR: set MISTRAL_API_KEY", file=sys.stderr)
+        return 2
+    if not config.NEON_DATABASE_URL:
+        print("ERROR: set NEON_DATABASE_URL", file=sys.stderr)
+        return 2
+    from .seed_reason_app import default_seed_reason_components
+    from .seed_reason_runner import run_seed_reasoning
+    store, reasoner = default_seed_reason_components()
+    summary = run_seed_reasoning(store, reasoner, limit=limit, now=_now_iso,
+                                 overwrite_ai=overwrite_ai)
+    print("Drafted:", summary)
+    return 0
+
+
+def _run_feel_score(limit):
+    """Phase-4 entry: score clip feel via Gemini. Returns exit code."""
+    if not config.NEON_DATABASE_URL:
+        print("ERROR: set NEON_DATABASE_URL", file=sys.stderr)
+        return 2
+    if not config.GEMINI_API_KEY:
+        print("ERROR: set GEMINI_API_KEY", file=sys.stderr)
+        return 2
+    from .feel_score_app import default_feel_components
+    from .feel_score_runner import run_feel_scoring
+    store, router = default_feel_components()
+    summary = run_feel_scoring(store, router, limit=limit, tmp_dir=config.TMP_DIR,
+                               now=_now_iso)
+    print("Feel-scored:", summary)
+    return 0
+
+
 def main(argv, *, runner=build_library, components_factory=default_components):
     args = build_arg_parser().parse_args(argv)
     if args.dry_run:
@@ -112,6 +155,10 @@ def main(argv, *, runner=build_library, components_factory=default_components):
         return 0
     if args.tag:
         return _run_tag(args.limit)
+    if args.seed_reason:
+        return _run_seed_reason(args.limit, args.overwrite_ai)
+    if args.feel_score:
+        return _run_feel_score(args.limit)
     missing = _missing_secrets()
     if missing:
         print("ERROR: set env vars: " + ", ".join(missing), file=sys.stderr)

video_library/app.py

@@ -146,6 +146,9 @@ def build_ui():
 
         with gr.Tab("Seed Reasons (Phase 3.5)"):
             from .seed_app import build_seed_tab
+            from .seed_reason_app import (
+                SEED_REASON_STATUS, format_seed_reason_status,
+                start_seed_reasoner_thread)
 
             # Reuse the curation store, but ensure the seed table exists too.
             _seed_state = {"store": None}
@@ -157,9 +160,63 @@ def build_ui():
                     _seed_state["store"] = store
                 return _seed_state["store"]
 
+            # AI-draft batch controls. Drafting is manual so it doesn't compete
+            # with Phase-1/2 for the Mistral rate limit; drafts land as source='ai'
+            # and you confirm them below by editing.
+            reason_box = gr.Textbox(
+                label="AI reason drafter",
+                value=format_seed_reason_status(SEED_REASON_STATUS),
+                lines=4, interactive=False)
+            with gr.Row():
+                def _start_reasoner():
+                    if SEED_REASON_STATUS.get("state") != "running":
+                        start_seed_reasoner_thread()
+                    return format_seed_reason_status(SEED_REASON_STATUS)
+
+                gr.Button("Draft reasons for decided clips").click(
+                    _start_reasoner, outputs=reason_box)
+                gr.Button("Refresh drafter").click(
+                    lambda: format_seed_reason_status(SEED_REASON_STATUS),
+                    outputs=reason_box)
+
             seed_init = build_seed_tab(_seed_store, now=_now_iso)
             demo.load(seed_init["init_fn"], seed_init["init_inputs"],
                       seed_init["init_outputs"])
+
+        with gr.Tab("Feel Score"):
+            from .feel_score_app import (
+                FEEL_SCORE_STATUS, format_feel_status,
+                start_feel_scorer_thread, build_feel_tab)
+
+            # Reuse the curation store, but ensure the feel table exists too.
+            _feel_state = {"store": None}
+
+            def _feel_store():
+                if _feel_state["store"] is None:
+                    store = _curation_store()
+                    store.ensure_feel_schema()
+                    _feel_state["store"] = store
+                return _feel_state["store"]
+
+            feel_status_box = gr.Textbox(
+                label="Gemini feel scorer",
+                value=format_feel_status(FEEL_SCORE_STATUS),
+                lines=4, interactive=False)
+            with gr.Row():
+                def _start_feel():
+                    if FEEL_SCORE_STATUS.get("state") != "running":
+                        start_feel_scorer_thread()
+                    return format_feel_status(FEEL_SCORE_STATUS)
+
+                gr.Button("Score clips (Gemini)").click(
+                    _start_feel, outputs=feel_status_box)
+                gr.Button("Refresh scorer").click(
+                    lambda: format_feel_status(FEEL_SCORE_STATUS),
+                    outputs=feel_status_box)
+
+            feel_init = build_feel_tab(_feel_store, now=_now_iso)
+            demo.load(feel_init["init_fn"], feel_init["init_inputs"],
+                      feel_init["init_outputs"])
     return demo
 
 

video_library/catalog_store.py

@@ -124,25 +124,63 @@ class CatalogStore(abc.ABC):
 
     @abc.abstractmethod
     def set_seed(self, provider: str, provider_id: str, *, verdict: str,
-                 reason: str, moods: list[str], clip_url: str, now: str) -> None:
-        """Upsert one seed row (verdict + written reason + context)."""
+                 reason: str, moods: list[str], clip_url: str, now: str,
+                 source: str = "human") -> None:
+        """Upsert one seed row (verdict + written reason + context).
+
+        `source` is "human" for a person-confirmed reason or "ai_draft" for an
+        LLM-generated draft awaiting review."""
         ...
 
     def set_seed_bulk(self, rows: list[dict], *, now: str) -> None:
         """Upsert many seed rows at once. Default = per-row set_seed; stores with
         real transactions override to commit in one round-trip. Each row dict:
-        {provider, provider_id, verdict, reason, moods, clip_url}."""
+        {provider, provider_id, verdict, reason, moods, clip_url, source?}."""
         for r in rows:
             self.set_seed(
                 r["provider"], r["provider_id"], verdict=r["verdict"],
                 reason=r.get("reason", ""), moods=r.get("moods") or [],
-                clip_url=r.get("clip_url", ""), now=now)
+                clip_url=r.get("clip_url", ""), now=now,
+                source=r.get("source", "human"))
 
     @abc.abstractmethod
     def load_seed(self) -> list[dict]:
         """All seed rows, for building the few-shot prompt later."""
         ...
 
+    # --- Phase-4 (feel scoring) ---
+
+    @abc.abstractmethod
+    def ensure_feel_schema(self) -> None:
+        """Create the feel_scores table if it doesn't exist."""
+        ...
+
+    @abc.abstractmethod
+    def load_for_feel_scoring(self, limit: int = 100) -> list[dict]:
+        """Mechanically-passed clip records with no feel_scores row yet."""
+        ...
+
+    @abc.abstractmethod
+    def set_feel_score(self, provider: str, provider_id: str, *, result,
+                       clip_url: str, now: str) -> None:
+        """UPSERT one clip's feel verdict."""
+        ...
+
+    @abc.abstractmethod
+    def load_feel_scores(self) -> list[dict]:
+        """All feel rows as dicts, sorted score DESC."""
+        ...
+
+    @abc.abstractmethod
+    def feel_page(self, *, decision=None, min_score=0, limit=6, offset=0) -> list[dict]:
+        """A page of feel rows, filtered by decision/min_score, sorted score DESC."""
+        ...
+
+    @abc.abstractmethod
+    def feel_counts(self) -> dict:
+        """{KEEP, BORDERLINE, REJECT, total, scored, unscored}."""
+        ...
+
 
 class InMemoryCatalogStore(CatalogStore):
     """Non-persistent store for tests and local dry runs."""
@@ -151,6 +189,7 @@ class InMemoryCatalogStore(CatalogStore):
         self._records: dict[tuple[str, str], dict] = {}
         self._usage: dict[tuple[str, str], int] = {}
         self._seed: dict[tuple[str, str], dict] = {}
+        self._feel: dict[tuple[str, str], dict] = {}
 
     def is_cached(self, provider: str, provider_id: str) -> bool:
         return (provider, provider_id) in self._records
@@ -278,15 +317,75 @@ class InMemoryCatalogStore(CatalogStore):
                 if curation_verdict(rec) in ("kept", "rejected")]
 
     def set_seed(self, provider, provider_id, *, verdict, reason, moods,
-                 clip_url, now):
+                 clip_url, now, source="human"):
         self._seed[(provider, provider_id)] = {
             "provider": provider, "provider_id": provider_id, "verdict": verdict,
             "reason": reason, "moods": list(moods or []), "clip_url": clip_url,
-            "at": now}
+            "source": source, "at": now}
 
     def load_seed(self) -> list[dict]:
         return list(self._seed.values())
 
+    # --- Phase-4 feel scoring ---
+
+    def ensure_feel_schema(self) -> None:
+        if not hasattr(self, "_feel"):
+            self._feel = {}
+
+    def load_for_feel_scoring(self, limit: int = 100) -> list[dict]:
+        self.ensure_feel_schema()
+        out = []
+        for rec in self._records.values():
+            if not (rec.get("mechanical_check") or {}).get("passed"):
+                continue
+            if (rec.get("provider"), rec.get("provider_id")) in self._feel:
+                continue
+            out.append(rec)
+            if len(out) >= limit:
+                break
+        return out
+
+    def set_feel_score(self, provider, provider_id, *, result, clip_url, now) -> None:
+        self.ensure_feel_schema()
+        self._feel[(provider, provider_id)] = {
+            "provider": provider, "provider_id": provider_id,
+            "decision": result.decision, "score": int(result.score),
+            "camera_motion": result.camera_motion,
+            "subject_motion": result.subject_motion,
+            "visual_complexity": result.visual_complexity,
+            "human_presence": result.human_presence, "mood": result.mood,
+            "reason": result.reason, "clip_url": clip_url,
+            "scored_at": now, "scored_by": result.scored_by,
+        }
+
+    def load_feel_scores(self) -> list[dict]:
+        self.ensure_feel_schema()
+        rows = list(self._feel.values())
+        rows.sort(key=lambda r: (-int(r["score"]), r["provider"], r["provider_id"]))
+        return [dict(r) for r in rows]
+
+    def feel_page(self, *, decision=None, min_score=0, limit=6, offset=0) -> list[dict]:
+        rows = self.load_feel_scores()
+        if decision and decision != "all":
+            rows = [r for r in rows if r["decision"] == decision]
+        if min_score:
+            rows = [r for r in rows if int(r["score"]) >= int(min_score)]
+        return rows[offset:offset + limit]
+
+    def feel_counts(self) -> dict:
+        self.ensure_feel_schema()
+        c = {"KEEP": 0, "BORDERLINE": 0, "REJECT": 0, "total": 0,
+             "scored": 0, "unscored": 0}
+        for r in self._feel.values():
+            if r["decision"] in c:
+                c[r["decision"]] += 1
+            c["total"] += 1
+        passed = sum(1 for rec in self._records.values()
+                     if (rec.get("mechanical_check") or {}).get("passed"))
+        c["scored"] = len(self._feel)
+        c["unscored"] = max(0, passed - len(self._feel))
+        return c
+
 
 def _default_connect(dsn: str):
     import psycopg
@@ -325,23 +424,32 @@ CREATE TABLE IF NOT EXISTS curation_seed (
     moods        JSONB NOT NULL DEFAULT '[]'::jsonb,
     clip_url     TEXT NOT NULL DEFAULT '',
     at           TEXT,
+    source       TEXT NOT NULL DEFAULT 'human',
     PRIMARY KEY (provider, provider_id)
 )
 """
 
+# Idempotent migration for tables created before `source` existed (the first ~23
+# human-labelled rows). Adds the column defaulting to 'human' so old rows read
+# correctly. Safe to run every boot.
+MIGRATE_SEED_SOURCE = """
+ALTER TABLE curation_seed ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'human'
+"""
+
 UPSERT_SEED = """
-INSERT INTO curation_seed (provider, provider_id, verdict, reason, moods, clip_url, at)
-VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s)
+INSERT INTO curation_seed (provider, provider_id, verdict, reason, moods, clip_url, at, source)
+VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s, %s)
 ON CONFLICT (provider, provider_id)
 DO UPDATE SET verdict = EXCLUDED.verdict,
               reason = EXCLUDED.reason,
               moods = EXCLUDED.moods,
               clip_url = EXCLUDED.clip_url,
-              at = EXCLUDED.at
+              at = EXCLUDED.at,
+              source = EXCLUDED.source
 """
 
 SELECT_SEED = """
-SELECT provider, provider_id, verdict, reason, moods, clip_url, at
+SELECT provider, provider_id, verdict, reason, moods, clip_url, at, source
 FROM curation_seed
 ORDER BY verdict, provider, provider_id
 """
@@ -356,6 +464,64 @@ WHERE (record->'mechanical_check'->>'passed')::bool = true
 ORDER BY analyzed_at
 """
 
+# --- Phase-4 feel scoring: a fresh, disposable table of AI feel verdicts. ---
+
+CREATE_FEEL_SCORES = """
+CREATE TABLE IF NOT EXISTS feel_scores (
+    provider          TEXT NOT NULL,
+    provider_id       TEXT NOT NULL,
+    decision          TEXT NOT NULL,
+    score             INT  NOT NULL DEFAULT 0,
+    camera_motion     TEXT NOT NULL DEFAULT '',
+    subject_motion    TEXT NOT NULL DEFAULT '',
+    visual_complexity TEXT NOT NULL DEFAULT '',
+    human_presence    TEXT NOT NULL DEFAULT '',
+    mood              TEXT NOT NULL DEFAULT '',
+    reason            TEXT NOT NULL DEFAULT '',
+    clip_url          TEXT NOT NULL DEFAULT '',
+    scored_at         TEXT,
+    scored_by         TEXT NOT NULL DEFAULT '',
+    PRIMARY KEY (provider, provider_id)
+)
+"""
+
+UPSERT_FEEL = """
+INSERT INTO feel_scores (provider, provider_id, decision, score, camera_motion,
+    subject_motion, visual_complexity, human_presence, mood, reason, clip_url,
+    scored_at, scored_by)
+VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
+ON CONFLICT (provider, provider_id) DO UPDATE SET
+    decision = EXCLUDED.decision, score = EXCLUDED.score,
+    camera_motion = EXCLUDED.camera_motion, subject_motion = EXCLUDED.subject_motion,
+    visual_complexity = EXCLUDED.visual_complexity,
+    human_presence = EXCLUDED.human_presence, mood = EXCLUDED.mood,
+    reason = EXCLUDED.reason, clip_url = EXCLUDED.clip_url,
+    scored_at = EXCLUDED.scored_at, scored_by = EXCLUDED.scored_by
+"""
+
+SELECT_FEEL = """
+SELECT provider, provider_id, decision, score, camera_motion, subject_motion,
+    visual_complexity, human_presence, mood, reason, clip_url, scored_at, scored_by
+FROM feel_scores
+ORDER BY score DESC, provider, provider_id
+"""
+
+# Mechanically-passed clips with no row yet in feel_scores.
+SELECT_FOR_FEEL = """
+SELECT c.record FROM clips c
+WHERE (c.record->'mechanical_check'->>'passed')::bool = true
+  AND NOT EXISTS (
+      SELECT 1 FROM feel_scores f
+      WHERE f.provider = c.provider AND f.provider_id = c.provider_id)
+ORDER BY c.analyzed_at
+LIMIT %s
+"""
+
+COUNT_FEEL_PASSED = """
+SELECT COUNT(*) FROM clips
+WHERE (record->'mechanical_check'->>'passed')::bool = true
+"""
+
 UPSERT_CLIP = """
 INSERT INTO clips (provider, provider_id, mood_bucket, record, analyzed_at)
 VALUES (%s, %s, %s, %s::jsonb, %s)
@@ -669,6 +835,7 @@ class NeonCatalogStore(CatalogStore):
         with self._connect(self._dsn) as conn:
             with conn.cursor() as cur:
                 cur.execute(CREATE_CURATION_SEED)
+                cur.execute(MIGRATE_SEED_SOURCE)
             conn.commit()
 
     def load_decided_clips(self) -> list[dict]:
@@ -685,10 +852,11 @@ class NeonCatalogStore(CatalogStore):
         return out
 
     def set_seed(self, provider, provider_id, *, verdict, reason, moods,
-                 clip_url, now):
+                 clip_url, now, source="human"):
         self.set_seed_bulk([{
             "provider": provider, "provider_id": provider_id, "verdict": verdict,
-            "reason": reason, "moods": moods, "clip_url": clip_url}], now=now)
+            "reason": reason, "moods": moods, "clip_url": clip_url,
+            "source": source}], now=now)
 
     def set_seed_bulk(self, rows: list[dict], *, now: str) -> None:
         if not rows:
@@ -699,7 +867,7 @@ class NeonCatalogStore(CatalogStore):
                     cur.execute(UPSERT_SEED, (
                         r["provider"], r["provider_id"], r["verdict"],
                         r.get("reason", ""), json.dumps(r.get("moods") or []),
-                        r.get("clip_url", ""), now))
+                        r.get("clip_url", ""), now, r.get("source", "human")))
             conn.commit()
 
     def load_seed(self) -> list[dict]:
@@ -708,10 +876,75 @@ class NeonCatalogStore(CatalogStore):
                 cur.execute(SELECT_SEED)
                 rows = cur.fetchall()
         out = []
-        for provider, provider_id, verdict, reason, moods, clip_url, at in rows:
+        for provider, provider_id, verdict, reason, moods, clip_url, at, source in rows:
             if isinstance(moods, str):
                 moods = json.loads(moods)
             out.append({"provider": provider, "provider_id": provider_id,
                         "verdict": verdict, "reason": reason,
-                        "moods": moods or [], "clip_url": clip_url, "at": at})
+                        "moods": moods or [], "clip_url": clip_url, "at": at,
+                        "source": source})
+        return out
+
+    # --- Phase-4 feel scoring ---
+
+    def ensure_feel_schema(self) -> None:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(CREATE_FEEL_SCORES)
+            conn.commit()
+
+    def load_for_feel_scoring(self, limit: int = 100) -> list[dict]:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(SELECT_FOR_FEEL, (limit,))
+                rows = cur.fetchall()
+        out = []
+        for row in rows:
+            rec = row[0]
+            if isinstance(rec, str):
+                rec = json.loads(rec)
+            out.append(rec)
         return out
+
+    def set_feel_score(self, provider, provider_id, *, result, clip_url, now) -> None:
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(UPSERT_FEEL, (
+                    provider, provider_id, result.decision, int(result.score),
+                    result.camera_motion, result.subject_motion,
+                    result.visual_complexity, result.human_presence, result.mood,
+                    result.reason, clip_url, now, result.scored_by))
+            conn.commit()
+
+    def load_feel_scores(self) -> list[dict]:
+        cols = ("provider", "provider_id", "decision", "score", "camera_motion",
+                "subject_motion", "visual_complexity", "human_presence", "mood",
+                "reason", "clip_url", "scored_at", "scored_by")
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(SELECT_FEEL)
+                rows = cur.fetchall()
+        return [dict(zip(cols, row)) for row in rows]
+
+    def feel_page(self, *, decision=None, min_score=0, limit=6, offset=0) -> list[dict]:
+        rows = self.load_feel_scores()
+        if decision and decision != "all":
+            rows = [r for r in rows if r["decision"] == decision]
+        if min_score:
+            rows = [r for r in rows if int(r["score"]) >= int(min_score)]
+        return rows[offset:offset + limit]
+
+    def feel_counts(self) -> dict:
+        c = {"KEEP": 0, "BORDERLINE": 0, "REJECT": 0, "total": 0,
+             "scored": 0, "unscored": 0}
+        for r in self.load_feel_scores():
+            if r["decision"] in c:
+                c[r["decision"]] += 1
+            c["total"] += 1
+        with self._connect(self._dsn) as conn:
+            with conn.cursor() as cur:
+                cur.execute(COUNT_FEEL_PASSED)
+                passed = int((cur.fetchone() or [0])[0])
+        c["scored"] = c["total"]
+        c["unscored"] = max(0, passed - c["total"])
+        return c

video_library/feel_score_app.py

@@ -0,0 +1,223 @@
+"""Phase-4 feel-scorer glue: pure card/status helpers, a background-thread trigger
+(mirrors seed_reason_app), and the Gradio 'Feel Score' results tab.
+
+Crash safety carried from the curation tab: video players autoplay=False, loop=True,
+page size 6, browsing repaints a small page, guarded render (no 500 on cold DB).
+"""
+from __future__ import annotations
+import threading
+import traceback
+
+from . import config
+from .feel_score_runner import run_feel_scoring
+from .__main__ import _now_iso
+
+FEEL_PAGE_SIZE = 6
+DECISION_CHOICES = ["all", "KEEP", "BORDERLINE", "REJECT"]
+
+_BADGE = {"KEEP": "✅ KEEP", "BORDERLINE": "\U0001f7e1 BORDERLINE",
+          "REJECT": "❌ REJECT"}
+
+
+def feel_caption(row: dict) -> str:
+    dec = row.get("decision", "")
+    badge = _BADGE.get(dec, dec)
+    parts = [f"**{badge} · {row.get('score', 0)}/100**"]
+    meta = []
+    if row.get("mood"):
+        meta.append(row["mood"])
+    if row.get("camera_motion"):
+        meta.append("cam: " + row["camera_motion"])
+    if row.get("subject_motion"):
+        meta.append("subj: " + row["subject_motion"])
+    if meta:
+        parts.append(" · ".join(meta))
+    if row.get("reason"):
+        parts.append("_" + row["reason"] + "_")
+    return "  \n".join(parts)
+
+
+def feel_card(row: dict) -> dict:
+    return {
+        "video_url": row.get("clip_url", ""),
+        "score": int(row.get("score", 0) or 0),
+        "decision": row.get("decision", ""),
+        "badge": _BADGE.get(row.get("decision", ""), row.get("decision", "")),
+        "caption": feel_caption(row),
+    }
+
+
+# --- background-thread trigger (mirrors seed_reason_app) ---
+
+FEEL_SCORE_STATUS = {"state": "idle", "summary": None, "error": None}
+
+
+def default_feel_components():
+    """Build the Neon store (feel schema ensured) + a FEEL_PROMPT vision router."""
+    from .catalog_store import NeonCatalogStore
+    from .vision_describe import GenAIBackend, VisionRouter
+    from .feel_scorer import FEEL_PROMPT, parse_feel_json
+    from .__main__ import _build_model_configs, _utc_date
+
+    store = NeonCatalogStore(config.NEON_DATABASE_URL)
+    store.ensure_schema()
+    store.ensure_feel_schema()
+    router = VisionRouter(
+        _build_model_configs(), store,
+        backend_factory=lambda cfg: GenAIBackend(
+            cfg.api_key or config.GEMINI_API_KEY, cfg.model_id,
+            strip_audio=cfg.strip_audio,
+            prompt=FEEL_PROMPT, parser=parse_feel_json),
+        today=_utc_date,
+    )
+    return store, router
+
+
+def run_feel_scorer(*, components_factory=default_feel_components,
+                    runner=run_feel_scoring, limit=10000):
+    if not config.NEON_DATABASE_URL:
+        FEEL_SCORE_STATUS.update(state="missing-key", error="set NEON_DATABASE_URL")
+        return FEEL_SCORE_STATUS
+    if not config.GEMINI_API_KEY:
+        FEEL_SCORE_STATUS.update(state="missing-key", error="set GEMINI_API_KEY")
+        return FEEL_SCORE_STATUS
+    FEEL_SCORE_STATUS.update(state="running", error=None, summary=None)
+    try:
+        store, router = components_factory()
+        summary = runner(store, router, limit=limit, tmp_dir=config.TMP_DIR,
+                         now=_now_iso)
+        FEEL_SCORE_STATUS.update(state="done", summary=summary)
+    except Exception as e:
+        FEEL_SCORE_STATUS.update(
+            state="error", error=repr(e) + "\n" + traceback.format_exc())
+    return FEEL_SCORE_STATUS
+
+
+def format_feel_status(status) -> str:
+    lines = ["Feel scorer: " + str(status.get("state"))]
+    summary = status.get("summary")
+    if summary:
+        lines.append(
+            "processed {processed} | scored {scored} | "
+            "skipped {skipped} | errors {errors}".format(**summary))
+        if summary.get("by_decision"):
+            lines.append("decisions: " + ", ".join(
+                f"{k} {v}" for k, v in summary["by_decision"].items()))
+        if summary.get("stopped_quota"):
+            lines.append("Stopped: Gemini quota hit; resume later.")
+    if status.get("error"):
+        lines.append("Error: " + str(status["error"]).splitlines()[0])
+    return "\n".join(lines)
+
+
+def start_feel_scorer_thread():
+    t = threading.Thread(target=run_feel_scorer, name="feel-scorer-loop", daemon=True)
+    t.start()
+    return t
+
+
+def build_feel_tab(store_factory, *, now):
+    """Construct the 'Feel Score' results tab. Imported lazily by app.py."""
+    import gradio as gr
+
+    session = {"rows": None, "decision": "all", "min_score": 0, "page": 1}
+
+    gr.Markdown(
+        "### Feel Score\nGemini watched every clip and scored how well it works as a "
+        "quiet background for narration. Sorted best-first. Filter by decision or "
+        "minimum score, then click a clip to play it.")
+
+    with gr.Row():
+        dec_dd = gr.Dropdown(DECISION_CHOICES, value="all", label="Decision")
+        min_slider = gr.Slider(0, 100, value=0, step=5, label="Min score")
+    header = gr.Markdown("")
+    page_label = gr.Markdown("")
+
+    slots = []
+    with gr.Column():
+        for _ in range(FEEL_PAGE_SIZE):
+            with gr.Row(visible=False) as row:
+                player = gr.Video(interactive=False, height=260, show_label=False,
+                                  autoplay=False, loop=True)
+                caption = gr.Markdown("")
+            slots.append({"row": row, "player": player, "caption": caption})
+
+    with gr.Row():
+        prev_btn = gr.Button("← Prev")
+        next_btn = gr.Button("Next →")
+
+    def _load_rows():
+        if session["rows"] is None:
+            session["rows"] = store_factory().load_feel_scores()
+        return session["rows"]
+
+    def _filtered():
+        rows = _load_rows()
+        dec, ms = session["decision"], session["min_score"]
+        if dec and dec != "all":
+            rows = [r for r in rows if r["decision"] == dec]
+        if ms:
+            rows = [r for r in rows if int(r["score"]) >= int(ms)]
+        return rows
+
+    def _outputs():
+        out = [header, page_label]
+        for slot in slots:
+            out += [slot["row"], slot["player"], slot["caption"]]
+        return out
+
+    def _render():
+        view = _filtered()
+        pages = max(1, (len(view) + FEEL_PAGE_SIZE - 1) // FEEL_PAGE_SIZE)
+        session["page"] = max(1, min(session["page"], pages))
+        off = (session["page"] - 1) * FEEL_PAGE_SIZE
+        clips = view[off:off + FEEL_PAGE_SIZE]
+        hdr = (f"{len(view)} clips shown · {len(_load_rows())} scored total")
+        updates = [gr.update(value=hdr),
+                   gr.update(value=f"Page {session['page']} / {pages}")]
+        for i, slot in enumerate(slots):
+            if i < len(clips):
+                card = feel_card(clips[i])
+                updates += [gr.update(visible=True),
+                            gr.update(value=card["video_url"]),
+                            gr.update(value=card["caption"])]
+            else:
+                updates += [gr.update(visible=False), gr.update(value=None),
+                            gr.update(value="")]
+        return updates
+
+    def _error_render(exc):
+        updates = [gr.update(value=f"⚠️ Could not load scores: {exc}"),
+                   gr.update(value="")]
+        for _ in slots:
+            updates += [gr.update(visible=False), gr.update(value=None),
+                        gr.update(value="")]
+        return updates
+
+    def _on_filter(dec, ms):
+        session["decision"], session["min_score"], session["page"] = dec, ms, 1
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    def _on_page(delta):
+        session["page"] = max(1, session["page"] + delta)
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    def _on_init():
+        try:
+            return _render()
+        except Exception as exc:
+            return _error_render(exc)
+
+    full_out = _outputs()
+    dec_dd.change(_on_filter, [dec_dd, min_slider], full_out)
+    min_slider.release(_on_filter, [dec_dd, min_slider], full_out)
+    prev_btn.click(lambda: _on_page(-1), None, full_out)
+    next_btn.click(lambda: _on_page(1), None, full_out)
+
+    return {"init_fn": _on_init, "init_inputs": None, "init_outputs": _outputs()}

video_library/feel_score_runner.py

@@ -0,0 +1,88 @@
+"""Phase-4 orchestrator: score every mechanically-passed clip's FEEL.
+
+Loop mirrors tag_runner / library_builder:
+  1. Load a batch of mechanically-passed, not-yet-scored clips.
+  2. Pick the smallest streamable file, download to temp.
+  3. router.describe(path) -> FeelResult (router carries FEEL_PROMPT).
+  4. store.set_feel_score(...); delete temp.
+  5. Sleep rpm_seconds.
+Resumable (each row written immediately) and quota-aware (NoModelAvailableError or
+a detected quota error stops the run cleanly; rerun to continue).
+"""
+from __future__ import annotations
+import datetime as _dt
+import time
+from typing import Callable
+
+from .downloader import download_to_temp, delete_quietly
+from .curation_app import pick_stream_url
+from .vision_describe import NoModelAvailableError, is_quota_error
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def run_feel_scoring(store, router, *,
+                     limit: int = 10_000,
+                     batch_size: int = 100,
+                     tmp_dir,
+                     now: Callable[[], str] = _now_iso,
+                     sleep: Callable[[float], None] = time.sleep,
+                     rpm_seconds: float = 2.5,
+                     downloader=download_to_temp,
+                     deleter=delete_quietly,
+                     url_picker=pick_stream_url,
+                     quota_detector: Callable[[Exception], bool] = is_quota_error,
+                     ) -> dict:
+    summary = {"processed": 0, "scored": 0, "skipped": 0, "errors": 0,
+               "stopped_quota": False, "by_decision": {}}
+    errored: set[tuple[str, str]] = set()
+    remaining = limit
+    while remaining > 0:
+        batch = store.load_for_feel_scoring(limit=min(batch_size, remaining))
+        batch = [r for r in batch
+                 if (r.get("provider"), r.get("provider_id")) not in errored]
+        if not batch:
+            break
+        for rec in batch:
+            if remaining <= 0:
+                break
+            provider = rec.get("provider")
+            provider_id = rec.get("provider_id")
+            summary["processed"] += 1
+            remaining -= 1
+
+            url = url_picker(rec)
+            if not url:
+                summary["skipped"] += 1
+                continue
+
+            temp_path = None
+            try:
+                temp_path = downloader(url, tmp_dir, provider_id)
+                result = router.describe(str(temp_path))
+            except NoModelAvailableError:
+                summary["stopped_quota"] = True
+                if temp_path is not None:
+                    deleter(temp_path)
+                return summary
+            except Exception as e:
+                if temp_path is not None:
+                    deleter(temp_path)
+                if quota_detector(e):
+                    summary["stopped_quota"] = True
+                    return summary
+                summary["errors"] += 1
+                errored.add((provider, provider_id))
+                continue
+            else:
+                deleter(temp_path)
+
+            store.set_feel_score(provider, provider_id, result=result,
+                                 clip_url=url, now=now())
+            summary["scored"] += 1
+            d = result.decision
+            summary["by_decision"][d] = summary["by_decision"].get(d, 0) + 1
+            sleep(rpm_seconds)
+    return summary

video_library/feel_scorer.py

@@ -0,0 +1,413 @@
+"""Phase-4 vision judge: watch a clip, score its FEEL for contemplative narration.
+
+Unlike the Phase-1 VISION_PROMPT (which forbids feel-judgment on purpose), this
+prompt asks the model to decide whether the clip can sit quietly *under* Osho-style
+narration without competing for attention. Output is strict JSON we parse into a
+FeelResult. The taste lives entirely in FEEL_PROMPT - no few-shot examples.
+"""
+from __future__ import annotations
+import json
+import re
+from dataclasses import dataclass
+
+FEEL_DECISIONS = ("KEEP", "BORDERLINE", "REJECT")
+
+FEEL_PROMPT = """# Role
+
+You are an expert curator of stock footage for a YouTube Shorts channel featuring slow, contemplative Osho narration.
+
+Your job is **not** to judge whether a clip is beautiful. Your job is to judge whether it quietly supports spoken wisdom without competing for the viewer's attention.
+
+The narration is always the hero.
+
+The footage exists only to create atmosphere.
+
+A perfect clip should almost disappear behind the words while gently deepening their emotional impact.
+
+---
+
+# Core Principle
+
+**Ask yourself one question above all else:**
+
+> **Would this clip make the viewer watch the visuals instead of listening to the narration?**
+
+If **yes**, reject it.
+
+If the visuals simply create a peaceful atmosphere while allowing the narration to remain the focus, keep it.
+
+---
+
+# The Most Important Distinction
+
+Evaluate **camera movement** and **subject movement** separately.
+
+### Good Camera Movement
+
+The camera may move if it slowly reveals the scene.
+
+Examples:
+
+* slow drone glide
+* slow dolly
+* gentle pan
+* slow tilt
+* slow push-in
+* smooth floating movement
+
+This creates immersion without distraction.
+
+### Bad Camera Movement
+
+Reject:
+
+* handheld shake
+* whip pans
+* rapid zooms
+* aggressive orbiting
+* jerky stabilization
+* fast tracking
+
+---
+
+### Good Subject Movement
+
+Minimal natural movement is ideal.
+
+Examples:
+
+* clouds drifting
+* mist moving
+* gentle waves
+* leaves lightly swaying
+* rain
+* smoke
+* calm flowing river
+
+The subject should feel almost still.
+
+### Bad Subject Movement
+
+Reject if the subject itself becomes the attraction.
+
+Examples:
+
+* people walking prominently
+* animals running
+* birds dominating the frame
+* vehicles
+* traffic
+* crowds
+* dancing
+* sports
+* waterfalls with violent motion
+* fireworks
+* explosions
+* anything energetic
+
+**Camera motion is good. Subject motion is usually bad.**
+
+---
+
+# Desired Visual Style
+
+Prefer clips with:
+
+* open landscapes
+* mountains
+* forests
+* rivers
+* lakes
+* ocean horizons
+* beaches
+* deserts
+* snow
+* fog
+* mist
+* clouds
+* sunrise
+* sunset
+* moonlight
+* empty temples
+* quiet interiors
+* empty roads
+* abandoned structures
+* fields
+* minimal architecture
+* cinematic nature
+
+The frame should evoke:
+
+* silence
+* spaciousness
+* solitude
+* timelessness
+* contemplation
+* stillness
+* peace
+* wonder
+* presence
+
+---
+
+# Visual Simplicity
+
+The composition should be clean and easy to absorb.
+
+Prefer:
+
+* few dominant elements
+* negative space
+* uncluttered backgrounds
+* natural lighting
+* cinematic framing
+* soft colors
+* slow visual rhythm
+
+Reject:
+
+* clutter
+* visual chaos
+* many competing objects
+* busy forests
+* dense city scenes
+* heavy visual texture
+* flashing lights
+* excessive detail that constantly pulls attention
+
+---
+
+# Human Presence
+
+People should rarely be the subject.
+
+KEEP:
+
+* tiny distant silhouette
+* lone person occupying very little of the frame
+* person integrated into the landscape
+* person remaining still
+
+REJECT:
+
+* close-up faces
+* portraits
+* talking people
+* emotional acting
+* influencers
+* vlog footage
+* people performing actions
+* crowds
+* dancing
+* weddings
+* celebrations
+
+If a viewer immediately notices the person before the environment, reject.
+
+---
+
+# Commercial or Stock Feel
+
+Reject clips that resemble:
+
+* advertisements
+* travel influencer footage
+* luxury lifestyle
+* product showcases
+* fashion videos
+* cinematic hero shots
+* corporate footage
+* generic motivational stock
+* obvious AI-generated visuals
+* overprocessed HDR
+* unrealistic colors
+
+The footage should feel authentic rather than promotional.
+
+---
+
+# Technical Rejections
+
+Automatically reject clips containing:
+
+* text
+* subtitles
+* logos
+* watermarks
+* UI elements
+* borders
+* split screens
+* multiple scenes
+* fast cuts
+* jump cuts
+* transitions
+* glitches
+* poor stabilization
+* distracting compression artifacts
+
+One uninterrupted shot is strongly preferred.
+
+---
+
+# Emotional Tone
+
+Desired emotions:
+
+* peaceful
+* reflective
+* meditative
+* calm
+* expansive
+* spiritual
+* quiet
+* gentle
+* grounded
+
+Avoid clips that feel:
+
+* exciting
+* urgent
+* dramatic
+* suspenseful
+* comedic
+* romantic
+* aggressive
+* adventurous
+* action-oriented
+* emotionally manipulative
+
+---
+
+# Scoring Rubric
+
+Score each clip out of 100.
+
+**Subject Motion (35 points)**
+
+* Minimal movement = high score
+* Active subjects = low score
+
+**Camera Movement (20 points)**
+
+* Smooth slow movement = high score
+* Fast or shaky movement = low score
+
+**Visual Simplicity (20 points)**
+
+* Clean composition = high score
+* Busy composition = low score
+
+**Mood & Atmosphere (15 points)**
+
+* Peaceful and contemplative = high score
+* Exciting or dramatic = low score
+
+**Human Presence (10 points)**
+
+* None or tiny silhouette = high score
+* Human as subject = low score
+
+---
+
+# Decision Rules
+
+**KEEP**
+
+* Score **80-100**
+* Strong contemplative background
+* Supports narration effortlessly
+
+**BORDERLINE**
+
+* Score **60-79**
+* Usable but contains mild distractions
+
+**REJECT**
+
+* Score **below 60**
+* Competes with narration or breaks the contemplative atmosphere
+
+---
+
+# Output Format
+
+Return only valid JSON.
+
+{
+  "decision": "KEEP | BORDERLINE | REJECT",
+  "score": 0,
+  "camera_motion": "",
+  "subject_motion": "",
+  "visual_complexity": "low | medium | high",
+  "human_presence": "none | distant | prominent",
+  "mood": "",
+  "reason": ""
+}
+
+Keep the "reason" concise (1-2 sentences), focusing on why the clip does or does not work as a quiet background for contemplative narration.
+
+Never judge the philosophical content. Judge only the visual suitability for an Osho-style spoken wisdom video."""
+
+
+class FeelParseError(Exception):
+    pass
+
+
+@dataclass
+class FeelResult:
+    decision: str
+    score: int
+    camera_motion: str = ""
+    subject_motion: str = ""
+    visual_complexity: str = ""
+    human_presence: str = ""
+    mood: str = ""
+    reason: str = ""
+    described_by: str = ""
+
+    @property
+    def scored_by(self) -> str:
+        return self.described_by
+
+
+def _coerce_score(v) -> int:
+    try:
+        n = int(float(v))
+    except (TypeError, ValueError):
+        raise FeelParseError(f"bad score: {v!r}")
+    return max(0, min(100, n))
+
+
+def _from_obj(obj: dict) -> FeelResult:
+    decision = str(obj.get("decision", "")).strip().upper()
+    if decision not in FEEL_DECISIONS:
+        raise FeelParseError(f"bad decision: {decision!r}")
+    return FeelResult(
+        decision=decision,
+        score=_coerce_score(obj.get("score", 0)),
+        camera_motion=str(obj.get("camera_motion", "") or ""),
+        subject_motion=str(obj.get("subject_motion", "") or ""),
+        visual_complexity=str(obj.get("visual_complexity", "") or ""),
+        human_presence=str(obj.get("human_presence", "") or ""),
+        mood=str(obj.get("mood", "") or ""),
+        reason=str(obj.get("reason", "") or ""),
+    )
+
+
+def parse_feel_json(text: str) -> FeelResult:
+    """Extract the first valid JSON object from model text -> FeelResult."""
+    if not text:
+        raise FeelParseError("empty response")
+    candidates = []
+    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+    if fence:
+        candidates.append(fence.group(1))
+    brace = re.search(r"\{.*\}", text, re.DOTALL)
+    if brace:
+        candidates.append(brace.group(0))
+    for c in candidates:
+        try:
+            return _from_obj(json.loads(c))
+        except (json.JSONDecodeError, TypeError):
+            continue
+    raise FeelParseError("no valid JSON object in response")

video_library/seed_app.py

@@ -22,12 +22,25 @@ SEED_VERDICTS = ("kept", "rejected")
 SEED_PAGE_SIZE = 6
 
 
-def _seed_caption(card: dict) -> str:
-    """Card caption showing subject/quality context (not editable, just reference)."""
-    return (f"**{', '.join(card['moods']) or '—'}** · q{card['quality']} · "
+def _seed_caption(card: dict, status: str = "") -> str:
+    """Card caption showing subject/quality context (not editable, just reference).
+
+    `status` is an optional marker line for the reason's state: an AI draft awaiting
+    review, or a human-confirmed reason."""
+    head = f"{status}\n\n" if status else ""
+    return (f"{head}**{', '.join(card['moods']) or '—'}** · q{card['quality']} · "
             f"{card['motion'] or '—'}\n\n{card['description']}")
 
 
+def _reason_status(saved_row: dict | None) -> str:
+    """Marker for a card's reason state, from its saved seed row (or None)."""
+    if not saved_row or not (saved_row.get("reason") or "").strip():
+        return ""
+    if (saved_row.get("source") or "human") == "ai":
+        return "🤖 *AI draft — glance & edit to confirm*"
+    return "✅ *confirmed*"
+
+
 class SeedSnapshot:
     """The decided-clip set (kept/rejected) loaded into RAM once.
 
@@ -114,10 +127,18 @@ def build_seed_tab(store_factory, *, now):
         snap = _ensure_snapshot()
         total = snap.count()
         # A reason is "done" only if it's non-empty; a bare verdict was already set
-        # in Phase 3, so it doesn't count as newly annotated here.
-        done = sum(1 for r in session["saved"].values() if (r.get("reason") or "").strip())
-        return (f"**{done} / {total}** clips have a written reason  ·  "
-                f"edits save automatically when you click Next / Prev")
+        # in Phase 3, so it doesn't count as newly annotated here. Split into
+        # human-confirmed vs AI drafts still awaiting your glance.
+        confirmed = drafts = 0
+        for r in session["saved"].values():
+            if not (r.get("reason") or "").strip():
+                continue
+            if (r.get("source") or "human") == "ai":
+                drafts += 1
+            else:
+                confirmed += 1
+        return (f"**{confirmed} confirmed** · 🤖 **{drafts} AI drafts to review** · "
+                f"{total} total  ·  edits save when you click Next / Prev")
 
     def _flush():
         """Write buffered seed rows to Neon in one batch, then clear the buffer.
@@ -167,9 +188,11 @@ def build_seed_tab(store_factory, *, now):
             if i < len(page_records):
                 card = card_for_record(page_records[i])
                 v, r = _buffered(card)
+                key = (card["provider"], card["provider_id"])
+                status = _reason_status(session.get("saved", {}).get(key))
                 updates += [gr.update(visible=True),
                             gr.update(value=card["video_url"]),
-                            gr.update(value=_seed_caption(card)),
+                            gr.update(value=_seed_caption(card, status)),
                             gr.update(value=v),
                             gr.update(value=r)]
             else:
@@ -214,10 +237,12 @@ def build_seed_tab(store_factory, *, now):
                     and (prev.get("reason") or "") == reason_val:
                 continue  # already saved with these exact values — nothing to do
 
+            # A human touched this card, so it's a confirmed reason now — even if
+            # the text came from an AI draft they left as-is by editing the verdict.
             session["pending"][key] = {
                 "provider": p, "provider_id": pid, "verdict": verdict_val,
                 "reason": reason_val, "moods": card["moods"],
-                "clip_url": card["video_url"]}
+                "clip_url": card["video_url"], "source": "human"}
 
     def _on_page(delta, *vals):
         try:

video_library/seed_reason_app.py

@@ -0,0 +1,72 @@
+"""Phase-3.5 seed-reasoner glue for the Gradio Space + CLI.
+
+Mirrors tag_app.py: builds the Neon store + a SeedReasoner from config, runs one
+batch pass in a background thread, and exposes a compact status string. Manual
+(button / CLI flag), never auto-started, so it doesn't compete with Phase-1/2 for
+the Mistral rate limit.
+"""
+from __future__ import annotations
+import threading
+import traceback
+
+from . import config
+from .seed_reason_runner import run_seed_reasoning
+from .__main__ import _now_iso
+
+
+SEED_REASON_STATUS = {"state": "idle", "summary": None, "error": None}
+
+
+def default_seed_reason_components():
+    """Build the Neon store (seed schema ensured) + a Mistral SeedReasoner."""
+    from .catalog_store import NeonCatalogStore
+    from .seed_reasoner import SeedReasoner
+
+    store = NeonCatalogStore(config.NEON_DATABASE_URL)
+    store.ensure_schema()
+    store.ensure_seed_schema()
+    reasoner = SeedReasoner(config.MISTRAL_API_KEY, config.MISTRAL_MODEL)
+    return store, reasoner
+
+
+def run_seed_reasoner(*, components_factory=default_seed_reason_components,
+                      runner=run_seed_reasoning, limit=10000, overwrite_ai=False):
+    """Run one full pass of the seed reasoner. Updates SEED_REASON_STATUS as it goes."""
+    if not config.MISTRAL_API_KEY:
+        SEED_REASON_STATUS.update(state="missing-key", error="set MISTRAL_API_KEY")
+        return SEED_REASON_STATUS
+    if not config.NEON_DATABASE_URL:
+        SEED_REASON_STATUS.update(state="missing-key", error="set NEON_DATABASE_URL")
+        return SEED_REASON_STATUS
+    SEED_REASON_STATUS.update(state="running", error=None, summary=None)
+    try:
+        store, reasoner = components_factory()
+        summary = runner(store, reasoner, limit=limit, now=_now_iso,
+                         overwrite_ai=overwrite_ai)
+        SEED_REASON_STATUS.update(state="done", summary=summary)
+    except Exception as e:
+        SEED_REASON_STATUS.update(
+            state="error", error=repr(e) + "\n" + traceback.format_exc())
+    return SEED_REASON_STATUS
+
+
+def format_seed_reason_status(status):
+    """Render SEED_REASON_STATUS as a compact human-readable string."""
+    lines = ["Seed reasoner: " + str(status.get("state"))]
+    summary = status.get("summary")
+    if summary:
+        lines.append(
+            "Processed {processed} | drafted {drafted} | "
+            "skipped {skipped} | errors {errors}".format(**summary))
+        if summary.get("stopped_quota"):
+            lines.append("Stopped: Mistral quota hit; resume later.")
+    if status.get("error"):
+        lines.append("Error: " + str(status["error"]).splitlines()[0])
+    return "\n".join(lines)
+
+
+def start_seed_reasoner_thread(*, overwrite_ai=False):
+    t = threading.Thread(target=lambda: run_seed_reasoner(overwrite_ai=overwrite_ai),
+                         name="seed-reasoner-loop", daemon=True)
+    t.start()
+    return t

video_library/seed_reason_runner.py

@@ -0,0 +1,114 @@
+"""Phase-3.5 orchestrator: draft a calibration reason for every decided clip.
+
+For each clip the human already kept/rejected (Phase 3), ask a SeedReasoner to draft a
+one-line principle-level reason, then upsert it into the curation_seed table as an AI
+DRAFT (source='ai'). The human later eyeballs drafts in the seed tab and, on any edit,
+the row is re-saved as source='human' (confirmed).
+
+Design mirrors tag_runner.py:
+  * Resumable — every clip's reason is written immediately, so killing the process
+    loses at most one in-flight draft.
+  * Quota-aware — a 429 stops the run cleanly; rerun later to continue.
+  * Idempotent by default — clips that already have a reason are skipped, so a rerun
+    only fills the gaps. Pass overwrite_ai=True to redraft existing AI drafts (but never
+    clobber a human-confirmed reason).
+
+The human's own rough note (if any already saved) is fed to the reasoner as signal.
+"""
+from __future__ import annotations
+
+import datetime as _dt
+import time
+from typing import Callable
+
+from .seed_reasoner import is_quota_error
+from .curation_app import pick_stream_url
+
+
+def _now_iso() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _clip_key(rec: dict) -> tuple[str, str]:
+    # Stored records are the flattened to_dict() shape: provider/provider_id are
+    # top-level, NOT nested under `candidate`.
+    return (rec.get("provider", ""), rec.get("provider_id", ""))
+
+
+def _verdict_of(rec: dict) -> str:
+    return (rec.get("curation") or {}).get("verdict", "")
+
+
+def run_seed_reasoning(store, reasoner, *,
+                       limit: int = 10_000,
+                       now: Callable[[], str] = _now_iso,
+                       sleep: Callable[[float], None] = time.sleep,
+                       rpm_seconds: float = 2.5,
+                       overwrite_ai: bool = False,
+                       quota_detector: Callable[[Exception], bool] = is_quota_error,
+                       ) -> dict:
+    """Draft reasons for up to `limit` decided clips. Returns a summary dict.
+
+    Summary keys:
+      processed     - decided clips inspected this run
+      drafted       - reasons written this run
+      skipped       - clips left as-is (already had a reason we won't touch)
+      errors        - non-quota failures; clip left for a later retry
+      stopped_quota - True if a quota error halted the run
+    """
+    summary = {"processed": 0, "drafted": 0, "skipped": 0, "errors": 0,
+               "stopped_quota": False}
+
+    # What's already in curation_seed, so we can skip / respect human reasons.
+    existing = {(r["provider"], r["provider_id"]): r for r in store.load_seed()}
+
+    decided = store.load_decided_clips()
+    remaining = limit
+    for rec in decided:
+        if remaining <= 0:
+            break
+        provider, provider_id = _clip_key(rec)
+        if not provider_id:
+            continue
+        verdict = _verdict_of(rec)
+        if verdict not in ("kept", "rejected"):
+            continue
+
+        summary["processed"] += 1
+        remaining -= 1
+
+        prior = existing.get((provider, provider_id))
+        note = ""
+        if prior:
+            has_reason = bool((prior.get("reason") or "").strip())
+            source = prior.get("source") or "human"
+            # Never overwrite a human-confirmed reason. Skip existing AI drafts
+            # unless explicitly asked to redraft. A prior human note with no verdict
+            # change is still useful signal for (re)drafting.
+            if has_reason and source == "human":
+                summary["skipped"] += 1
+                continue
+            if has_reason and source == "ai" and not overwrite_ai:
+                summary["skipped"] += 1
+                continue
+            note = prior.get("reason") or ""
+
+        description = rec.get("description") or {}
+        moods = rec.get("moods") or []
+        clip_url = pick_stream_url(rec)
+
+        try:
+            reason = reasoner.draft(description, verdict, note)
+        except Exception as e:
+            if quota_detector(e):
+                summary["stopped_quota"] = True
+                return summary
+            summary["errors"] += 1
+            continue
+
+        store.set_seed(provider, provider_id, verdict=verdict, reason=reason,
+                       moods=moods, clip_url=clip_url, now=now(), source="ai")
+        summary["drafted"] += 1
+        sleep(rpm_seconds)
+
+    return summary

video_library/seed_reasoner.py

@@ -0,0 +1,188 @@
+"""Phase-3.5 seed reasoner: draft a one-line calibration reason for a decided clip.
+
+The human already gave the ground-truth signal in Phase 3 (kept / rejected). This
+module turns that verdict — plus the clip's neutral description and any rough note the
+human typed — into a clean, PRINCIPLE-LEVEL reason that a later AI visual scorer can
+learn from without over-fixating on literal details.
+
+Taste (the Osho / spiritual channel calibration) is encoded in REASONER_PROMPT:
+  WANT  natural/nature footage, solitude/stillness/aesthetic calm, CAMERA movement that
+        reveals beauty, clean low-clutter composition.
+  AVOID in-frame subject motion & clutter, product/studio look, text/logos/faces, generic.
+  KEY   camera movement = good; in-frame subject motion & clutter = bad (the subtle one).
+
+We reuse the same plain-HTTP Mistral plumbing as mistral_backend.py (no SDK, injectable
+session so tests never hit the network). The reason text comes back as strict JSON so we
+can parse it deterministically.
+"""
+from __future__ import annotations
+
+import json
+import re
+import time
+
+MISTRAL_URL = "https://api.mistral.ai/v1/chat/completions"
+
+# The description fields the reasoner is allowed to see. We deliberately hand it the
+# neutral facts only — never the mood tags or search query — so the reason reflects the
+# clip itself, matching how the future scorer will work.
+_DESC_FIELDS = ("summary", "subjects", "environment", "lighting", "color_palette",
+                "motion", "camera", "people", "faces", "text_or_logo", "watermark",
+                "visual_quality", "raw_caption")
+
+REASONER_PROMPT = """You calibrate a taste model for a spiritual (Osho) short-form video \
+channel. The channel pairs slow, contemplative narration with a single beautiful moving \
+image. You are given ONE clip's neutral description JSON and the human's verdict (kept or \
+rejected). A rough human note may also appear — it is weak background signal only.
+
+YOUR JOB: write ONE short reason (max ~14 words) that states, as a GENERAL PRINCIPLE, why \
+a clip like this belongs (or doesn't) on the channel. The verdict is ground truth; explain \
+the quality behind it — never argue with it.
+
+THE TASTE (this, not the note, is the authority):
+KEEP a clip when it delivers a single calm, unhurried image that lets narration breathe — \
+open natural scenes (landscape, water, sky, forest, desert, still interiors), a feeling of \
+solitude and spaciousness, and CAMERA motion that slowly unveils the scene (a drift, glide, \
+drone, or slow pan). Clean composition, few elements, natural light.
+REJECT a clip that fights the narration — busy in-frame subject motion, clutter or visual \
+chaos, fast cuts; a product-shot / studio / commercial look; on-screen text, logos, or \
+watermarks; prominent faces or people as the subject; anything generic, stocky, or \
+algorithmic.
+THE SUBTLE DISTINCTION most reasons hinge on: CAMERA movement is good (it reveals calm). \
+SUBJECT movement inside the frame is bad (it distracts). Judge those two separately — a \
+still subject under a gliding camera is ideal; a static camera on a fidgeting subject is not.
+
+HOW TO PHRASE THE REASON:
+- Name the QUALITY, never the literal contents. The scorer must generalize, so a reason \
+tied to specific objects teaches it the wrong lesson.
+- WRONG (too literal): "a bird flies over a lake at sunset"
+  RIGHT (principle):    "calm open nature, camera glides to reveal it, uncluttered"
+- WRONG (too literal): "person in a red jacket walks through a busy market"
+  RIGHT (principle):    "busy subject motion and clutter compete with the narration"
+- If a human note is literal, distill it to the underlying quality; do not repeat its words.
+
+Return STRICT JSON and nothing else:
+{"reason": "one short principle-level line"}
+No markdown, no commentary outside the JSON."""
+
+
+class SeedReasonParseError(Exception):
+    pass
+
+
+class MistralQuotaError(Exception):
+    """Raised on an HTTP 429 so a runner can stop/fall through cleanly."""
+
+
+def is_quota_error(exc: Exception) -> bool:
+    """True if an exception looks like a 429 / rate / quota from Mistral."""
+    s = str(exc).upper()
+    return ("429" in s) or ("RATE" in s and "LIMIT" in s) or ("QUOTA" in s) \
+        or ("RESOURCE_EXHAUSTED" in s) or ("TOO MANY REQUESTS" in s)
+
+
+def _slim_description(description: dict) -> dict:
+    """Keep only the neutral fields; drop provenance and anything not decision-relevant."""
+    return {k: description[k] for k in _DESC_FIELDS if k in description}
+
+
+def _user_message(description: dict, verdict: str, note: str) -> str:
+    payload = {
+        "verdict": verdict,
+        "description": _slim_description(description or {}),
+        # Weak background signal only — a rough human aside, often literal. The
+        # reason must reflect the taste spec and the description, not echo this.
+        "human_note_weak_signal": (note or "").strip(),
+    }
+    return ("Draft the principle-level reason for this clip. The verdict is ground "
+            "truth. Data follows:\n" + json.dumps(payload, indent=2))
+
+
+def parse_reason_json(text: str) -> str:
+    """Extract the first valid JSON object from model text -> the reason string."""
+    if not text:
+        raise SeedReasonParseError("empty response")
+    candidates: list[str] = []
+    fence = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", text, re.DOTALL)
+    if fence:
+        candidates.append(fence.group(1))
+    brace = re.search(r"\{.*\}", text, re.DOTALL)
+    if brace:
+        candidates.append(brace.group(0))
+    for c in candidates:
+        try:
+            obj = json.loads(c)
+        except json.JSONDecodeError:
+            continue
+        if not isinstance(obj, dict):
+            continue
+        reason = obj.get("reason")
+        if isinstance(reason, str) and reason.strip():
+            return reason.strip()
+    raise SeedReasonParseError("no valid reason in response")
+
+
+class SeedReasoner:
+    """One call: (description, verdict, note) -> a principle-level reason string."""
+
+    def __init__(self, api_key, model="mistral-medium-2505", *,
+                 session=None, sleep=time.sleep, max_retries=2, timeout=60):
+        self._key = api_key
+        self._model = model
+        self._session = session
+        self._sleep = sleep
+        self._max_retries = max_retries
+        self._timeout = timeout
+
+    def _ensure_session(self):
+        if self._session is None:
+            import requests
+            self._session = requests.Session()
+        return self._session
+
+    def _post(self, body):
+        sess = self._ensure_session()
+        headers = {"Authorization": "Bearer " + self._key,
+                   "Content-Type": "application/json"}
+        resp = sess.post(MISTRAL_URL, headers=headers, json=body, timeout=self._timeout)
+        status = getattr(resp, "status_code", 200)
+        if status == 429:
+            raise MistralQuotaError("429 Too Many Requests")
+        if status >= 400:
+            text = getattr(resp, "text", "")
+            raise RuntimeError("Mistral HTTP " + str(status) + ": " + str(text)[:300])
+        return resp.json()
+
+    @staticmethod
+    def _text_of(data):
+        try:
+            return data["choices"][0]["message"]["content"] or ""
+        except (KeyError, IndexError, TypeError):
+            return ""
+
+    def draft(self, description: dict, verdict: str, note: str = "") -> str:
+        body = {
+            "model": self._model,
+            "messages": [
+                {"role": "system", "content": REASONER_PROMPT},
+                {"role": "user", "content": _user_message(description, verdict, note)},
+            ],
+            "response_format": {"type": "json_object"},
+        }
+        attempt = 0
+        last_exc = None
+        while attempt <= self._max_retries:
+            try:
+                data = self._post(body)
+                return parse_reason_json(self._text_of(data))
+            except MistralQuotaError:
+                raise
+            except Exception as e:
+                if is_quota_error(e):
+                    raise
+                last_exc = e
+                attempt += 1
+                if attempt > self._max_retries:
+                    break
+                self._sleep(2 ** attempt)
+        raise last_exc if last_exc else RuntimeError("SeedReasoner: exhausted retries")

video_library/vision_describe.py

@@ -105,7 +105,8 @@ class GenAIBackend(VisionBackend):
     def __init__(self, api_key: str, model: str, *, strip_audio: bool = False,
                  client=None, audio_stripper=strip_audio_track,
                  sleep=time.sleep, poll_interval: float = 5.0, max_poll: int = 60,
-                 gen_config=None):
+                 gen_config=None, prompt: str = VISION_PROMPT,
+                 parser=parse_vision_json):
         self._key = api_key
         self._model = model
         self._strip_audio = strip_audio
@@ -115,6 +116,11 @@ class GenAIBackend(VisionBackend):
         self._poll_interval = poll_interval
         self._max_poll = max_poll
         self._gen_config = gen_config
+        # Prompt + parser are injectable so the same backend/router can run the
+        # Phase-4 feel judge (FEEL_PROMPT + parse_feel_json). Defaults preserve
+        # the neutral Phase-1 behavior for existing callers.
+        self._prompt = prompt
+        self._parser = parser
 
     def _ensure_client(self):
         if self._client is None:
@@ -164,10 +170,10 @@ class GenAIBackend(VisionBackend):
             uploaded = self._wait_active(client, uploaded)
             resp = client.models.generate_content(
                 model=self._model,
-                contents=[uploaded, VISION_PROMPT],
+                contents=[uploaded, self._prompt],
                 config=self._gen_config_obj(),
             )
-            desc = parse_vision_json(self._text_of(resp))
+            desc = self._parser(self._text_of(resp))
             desc.described_by = self._model
             return desc
         finally: