Auto-extract: generate the LLM step outline first, anchor frames to step times

Auto-extract now runs guide generation (after transcription) when no draft exists,
then extracts frames AT the LLM step timestamps — the same suggestions the per-step
weighted selection relies on. Falls back to narration-gated scenes when there is no
token/draft. The generated transcript + outline are surfaced in the UI (reused, not
recomputed). Smoke test reordered to generate before extracting.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app.py

@@ -171,39 +171,73 @@ def on_capture(session: str, frames: list[dict], data_url: str, current_time: fl
     )
 
 
-def on_auto(session: str, frames: list[dict], video_path: str, transcript_obj, progress=gr.Progress()):
-    # Outputs: gallery, frames_state, transcript_box, transcript_state, status
+def on_auto(
+    session: str,
+    frames: list[dict],
+    video_path: str,
+    transcript_obj,
+    draft_obj,
+    hf_token: str,
+    progress=gr.Progress(),
+):
+    # Outputs: gallery, frames_state, transcript_box, transcript_state, guide_md, draft_state, status
     if not video_path:
-        return _gallery_value(frames), frames, gr.update(), gr.update(), "Upload a video first."
+        return (_gallery_value(frames), frames, gr.update(), transcript_obj,
+                gr.update(), draft_obj, "Upload a video first.")
+
+    token = config.apply_token(hf_token)
+    notes: list[str] = []
 
-    # Transcribe first if we don't have a transcript yet, so frames can be gated
-    # to the narration (otherwise recorder intro/idle screens get captured).
-    auto_transcribed = False
+    # 1) Transcript — needed to anchor/gate frames to the narration.
+    auto_tr = False
     if not (transcript_obj and getattr(transcript_obj, "segments", None)):
-        progress(0.0, "No transcript yet — transcribing first…")
+        progress(0.0, "Transcribing first…")
         transcript_obj = _run_transcription(session, video_path, progress)
-        auto_transcribed = True
+        auto_tr = True
+
+    # 2) LLM step outline — so frames anchor to the actual guide steps (the same
+    #    LLM timestamps the per-step selection weights heavily).
+    auto_draft = False
+    if not (draft_obj and getattr(draft_obj, "steps", None)):
+        if token:
+            progress(0.5, "Generating step outline (LLM)…")
+            new_draft, msg = _generate_draft(transcript_obj, token, progress)
+            if new_draft:
+                draft_obj, auto_draft = new_draft, True
+            else:
+                notes.append(msg)
+        else:
+            notes.append("add your HF token for step-aligned frames")
 
+    # 3) Extract — at step timestamps when available, else narration-gated scenes.
     spoken = (
         [(s.start, s.end) for s in transcript_obj.segments]
-        if transcript_obj and transcript_obj.segments
-        else None
+        if transcript_obj and transcript_obj.segments else None
+    )
+    steps_ts = (
+        [s.approx_timestamp for s in draft_obj.steps if s.approx_timestamp is not None]
+        if draft_obj and getattr(draft_obj, "steps", None) else None
+    )
+    progress(0.9, "Extracting frames…")
+    recs = extract_auto_frames(
+        video_path, config.session_dir(session),
+        spoken_intervals=spoken, step_timestamps=steps_ts,
     )
-    progress(0.85, "Detecting scenes…")
-    recs = extract_auto_frames(video_path, config.session_dir(session), spoken_intervals=spoken)
     merged = frames + [asdict(r) for r in recs]
     progress(1.0, "Done.")
 
-    # If we just transcribed, surface it in the box + state so it's reused (and
-    # don't clobber a transcript the user may have already edited).
-    box_out = transcript_obj.to_timestamped_text() if auto_transcribed else gr.update()
-    note = " (auto-transcribed first)" if auto_transcribed else ""
+    kind = "step-aligned" if steps_ts else ("narration-gated" if spoken else "scene")
+    box_out = transcript_obj.to_timestamped_text() if auto_tr else gr.update()
+    md_out = _draft_to_md(draft_obj) if auto_draft else gr.update()
+    note = ("  ·  " + "; ".join(notes)) if notes else ""
     return (
         _gallery_value(merged),
         merged,
         box_out,
         transcript_obj,
-        f"Auto-extracted {len(recs)} frames{note} ({len(merged)} total).",
+        md_out,
+        draft_obj,
+        f"Auto-extracted {len(recs)} {kind} frames ({len(merged)} total).{note}",
     )
 
 
@@ -258,6 +292,18 @@ def on_token_set(hf_token: str):
     return "Enter your HuggingFace token to generate the guide."
 
 
+def _generate_draft(tr, token: str, progress):
+    """Build the LLM step draft. Returns (draft|None, message). Shared by the
+    Generate button and Auto-extract."""
+    try:
+        draft = llm.build_guide_draft(tr, token=token, progress=progress)
+    except RuntimeError as exc:
+        return None, f"⚠️ {exc}"
+    if not draft.steps:
+        return None, "The LLM returned no steps — try a different DOCUMAKER_LLM_MODEL."
+    return draft, f"Drafted {len(draft.steps)} steps."
+
+
 def on_generate(transcript_text: str, transcript_obj, hf_token: str, progress=gr.Progress()):
     token = config.apply_token(hf_token)
     if not token:
@@ -265,13 +311,10 @@ def on_generate(transcript_text: str, transcript_obj, hf_token: str, progress=gr
     tr = _parse_timestamped_text(transcript_text) if transcript_text.strip() else transcript_obj
     if tr is None or not tr.segments:
         return "", None, "Transcribe the audio first (or paste a transcript)."
-    try:
-        draft = llm.build_guide_draft(tr, token=token, progress=progress)
-    except RuntimeError as exc:
-        return "", None, f"⚠️ {exc}"
-    if not draft.steps:
-        return "", None, "The LLM returned no steps — try a different DOCUMAKER_LLM_MODEL."
-    return _draft_to_md(draft), draft, f"Drafted {len(draft.steps)} steps. Review, then build the DOCX."
+    draft, msg = _generate_draft(tr, token, progress)
+    if draft is None:
+        return "", None, msg
+    return _draft_to_md(draft), draft, msg + " Review, then build the DOCX."
 
 
 def on_build(
@@ -363,8 +406,8 @@ def build_ui() -> gr.Blocks:
             with gr.Column(scale=2):
                 gr.Markdown(
                     "### Captured frames\n"
-                    "_**Auto-extract** transcribes first automatically, so frames snap to "
-                    "the narration and skip recorder intro/idle screens._"
+                    "_**Auto-extract** runs transcription + the step outline first, so frames "
+                    "anchor to the guide steps (and skip recorder intro/idle screens)._"
                 )
                 gallery = gr.Gallery(
                     label="Frames pool — click an image to enlarge / select it",
@@ -414,8 +457,8 @@ def build_ui() -> gr.Blocks:
         )
         auto_btn.click(
             on_auto,
-            [session_state, frames_state, video_state, transcript_state],
-            [gallery, frames_state, transcript_box, transcript_state, status],
+            [session_state, frames_state, video_state, transcript_state, draft_state, hf_token],
+            [gallery, frames_state, transcript_box, transcript_state, guide_md, draft_state, status],
         )
         gallery.select(on_select_frame, None, [selected_state, status])
         delete_btn.click(

scripts/smoke_test.py

@@ -54,13 +54,7 @@ def main() -> None:
     print(f"  device={tr.device} segments={len(tr.segments)} text={tr.text[:120]!r}")
     assert tr.text.strip(), "Transcript is empty"
 
-    print("[2/5] Auto-extract frames (narration-gated)…")
-    spoken = [(s.start, s.end) for s in tr.segments] if tr.segments else None
-    recs = frames_lib.extract_auto_frames(sample, sdir, spoken_intervals=spoken)
-    print(f"  frames={len(recs)}")
-    assert recs, "No frames were extracted"
-
-    print("[3/5] Build guide draft (LLM)…")
+    print("[2/5] Build guide draft (LLM)…")
     try:
         draft = llm.build_guide_draft(tr, token=token)
         if not draft.steps:
@@ -71,6 +65,15 @@ def main() -> None:
         draft = naive_draft(tr)
     assert draft.steps, "No steps in draft"
 
+    print("[3/5] Auto-extract frames (step-aligned)…")
+    spoken = [(s.start, s.end) for s in tr.segments] if tr.segments else None
+    step_ts = [s.approx_timestamp for s in draft.steps if s.approx_timestamp is not None]
+    recs = frames_lib.extract_auto_frames(
+        sample, sdir, spoken_intervals=spoken, step_timestamps=step_ts or None
+    )
+    print(f"  frames={len(recs)} (from {len(step_ts)} step timestamps)")
+    assert recs, "No frames were extracted"
+
     print("[4/5] Assemble (align + caption)…")
     spoken_range = (
         (min(s.start for s in tr.segments), max(s.end for s in tr.segments))

src/frames.py

@@ -80,22 +80,45 @@ def _scene_timestamps(
     return timestamps
 
 
+def _dedup_close(timestamps: list[float], min_gap: float = 1.5) -> list[float]:
+    """Collapse timestamps that are closer than ``min_gap`` seconds."""
+    out: list[float] = []
+    for t in sorted(timestamps):
+        if not out or t - out[-1] >= min_gap:
+            out.append(t)
+    return out
+
+
 def extract_auto_frames(
     video_path: str | Path,
     session_dir: str | Path,
     max_frames: int = 40,
     spoken_intervals: list[tuple[float, float]] | None = None,
+    step_timestamps: list[float] | None = None,
 ) -> list[FrameRecord]:
-    """Extract one representative frame per detected scene, then dedup.
-
-    When ``spoken_intervals`` (from the transcript) are given, frames are gated to
-    the narrated time range so recorder intro/idle screens are not captured.
+    """Extract representative frames, then dedup.
+
+    Priority of anchors:
+    1. ``step_timestamps`` (from the LLM step draft) — extract at the exact moments
+       the guide refers to, so the pool matches the steps.
+    2. ``spoken_intervals`` (from the transcript) — scene frames gated to the
+       narrated time range, dropping recorder intro/idle screens.
+    3. Otherwise — scene midpoints (or uniform sampling).
     """
     frames_dir = Path(session_dir) / "frames"
     frames_dir.mkdir(parents=True, exist_ok=True)
 
+    if step_timestamps:
+        timestamps = _dedup_close([t for t in step_timestamps if t is not None and t >= 0])
+        if len(timestamps) > max_frames:
+            last = len(timestamps) - 1
+            picks = sorted({round(i * last / (max_frames - 1)) for i in range(max_frames)})
+            timestamps = [timestamps[i] for i in picks]
+    else:
+        timestamps = _scene_timestamps(video_path, max_frames, spoken_intervals)
+
     records: list[FrameRecord] = []
-    for i, ts in enumerate(_scene_timestamps(video_path, max_frames, spoken_intervals)):
+    for i, ts in enumerate(timestamps):
         out = frames_dir / f"auto_{i:03d}_{int(ts * 1000):08d}.png"
         try:
             video.extract_frame(video_path, ts, out)