Polish Rhythma runtime and verify the redesigned session flow

Finish the redesign by making the session profile drive the actual generation pipeline, locking the full path with an integration-style test, and preserving override behavior for legacy analysis payloads. This turns the earlier design work into a coherent end-to-end session flow instead of a collection of isolated upgrades.

Constraint: Final verification must stay honest about the missing local gradio runtime
Constraint: Legacy analysis payloads without session_profile must still respect explicit overrides
Rejected: Leaving the session profile disconnected from generation | would make the UI and audio disagree about the active session
Confidence: medium
Scope-risk: moderate
Reversibility: clean
Directive: Keep runtime session shaping and UI session copy in sync if later work expands controls or fallbacks
Tested: python -m py_compile app.py rhythma.py rhythma_analysis.py rhythma_engine.py tests/test_rhythma_profiles.py tests/test_rhythma_layered_audio.py tests/test_rhythma_regression.py tests/test_rhythma_ui_copy.py
Tested: PYTEST_DISABLE_PLUGIN_AUTOLOAD=1 python -c "import pytest; raise SystemExit(pytest.main(['tests/test_rhythma_profiles.py','tests/test_rhythma_layered_audio.py','tests/test_rhythma_regression.py','tests/test_rhythma_ui_copy.py','-q','-p','no:cacheprovider']))"
Not-tested: Live Gradio startup in this environment because gradio is not installed locally

app.py

@@ -57,16 +57,35 @@ def generate_modulated_experience(analysis_result, base_freq=None, modulation_ty
         return error_msg, None, None, None, None
 
     emotional_state = analysis_result.get("emotional_state", "neutral")
-    final_rhythm_pattern = rhythm_pattern or analysis_result.get("rhythm_pattern", "calm")
-    final_base_freq = base_freq if base_freq and base_freq > 0 else None
+    session_profile = build_runtime_session_profile(
+        analysis_result,
+        base_freq=base_freq,
+        modulation_type=modulation_type,
+        rhythm_pattern=rhythm_pattern,
+    )
+    final_rhythm_pattern = (
+        session_profile.get("pattern")
+        or rhythm_pattern
+        or analysis_result.get("rhythm_pattern")
+        or "calm"
+    )
+    final_modulation_type = session_profile.get("modulation_type") or modulation_type
+    final_base_freq = session_profile.get("tone_center")
+    if not final_base_freq:
+        final_base_freq = base_freq if base_freq and base_freq > 0 else None
 
     try:
         engine = RhythmaModulationEngine(
             base_freq=final_base_freq,
-            modulation_type=modulation_type,
+            modulation_type=final_modulation_type,
             rhythm_pattern=final_rhythm_pattern,
             emotional_state=emotional_state if not final_base_freq else None,
         )
+        if session_profile:
+            engine.generate_modulated_wave = (
+                lambda requested_duration, _engine=engine, _profile=dict(session_profile):
+                _engine.render_session(_profile, requested_duration)
+            )
 
         timestamp = int(time.time())
         temp_dir = tempfile.gettempdir()
@@ -79,8 +98,18 @@ def generate_modulated_experience(analysis_result, base_freq=None, modulation_ty
 
         fig = engine.visualize_waveform(duration)
         waveform_image = engine.get_waveform_image()
-        analysis_text = engine.get_complete_analysis()
-        symbolic = engine.get_symbolic_interpretation()
+        if session_profile:
+            analysis_text = "\n\n".join(
+                [
+                    f"Session: {session_profile.get('title', 'Session')}",
+                    f"Tone: {session_profile.get('emotional_tone', emotional_state.title())}",
+                    f"Listening Path: {session_profile.get('guidance', '')}",
+                ]
+            )
+            symbolic = session_profile.get("reflection") or engine.get_symbolic_interpretation()
+        else:
+            analysis_text = engine.get_complete_analysis()
+            symbolic = engine.get_symbolic_interpretation()
 
         return analysis_text, saved_audio_path, fig, waveform_image, symbolic
 
@@ -104,6 +133,35 @@ def normalize_rhythm_override(value):
     return value
 
 
+def build_runtime_session_profile(
+    analysis_result,
+    base_freq=None,
+    modulation_type="sine",
+    rhythm_pattern=None,
+):
+    analysis_result = analysis_result if isinstance(analysis_result, dict) else {}
+    profile = analysis_result.get("session_profile") or {}
+    if not profile:
+        return {}
+
+    if symphai_core is not None:
+        return symphai_core.apply_profile_overrides(
+            profile,
+            tone_center=base_freq,
+            modulation_type=modulation_type,
+            session_pattern=rhythm_pattern,
+        )
+
+    shaped_profile = dict(profile)
+    if base_freq is not None and base_freq > 0:
+        shaped_profile["tone_center"] = base_freq
+    if modulation_type:
+        shaped_profile["modulation_type"] = modulation_type
+    if rhythm_pattern:
+        shaped_profile["pattern"] = rhythm_pattern
+    return shaped_profile
+
+
 def build_session_copy(analysis_result):
     analysis_result = analysis_result if isinstance(analysis_result, dict) else {}
     profile = analysis_result.get("session_profile") or {}

tests/test_rhythma_ui_copy.py

@@ -127,3 +127,112 @@ def test_rhythma_experience_degrades_copy_consistently_on_generation_failure(mon
 
     assert outputs[2] == "Generation unavailable"
     assert outputs[6] == "Generation unavailable"
+
+
+def test_rhythma_experience_uses_real_session_profile_across_pipeline(monkeypatch):
+    app = import_app_with_gradio_stub()
+    captured = {}
+
+    class FakeEngine:
+        def __init__(self, **kwargs):
+            captured["init_kwargs"] = kwargs
+            self.generated_audio = None
+
+        def render_session(self, profile, duration):
+            captured["render_profile"] = profile
+            captured["render_duration"] = duration
+            return "layered-session-audio"
+
+        def generate_modulated_wave(self, duration):
+            captured["generated_duration"] = duration
+            return "legacy-generated-audio"
+
+        def save_audio(self, duration, file_path):
+            captured["save_duration"] = duration
+            captured["save_path"] = file_path
+            self.generated_audio = self.generate_modulated_wave(duration)
+            return file_path
+
+        def visualize_waveform(self, duration):
+            captured["visualize_duration"] = duration
+            return "plot"
+
+        def get_waveform_image(self):
+            return "image"
+
+        def get_complete_analysis(self):
+            return "legacy analysis"
+
+        def get_symbolic_interpretation(self):
+            return "legacy symbolic"
+
+    monkeypatch.setattr(app, "RhythmaModulationEngine", FakeEngine)
+
+    outputs = app.rhythma_experience(
+        "I want to focus on deep work",
+        None,
+        override_freq=0,
+        override_modulation="sine",
+        override_rhythm="Automatic",
+        duration=5,
+    )
+
+    assert outputs[0] == "### Clear Horizon"
+    assert outputs[1] == "Attentive and composed"
+    assert outputs[2].startswith("Attentive and composed")
+    assert outputs[6] == "This session narrows motion to support sustained attention."
+    assert outputs[7] == ""
+    assert captured["init_kwargs"] == {
+        "base_freq": 512.0,
+        "modulation_type": "sine",
+        "rhythm_pattern": "focused",
+        "emotional_state": None,
+    }
+    assert captured["render_profile"]["title"] == "Clear Horizon"
+    assert captured["render_profile"]["tone_center"] == 512.0
+    assert captured["render_profile"]["pattern"] == "focused"
+    assert captured["render_duration"] == 5
+    assert captured["save_duration"] == 5
+    assert captured["visualize_duration"] == 5
+
+
+def test_generate_modulated_experience_prefers_explicit_rhythm_override_without_session_profile(monkeypatch):
+    app = import_app_with_gradio_stub()
+    captured = {}
+
+    class FakeEngine:
+        def __init__(self, **kwargs):
+            captured["init_kwargs"] = kwargs
+
+        def save_audio(self, duration, file_path):
+            return file_path
+
+        def visualize_waveform(self, duration):
+            return "plot"
+
+        def get_waveform_image(self):
+            return "image"
+
+        def get_complete_analysis(self):
+            return "legacy analysis"
+
+        def get_symbolic_interpretation(self):
+            return "legacy symbolic"
+
+    monkeypatch.setattr(app, "RhythmaModulationEngine", FakeEngine)
+
+    analysis = {
+        "emotional_state": "neutral",
+        "rhythm_pattern": "calm",
+        "transcription": "",
+    }
+
+    result = app.generate_modulated_experience(
+        analysis,
+        modulation_type="sine",
+        rhythm_pattern="focused",
+        duration=5,
+    )
+
+    assert result[0] == "legacy analysis"
+    assert captured["init_kwargs"]["rhythm_pattern"] == "focused"