Keep manual edits to a song across rebuilds

The groundwork for an in-app editor. Edits live in edits.json in the work
dir, which no stage produces, so no rebuild removes them. An edit is a
whole lyric line as the user left it, in seconds, so a rebuild with a new
BPM re-snaps it. grid lays the edits over the freshly generated notes
(neighbouring lines give way to an edited one), writes the result to
notes_final.json with warnings brought up to date, and charts that. When
the lyrics change, edits follow their lines by text and order; an edit
whose line's words changed is set aside as stale and reported, not applied.

Notes gain a kind, so golden, freestyle and rap notes can be charted. The
review endpoint returns the merged notes with their syllables, kinds and
edit state, plus the chart's beat grid. PUT/DELETE
/api/jobs/{id}/lines/{n} save and revert a line, DELETE
/api/jobs/{id}/stale/{n} discards a stale edit; each rewrites the work and
packaged charts in place with the tempo already chosen, so a save
re-encodes nothing.

.dejima/allowed-domains

@@ -0,0 +1,6 @@
+deb.debian.org
+security.debian.org
+pypi.org
+files.pythonhosted.org
+download.pytorch.org
+download-r2.pytorch.org

.gitea/workflows/tests.yml

@@ -0,0 +1,86 @@
+name: tests
+
+# `pull_request_target` reads this file from the base branch, so a pull
+# request cannot change what runs or where (including `runs-on:`). The run's
+# ref is therefore the base branch, and the checkout has to name the head.
+on:
+  pull_request_target:
+
+jobs:
+  test:
+    # No Docker socket, and egress limited to .dejima/allowed-domains on main.
+    runs-on: dejima-confined
+
+    # Node for actions/checkout; Debian bookworm's python3 is the 3.11 the
+    # Dockerfile ships.
+    container:
+      image: node:22-bookworm
+
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+
+      # Inline rather than a script in the repository: the files come from the
+      # branch under test, so a script would be missing from older branches.
+      - name: Check the hosts this job fetches from are reachable
+        run: |
+          set -u
+          bad=""
+          for host in deb.debian.org security.debian.org pypi.org \
+                      files.pythonhosted.org download.pytorch.org \
+                      download-r2.pytorch.org; do
+            resolved=""
+            for attempt in 1 2 3 4; do
+              getent hosts "$host" >/dev/null 2>&1 && { resolved=yes; break; }
+              sleep 2
+            done
+            if [ -z "$resolved" ]; then
+              echo "no    $host - does not resolve"; bad="$bad $host(dns)"; continue
+            fi
+            # Any HTTP answer proves the connection; curl reports 000 for none.
+            code="$(curl -sS -o /dev/null --max-time 20 --retry 2 -w '%{http_code}' "https://$host/" 2>/dev/null || true)"
+            case "${code:-000}" in
+              000) echo "no    $host - resolves, but nothing answered over https"
+                   bad="$bad $host(net)" ;;
+              *)   echo "ok    $host (http $code)" ;;
+            esac
+          done
+          [ -z "$bad" ] && exit 0
+          {
+            echo
+            echo "Unreachable:$bad"
+            echo
+            echo "The confined tier reaches only what .dejima/allowed-domains on"
+            echo "main declares, rendered onto the runner separately from the"
+            echo "merge. On the runner:"
+            echo "  sudo dejima-ci-allowlist show    - is the name listed?"
+            echo "  sudo dejima-ci-allowlist check   - is a change pending?"
+            echo "  sudo dejima-ci-allowlist render  - apply it now"
+            echo "A (dns) failure is the resolver; a (net) one is the packet"
+            echo "filter, which sudo dejima-ci-firewall reapplies."
+            echo
+            echo "download.pytorch.org redirects its wheels to"
+            echo "download-r2.pytorch.org, so both have to be allowed."
+          } >&2
+          exit 1
+
+      - name: Install Python
+        run: |
+          apt-get update
+          apt-get install -y --no-install-recommends python3 python3-venv
+          python3 -m venv /opt/venv
+          # bookworm's pip 23.0 rejects the PyTorch index's typing_extensions
+          # wheel over its name's spelling, then cannot build it from source
+          /opt/venv/bin/pip install --retries 5 --upgrade pip
+          echo /opt/venv/bin >> "$GITHUB_PATH"
+
+      # The CPU build first: torch from PyPI is the CUDA build, several GB of
+      # libraries no test uses. The install below then finds torch satisfied.
+      - name: Install CPU-only torch
+        run: pip install --retries 5 torch torchaudio --index-url https://download.pytorch.org/whl/cpu
+
+      - name: Install songtrace
+        run: pip install --retries 5 -e '.[dev]'
+
+      - run: pytest

src/songtrace/edits.py

@@ -0,0 +1,173 @@
+"""Manual edits to a song, kept apart from what the pipeline generates.
+
+Edits live in work/<slug>/edits.json, which no stage produces, so no rebuild
+removes them; grid lays them over the freshly generated notes every time. An
+edit is a whole lyric line as the user left it, timed in seconds so that a
+rebuild with a different BPM re-snaps it rather than breaking it.
+"""
+import difflib
+import json
+import time
+from pathlib import Path
+
+from .notes import MIN_NOTE_SEC, TimedNote, low_confidence_warnings
+
+EDITS_FILE = "edits.json"
+FINAL_FILE = "notes_final.json"
+KINDS = {":", "*", "F", "R", "G"}
+
+
+def _empty(lyrics: list[list[str]]) -> dict:
+    return {"lyrics": lyrics, "lines": {}, "stale": []}
+
+
+def load(work: Path, lyrics: list[list[str]]) -> dict:
+    """The song's edits, keyed to `lyrics`.
+
+    Edits are stored with the lyrics they were made against. When those
+    differ, each edited line is matched to its line in the new lyrics by
+    text and order; an edit whose line's words changed is set aside as stale
+    rather than applied to words it was not made for.
+    """
+    path = work / EDITS_FILE
+    if not path.exists():
+        return _empty(lyrics)
+    data = json.loads(path.read_text())
+    if data["lyrics"] != lyrics:
+        data = _remap(data, lyrics)
+        save(work, data)
+    return data
+
+
+def save(work: Path, data: dict) -> None:
+    path = work / EDITS_FILE
+    if not data["lines"] and not data["stale"]:
+        path.unlink(missing_ok=True)
+        return
+    tmp = path.with_suffix(".tmp")
+    tmp.write_text(json.dumps(data))
+    tmp.replace(path)
+
+
+def _remap(data: dict, lyrics: list[list[str]]) -> dict:
+    old = [tuple(line) for line in data["lyrics"]]
+    new = [tuple(line) for line in lyrics]
+    moved = {}
+    for tag, i1, i2, j1, _ in difflib.SequenceMatcher(
+            a=old, b=new, autojunk=False).get_opcodes():
+        if tag == "equal":
+            moved.update({i1 + k: j1 + k for k in range(i2 - i1)})
+    lines, stale = {}, list(data["stale"])
+    for key, edit in data["lines"].items():
+        to = moved.get(int(key))
+        if to is None:
+            stale.append(edit)
+        else:
+            lines[str(to)] = {**edit, "line_index": to}
+    return {"lyrics": lyrics, "lines": lines, "stale": stale}
+
+
+def validate_notes(notes: list[dict]) -> list[dict]:
+    """The notes of a line edit, checked and reduced to what is stored."""
+    clean = []
+    for n in notes:
+        try:
+            note = {"start": float(n["start"]), "duration": float(n["duration"]),
+                    "pitch": int(n["pitch"]), "text": str(n["text"]),
+                    "kind": str(n.get("kind", ":"))}
+        except (KeyError, TypeError, ValueError):
+            raise ValueError(f"malformed note {n!r}")
+        if note["start"] < 0 or note["duration"] <= 0:
+            raise ValueError(f"a note needs start >= 0 and duration > 0: {n!r}")
+        if note["kind"] not in KINDS:
+            raise ValueError(f"unknown note kind {note['kind']!r}")
+        if not note["text"].strip():
+            raise ValueError("a note needs text")
+        clean.append(note)
+    return sorted(clean, key=lambda n: n["start"])
+
+
+def set_line(data: dict, index: int, notes: list[dict]) -> None:
+    data["lines"][str(index)] = {
+        "line_index": index, "words": data["lyrics"][index],
+        "notes": validate_notes(notes), "saved_at": time.time()}
+
+
+def clear_line(data: dict, index: int) -> bool:
+    return data["lines"].pop(str(index), None) is not None
+
+
+def merge(notes: list[dict], breaks: list[float], data: dict
+          ) -> tuple[list[dict], list[float]]:
+    """Generated notes and breaks with every edited line laid over them.
+
+    An edited line replaces the generated notes of that line, and the user's
+    timing wins: generated notes of earlier lines are cut back to end where it
+    starts, and those of later lines to start where it ends. Line breaks next
+    to an edited line move to the middle of the gap it leaves.
+    """
+    edited = {int(k): e["notes"] for k, e in data["lines"].items()}
+    if not edited:
+        return notes, breaks
+    spans = {li: (ns[0]["start"], ns[-1]["start"] + ns[-1]["duration"])
+             for li, ns in edited.items() if ns}
+    kept = []
+    for n in notes:
+        li = n["line_index"]
+        if li in edited:
+            continue
+        start, end = n["start"], n["start"] + n["duration"]
+        for eli, (a, b) in spans.items():
+            if li < eli:
+                end = min(end, a)
+            elif li > eli:
+                start = max(start, b)
+        if end - start >= MIN_NOTE_SEC:
+            kept.append({**n, "start": start, "duration": end - start})
+    for li, ns in edited.items():
+        kept += [{**n, "line_index": li, "low_confidence": False} for n in ns]
+    kept.sort(key=lambda n: (n["start"], n["line_index"]))
+
+    def line_after(t: float) -> int | None:
+        return next((n["line_index"] for n in notes if n["start"] >= t), None)
+
+    first = {}
+    last = {}
+    for n in kept:
+        first.setdefault(n["line_index"], n)
+        last[n["line_index"]] = n
+    order = sorted(first, key=lambda li: first[li]["start"])
+    out_breaks = []
+    for prev, nxt in zip(order, order[1:]):
+        if prev not in edited and nxt not in edited:
+            # the generated break between these two lines, untouched
+            gen = [t for t in breaks if line_after(t) == nxt]
+            if gen:
+                out_breaks.append(gen[0])
+                continue
+        end = last[prev]["start"] + last[prev]["duration"]
+        out_breaks.append((end + first[nxt]["start"]) / 2)
+    return kept, out_breaks
+
+
+def finalize(work: Path, lyrics: list[list[str]]) -> dict:
+    """The song as it is to be charted: generated notes with the edits laid
+    over them, warnings brought up to date, written to notes_final.json."""
+    generated = json.loads((work / "notes_sec.json").read_text())
+    data = load(work, lyrics)
+    notes, breaks = merge(generated["notes"], generated["breaks"], data)
+    warnings = generated.get("warnings", [])
+    if data["lines"]:
+        # low-confidence warnings must describe the merged notes, so a line
+        # the user fixed stops being reported; the others are about lyrics
+        warnings = (low_confidence_warnings([TimedNote(**n) for n in notes])
+                    + [w for w in warnings if "low-confidence" not in w])
+    if data["stale"]:
+        n = len(data["stale"])
+        warnings.append(f"{n} edited line{' no longer matches' if n == 1 else 's no longer match'}"
+                        " the lyrics; set aside in the editor")
+    final = {"notes": notes, "breaks": breaks, "warnings": warnings,
+             "edited_lines": sorted(int(k) for k in data["lines"]),
+             "stale": data["stale"]}
+    (work / FINAL_FILE).write_text(json.dumps(final))
+    return final

src/songtrace/grid.py

@@ -3,7 +3,8 @@ from pathlib import Path
 
 import numpy as np
 
-from .ultrastar import LineBreak, Note, UltraStarSong, format_bpm
+from .ultrastar import (LineBreak, Note, UltraStarSong, format_bpm, get_header,
+                        parse_bpm, parse_ultrastar, serialize_ultrastar)
 
 
 def detect_bpm(instrumental_path: Path) -> float:
@@ -49,10 +50,24 @@ def to_ultrastar(notes: list[dict], breaks: list[float], file_bpm: float,
         else:
             b = max(beat(t), cursor)
             length = max(1, beat(t + n["duration"]) - beat(t))
-            body.append(Note(kind=":", beat=b, length=length, pitch=n["pitch"],
-                             text=n["text"]))
+            body.append(Note(kind=n.get("kind", ":"), beat=b, length=length,
+                             pitch=n["pitch"], text=n["text"]))
             cursor = b + length
     out_headers = list(headers)
     out_headers.append(("BPM", format_bpm(file_bpm)))
     out_headers.append(("GAP", str(gap_ms)))
     return UltraStarSong(headers=out_headers, body=body)
+
+
+def rechart(txt_path: Path, notes: list[dict], breaks: list[float]) -> None:
+    """Replace a chart's notes, keeping its headers and its BPM.
+
+    For an edit saved from the editor: the tempo was settled when the song
+    was built, and the packaged chart's headers may differ from the work
+    copy's (a re-encoded #VIDEO, a dropped #COVER), so each keeps its own.
+    """
+    song = parse_ultrastar(txt_path.read_text(encoding="utf-8"))
+    bpm = parse_bpm(get_header(song, "BPM"))
+    headers = [(k, v) for k, v in song.headers if k.upper() not in ("BPM", "GAP")]
+    txt_path.write_text(serialize_ultrastar(to_ultrastar(notes, breaks, bpm, headers)),
+                        encoding="utf-8")

src/songtrace/notes.py

@@ -7,7 +7,17 @@ import pyphen
 from .languages import LANGUAGES
 
 CONF_MASK = 0.5          # pitch frames below this are unvoiced
-LOW_SCORE = 0.5          # word alignment score below this -> low_confidence
+# A word is low-confidence when it scores below this share of the song's
+# median alignment score: the scale differs a lot from song to song, so no
+# fixed cutoff holds across them.
+LOW_SCORE_OF_MEDIAN = 0.25
+# ...or when no one sings a word that fast, or it was stretched over more than
+# it can be holding.
+MIN_WORD_SEC = 0.08
+MAX_WORD_SEC = 3.0
+# Flagged notes closer than this share one warning, so the list reads as
+# places to look rather than one line per note.
+WARN_GAP_SEC = 2.0
 SPLIT_SEMITONES = 0.8
 MIN_SEG_SEC = 0.05
 MIN_NOTE_SEC = 0.05      # alignment may hand us zero-length words
@@ -24,6 +34,7 @@ class TimedNote:
     text: str             # display text incl. leading space / "~"
     low_confidence: bool
     line_index: int
+    kind: str = ":"       # UltraStar note type; generated notes are all normal
 
 
 def hz_to_midi_float(hz: float) -> float:
@@ -84,6 +95,7 @@ def build_notes(rows: list[dict], times, f0, conf, language: str | None,
                 ) -> tuple[list[TimedNote], list[float], list[str]]:
     notes: list[TimedNote] = []
     breaks: list[float] = []
+    floor = LOW_SCORE_OF_MEDIAN * float(np.median([r["score"] for r in rows])) if rows else 0.0
     prev_line, prev_end = None, 0.0
     for row in rows:
         if prev_line is not None and row["line_index"] > prev_line:
@@ -97,7 +109,8 @@ def build_notes(rows: list[dict], times, f0, conf, language: str | None,
                                _text(row["word"], 0, line_initial), True,
                                row["line_index"]))
             continue
-        low = row["score"] < LOW_SCORE
+        low = (row["score"] < floor
+               or not MIN_WORD_SEC <= row["end"] - row["start"] <= MAX_WORD_SEC)
         sylls = syllabify(row["word"], language)
         if len(segs) < len(sylls):
             segs = _subdivide(segs, len(sylls))
@@ -158,18 +171,25 @@ def _fmt(t: float) -> str:
     return f"{int(t // 60)}:{int(t % 60):02d}"
 
 
+def low_confidence_warnings(notes: list[TimedNote]) -> list[str]:
+    runs: list[list[TimedNote]] = []
+    for n in notes:
+        if not n.low_confidence:
+            continue
+        last = runs[-1][-1] if runs else None
+        if last is not None and n.start - (last.start + last.duration) < WARN_GAP_SEC:
+            runs[-1].append(n)
+        else:
+            runs.append([n])
+    return [
+        f"{len(run)} low-confidence note{'' if len(run) == 1 else 's'} "
+        f"near {_fmt(run[0].start)}–{_fmt(run[-1].start + run[-1].duration)}"
+        for run in runs]
+
+
 def _warnings(notes: list[TimedNote], dropped_lines=(), missing_spans=(),
              ref_lines=()) -> list[str]:
-    warnings, run = [], []
-    for n in list(notes) + [None]:
-        if n is not None and n.low_confidence:
-            run.append(n)
-        elif run:
-            first, last = run[0], run[-1]
-            warnings.append(
-                f"{len(run)} low-confidence note{'' if len(run) == 1 else 's'} "
-                f"near {_fmt(first.start)}–{_fmt(last.start + last.duration)}")
-            run = []
+    warnings = low_confidence_warnings(notes)
     for group in _consecutive_runs(dropped_lines):
         n = len(group)
         text = " ".join(ref_lines[group[0]]) if group[0] < len(ref_lines) else ""

src/songtrace/pipeline.py

@@ -47,7 +47,11 @@ def song_slug(inp: SongInput) -> str:
     return name.replace("/", "").replace("\0", "")
 
 def read_warnings(work: Path) -> list[str]:
-    notes_file = work / "notes_sec.json"
+    # grid's copy accounts for the user's edits; notes' is all a song built
+    # before edits existed has
+    notes_file = work / "notes_final.json"
+    if not notes_file.exists():
+        notes_file = work / "notes_sec.json"
     if not notes_file.exists():
         return []
     return json.loads(notes_file.read_text()).get("warnings", [])

src/songtrace/stages/align.py

@@ -1,12 +1,24 @@
 import json
 from pathlib import Path
 
+import numpy as np
+import soundfile as sf
+
 from ..lyrics import alignment_token, normalize_lyrics
 from ..models import on_release
 from ..pipeline import Ctx, PipelineError, Stage
-from .transcribe import classify_lyrics_match, load_transcript, lyrics_similarity, vocal_sections
+from ..whisper_client import TranscriptSegment
+from .transcribe import (classify_lyrics_match, load_transcript, lyrics_similarity,
+                         recovered_spans, vocal_sections)
 
 SIMILARITY_THRESHOLD = 0.4
+# MMS_FA's wildcard token: any audio the transcript does not cover
+STAR = "*"
+# Singing is audio within this much of the stem's loud level; separation
+# bleed sits far below it.
+SINGING_BELOW_LOUD_DB = 25.0
+# Shorter stretches are breaths, ad-libs or Whisper's loose segment edges.
+MIN_UNHEARD_SEC = 3.0
 
 # a stage aligns section by section, so the model is cached across the calls
 # within one run and dropped at the end of it
@@ -40,7 +52,7 @@ def force_align_section(vocals_path: Path, start: float, end: float,
         waveform[:, lo:hi], sr, bundle.sample_rate)
     # MMS_FA only models a-z and apostrophe; anything else becomes the star token.
     vocab = bundle.get_dict()
-    tokens = ["".join(c if c in vocab else "*" for c in t) for t in tokens]
+    tokens = ["".join(c if c in vocab else STAR for c in t) for t in tokens]
     with torch.inference_mode():
         emission, _ = _get_model(device)(waveform.to(device))
         spans = bundle.get_aligner()(emission[0], bundle.get_tokenizer()(tokens))
@@ -50,6 +62,39 @@ def force_align_section(vocals_path: Path, start: float, end: float,
              sum(t.score for t in s) / len(s)) for s in spans]
 
 
+def untranscribed_singing(vocals_path: Path, segments,
+                          frame: float = 0.05) -> list[tuple[float, float]]:
+    """Stretches of the vocal stem with singing in them but no transcript."""
+    data, sr = sf.read(str(vocals_path), dtype="float32", always_2d=True)
+    hop = int(sr * frame)
+    n = len(data) // hop
+    if n == 0:
+        return []
+    rms = np.sqrt((data[:n * hop].mean(1).reshape(n, hop) ** 2).mean(1))
+    db = 20 * np.log10(rms + 1e-10)
+    sung = db >= np.quantile(db, 0.9) - SINGING_BELOW_LOUD_DB
+    times = np.arange(n) * frame
+    for s in segments:
+        sung[(times >= min(s.start, s.end)) & (times < max(s.start, s.end))] = False
+    spans, start = [], None
+    for t, on in zip(times, sung):
+        if on and start is None:
+            start = float(t)
+        elif not on and start is not None:
+            spans.append((start, float(t)))
+            start = None
+    if start is not None:
+        spans.append((start, float(times[-1] + frame)))
+    # a breath or a quiet syllable must not split one unsung passage in two
+    merged: list[list[float]] = []
+    for a, b in spans:
+        if merged and a - merged[-1][1] < 1.0:
+            merged[-1][1] = b
+        else:
+            merged.append([a, b])
+    return [(a, b) for a, b in merged if b - a >= MIN_UNHEARD_SEC]
+
+
 def _assign_lines_to_sections(line_indices: list[int], sections) -> list[list[int]]:
     """Split matched lyric line indices across vocal sections proportionally to section duration."""
     total = sum(b - a for a, b in sections) or 1.0
@@ -92,9 +137,20 @@ def _enforce_monotonic(rows: list[dict]) -> None:
 def _align_section(vocals: Path, sec: tuple[float, float], words: list[tuple[str, int]],
                    language: str | None, device: str) -> list[dict]:
     tokens = [alignment_token(w, language) for w, _ in words]
-    alignable = [t for t in tokens if t]
-    aligned = iter(force_align_section(vocals, sec[0], sec[1], alignable, device)
-                   if alignable else [])
+    # A star before, between and after lines lets speech, laughter or an
+    # ad-lib there go unexplained rather than claim the next line's letters.
+    seq, prev_line = [], None
+    for (_, line_index), token in zip(words, tokens):
+        if not token:
+            continue
+        if line_index != prev_line:
+            seq.append(STAR)
+            prev_line = line_index
+        seq.append(token)
+    if seq:
+        seq.append(STAR)
+    spans = force_align_section(vocals, sec[0], sec[1], seq, device) if seq else []
+    aligned = iter(span for span, token in zip(spans, seq) if token != STAR)
     rows, blank = [], (None, None, 0.0)
     for (word, line_index), token in zip(words, tokens):
         start, end, score = next(aligned, blank) if token else blank
@@ -112,11 +168,18 @@ def _run(ctx: Ctx) -> None:
         raise PipelineError(
             f"lyrics don't match audio (similarity {sim:.2f}) — "
             "wrong lyrics or wrong recording?")
-    # normalised so an inverted span can never collapse a section onto one instant
-    sections = [(min(a, b), max(a, b)) for a, b in vocal_sections(segments)]
-    dropped_lines, missing_spans = classify_lyrics_match(segments, ref)
-    matched = [i for i in range(len(ref.lines)) if i not in set(dropped_lines)]
     vocals = ctx.work / "vocals.wav"
+    unheard = untranscribed_singing(vocals, segments)
+    dropped_lines, missing_spans = classify_lyrics_match(segments, ref,
+                                                         unheard=unheard)
+    # lines Whisper missed are aligned against the singing it had no text for;
+    # only that singing, so an untranscribed ad-lib cannot draw lines into it
+    recovered = [TranscriptSegment(a, b, "")
+                 for a, b in recovered_spans(segments, ref, unheard)]
+    # normalised so an inverted span can never collapse a section onto one instant
+    sections = [(min(a, b), max(a, b))
+                for a, b in vocal_sections(segments + recovered)]
+    matched = [i for i in range(len(ref.lines)) if i not in set(dropped_lines)]
     rows = []
     for sec, line_idxs in zip(sections, _assign_lines_to_sections(matched, sections)):
         words = [(w, li) for li in line_idxs for w in ref.lines[li]]

src/songtrace/stages/grid.py

@@ -1,13 +1,16 @@
 import json
 from pathlib import Path
 
-from .. import grid
+from .. import edits, grid
+from ..lyrics import normalize_lyrics
 from ..pipeline import Ctx, PipelineError, Stage, song_slug
 from ..ultrastar import serialize_ultrastar
 
 
 def _run(ctx: Ctx) -> None:
-    data = json.loads((ctx.work / "notes_sec.json").read_text())
+    data = edits.finalize(ctx.work, normalize_lyrics(ctx.inp.lyrics_text).lines)
+    if data["edited_lines"]:
+        ctx.log(f"[grid] kept {len(data['edited_lines'])} edited line(s)")
     meta = json.loads((ctx.work / "meta.json").read_text())
     if ctx.inp.bpm_override is not None:
         bpm = ctx.inp.bpm_override
@@ -36,5 +39,6 @@ def _run(ctx: Ctx) -> None:
 
 def grid_stage() -> Stage:
     return Stage(name="grid",
-                 outputs=lambda ctx: [ctx.work / "song.txt"],
+                 outputs=lambda ctx: [ctx.work / "song.txt",
+                                      ctx.work / edits.FINAL_FILE],
                  run=_run)

src/songtrace/stages/transcribe.py

@@ -39,28 +39,74 @@ def lyrics_similarity(segments, ref: RefLyrics) -> float:
     return matcher.ratio()
 
 
-def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0
+def _line_times(segments, matcher, seg_of, line_of) -> dict[int, tuple[float, float]]:
+    """For each reference line that was heard, the start of the first and the
+    end of the last transcript segment its words were heard in. Words in a
+    'replace' block are paired with heard words proportionally."""
+    times: dict[int, tuple[float, float]] = {}
+    for tag, ai1, ai2, bj1, bj2 in matcher.get_opcodes():
+        if tag not in ("equal", "replace"):
+            continue
+        for j in range(bj1, bj2):
+            seg = segments[seg_of[ai1 + (j - bj1) * (ai2 - ai1) // (bj2 - bj1)]]
+            a, b = times.get(line_of[j], (seg.start, seg.end))
+            times[line_of[j]] = (min(a, seg.start), max(b, seg.end))
+    return times
+
+
+def _unheard_runs(segments, ref: RefLyrics, unheard
+                  ) -> tuple[list[list[int]], list[list[tuple[float, float]]]]:
+    """Runs of reference lines with nothing heard for them, each with the
+    untranscribed singing between the heard lines either side of it."""
+    matcher, seg_of, line_of = match_lyrics(segments, ref)
+    times = _line_times(segments, matcher, seg_of, line_of)
+    alignable = set(line_of)
+    runs, spans, run, prev_end = [], [], [], 0.0
+
+    def close(next_start: float) -> None:
+        if run:
+            runs.append(list(run))
+            spans.append([(max(a, prev_end), min(b, next_start)) for a, b in unheard
+                          if a < next_start and b > prev_end])
+            run.clear()
+
+    for li in range(len(ref.lines)):
+        if li in times:
+            close(times[li][0])
+            prev_end = times[li][1]
+        elif li in alignable:
+            run.append(li)
+    close(float("inf"))
+    return runs, spans
+
+
+def recovered_spans(segments, ref: RefLyrics, unheard) -> list[tuple[float, float]]:
+    """The untranscribed singing that stands in for lines Whisper missed, so
+    alignment can place them there."""
+    return _merge_spans([sp for spans in _unheard_runs(segments, ref, unheard)[1]
+                         for sp in spans])
+
+
+def classify_lyrics_match(segments, ref: RefLyrics, pad: float = 1.0, unheard=()
                           ) -> tuple[list[int], list[tuple[float, float]]]:
     """Split reference lines into matched vs. dropped, and flag sung spans no
     reference line explains.
 
     dropped_lines: reference line indices covered only by 'insert' opcodes
-    (present in the lyrics, never heard). missing_spans: the padded, merged
-    bounds of the heard segments behind 'delete' opcodes (heard, but matching
-    no reference word). A segment can also hold matched lines, so a span
-    marks where to look, not words to discard.
-    """
-    matcher, seg_of, line_of = match_lyrics(segments, ref)
-    covered = [False] * len(line_of)
-    for tag, _, _, bj1, bj2 in matcher.get_opcodes():
-        if tag in ("equal", "replace"):
-            for j in range(bj1, bj2):
-                covered[j] = True
-    by_line = {}
-    for j, li in enumerate(line_of):
-        by_line.setdefault(li, []).append(covered[j])
-    dropped_lines = sorted(li for li, cov in by_line.items() if not any(cov))
+    (present in the lyrics, never heard), unless `unheard` (sung spans the
+    transcript has no text for) has singing between the heard lines either
+    side of them: then Whisper missed them, not the singer.
 
+    missing_spans: the padded, merged bounds of the heard segments behind
+    'delete' opcodes (heard, but matching no reference word). A segment can
+    also hold matched lines, so a span marks where to look, not words to
+    discard.
+    """
+    runs, run_spans = _unheard_runs(segments, ref, unheard)
+    dropped_lines = sorted(li for run, sp in zip(runs, run_spans) if not sp
+                           for li in run)
+
+    matcher, seg_of, _ = match_lyrics(segments, ref)
     spans = []
     for tag, ai1, ai2, _, _ in matcher.get_opcodes():
         if tag != "delete":

src/songtrace/web.py

@@ -9,11 +9,12 @@ import uuid
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
 
-from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi import Body, FastAPI, File, Form, HTTPException, UploadFile
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import FileResponse
 from fastapi.staticfiles import StaticFiles
 
+from . import edits, grid
 from . import rebuild as rebuild_policy
 from .config import Config
 from .lyrics import normalize_lyrics
@@ -77,7 +78,7 @@ def _cleanup_uploads(cfg: Config, job_id: str, video_path: Path | None) -> None:
     Covers are left behind: they are small, and a rebuild inherits cover_path, so
     removing one would silently strip #COVER from the rebuilt song. A rebuild can
     also inherit its parent's upload, so the directory is not always uploads/<job_id>.
-    Never raises — the caller has already marked the job done.
+    Never raises, so cleaning up cannot fail a job that otherwise succeeded.
     """
     uploads = (cfg.work_dir / "uploads").resolve()
     dirs = {uploads / job_id}
@@ -132,6 +133,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
     jobs: dict[str, Job] = {}
     order: list[str] = []
     lock = threading.Lock()
+    edit_lock = threading.Lock()
     todo: queue.Queue[str] = queue.Queue()
 
     for job in _discover_jobs(cfg):
@@ -193,6 +195,7 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
                     if m := STAGE_LINE.match(msg):
                         job.stage = m[1]
 
+            status, error = "done", None
             try:
                 force_from = _prepare_rebuild(job, log)
                 _write_sidecar(job)
@@ -201,18 +204,15 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
                     job.output_dir = str(result.output_dir)
                     job.warnings = list(result.warnings)
                 _drop_superseded_output(job, log)
-                with lock:
-                    job.status = "done"
                 # uploads survive a failure so the job stays rerunnable; acquire
                 # has already copied the video into the work dir by now.
                 _cleanup_uploads(cfg, job_id, job.inp.video_path)
             except BaseException as e:
-                with lock:
-                    job.status = "failed"
-                    job.error = str(e)
-            finally:
-                with lock:
-                    job.finished_at = time.time()
+                status, error = "failed", str(e)
+            # last, and all at once: a client that sees the job finished must
+            # find everything finishing it does already done
+            with lock:
+                job.status, job.error, job.finished_at = status, error, time.time()
 
     threading.Thread(target=worker, daemon=True).start()
 
@@ -468,7 +468,9 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
         if job.status != "done":
             raise HTTPException(404)
         work = cfg.work_dir / song_slug(job.inp)
-        notes_file = work / "notes_sec.json"
+        notes_file = work / edits.FINAL_FILE
+        if not notes_file.exists():
+            notes_file = work / "notes_sec.json"
         if not notes_file.exists():
             raise HTTPException(404)
         notes_data = json.loads(notes_file.read_text())
@@ -488,13 +490,96 @@ def create_app(cfg: Config, runner=run_pipeline) -> FastAPI:
         for n in notes_data["notes"]:
             by_line.setdefault(n["line_index"], []).append({
                 "start": n["start"], "end": n["start"] + n["duration"],
-                "pitch": n["pitch"], "low_confidence": n["low_confidence"]})
+                "pitch": n["pitch"], "low_confidence": n["low_confidence"],
+                "text": n["text"], "kind": n.get("kind", ":")})
+        edited = set(notes_data.get("edited_lines", []))
         lines = [{"index": i, "text": " ".join(words), "dropped": i in dropped,
+                  "edited": i in edited,
                   "notes": by_line.get(i, []), "words": words_by_line.get(i, [])}
                  for i, words in enumerate(ref.lines)]
+        stale = [{"text": " ".join(e["words"]), "notes": e["notes"]}
+                 for e in notes_data.get("stale", [])]
         media = {kind: _resolve_media(job, header) is not None
                  for kind, header in MEDIA_HEADERS.items()}
-        return {"lines": lines, "missing_spans": missing_spans, "media": media}
+        return {"lines": lines, "missing_spans": missing_spans, "media": media,
+                "stale": stale, "grid": _chart_grid(work)}
+
+    def _chart_grid(work: Path) -> dict | None:
+        """The beat grid the chart was built on, for snapping edits to it."""
+        txt = work / "song.txt"
+        if not txt.exists():
+            return None
+        song = parse_ultrastar(txt.read_text(encoding="utf-8"))
+        bpm, gap = get_header(song, "BPM"), get_header(song, "GAP")
+        if bpm is None or gap is None:
+            return None
+        return {"bpm": float(bpm.replace(",", ".")), "gap_ms": float(gap.replace(",", "."))}
+
+    def _editable(job_id: str) -> Job:
+        job = _finished_job(job_id)
+        slug = song_slug(job.inp)
+        with lock:
+            # a build of this song would overwrite the chart being edited
+            if any(j.status in ("queued", "running") and song_slug(j.inp) == slug
+                   for j in jobs.values()):
+                raise HTTPException(409, "this song is being rebuilt")
+        if job.status != "done":
+            raise HTTPException(409, "only a finished song can be edited")
+        return job
+
+    def _edit(job_id: str, change) -> dict:
+        """Apply `change` to the song's edits, then rewrite its charts."""
+        job = _editable(job_id)
+        work = cfg.work_dir / song_slug(job.inp)
+        lyrics = normalize_lyrics(job.inp.lyrics_text).lines
+        with edit_lock:
+            data = edits.load(work, lyrics)
+            try:
+                change(data, lyrics)
+            except ValueError as e:
+                raise HTTPException(400, str(e))
+            edits.save(work, data)
+            final = edits.finalize(work, lyrics)
+            charts = [work / "song.txt"]
+            if job.output_dir:
+                charts.append(Path(job.output_dir) / f"{song_slug(job.inp)}.txt")
+            for txt in charts:
+                if txt.exists():
+                    grid.rechart(txt, final["notes"], final["breaks"])
+        with lock:
+            job.warnings = list(final["warnings"])
+        return get_review(job_id)
+
+    def _line(lyrics, index: int) -> int:
+        if not 0 <= index < len(lyrics):
+            raise HTTPException(404, f"no line {index}")
+        return index
+
+    @app.put("/api/jobs/{job_id}/lines/{index}")
+    def save_line(job_id: str, index: int, body: dict = Body(...)):
+        """Save a line as edited: its full note list, in seconds."""
+        notes = body.get("notes")
+        if not isinstance(notes, list):
+            raise HTTPException(400, "body must be {\"notes\": [...]}")
+        return _edit(job_id, lambda data, lyrics: edits.set_line(
+            data, _line(lyrics, index), notes))
+
+    @app.delete("/api/jobs/{job_id}/lines/{index}")
+    def revert_line(job_id: str, index: int):
+        """Drop a line's edit, so it follows what the pipeline generates."""
+        def change(data, lyrics):
+            if not edits.clear_line(data, _line(lyrics, index)):
+                raise HTTPException(404, f"line {index} has no edit")
+        return _edit(job_id, change)
+
+    @app.delete("/api/jobs/{job_id}/stale/{n}")
+    def discard_stale(job_id: str, n: int):
+        """Throw away an edit whose line's lyrics have since changed."""
+        def change(data, lyrics):
+            if not 0 <= n < len(data["stale"]):
+                raise HTTPException(404, f"no stale edit {n}")
+            data["stale"].pop(n)
+        return _edit(job_id, change)
 
     return app
 

tests/test_align.py

@@ -1,11 +1,23 @@
 import json
 
+import numpy as np
 import pytest
+import soundfile as sf
 
 from songtrace.config import Config
 from songtrace.lyrics import normalize_lyrics
 from songtrace.pipeline import Ctx, PipelineError, SongInput
 from songtrace.stages import align
+from songtrace.whisper_client import TranscriptSegment
+
+REAL_UNTRANSCRIBED_SINGING = align.untranscribed_singing
+
+
+@pytest.fixture(autouse=True)
+def no_untranscribed_singing(monkeypatch):
+    """The tests have no vocals.wav; those about missed singing set their own."""
+    monkeypatch.setattr(align, "untranscribed_singing", lambda vocals, segments: [])
+
 
 LYRICS = "it's one two three\nfour five six seven eight"
 
@@ -160,3 +172,66 @@ def test_align_flags_missing_span_for_unmatched_singing(tmp_path, monkeypatch):
     meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
     assert meta["dropped_lines"] == []
     assert meta["missing_spans"] == [[19.0, 24.0]]
+
+
+def _tone(seconds, sr, level):
+    t = np.arange(int(seconds * sr)) / sr
+    return level * np.sin(2 * np.pi * 220 * t)
+
+
+def test_untranscribed_singing_finds_sung_audio_the_transcript_skips(
+        tmp_path):
+    sr = 16000
+    rng = np.random.default_rng(0)
+    audio = np.concatenate([
+        _tone(2, sr, 0.3),           # 0-2s: sung and transcribed
+        np.zeros(2 * sr),            # 2-4s: silence
+        _tone(5, sr, 0.3),           # 4-9s: sung, never transcribed
+        np.zeros(3 * sr),            # 9-12s: silence
+        _tone(1, sr, 0.3),           # 12-13s: too short to count
+        np.zeros(3 * sr),
+    ]) + 0.001 * rng.standard_normal(16 * sr)   # separation bleed
+    path = tmp_path / "vocals.wav"
+    sf.write(path, audio.astype(np.float32), sr)
+    segments = [TranscriptSegment(0.0, 2.0, "one two")]
+    spans = REAL_UNTRANSCRIBED_SINGING(path, segments)
+    assert len(spans) == 1
+    assert spans[0][0] == pytest.approx(4.0, abs=0.1)
+    assert spans[0][1] == pytest.approx(9.0, abs=0.1)
+
+
+def test_lines_whisper_missed_are_aligned_in_the_untranscribed_singing(
+        tmp_path, monkeypatch):
+    lyrics = "one two three\nfour five six\nseven eight nine"
+    transcript = [{"start": 1.0, "end": 3.0, "text": "one two three"},
+                  {"start": 40.0, "end": 42.0, "text": "seven eight nine"}]
+    monkeypatch.setattr(align, "untranscribed_singing",
+                        lambda vocals, segments: [(20.0, 26.0)])
+    rows = run_align(tmp_path, monkeypatch, lyrics, transcript)
+    check_invariants(rows, lyrics)
+    middle = [r for r in rows if r["line_index"] == 1]
+    assert all(19.0 <= r["start"] and r["end"] <= 27.0 for r in middle)
+    meta = json.loads((tmp_path / "w" / "align_meta.json").read_text())
+    assert meta["dropped_lines"] == []
+
+
+def test_stars_bracket_each_line_and_take_no_words(tmp_path, monkeypatch):
+    lyrics = "one two\n...\nthree four"
+    transcript = [{"start": 0.0, "end": 10.0, "text": "one two three four"}]
+    seen = []
+
+    def spy(vocals_path, start, end, tokens, device):
+        seen.append(tokens)
+        # each star claims a long stretch, so a word given a star's span shows
+        spans = [(start + i, start + i + (5.0 if t == "*" else 0.5), 0.9)
+                 for i, t in enumerate(tokens)]
+        return spans
+
+    monkeypatch.setattr(align, "force_align_section", spy)
+    ctx = make_ctx(tmp_path, lyrics, transcript)
+    align.align_stage().run(ctx)
+    rows = json.loads((ctx.work / "alignment.json").read_text())
+    assert seen == [["*", "one", "two", "*", "three", "four", "*"]]
+    timed = [r for r in rows if r["word"] != "..."]
+    assert all(r["end"] - r["start"] == pytest.approx(0.5) for r in timed)
+    check_invariants(rows, lyrics)

tests/test_edits.py

@@ -0,0 +1,124 @@
+import json
+
+import pytest
+
+from songtrace import edits
+
+LYRICS = [["one", "two"], ["three", "four"], ["five", "six"]]
+
+
+def gen(start, line, pitch=0, dur=0.4, low=False, text="la"):
+    return {"start": start, "duration": dur, "pitch": pitch, "text": text,
+            "low_confidence": low, "line_index": line}
+
+
+def edited(*lines):
+    data = edits._empty(LYRICS)
+    for index, notes in lines:
+        edits.set_line(data, index, notes)
+    return data
+
+
+def note(start, pitch=5, dur=0.4, text="la", kind=":"):
+    return {"start": start, "duration": dur, "pitch": pitch, "text": text, "kind": kind}
+
+
+GENERATED = [gen(1.0, 0), gen(1.5, 0), gen(4.0, 1), gen(4.5, 1), gen(8.0, 2), gen(8.5, 2)]
+BREAKS = [3.0, 6.0]
+
+
+def test_no_edits_leaves_the_generated_song_alone():
+    assert edits.merge(GENERATED, BREAKS, edits._empty(LYRICS)) == (GENERATED, BREAKS)
+
+
+def test_an_edited_line_replaces_its_generated_notes():
+    notes, _ = edits.merge(GENERATED, BREAKS, edited(
+        (1, [note(4.2, pitch=7, kind="*"), note(4.8, pitch=9)])))
+    line1 = [n for n in notes if n["line_index"] == 1]
+    assert [(n["start"], n["pitch"], n["kind"]) for n in line1] == [
+        (4.2, 7, "*"), (4.8, 9, ":")]
+    assert not any(n["low_confidence"] for n in line1)
+    assert [n["line_index"] for n in notes] == [0, 0, 1, 1, 2, 2]
+
+
+def test_neighbours_give_way_to_an_edited_line():
+    # line 1 edited to run from 1.7s to 8.6s, over its neighbours' notes
+    notes, _ = edits.merge(GENERATED, BREAKS, edited(
+        (1, [note(1.7), note(8.2, dur=0.4)])))
+    by_line = {}
+    for n in notes:
+        by_line.setdefault(n["line_index"], []).append(
+            (round(n["start"], 2), round(n["start"] + n["duration"], 2)))
+    assert by_line[0] == [(1.0, 1.4), (1.5, 1.7)]     # cut back to end at 1.7s
+    assert by_line[2] == [(8.6, 8.9)]                 # 8.0 note gone, 8.5 starts at 8.6
+
+
+def test_breaks_beside_an_edited_line_move_and_the_rest_stay():
+    data = edited((2, [note(9.0), note(9.5)]))
+    _, breaks = edits.merge(GENERATED, BREAKS, data)
+    assert breaks[0] == 3.0                           # lines 0-1: generated
+    assert breaks[1] == pytest.approx((4.9 + 9.0) / 2)  # lines 1-2: re-centred
+
+
+def test_an_edit_follows_its_line_when_lines_are_added_above(tmp_path):
+    data = edited((1, [note(4.0)]))
+    edits.save(tmp_path, data)
+    moved = edits.load(tmp_path, [["zero"]] + LYRICS)
+    assert list(moved["lines"]) == ["2"] and moved["lines"]["2"]["line_index"] == 2
+    assert moved["stale"] == []
+    assert json.loads((tmp_path / edits.EDITS_FILE).read_text())["lyrics"][0] == ["zero"]
+
+
+def test_an_edit_whose_words_changed_is_set_aside(tmp_path):
+    edits.save(tmp_path, edited((1, [note(4.0)]), (2, [note(8.0)])))
+    changed = [LYRICS[0], ["three", "more"], LYRICS[2]]
+    data = edits.load(tmp_path, changed)
+    assert list(data["lines"]) == ["2"]
+    assert [s["words"] for s in data["stale"]] == [["three", "four"]]
+
+
+def test_validate_notes_checks_and_orders():
+    assert [n["start"] for n in edits.validate_notes([note(2.0), note(1.0)])] == [1.0, 2.0]
+    for bad in (note(1.0, kind="X"), note(1.0, dur=0), note(-1.0), note(1.0, text=" "),
+                {"start": 1.0}):
+        with pytest.raises(ValueError):
+            edits.validate_notes([bad])
+
+
+def test_saving_no_edits_removes_the_file(tmp_path):
+    edits.save(tmp_path, edited((0, [note(1.0)])))
+    data = edits.load(tmp_path, LYRICS)
+    edits.clear_line(data, 0)
+    edits.save(tmp_path, data)
+    assert not (tmp_path / edits.EDITS_FILE).exists()
+
+
+def _write_generated(work, notes, warnings):
+    (work / "notes_sec.json").write_text(json.dumps(
+        {"notes": notes, "breaks": BREAKS, "warnings": warnings}))
+
+
+def test_finalize_without_edits_passes_warnings_through(tmp_path):
+    _write_generated(tmp_path, GENERATED, ["anything at all"])
+    final = edits.finalize(tmp_path, LYRICS)
+    assert final["warnings"] == ["anything at all"]
+    assert final["edited_lines"] == [] and final["stale"] == []
+    assert json.loads((tmp_path / edits.FINAL_FILE).read_text()) == final
+
+
+def test_finalize_stops_reporting_a_line_the_user_fixed(tmp_path):
+    notes = [gen(1.0, 0), gen(4.0, 1, low=True), gen(8.0, 2)]
+    _write_generated(tmp_path, notes, ["1 low-confidence note near 0:04–0:04",
+                                       "missing lyrics near 0:20–0:25"])
+    edits.save(tmp_path, edited((1, [note(4.0)])))
+    final = edits.finalize(tmp_path, LYRICS)
+    assert final["warnings"] == ["missing lyrics near 0:20–0:25"]
+    assert final["edited_lines"] == [1]
+
+
+def test_finalize_reports_stale_edits(tmp_path):
+    _write_generated(tmp_path, GENERATED, [])
+    edits.save(tmp_path, edited((1, [note(4.0)])))
+    final = edits.finalize(tmp_path, [LYRICS[0], ["other"], LYRICS[2]])
+    assert final["warnings"] == ["1 edited line no longer matches the lyrics; "
+                                 "set aside in the editor"]

tests/test_grid.py

@@ -169,7 +169,7 @@ def test_grid_stage_writes_song_txt(tmp_path, monkeypatch):
     monkeypatch.setattr(grid_stage_mod.grid, "detect_bpm", lambda p: 120.0)
     ctx = make_ctx(work, language="English", genre="Pop", year=1999)
     stage = grid_stage_mod.grid_stage()
-    assert stage.outputs(ctx) == [work / "song.txt"]
+    assert stage.outputs(ctx) == [work / "song.txt", work / "notes_final.json"]
     stage.run(ctx)
     song = parse_ultrastar((work / "song.txt").read_text(encoding="utf-8"))
     assert [k for k, _ in song.headers] == [
@@ -257,3 +257,45 @@ def test_grid_stage_reports_no_notes_as_pipeline_error(tmp_path, monkeypatch):
 def test_grid_stage_registered():
     names = [s.name for s in all_stages()]
     assert names[-1] == "package" and names[-2] == "grid"
+
+
+def _regenerate(work, pitch):
+    """What a rebuild leaves: fresh notes for both lines, at `pitch`."""
+    (work / "notes_sec.json").write_text(json.dumps({
+        "notes": [{**note(1.0, 0.5, pitch, "one"), "line_index": 0},
+                  {**note(3.0, 0.5, pitch, "three"), "line_index": 1}],
+        "breaks": [2.0], "warnings": []}))
+
+
+def test_an_edited_line_survives_a_rebuild(tmp_path, monkeypatch):
+    from songtrace import edits
+    work = tmp_path / "w"
+    write_work(work)
+    monkeypatch.setattr(grid_stage_mod.grid, "detect_bpm", lambda p: 120.0)
+    ctx = make_ctx(work)
+    ctx.inp.lyrics_text = "one\nthree"
+    data = edits.load(work, [["one"], ["three"]])
+    edits.set_line(data, 1, [{"start": 3.2, "duration": 0.6, "pitch": 9,
+                              "text": "three", "kind": "*"}])
+    edits.save(work, data)
+    for regenerated_pitch in (0, 4):
+        _regenerate(work, regenerated_pitch)
+        grid_stage_mod.grid_stage().run(ctx)
+        song = parse_ultrastar((work / "song.txt").read_text(encoding="utf-8"))
+        notes = [n for n in song.body if isinstance(n, Note)]
+        assert [(n.kind, n.pitch, n.text) for n in notes] == [
+            (":", regenerated_pitch, "one"), ("*", 9, "three")]
+
+
+def test_rechart_replaces_the_notes_and_keeps_headers_and_tempo(tmp_path):
+    from songtrace.grid import rechart
+    from songtrace.ultrastar import serialize_ultrastar
+    txt = tmp_path / "song.txt"
+    txt.write_text(serialize_ultrastar(to_ultrastar(
+        [note(1.0, 0.5)], [], 300.0,
+        [("TITLE", "Song"), ("VIDEO", "Band - Song.mp4")])), encoding="utf-8")
+    rechart(txt, [{**note(2.0, 0.5, 7, "new"), "kind": "F"}], [])
+    song = parse_ultrastar(txt.read_text(encoding="utf-8"))
+    assert [k for k, _ in song.headers] == ["TITLE", "VIDEO", "BPM", "GAP"]
+    assert get_header(song, "BPM") == "300" and get_header(song, "GAP") == "2000"
+    assert [(n.kind, n.pitch, n.text) for n in song.body] == [("F", 7, "new")]

tests/test_notes.py

@@ -4,7 +4,7 @@ import numpy as np
 import pytest
 from songtrace.config import Config
 from songtrace.notes import (
-    MIN_NOTE_SEC, TimedNote, build_notes, hz_to_ultrastar, segment_span,
+    MIN_NOTE_SEC, TimedNote, _warnings, build_notes, hz_to_ultrastar, segment_span,
     syllabify,
 )
 from songtrace.pipeline import Ctx, SongInput
@@ -127,11 +127,36 @@ def test_build_notes_unvoiced_word_flagged():
     assert warnings[0] == "1 low-confidence note near 0:01–0:01"
 
 
-def test_build_notes_low_score_word_flagged():
+def _scored_rows(scores, dur=0.4):
+    return [{"word": w, "start": 1.0 + i, "end": 1.0 + i + dur, "score": sc,
+             "line_index": 0}
+            for i, (w, sc) in enumerate(zip(["one", "two", "three", "four", "five"], scores))]
+
+
+def test_build_notes_flags_a_word_scoring_far_below_the_rest():
     times, f0, conf = flat_pitch()
-    rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.1, "line_index": 0}]
-    notes, _, warnings = build_notes(rows, times, f0, conf, "English")
-    assert all(n.low_confidence for n in notes) and len(warnings) == 1
+    notes, _, warnings = build_notes(_scored_rows([0.4, 0.4, 0.05, 0.4, 0.4]),
+                                     times, f0, conf, "English")
+    assert [n.low_confidence for n in notes] == [False, False, True, False, False]
+    assert len(warnings) == 1
+
+
+def test_build_notes_judges_scores_against_the_song_not_a_fixed_cutoff():
+    # sung vocals score low throughout; uniformly low is not a reason to flag
+    times, f0, conf = flat_pitch()
+    notes, _, warnings = build_notes(_scored_rows([0.05] * 5), times, f0, conf, "English")
+    assert not any(n.low_confidence for n in notes) and warnings == []
+
+
+def test_build_notes_flags_words_too_short_or_too_long_to_be_sung():
+    times, f0, conf = flat_pitch()
+    for dur in (0.05, 3.5):
+        rows = _scored_rows([0.4] * 3)
+        rows[1]["end"] = rows[1]["start"] + dur
+        rows[2]["start"], rows[2]["end"] = rows[1]["end"] + 0.1, rows[1]["end"] + 0.5
+        notes, _, _ = build_notes(rows, times, f0, conf, "English")
+        # flat pitch: one note per word
+        assert [n.low_confidence for n in notes] == [False, True, False], dur
 
 
 def test_build_notes_unvoiced_word_inherits_previous_pitch():
@@ -334,3 +359,13 @@ def test_notes_carry_line_index():
     rows = [{"word": "hey", "start": 1.0, "end": 1.4, "score": 0.9, "line_index": 3}]
     notes, _, _ = build_notes(rows, times, f0, conf, "English")
     assert notes[0].line_index == 3
+
+
+def test_warnings_group_flagged_notes_close_together():
+    def note(start, low):
+        return TimedNote(start, 0.3, 0, "la", low, 0)
+    notes = [note(10.0, True), note(10.5, False), note(11.0, True),   # 0.7s apart
+             note(20.0, True)]                                       # far from both
+    assert _warnings(notes) == [
+        "2 low-confidence notes near 0:10–0:11",
+        "1 low-confidence note near 0:20–0:20"]

tests/test_transcribe.py

@@ -1,6 +1,7 @@
 from songtrace.lyrics import normalize_lyrics
 from songtrace.whisper_client import TranscriptSegment
-from songtrace.stages.transcribe import classify_lyrics_match, lyrics_similarity, vocal_sections
+from songtrace.stages.transcribe import (classify_lyrics_match, lyrics_similarity,
+                                         recovered_spans, vocal_sections)
 
 def seg(s, e, t=""):
     return TranscriptSegment(start=s, end=e, text=t)
@@ -64,3 +65,31 @@ def test_classify_lyrics_match_scopes_missing_span_to_adlib_not_whole_section():
     assert missing == [(5.0, 9.0)]
     for a, b in missing:
         assert not (a <= 1.0 < b)        # matched line's start (1.0) untouched
+
+
+THREE_LINES = "one two three\nfour five six\nseven eight nine"
+
+
+def test_lines_whisper_missed_are_kept_when_singing_went_untranscribed():
+    ref = normalize_lyrics(THREE_LINES)
+    segs = [seg(0, 2, "one two three"), seg(20, 22, "seven eight nine")]
+    assert classify_lyrics_match(segs, ref)[0] == [1]
+    dropped, _ = classify_lyrics_match(segs, ref, unheard=[(8.0, 14.0)])
+    assert dropped == []
+
+
+def test_untranscribed_singing_elsewhere_does_not_keep_a_line():
+    ref = normalize_lyrics(THREE_LINES)
+    segs = [seg(0, 2, "one two three"), seg(20, 22, "seven eight nine")]
+    dropped, _ = classify_lyrics_match(segs, ref, unheard=[(30.0, 35.0)])
+    assert dropped == [1]
+
+
+def test_recovered_spans_are_the_singing_between_heard_neighbours():
+    ref = normalize_lyrics(THREE_LINES)
+    segs = [seg(10, 12, "four five six"), seg(20, 22, "seven eight nine")]
+    unheard = [(3.0, 9.0), (14.0, 18.0), (30.0, 35.0)]
+    # line 0 comes before anything heard, so only the singing before line 1
+    # can stand in for it; the other spans have no missing line around them
+    assert recovered_spans(segs, ref, unheard) == [(3.0, 9.0)]
+    assert classify_lyrics_match(segs, ref, unheard=unheard)[0] == []

tests/test_web.py

@@ -1412,13 +1412,16 @@ def test_review_endpoint_groups_notes_by_line(tmp_path):
     wait_status(client, job_id, "done")
     body = client.get(f"/api/jobs/{job_id}/review").json()
     assert body["lines"] == [
-        {"index": 0, "text": "hi", "dropped": False,
-         "notes": [{"start": 1.0, "end": 1.4, "pitch": 0, "low_confidence": False}],
+        {"index": 0, "text": "hi", "dropped": False, "edited": False,
+         "notes": [{"start": 1.0, "end": 1.4, "pitch": 0, "low_confidence": False,
+                    "text": "hi", "kind": ":"}],
          "words": []},
-        {"index": 1, "text": "there", "dropped": False,
-         "notes": [{"start": 5.0, "end": 5.3, "pitch": 2, "low_confidence": True}],
+        {"index": 1, "text": "there", "dropped": False, "edited": False,
+         "notes": [{"start": 5.0, "end": 5.3, "pitch": 2, "low_confidence": True,
+                    "text": "there", "kind": ":"}],
+         "words": []},
+        {"index": 2, "text": "gone", "dropped": True, "edited": False, "notes": [],
          "words": []},
-        {"index": 2, "text": "gone", "dropped": True, "notes": [], "words": []},
     ]
     assert body["missing_spans"] == [{"start": 8.0, "end": 9.5}]
 
@@ -1852,3 +1855,125 @@ def test_rebuild_all_reports_a_selected_job_that_was_since_rebuilt(tmp_path):
     body = client.post("/api/rebuild-all", data={"force_from": "pitch", "ids": [old]}).json()
     assert body == {"ids": [], "skipped": [
         {"id": old, "reason": "no longer the current job for its song"}]}
+
+
+def test_a_job_is_only_reported_finished_once_it_has_finished(tmp_path, monkeypatch):
+    """Done must mean done: a slow upload cleanup used to show a 'done' job
+    with its upload still there and no finished_at."""
+    from songtrace import web
+    real_cleanup = web._cleanup_uploads
+
+    def slow_cleanup(*args):
+        time.sleep(0.3)
+        real_cleanup(*args)
+
+    monkeypatch.setattr(web, "_cleanup_uploads", slow_cleanup)
+    client, cfg = make_client(tmp_path, ok_runner)
+    job_id = client.post("/api/jobs", data={"title": "T", "artist": "A", "lyrics": "x"},
+                         files={"video": ("v.mp4", io.BytesIO(b"data"), "video/mp4")}
+                         ).json()["id"]
+    job = wait_status(client, job_id, "done")
+    assert job["finished_at"] is not None
+    assert not (cfg.work_dir / "uploads" / job_id / "v.mp4").exists()
+
+
+def chart_runner(inp, cfg, log=print, force_from=None, stages=None):
+    """Leaves what a real build does for the editor: generated notes and a
+    chart in the work dir, and the packaged chart."""
+    from songtrace.grid import to_ultrastar
+    from songtrace.ultrastar import serialize_ultrastar
+    work = cfg.work_dir / song_slug(inp)
+    work.mkdir(parents=True, exist_ok=True)
+    notes = [{"start": 1.0, "duration": 0.5, "pitch": 0, "text": "one",
+              "low_confidence": False, "line_index": 0},
+             {"start": 3.0, "duration": 0.5, "pitch": 2, "text": "three",
+              "low_confidence": True, "line_index": 1}]
+    warnings = ["1 low-confidence note near 0:03–0:03"]
+    (work / "notes_sec.json").write_text(json.dumps(
+        {"notes": notes, "breaks": [2.0], "warnings": warnings}))
+    chart = serialize_ultrastar(to_ultrastar(notes, [2.0], 300.0, [("TITLE", inp.title)]))
+    (work / "song.txt").write_text(chart, encoding="utf-8")
+    out = cfg.output_dir / song_slug(inp)
+    out.mkdir(parents=True, exist_ok=True)
+    (out / f"{song_slug(inp)}.txt").write_text(chart, encoding="utf-8")
+    return PipelineResult(output_dir=out, warnings=warnings)
+
+
+def make_charted_job(client, lyrics="one\nthree"):
+    job_id = client.post("/api/jobs", data={
+        "title": "T", "artist": "A", "lyrics": lyrics, "youtube_url": "http://y"}).json()["id"]
+    wait_status(client, job_id, "done")
+    return job_id
+
+
+EDIT = {"notes": [{"start": 3.1, "duration": 0.5, "pitch": 9, "text": "three",
+                   "kind": "*"}]}
+
+
+def test_saving_a_line_rewrites_both_charts(tmp_path):
+    from songtrace.ultrastar import Note, parse_ultrastar
+    client, cfg = make_client(tmp_path, chart_runner)
+    job_id = make_charted_job(client)
+    resp = client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT)
+    assert resp.status_code == 200
+    line = resp.json()["lines"][1]
+    assert line["edited"] and line["notes"][0]["kind"] == "*" and line["notes"][0]["pitch"] == 9
+    assert resp.json()["grid"] == {"bpm": 300.0, "gap_ms": 1000.0}
+    for txt in (cfg.work_dir / "A - T" / "song.txt", cfg.output_dir / "A - T" / "A - T.txt"):
+        notes = [n for n in parse_ultrastar(txt.read_text(encoding="utf-8")).body
+                 if isinstance(n, Note)]
+        assert [(n.kind, n.pitch) for n in notes] == [(":", 0), ("*", 9)], txt
+    # the fixed line was the one being warned about
+    assert client.get(f"/api/jobs/{job_id}").json()["warnings"] == []
+
+
+def test_reverting_a_line_returns_it_to_the_generated_notes(tmp_path):
+    client, _ = make_client(tmp_path, chart_runner)
+    job_id = make_charted_job(client)
+    client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT)
+    resp = client.delete(f"/api/jobs/{job_id}/lines/1")
+    assert resp.status_code == 200
+    line = resp.json()["lines"][1]
+    assert not line["edited"] and line["notes"][0]["pitch"] == 2
+    assert client.delete(f"/api/jobs/{job_id}/lines/1").status_code == 404
+
+
+def test_line_edits_are_checked(tmp_path):
+    client, _ = make_client(tmp_path, chart_runner)
+    job_id = make_charted_job(client)
+    assert client.put(f"/api/jobs/{job_id}/lines/5", json=EDIT).status_code == 404
+    bad = {"notes": [{**EDIT["notes"][0], "kind": "X"}]}
+    assert client.put(f"/api/jobs/{job_id}/lines/1", json=bad).status_code == 400
+    assert client.put(f"/api/jobs/{job_id}/lines/1", json={"x": 1}).status_code == 400
+
+
+def test_a_song_cannot_be_edited_while_it_is_being_rebuilt(tmp_path):
+    gate = threading.Event()
+
+    def runner(inp, cfg, log=print, force_from=None, stages=None):
+        if force_from:
+            gate.wait(timeout=5)
+        return chart_runner(inp, cfg)
+
+    client, _ = make_client(tmp_path, runner)
+    try:
+        job_id = make_charted_job(client)
+        client.post(f"/api/jobs/{job_id}/rebuild-full")
+        assert client.put(f"/api/jobs/{job_id}/lines/1", json=EDIT).status_code == 409
+    finally:
+        gate.set()
+
+
+def test_a_stale_edit_can_be_discarded(tmp_path):
+    from songtrace import edits
+    client, cfg = make_client(tmp_path, chart_runner)
+    first = make_charted_job(client)
+    client.put(f"/api/jobs/{first}/lines/1", json=EDIT)
+    # the same song again with that line's words changed
+    second = make_charted_job(client, lyrics="one\nthree more")
+    work = cfg.work_dir / "A - T"
+    assert len(edits.load(work, [["one"], ["three", "more"]])["stale"]) == 1
+    resp = client.delete(f"/api/jobs/{second}/stale/0")
+    assert resp.status_code == 200 and resp.json()["stale"] == []
+    assert not (work / edits.EDITS_FILE).exists()
+    assert client.delete(f"/api/jobs/{second}/stale/0").status_code == 404