Upgrade BeatForge Studio and add authenticated MCP connector

.codex/skills/beat-saber-mapping/scripts/choreography.py

@@ -25,6 +25,7 @@
 )
 from kinematics import bomb_on_swing_path, double_is_safe, swing_pose, transition_finding
 from official_corpus import map_features, normalize_beatmap
+from mapping_plan import build_section_plan, normalize_mapping_plan
 from safety_contract import (
     RECOVERY_BEATS,
     hold_span_beats,
@@ -184,6 +185,8 @@ def select_intents(analysis: dict[str, Any], sections: dict[str, Any], difficult
         )
         gap = config.peak_min_gap if is_peak else config.min_gap
         adaptive_gap = max(gap, gap * (1.15 - 0.35 * intensity))
+        density = float(section.get("density", 1.0))
+        adaptive_gap = max(gap, adaptive_gap / max(0.35, density))
         if beat - last_beat + 1e-9 < adaptive_gap:
             continue
         if any(
@@ -194,13 +197,16 @@ def select_intents(analysis: dict[str, Any], sections: dict[str, Any], difficult
             continue
         strength = float(event.get("strength", 0.0))
         energies = event.get("stemEnergy") if isinstance(event.get("stemEnergy"), dict) else {}
+        dominant = str(section.get("dominantInstrument", "auto"))
+        if dominant != "auto" and dominant in energies:
+            strength *= 0.7 + 1.1 * max(0.0, min(1.0, float(energies[dominant])))
         layer = str(event.get("layer") or "")
         instrument = float(energies.get("drums") or 0.0) + float(energies.get("bass") or 0.0) + float(energies.get("guitar") or 0.0)
         vocal = float(energies.get("vocals") or 0.0)
         layered = bool(energies) and instrument >= vocal * 1.15 and layer in {"drums", "bass", "guitar", "piano", "other"}
         if difficulty in {"Expert", "ExpertPlus"} and layered:
             strength *= 1.35
-        threshold = base_threshold
+        threshold = base_threshold / max(0.35, density)
         if is_peak:
             threshold *= 0.88 if difficulty in {"Expert", "ExpertPlus"} else 0.94
         elif intensity < 0.4:
@@ -256,6 +262,15 @@ def select_intents(analysis: dict[str, Any], sections: dict[str, Any], difficult
             "layer": layer or "mix",
             "stemEnergy": energies,
         }
+        if "motifId" in section:
+            relative = beat - float(section.get("startBeat", 0.0))
+            motif_offset = int(section.get("motifOffset", 0))
+            payload.update({
+                "sectionId": section["id"], "motifId": section["motifId"],
+                "patternBeat": relative + motif_offset * 4.0,
+                "patternIndex": int(round(relative * 2.0)) + motif_offset,
+                "style": section.get("style", "balanced"),
+            })
         selected.append(payload)
         last_beat = beat
     return selected
@@ -433,15 +448,17 @@ def _candidate_notes(
     index: int,
     intensity: float,
     beat: float,
+    preference_beat: float | None = None,
 ) -> list[dict[str, Any]]:
     """Enumerate combo-first arrow poses. Dots are a rare reset, not the default cut."""
 
     if color == 0:
         home = (0, 1)
     else:
         home = (3, 2)
-    preferred = swing_family(color, parity, beat)
-    lead = cut_for(color, parity, index, beat)
+    phase = beat if preference_beat is None else preference_beat
+    preferred = swing_family(color, parity, phase)
+    lead = cut_for(color, parity, index, phase)
     notes: list[dict[str, Any]] = []
     seen: set[tuple[int, int, int]] = set()
 
@@ -627,9 +644,14 @@ def _pose_domain(
     intensity = float(intent.get("intensity", 0.5))
     kind = intent["kind"]
     duration = float(intent.get("duration") or 0.0)
-    heads = _candidate_notes(color, parity, index, intensity, beat)
+    index = int(intent.get("patternIndex", index)) + int(intent.get("candidateVariant", 0))
+    phase = float(intent.get("patternBeat", beat))
+    heads = _candidate_notes(color, parity, index, intensity, beat, preference_beat=phase)
+    if intent.get("style") in {"flow", "chill"}:
+        home_x = 0.5 if color == 0 else 2.5
+        heads.sort(key=lambda head: abs(float(head["x"]) - home_x) + (0.5 if int(head["d"]) in {2, 3, 8} else 0.0))
     if str(intent.get("layer") or "") == "combo-stack":
-        lead = cut_for(color, parity, index, beat)
+        lead = cut_for(color, parity, index, phase)
         if color == 0:
             pair_cells = ((1, 1), (0, 1), (1, 0), (1, 2))
         else:
@@ -667,7 +689,9 @@ def _joint_combined_domain(
 
     beat = float(intent["beat"])
     stack = str(intent.get("layer") or "") == "combo-stack"
-    lead_parity = int(beat // 4.0) % 2 if stack else (index + int(beat // 2.0)) % 2
+    pattern_beat = float(intent.get("patternBeat", beat))
+    pattern_index = int(intent.get("patternIndex", index)) + int(intent.get("candidateVariant", 0))
+    lead_parity = int(pattern_beat // 4.0) % 2 if stack else (pattern_index + int(pattern_beat // 2.0)) % 2
     buckets = {
         (color_id, parity): _pose_domain(intent, color_id, parity, index, bpm, recovery_beats)[:6]
         for color_id in (0, 1)
@@ -2500,19 +2524,22 @@ def score_candidate(database_path: Path | None, features: dict[str, Any]) -> dic
     }
 
 
-def generate_all(
+def _generate_single_pass(
     analysis: dict[str, Any],
     sections: dict[str, Any],
     seed: int,
     corpus_database: Path | None = None,
     difficulties: Sequence[str] | None = None,
+    candidate_variant: int = 0,
 ) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]:
     generated: dict[str, dict[str, Any]] = {}
     report: dict[str, Any] = {"solver": {}, "difficulties": {}, "officialReferences": {}}
     chosen = tuple(name for name in DIFFICULTIES if name in set(difficulties or DIFFICULTIES)) or DIFFICULTIES
     for difficulty in chosen:
         emit_progress("choreography", f"selecting {difficulty} intents from onsets and sections")
         intents = select_intents(analysis, sections, difficulty)
+        if candidate_variant:
+            intents = [{**intent, "candidateVariant": candidate_variant} for intent in intents]
         joint = None
         try:
             emit_progress("choreography", f"solving joint CP-SAT for {difficulty} ({len(intents)} events)")
@@ -2559,3 +2586,97 @@ def generate_all(
         )
         report["difficulties"][difficulty]["corpusRank"] = score_candidate(corpus_database, candidate_features)
     return generated, report
+
+
+def rank_choreography_candidate(
+    beatmap: dict[str, Any], difficulty: str, bpm: float,
+    corpus_rank: dict[str, Any], reference: dict[str, Any],
+    expected_count: int,
+) -> dict[str, Any]:
+    """Score complete candidates only after running the unchanged structural gates."""
+    from beatforge_core import ValidationReport
+    from validate_map import validate_v3
+
+    validation = ValidationReport()
+    validate_v3(beatmap, f"{difficulty}Standard.dat", validation, bpm=bpm, difficulty=difficulty)
+    notes = beatmap.get("colorNotes", [])
+    errors = [issue.code for issue in validation.errors]
+    if expected_count and not notes:
+        errors.append("NO_PLAYABLE_NOTES")
+    events, _ = normalize_beatmap(beatmap)
+    features = map_features(events, bpm)
+    profile = reference.get("profile", {})
+    comparable = [name for name in ("p95Reach", "centerVisionRate", "medianSameHandGap") if name in profile]
+    profile_distance = sum(abs(float(features.get(name, 0.0)) - float(profile[name])) / max(0.25, abs(float(profile[name]))) for name in comparable) / max(1, len(comparable))
+    reference_score = 1.0 / (1.0 + profile_distance) if comparable else 0.5
+    corpus_score = float(corpus_rank.get("score", 50.0)) / 100.0
+    coverage = min(1.0, len(notes) / max(1, expected_count))
+    signatures = [(int(note["x"]), int(note["y"]), int(note["d"])) for note in notes]
+    repeats = sum(left == right for left, right in zip(signatures, signatures[1:]))
+    variety = 1.0 - repeats / max(1, len(signatures) - 1)
+    score = 100.0 * (0.4 * corpus_score + 0.25 * reference_score + 0.25 * coverage + 0.1 * variety)
+    return {
+        "score": round(score, 6), "hardErrors": sorted(set(errors)),
+        "eligible": not errors, "corpusScore": round(corpus_score * 100.0, 6),
+        "referenceScore": round(reference_score * 100.0, 6),
+        "eventCoverage": round(coverage, 6), "variety": round(variety, 6),
+        "features": features,
+    }
+
+
+def select_ranked_candidate(candidates: list[dict[str, Any]]) -> dict[str, Any]:
+    eligible = [candidate for candidate in candidates if candidate.get("eligible")]
+    if not eligible:
+        codes = sorted({code for candidate in candidates for code in candidate.get("hardErrors", [])})
+        raise ValueError("No candidate passed the unchanged safety gates: " + ", ".join(codes))
+    return max(eligible, key=lambda candidate: (candidate["score"], -candidate["index"]))
+
+
+def generate_all(
+    analysis: dict[str, Any], sections: dict[str, Any], seed: int,
+    corpus_database: Path | None = None,
+    difficulties: Sequence[str] | None = None,
+    mapping_plan: dict[str, Any] | None = None,
+) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]:
+    if mapping_plan is None:
+        return _generate_single_pass(analysis, sections, seed, corpus_database, difficulties)
+    controls = normalize_mapping_plan(mapping_plan)
+    planned = build_section_plan(analysis, sections, controls)
+    chosen = tuple(name for name in DIFFICULTIES if name in set(difficulties or DIFFICULTIES)) or DIFFICULTIES
+    generated: dict[str, dict[str, Any]] = {}
+    report: dict[str, Any] = {"solver": {}, "difficulties": {}, "officialReferences": {}, "candidateRanking": {}, "mappingPlan": planned}
+    for difficulty in chosen:
+        # Retrieve before generation so every candidate is compared with the same reference family.
+        reference = retrieve_official_references(corpus_database, difficulty, float(analysis["bpm"]))
+        intents = select_intents(analysis, planned, difficulty)
+        candidates = []
+        payloads: dict[int, tuple[dict[str, Any], dict[str, Any]]] = {}
+        for index in range(controls["candidateCount"]):
+            emit_progress("choreography", f"{difficulty}: generating candidate {index + 1}/{controls['candidateCount']}")
+            try:
+                maps, trace = _generate_single_pass(analysis, planned, seed + index * 7919, corpus_database, (difficulty,), candidate_variant=index)
+                beatmap = maps[difficulty]
+                if controls["noBombs"]:
+                    beatmap["bombNotes"] = []
+                if controls["noWalls"]:
+                    beatmap["obstacles"] = []
+                ranked_map = beatmap
+                if len(analysis.get("beatGrid", [])) >= 2:
+                    from timing_export import project_map_timing
+                    ranked_map = project_map_timing(beatmap, analysis)
+                ranked_events, _ = normalize_beatmap(ranked_map)
+                corpus_rank = score_candidate(corpus_database, map_features(ranked_events, float(analysis["bpm"])))
+                rank = rank_choreography_candidate(ranked_map, difficulty, float(analysis["bpm"]), corpus_rank, reference, len(intents))
+                trace["difficulties"][difficulty]["corpusRank"] = corpus_rank
+                candidates.append({"index": index, **rank})
+                payloads[index] = (beatmap, trace)
+            except (RuntimeError, ValueError) as error:
+                candidates.append({"index": index, "eligible": False, "score": 0.0, "hardErrors": [str(error)]})
+        winner = select_ranked_candidate(candidates)
+        beatmap, trace = payloads[winner["index"]]
+        generated[difficulty] = beatmap
+        report["solver"][difficulty] = trace["solver"][difficulty]
+        report["difficulties"][difficulty] = trace["difficulties"][difficulty]
+        report["officialReferences"][difficulty] = reference
+        report["candidateRanking"][difficulty] = {"selectedIndex": winner["index"], "candidates": candidates}
+    return generated, report

.codex/skills/beat-saber-mapping/scripts/generate_map.py

@@ -17,6 +17,8 @@
 
 from beatforge_core import DIFFICULTIES, TOOL_VERSION, default_cache_dir, difficulty_rank, emit_progress, encode_ogg, find_ffmpeg, load_json, parse_difficulties, sha256_file, write_json
 from choreography import CONFIGS, generate_all
+from mapping_plan import build_section_plan, normalize_mapping_plan
+from timing_export import project_map_timing, project_plan_timing
 from artwork import derive_palette, extract_embedded_artwork, lookup_itunes_cover, lookup_release_cover, palette_from_rgb, write_palette_cover
 from official_corpus import corpus_is_fresh, detect_game_root, sync as sync_official_corpus
 
@@ -373,6 +375,7 @@ def main() -> int:
     parser.add_argument("--profile", default="official-premium", choices=["official-premium", "official-rl"])
     parser.add_argument("--engine", default="cp-sat", choices=["cp-sat", "rl", "hybrid"], help="Choreography engine to use (cp-sat, rl, hybrid)")
     parser.add_argument("--rl-model", type=Path, help="Path to trained PyTorch RL policy checkpoint")
+    parser.add_argument("--mapping-plan", type=Path, help="Version 1 structured creative controls JSON")
     parser.add_argument("--environment", default=None, help="Beat Saber environment name (e.g. DaftPunkEnvironment, DefaultEnvironment)")
     parser.add_argument("--full-spread", action="store_true", default=True)
     parser.add_argument(
@@ -393,6 +396,13 @@ def main() -> int:
     parser.add_argument("--game-root", type=Path, help="Beat Saber install root; auto-detected on common Oculus/Steam paths")
     parser.add_argument("--seed", type=int)
     args = parser.parse_args()
+    try:
+        creative_controls = normalize_mapping_plan(load_json(args.mapping_plan)) if args.mapping_plan else None
+        if (args.engine == "rl" or args.profile == "official-rl") and (args.rl_model is None or not args.rl_model.is_file()):
+            raise ValueError("The RL engine requires an existing trained checkpoint supplied with --rl-model")
+    except (ValueError, OSError) as error:
+        print(f"ERROR: {error}", file=sys.stderr)
+        return 2
     if not args.audio.is_file():
         print(f"ERROR: audio file does not exist: {args.audio}", file=sys.stderr)
         return 2
@@ -423,6 +433,7 @@ def main() -> int:
         emit_progress("timing", "reusing existing analysis.json")
     analysis = load_json(analysis_file)
     sections = load_json(sections_file)
+    creative_plan = build_section_plan(analysis, sections, creative_controls) if creative_controls is not None else None
     unconfirmed = bool(args.continue_unconfirmed)
     if analysis.get("status") != "timing_verified":
         if not unconfirmed:
@@ -496,54 +507,33 @@ def main() -> int:
     emit_progress("choreography", f"solving {', '.join(chosen)}")
     if args.engine == "rl" or args.profile == "official-rl":
         from rl.policy_generator import RLMapGenerator
+        from rl.features import load_analysis_features
 
         rl_gen = RLMapGenerator(model_path=args.rl_model)
-        raw_bg = load_json(analysis_dir / "beat_grid.json") if (analysis_dir / "beat_grid.json").exists() else None
-        if isinstance(raw_bg, dict) and "tempoRegions" in raw_bg:
-            max_beat = max((float(r.get("endBeat", 0.0)) for r in raw_bg.get("tempoRegions", [])), default=0.0)
-            beat_grid = [round(i * 0.25, 4) for i in range(max(16, int(max_beat * 4)))]
-        elif isinstance(raw_bg, list):
-            beat_grid = [float(b) for b in raw_bg]
-        else:
-            duration_s = float(analysis.get("durationSeconds", analysis.get("duration", 240.0)))
-            bpm_val = float(analysis.get("bpm", 125.0))
-            beat_grid = [round(i * 0.25, 4) for i in range(max(16, int(duration_s * bpm_val / 60.0 * 4)))]
-
-        grid_len = len(beat_grid)
-        try:
-            import numpy as np
-            from beatforge_core import frame_features, load_audio
-            audio_buf = load_audio(args.audio)
-            feat = frame_features(audio_buf.samples)
-            flux = feat["flux"]
-            onsets = np.interp(np.linspace(0, len(flux), grid_len), np.arange(len(flux)), flux)
-            onsets = ((onsets - np.min(onsets)) / (np.max(onsets) - np.min(onsets) + 1e-6)).tolist()
-        except Exception:
-            onsets = [0.5] * grid_len
-
-        audio_features = {
-            "onsets": onsets,
-            "flux": onsets,
-            "stems": {
-                "drums": onsets,
-                "bass": onsets,
-                "vocals": [o * 0.7 for o in onsets],
-                "guitar": [o * 0.5 for o in onsets],
-                "piano": [o * 0.5 for o in onsets],
-                "other": [o * 0.4 for o in onsets],
-            },
-            "sections": [s.get("type", "verse") for s in sections] if isinstance(sections, list) else ["verse"] * grid_len,
-        }
+        bundle = load_analysis_features(analysis_dir, require_verified=not unconfirmed)
         maps = {}
         for diff in chosen:
-            maps[diff] = rl_gen.generate_difficulty(audio_features, beat_grid, float(analysis["bpm"]), difficulty=diff)
+            maps[diff] = rl_gen.generate_difficulty(bundle.audio_features, bundle.beat_grid, bundle.bpm, difficulty=diff, seed=seed)
+            if creative_controls and creative_controls["noBombs"]:
+                maps[diff]["bombNotes"] = []
+            if creative_controls and creative_controls["noWalls"]:
+                maps[diff]["obstacles"] = []
         choreography_report = {
             "poseSolver": "rl-policy-v1",
             "solver": "rl-policy-v1",
             "difficulties": chosen,
+            "checkpoint": rl_gen.checkpoint_metadata,
+            "features": bundle.provenance,
+            "mappingPlan": creative_plan,
+            "creativeControlSupport": "RL applies noBombs and noWalls only; phrase, density, and style controls require the premium engine",
         }
     else:
-        maps, choreography_report = generate_all(analysis, sections, seed, args.corpus_database, difficulties=chosen)
+        maps, choreography_report = generate_all(analysis, sections, seed, args.corpus_database, difficulties=chosen, mapping_plan=creative_controls)
+    maps = {difficulty: project_map_timing(payload, analysis) for difficulty, payload in maps.items()}
+    if creative_plan is not None:
+        creative_plan = project_plan_timing(creative_plan, analysis)
+        choreography_report["mappingPlan"] = creative_plan
+        write_json(package_analysis_dir / "mapping_plan.json", creative_plan)
     for difficulty, payload in maps.items():
         write_json(args.out / f"{difficulty}Standard.dat", payload)
     emit_progress("audio", "encoding song.ogg for the Beat Saber pack")
@@ -590,6 +580,7 @@ def main() -> int:
             },
         },
         "choreography": choreography_report,
+        "mappingPlan": creative_plan,
         "officialCorpus": {"database": str(args.corpus_database.resolve()), **corpus_report},
         "documentation": {
             "manifest": "_beatforge/documentation_manifest.json",