feat(foundry): window-locked rotation — co-exported stems rotate by ONE shared slot

Refinement caught before re-cutting the real kits: superfreak drums and
bass ship from the SAME window (107.21s, 4 bars). If each stem rotated
to its own strongest attack, their relative groove could shift by whole
beats — bass's phrase-start is not necessarily on the drums' 1, and the
kit is played together (loopAt on a common cycle), so relative phase
between kit loops is musical content.

export_take now derives the downbeat slot ONCE per shared time window
from its most rhythmic stem (drums > bass > other > vocals priority)
and applies that same slot to every stem of the window
(_rotate_to_downbeat force_slot; per-stem roll still zc-snaps within
±10 ms on the stem's own mono ref). Stems at different windows keep
independent rotation. Stem reads are cached (each stem was loaded twice
otherwise).

Test: two-stem synthetic window, drums downbeat at slot 3, other's own
loudest attack at slot 6 — drums' marker must land at 0 and other's at
3 (= 6−3, following drums), not 0. 86 passed.

tools/foundry/engine/loops.py

@@ -473,7 +473,8 @@ ROTATE_ATTACK_WIN_S = 0.030       # attack window each side of a beat slot
 ROTATE_LOWBAND_HZ = 200.0         # downbeat = strongest LOW-band attack (kick-weighted)
 
 
-def _rotate_to_downbeat(clip: np.ndarray, sr: int, n_beats: int) -> tuple[np.ndarray, int, int]:
+def _rotate_to_downbeat(clip: np.ndarray, sr: int, n_beats: int,
+                        force_slot: Optional[int] = None) -> tuple[np.ndarray, int, int]:
     """Rotate a bar-exact, clean-seam loop so the musical downbeat sits at position 0.
 
     PLN's ear-finding (2026-07-11): loops came out 'timed ok but cut BCDA/DABC' — the PLP
@@ -492,6 +493,11 @@ def _rotate_to_downbeat(clip: np.ndarray, sr: int, n_beats: int) -> tuple[np.nda
     winning slot becomes position 0; the exact roll offset is zc-snapped (B1) on the mono
     reference so a re-trigger starts at a zero crossing. Length untouched (np.roll).
 
+    `force_slot` skips the scoring and rotates to that beat slot — used to keep stems
+    that share one time window LOCKED to a common rotation (the downbeat is derived once
+    from the most rhythmic stem of the window and applied to all of them; independent
+    per-stem rotation could shift their relative groove by whole beats).
+
     `clip` is (n,) or (n, ch). Returns (rotated, best_slot, roll_samples); slot 0 ⇒ no-op.
     """
     c2 = clip if clip.ndim == 2 else clip[:, None]
@@ -499,6 +505,7 @@ def _rotate_to_downbeat(clip: np.ndarray, sr: int, n_beats: int) -> tuple[np.nda
     if n_beats <= 1 or n < n_beats * 8:
         return clip, 0, 0
     mono = c2.mean(axis=1).astype(np.float64)
+    if force_slot is None:
         from scipy.signal import butter, sosfiltfilt
         sos = butter(4, ROTATE_LOWBAND_HZ / (sr / 2), btype="low", output="sos")
         low = sosfiltfilt(sos, mono)
@@ -516,9 +523,11 @@ def _rotate_to_downbeat(clip: np.ndarray, sr: int, n_beats: int) -> tuple[np.nda
             score = att_low + 0.25 * att_full
             if score > best_score:
                 best_score, best_k = score, k
+    else:
+        best_k = int(force_slot) % n_beats
     if best_k == 0:
         return clip, 0, 0
-    r = _snap_zc(c2.mean(axis=1), int(round(best_k * n / n_beats)), sr)
+    r = _snap_zc(mono, int(round(best_k * n / n_beats)), sr)
     rotated = np.roll(c2, -r, axis=0)
     return (rotated if clip.ndim == 2 else rotated[:, 0]), best_k, r
 
@@ -576,11 +585,15 @@ def export_take(take: Take, kit: str, workspace: Path, catch: Catch, *,
     catch_dir = catch.dir(workspace)
     stem_path = {st.name: catch_dir / st.path for st in catch.stems}
     written: list[Path] = []
-    for i, sl in enumerate(s for s in take.stems if s.name in keep):
-        src = stem_path.get(sl.name)
-        if not src or not src.exists():
-            continue
-        y, sr = sf.read(str(src), always_2d=True, dtype="float32")   # (n, ch)
+    n_beats = take.bars * BEATS_PER_BAR
+
+    def _load(name: str, _cache={}):
+        if name not in _cache:
+            _cache[name] = sf.read(str(stem_path[name]), always_2d=True, dtype="float32")
+        return _cache[name]
+
+    def _slice(sl: StemSlice) -> tuple[np.ndarray, int]:
+        y, sr = _load(sl.name)
         a = int(sl.start_s * sr)
         mono_ref = y[:, 0]
         # exact bar-multiple length from the take's grid; fall back to the raw span if the
@@ -591,9 +604,30 @@ def export_take(take: Take, kit: str, workspace: Path, catch: Catch, *,
         else:
             b = int(sl.end_s * sr)
             a, b = _snap_zc(mono_ref, a, sr), _snap_zc(mono_ref, b, sr)
-        clip = y[a:b]
+        return y[a:b], sr
+
+    slices = [s for s in take.stems if s.name in keep and
+              (stem_path.get(s.name) or Path("/nonexistent")).exists()]
+
+    # downbeat slot per shared time window, derived ONCE from the most rhythmic stem of
+    # the window (drums > bass > other > vocals) and applied to every stem in it — so
+    # co-exported stems keep their relative groove instead of each rotating to its own 1.
+    group_slot: dict[float, int] = {}
+    if rotate and take.bars > 0:
+        prio = {"drums": 0, "bass": 1, "other": 2, "vocals": 3}
+        for sl in sorted(slices, key=lambda s: prio.get(s.name, 9)):
+            key = round(sl.start_s, 3)
+            if key in group_slot:
+                continue
+            clip, sr = _slice(sl)
+            _, slot, _ = _rotate_to_downbeat(clip, sr, n_beats)
+            group_slot[key] = slot
+
+    for i, sl in enumerate(slices):
+        clip, sr = _slice(sl)
         if rotate and take.bars > 0:                                  # start on the downbeat
-            clip, _slot, _roll = _rotate_to_downbeat(clip, sr, take.bars * BEATS_PER_BAR)
+            slot = group_slot.get(round(sl.start_s, 3), 0)
+            clip, _slot, _roll = _rotate_to_downbeat(clip, sr, n_beats, force_slot=slot)
         if sl.name == "bass" and clip.shape[1] > 1:                   # B6 mono-sum
             clip = clip.mean(axis=1, keepdims=True)
         if peak_norm:                                                 # B10: off by default

tools/foundry/tests/test_loops.py

@@ -181,6 +181,55 @@ def test_rotation_never_breaks_a_clean_seam():
     assert seam > 0.6, "rotating a continuous loop must not introduce a wrap discontinuity"
 
 
+def test_export_rotation_locks_stems_sharing_a_window(tmp_path):
+    # stems exported from the SAME window must rotate by the SAME slot (anchored on
+    # drums), or their relative groove shifts by whole beats. drums' downbeat is at
+    # slot 3; the other stem's own loudest attack is at slot 6 — both must come out
+    # rotated by drums' slot 3, so the other stem's marker lands at slot 3 (6−3), NOT 0.
+    from engine.model import Catch, Stem
+    sr = 22050
+    bpm = 120.0
+    n_beats, beat_s = 8, 0.5
+    total = int(20 * sr)
+    drums = 0.01 * np.random.default_rng(2).standard_normal(total).astype(np.float32)
+    other = 0.01 * np.random.default_rng(3).standard_normal(total).astype(np.float32)
+    win_start = 5.0
+    L_hit = int(0.05 * sr)
+    env = np.exp(-np.linspace(0, 10, L_hit)).astype(np.float32)
+    kick = 0.9 * env * np.sin(2 * np.pi * 60 * np.arange(L_hit) / sr).astype(np.float32)
+    stab = 0.9 * env * np.sin(2 * np.pi * 800 * np.arange(L_hit) / sr).astype(np.float32)
+    drums[int((win_start + 3 * beat_s) * sr):int((win_start + 3 * beat_s) * sr) + L_hit] += kick
+    other[int((win_start + 6 * beat_s) * sr):int((win_start + 6 * beat_s) * sr) + L_hit] += stab
+
+    ws = tmp_path / "ws"
+    cd = ws / "c" / "stems"; cd.mkdir(parents=True)
+    sf.write(str(cd / "drums.wav"), drums, sr, subtype="PCM_24")
+    sf.write(str(cd / "other.wav"), other, sr, subtype="PCM_24")
+    catch = Catch(slug="c", title="t", source_path="s.wav",
+                  stems=[Stem(name="drums", path="stems/drums.wav"),
+                         Stem(name="other", path="stems/other.wav")])
+    take = L.Take(start_s=win_start, end_s=win_start + n_beats * beat_s, bars=2, bpm=bpm,
+                  tempo_unstable=False, score=0.9, n_stems=2,
+                  stems=[L.StemSlice(name="drums", start_s=win_start,
+                                     end_s=win_start + n_beats * beat_s, score=0.9,
+                                     rms_dbfs=-12.0, is_one_shot=False),
+                         L.StemSlice(name="other", start_s=win_start,
+                                     end_s=win_start + n_beats * beat_s, score=0.8,
+                                     rms_dbfs=-12.0, is_one_shot=False)])
+    written = L.export_take(take, "rotkit", ws, catch, keep=("drums", "other"),
+                            samples_root=tmp_path / "Samples")
+    assert len(written) == 2
+    outs = {p.stem.split("_")[1]: p for p in written}
+
+    def marker_slot(path):
+        y, srf = sf.read(str(path), always_2d=True); m = np.abs(y[:, 0])
+        n = len(m)
+        return int(round(int(np.argmax(m)) / (n / n_beats))) % n_beats
+
+    assert marker_slot(outs["drums"]) == 0, "drums downbeat rotated to the loop start"
+    assert marker_slot(outs["other"]) == 3, "other follows drums' rotation (6−3), not its own"
+
+
 def test_kit_multiples_check_flags_off_grid(tmp_path):
     sr = 44100
     bpm = 120.0