Adding improved scaffolding of videos and hopefully a fix of the narration

gnommo/cli.py

@@ -5149,7 +5149,17 @@ def _cmd_render_impl(
             _scaffold.reinterpolate_events(_events)
         # final_time = (narration + adjustment) shifted by preceding pauses.
         _scaffold.compute_final_times(_events)
-        _scaffold.write_events(project_path, _events)
+        # events.json carries representation-only narration-track events so a GUI
+        # treats every track uniformly. scaffold.json (the compiled render timeline)
+        # stays marker-only, and the render never reads these back, so behaviour is
+        # unchanged — this is a read-only mirror of the narration backbone.
+        _events_gui = _events + _scaffold.derive_narration_events(
+            narration_schedule,
+            plan.narration_videos,
+            plan.narration_pauses,
+            plan.total_duration,
+        )
+        _scaffold.write_events(project_path, _events_gui)
         _scaffold.write_scaffold(
             project_path, _events, transcription, narration_schedule, plan.total_duration
         )

gnommo/narration.py

@@ -118,3 +118,42 @@ def build_narration_schedule(
         offset += eff
 
     return segments, merged
+
+
+def slice_schedule(
+    schedule: list[NarrationSegment],
+    window_start: float,
+    window_end: float,
+) -> list[NarrationSegment]:
+    """Return the sub-schedule covering ``[window_start, window_end]`` of the
+    combined narration timeline — for a partial (chunked) render.
+
+    Each kept segment's ``skip``/``take``/``offset`` is adjusted so it seeks within
+    its OWN file: segments entirely outside the window are dropped, the first/last
+    kept segments are trimmed to the window edges, and offsets are re-based so the
+    sliced narration starts at 0. ``input_seek_time`` therefore stays 0 in concat
+    mode instead of a combined-timeline offset being (wrongly) applied to every
+    segment file.
+
+    This is what makes chunking bulletproof: because each chunk seeks its first
+    segment to the true source sample, ``render(A:B) ++ render(B:C)`` lands on the
+    exact same narration as ``render(A:C)``. Slicing to the full ``[0, total]`` is a
+    no-op, so full renders are unaffected.
+    """
+    import copy
+
+    out: list[NarrationSegment] = []
+    for seg in schedule:
+        seg_start = seg.offset
+        seg_end = seg.offset + seg.duration
+        keep_start = max(window_start, seg_start)
+        keep_end = min(window_end, seg_end)
+        if keep_end - keep_start <= 1e-6:
+            continue  # segment lies entirely outside the window
+        new = copy.copy(seg)
+        new.skip = round(seg.skip + (keep_start - seg_start), 6)
+        new.take = round(keep_end - keep_start, 6)
+        new.duration = new.take
+        new.offset = round(keep_start - window_start, 6)
+        out.append(new)
+    return out

gnommo/scaffold.py

@@ -201,6 +201,85 @@ def derive_events(
     return events
 
 
+def derive_narration_events(
+    narration_schedule: list,
+    narration_videos: list,
+    pauses: list,
+    total_duration: float,
+) -> list[dict]:
+    """Representation-only events mirroring the always-visible talking-head track.
+
+    One event per narration segment (or a single event for legacy single-file
+    narration), shaped like a `video` event — handle, cutout, layer, source_file,
+    skip/take, timing — so a GUI can render and lay out every track uniformly
+    instead of treating the narration backbone as invisible.
+
+    These are NOT read back into the render: events_to_marker_timings skips
+    `type == "narration"`, because narration timing is owned by the aligner and the
+    schedule (it is the clock, and it is multi-file). So this is a read-only mirror,
+    added purely for the editing surface — it never changes what renders.
+    """
+    # Cutout/layer come from the resolved narration video source (defaults match the
+    # talking-head convention used by import/transformer).
+    cutout = "talkinghead"
+    layer = "below"
+    if narration_videos:
+        _vs = narration_videos[0][1]
+        cutout = getattr(_vs, "cutout", None) or cutout
+        layer = getattr(_vs, "layer", None) or layer
+
+    pause_list = [
+        (float(p.narration_time), float(p.duration)) for p in (pauses or [])
+    ]
+
+    def _final(offset: float) -> float:
+        # Same shift compute_final_times applies: push forward by every pause at or
+        # before this point on the narration timeline.
+        return round(offset + sum(d for pn, d in pause_list if pn <= offset), 3)
+
+    def _base(seg_id, source_name, offset, duration, skip, take):
+        return {
+            "type": "narration",
+            "id": seg_id,
+            "narration_time": round(offset, 3),
+            "adjustment": 0.0,
+            "final_time": _final(offset),
+            "mapping": MAPPING_EXACT,
+            "confidence": 1.0,
+            "context": "(talking-head narration)",
+            "handle": seg_id,
+            "source_file": source_name,
+            "cutout": cutout,
+            "layer": layer,
+            "always_visible": True,
+            "end_on": "next_video",
+            "skip": round(skip or 0.0, 3),
+            "take": (round(take, 3) if take is not None else None),
+            "duration": round(duration, 3),
+        }
+
+    events: list[dict] = []
+    if narration_schedule:
+        for seg in narration_schedule:
+            src = getattr(seg.source_path, "name", None) or str(seg.source_path)
+            events.append(
+                _base(seg.seg_id, src, seg.offset, seg.duration, seg.skip, seg.take)
+            )
+    elif narration_videos:
+        _id, _vs, _ = narration_videos[0]
+        events.append(
+            _base(
+                _id,
+                getattr(_vs, "source_file", "") or _id,
+                0.0,
+                total_duration,
+                getattr(_vs, "skip", 0.0),
+                getattr(_vs, "take", None),
+            )
+        )
+    return events
+
+
 def _interpolate_narration(events: list[dict]) -> None:
     """Fill `narration_time: None` entries by linear interpolation between placed
     neighbours. Head/tail runs spread at +1s steps from the nearest known time (or
@@ -331,6 +410,10 @@ def events_to_marker_timings(events: list[dict]) -> list[MarkerTiming]:
     """
     timings: list[MarkerTiming] = []
     for e in events:
+        # Representation-only base-track events (derive_narration_events) are a
+        # read-only mirror of the narration backbone — never fed back as markers.
+        if e.get("type") == "narration":
+            continue
         n = e.get("narration_time")
         eff = (n + e.get("adjustment", 0.0)) if n is not None else -1.0
         # Carry any stored presentation as overrides so render honors the atomic event

gnommo/transformer.py

@@ -1053,6 +1053,23 @@ def build_render_plan(
     slides_json_path = project_path / config.slides_path.lower()
     slides_dir = slides_json_path.parent
 
+    # Concat-narration partial render: slice the schedule to the render window so
+    # each segment seeks within its OWN file (not by the combined-timeline offset,
+    # which over-seeks every file past the first). The offset is baked into each
+    # segment's skip, so narration input_seek_time stays 0. This runs on the shared
+    # slide_range path, so _chunked_render's per-chunk cmd_render and a hand-typed
+    # --slides use identical logic → render(A:B)++render(B:C) == render(A:C).
+    # Single-file narration keeps input_seek_time = time_offset (seeks that one file).
+    if narration_schedule:
+        from .narration import slice_schedule
+
+        narration_schedule = slice_schedule(
+            narration_schedule, time_offset, render_end_time
+        )
+        narration_input_seek = 0.0
+    else:
+        narration_input_seek = time_offset
+
     plan = RenderPlan(
         project_path=project_path,
         config=config,
@@ -1070,7 +1087,7 @@ def build_render_plan(
         camera_events=camera_events,
         time_offset=time_offset,
         initial_camera_state=initial_camera_state,
-        input_seek_time=time_offset,
+        input_seek_time=narration_input_seek,
         shared_assets_dir=shared_assets_dir,
         narration_pauses=narration_pauses,
         narration_segments=narration_schedule or [],

tests/test_narration_slice.py

@@ -0,0 +1,84 @@
+"""Invariant tests for narration schedule slicing (chunked-render correctness).
+
+The property that must hold: render(A:C) uses the same per-segment source samples
+as render(A:B) ++ render(B:C). Since the render seeks each narration segment by its
+(sliced) skip, that reduces to: slice([A,C]) covers the same source spans as
+slice([A,B]) followed by slice([B,C]).
+"""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from gnommo.narration import NarrationSegment, slice_schedule
+
+
+def seg(sid, skip, dur, offset):
+    return NarrationSegment(
+        seg_id=sid, source_path=Path(f"{sid}.mov"),
+        skip=skip, take=dur, duration=dur, offset=offset,
+    )
+
+
+# Combined timeline:
+#   s1: source skip 5,  plays 100  -> combined [0,100],   source [5,105]
+#   s2: source skip 10, plays 200  -> combined [100,300], source [10,210]
+#   s3: source skip 2,  plays 50   -> combined [300,350], source [2,52]
+SCHED = [seg("s1", 5, 100, 0), seg("s2", 10, 200, 100), seg("s3", 2, 50, 300)]
+
+
+def _spans(segs):
+    return [(s.seg_id, round(s.skip, 6), round(s.skip + s.take, 6)) for s in segs]
+
+
+def _merge(spans):
+    """Fuse adjacent spans of the same segment (the boundary segment split across
+    two chunks) so a chunked coverage can be compared to a single-window one."""
+    out = []
+    for sid, a, b in spans:
+        if out and out[-1][0] == sid and abs(out[-1][2] - a) < 1e-6:
+            out[-1] = (sid, out[-1][1], b)
+        else:
+            out.append((sid, a, b))
+    return out
+
+
+def check(name, cond):
+    print(f"  {'PASS' if cond else 'FAIL'}  {name}")
+    assert cond, name
+
+
+# Full-window slice is a no-op (full renders unaffected).
+full = slice_schedule(SCHED, 0, 350)
+check("full-window slice is identity", _spans(full) == _spans(SCHED))
+
+# Window drops the out-of-range segment and trims the edges into the files.
+w = slice_schedule(SCHED, 150, 320)
+check("drops out-of-window segment s1", [s.seg_id for s in w] == ["s2", "s3"])
+check("first kept seg seeks into its file (s2 skip 60, take 150, offset 0)",
+      (w[0].skip, w[0].take, w[0].offset) == (60, 150, 0))
+check("last kept seg trimmed to window (s3 skip 2, take 20, offset 150)",
+      (w[1].skip, w[1].take, w[1].offset) == (2, 20, 150))
+
+# Every seek stays within its own file's real bounds.
+for orig, sl in ((SCHED[1], w[0]), (SCHED[2], w[1])):
+    check(f"{sl.seg_id} seek within file bounds",
+          sl.skip >= orig.skip and sl.skip + sl.take <= orig.skip + orig.duration + 1e-6)
+
+# THE invariant: A:C == A:B ++ B:C in source coverage, for boundaries that land
+# mid-segment, on a segment edge, and spanning multiple segments.
+for A, B, C in [(150, 250, 340), (150, 300, 340), (50, 100, 350), (0, 300, 350)]:
+    whole = _spans(slice_schedule(SCHED, A, C))
+    joined = _merge(_spans(slice_schedule(SCHED, A, B)) + _spans(slice_schedule(SCHED, B, C)))
+    check(f"slice({A}:{C}) == slice({A}:{B})++slice({B}:{C})", joined == whole)
+
+# Offsets are contiguous and cover the window with no gap/overlap.
+for A, C in [(150, 320), (0, 350), (120, 340)]:
+    sl = slice_schedule(SCHED, A, C)
+    ok = abs(sl[0].offset) < 1e-6
+    for prev, cur in zip(sl, sl[1:]):
+        ok = ok and abs((prev.offset + prev.duration) - cur.offset) < 1e-6
+    ok = ok and abs((sl[-1].offset + sl[-1].duration) - (C - A)) < 1e-6
+    check(f"contiguous coverage of window [{A},{C}]", ok)
+
+print("\nAll narration-slice invariants passed.")