Render `_s*` model plots at expected variability + per-sample NDT

The cartoon and trajectory plots for DDM variants with random across-trial
parameters (ddm_sdv, ddm_st, ddm_truncnormt, ddm_rayleight) were showing
misleading information:

1. The noiseless "model cartoon" line rendered at slope `v + epsilon` for
one random epsilon ~ Normal(0, sv) instead of the deterministic skeleton
at slope `v`. Reason: ssm-simulators `no_noise=True` only zeros Brownian
noise, not across-trial variability draws.

2. Sample trajectories all started at the input `t` even when each actual
per-sample NDT was drawn from a random distribution (Uniform/TruncNorm/
Rayleigh). Reason: simulator metadata records the input `t`, not the
per-sample NDT actually drawn from `t_dist`.

GUI-side fixes:
- `_VARIABILITY_COLLAPSE` overrides each `_s*` model's variability params
before the cartoon-only simulator call.
- `_EXPECTED_T_POST_SHIFT` post-adjusts `metadata['t']` for distributions
whose param=0 collapse != expectation (ddm_rayleight: st * sqrt(pi/2)).
- `_patch_trajectory_t_with_actual_ndt` back-derives per-sample NDT from
`rt - decision_time` and overrides `metadata['t']` per trajectory.
- `expected_random_params` flag (default True) on plot_func_model{,_n},
surfaced as a sidebar checkbox in app.py.

Both fixes are GUI-scoped, no ssm-simulators changes needed. TODO
comments mark the lift points for eventual upstream relocation.

Build infra: migrate to uv (pyproject.toml + uv.lock), update Dockerfile
to install via `uv sync --frozen --no-dev` and entrypoint via `uv run`,
drop requirements.txt. Local workflow: `uv sync && uv run pytest`.

Tests: 17 new tests covering helpers and end-to-end behavior on each
affected model.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+.pytest_cache/
+.DS_Store

Dockerfile

@@ -8,13 +8,17 @@ RUN apt-get update && apt-get install -y \
     git \
     && rm -rf /var/lib/apt/lists/*
 
-COPY requirements.txt ./
-COPY src/ ./src/
+# Install uv (multi-stage copy from the official image — no curl/install scripts).
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+# Resolve and install dependencies from the lockfile (production: no dev group).
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev
 
-RUN pip3 install -r requirements.txt
+COPY src/ ./src/
 
 EXPOSE 8501
 
 HEALTHCHECK CMD curl --fail http://localhost:8501/_stcore/health
 
-ENTRYPOINT ["streamlit", "run", "src/app.py", "--server.port=8501", "--server.address=0.0.0.0"]
+ENTRYPOINT ["uv", "run", "--no-sync", "streamlit", "run", "src/app.py", "--server.port=8501", "--server.address=0.0.0.0"]

pyproject.toml

@@ -0,0 +1,22 @@
+[project]
+name = "ssms-gui"
+version = "0.1.0"
+description = "Streamlit dashboard for visualizing sequential sampling models from ssm-simulators."
+requires-python = ">=3.13"
+dependencies = [
+    "altair",
+    "cloudpickle",
+    "matplotlib",
+    "pandas",
+    "seaborn",
+    "ssm-simulators>=0.12.2",
+    "streamlit>1.30.0",
+]
+
+[dependency-groups]
+dev = [
+    "pytest",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]

requirements.txt

@@ -1,7 +0,0 @@
-altair
-pandas
-ssm-simulators>=0.12.2
-cloudpickle
-streamlit>1.30.0
-matplotlib
-seaborn

src/app.py

@@ -170,11 +170,22 @@ def create_styling_selectors(model_num: int = 1):
             key=f"show_ndt_{model_num}_{st.session_state['styling_version']}"
         )
         styling_config["add_data_model_keep_starting_point"] = st.checkbox(
-            "Show Starting Point", 
+            "Show Starting Point",
             value=True,
             key=f"show_start_{model_num}_{st.session_state['styling_version']}"
         )
 
+        styling_config["expected_random_params"] = st.checkbox(
+            "Cartoon at expected variability params",
+            value=True,
+            help=(
+                "When on, the noiseless model cartoon reflects the expected slope/NDT "
+                "instead of one random realization of across-trial variability "
+                "(e.g. sv, st in ddm_sdv, ddm_st, ddm_truncnormt, ddm_rayleight)."
+            ),
+            key=f"expected_random_params_{model_num}_{st.session_state['styling_version']}",
+        )
+
         # Axis Limits Section
         st.markdown("**Axis Limits**")
         styling_config["xlim_min"] = st.number_input(
@@ -288,14 +299,15 @@ def get_filtered_styling_config(styling_config, plot_type="plot_func_model"):
     elif plot_type == "plot_func_model_n":
         # plot_func_model_n only accepts a subset of parameters
         allowed_params = {
-            'linewidth_histogram', 'linewidth_model', 'bin_size', 
-            'alpha', 'legend_fontsize', 'legend_location', 'legend_shadow', 
+            'linewidth_histogram', 'linewidth_model', 'bin_size',
+            'alpha', 'legend_fontsize', 'legend_location', 'legend_shadow',
             'add_legend', 'add_data_model_markersize_starting_point',
             'add_data_model_markertype_starting_point',
             'add_data_model_keep_starting_point',
             'add_data_model_keep_boundary',
-            'add_data_model_keep_slope', 
-            'add_data_model_keep_ndt'
+            'add_data_model_keep_slope',
+            'add_data_model_keep_ndt',
+            'expected_random_params',
         }
         return {k: v for k, v in styling_config.items() if k in allowed_params}
     

src/utils/utils.py

@@ -8,6 +8,81 @@ _model_config = get_model_config()
 from matplotlib.lines import Line2D
 
 
+# TODO(ssm-simulators): upstream this as Simulator.simulate(variability_at_expectation=True).
+# For models with across-trial random parameters (e.g. ddm_sdv: v ~ N(v, sv)), `no_noise=True`
+# only zeros Brownian noise — the variability draw still happens, so the cartoon shows one
+# random realization of the variability term instead of the model's deterministic skeleton.
+# These tables let us collapse those draws to their expected value for the cartoon-only path.
+#
+# Maps model_name -> {variability_param_name: collapsed_value}.
+_VARIABILITY_COLLAPSE = {
+    "ddm_sdv":        {"sv": 0.0},     # N(0, 0)         -> 0 = E[N(0, sv)]
+    "ddm_st":         {"st": 0.0},     # U(0, 0)         -> 0 = E[U(-st, st)]
+    "ddm_truncnormt": {"st": 1e-9},    # TruncN(mt, ~0)  -> mt ≈ E[TruncN(mt, st)]; eps avoids div-by-zero in the config lambda (a=-mt/st)
+    "ddm_rayleight":  {"st": 0.0},     # Rayleigh(0)     -> 0; metadata['t'] is then post-shifted to st·sqrt(pi/2)
+}
+
+# Models where collapsing variability does NOT yield the distribution's expectation.
+# Shift `metadata['t']` after simulation by the expected NDT contribution.
+_EXPECTED_T_POST_SHIFT = {
+    "ddm_rayleight": lambda theta_d: theta_d["st"] * np.sqrt(np.pi / 2.0),
+}
+
+
+def _collapse_theta_for_cartoon(model_name, theta):
+    """Return a copy of `theta` with random-variability params replaced for the cartoon-only path.
+
+    `theta` is the list-of-lists shape that `app.py` passes to `Simulator.simulate`: positional
+    in `_model_config[model_name]['params']` order. Models without an entry in the table are
+    returned unchanged.
+    """
+    overrides = _VARIABILITY_COLLAPSE.get(model_name)
+    if not overrides:
+        return theta
+    params = _model_config[model_name]["params"]
+    inner = list(theta[0])
+    for name, value in overrides.items():
+        inner[params.index(name)] = value
+    return [inner]
+
+
+def _apply_expected_t_shift(model_name, theta_dict, sim_out):
+    """Mutate `sim_out['metadata']['t']` in place to reflect E[NDT] for models that need it.
+
+    Required for distributions whose param=0 collapse does not equal the expectation
+    (currently only ddm_rayleight: Rayleigh(0) = 0 but E[Rayleigh(scale=st)] = st·sqrt(pi/2)).
+    """
+    shift_fn = _EXPECTED_T_POST_SHIFT.get(model_name)
+    if shift_fn is None:
+        return
+    shift = shift_fn(theta_dict)
+    t_arr = np.asarray(sim_out["metadata"]["t"])
+    sim_out["metadata"]["t"] = (t_arr + shift).astype(t_arr.dtype, copy=False)
+
+
+# TODO(ssm-simulators): expose t_samplewise (and v_samplewise, z_samplewise) in metadata
+# so consumers don't have to back-derive per-sample NDT from RT and the trajectory.
+def _patch_trajectory_t_with_actual_ndt(sim_out, delta_t=0.001):
+    """Override `sim_out['metadata']['t']` with the per-sample NDT actually used for this trajectory.
+
+    The simulator records the *input* `t` in metadata, not the per-sample draw from `t_dist`
+    (e.g. Uniform/TruncNormal/Rayleigh for ddm_st / ddm_truncnormt / ddm_rayleight).
+    Without this patch, every trajectory plotter starts at the input `t` and ignores NDT
+    variability. We back-derive the per-sample NDT from `RT - decision_time`, where
+    `decision_time = (last valid trajectory index) * delta_t`. Trajectory calls in this
+    module use `smooth_unif=False`, so `RT = decision_time + NDT` exactly.
+
+    No-op for models without NDT variability (actual_ndt == input t).
+    """
+    traj = np.asarray(sim_out["metadata"]["trajectory"]).flatten()
+    valid_idx = np.where(traj > -999)[0]
+    decision_time = float(valid_idx[-1] * delta_t) if len(valid_idx) else 0.0
+    rt = float(np.asarray(sim_out["rts"]).flat[0])
+    actual_ndt = max(0.0, rt - decision_time)
+    t_arr = np.asarray(sim_out["metadata"]["t"])
+    sim_out["metadata"]["t"] = np.full_like(t_arr, actual_ndt, dtype=t_arr.dtype)
+
+
 def plot_func_model(
     model_name,
     theta,
@@ -35,6 +110,7 @@ def plot_func_model(
     delta_t_model=0.001,
     random_state=None,
     add_legend=True,  # keep_frame=False,
+    expected_random_params=True,
     **kwargs,
 ):
     """Calculate posterior predictive for a certain bottom node.
@@ -134,10 +210,16 @@ def plot_func_model(
         sim_out_traj[i] = sim.simulate(theta=theta, n_samples=1,
                                        no_noise=False, delta_t=0.001,
                                        random_state=rand_int, smooth_unif=False)
+        _patch_trajectory_t_with_actual_ndt(sim_out_traj[i], delta_t=0.001)
 
-    sim_out_no_noise = sim.simulate(theta=theta, n_samples=1,
+    theta_cartoon = _collapse_theta_for_cartoon(model_name, theta) if expected_random_params else theta
+    sim_out_no_noise = sim.simulate(theta=theta_cartoon, n_samples=1,
                                     no_noise=True, delta_t=0.001,
                                     smooth_unif=False)
+    if expected_random_params:
+        params = _model_config[model_name]["params"]
+        theta_dict = dict(zip(params, theta[0]))
+        _apply_expected_t_shift(model_name, theta_dict, sim_out_no_noise)
 
     # ADD DATA HISTOGRAMS
     weights_up = np.tile(
@@ -453,6 +535,7 @@ def plot_func_model_n(
     alpha=1,
     keep_frame=False,
     random_state=None,
+    expected_random_params=True,
     **kwargs,
 ):
     """Calculate posterior predictive for a certain bottom node.
@@ -565,10 +648,16 @@ def plot_func_model_n(
         sim_out_traj[i] = sim.simulate(theta=theta, n_samples=1,
                                        no_noise=False, delta_t=0.001,
                                        random_state=rand_int, smooth_unif=False)
+        _patch_trajectory_t_with_actual_ndt(sim_out_traj[i], delta_t=0.001)
 
-    sim_out_no_noise = sim.simulate(theta=theta, n_samples=1,
+    theta_cartoon = _collapse_theta_for_cartoon(model_name, theta) if expected_random_params else theta
+    sim_out_no_noise = sim.simulate(theta=theta_cartoon, n_samples=1,
                                     no_noise=True, delta_t=0.001,
                                     smooth_unif=False)
+    if expected_random_params:
+        params = _model_config[model_name]["params"]
+        theta_dict = dict(zip(params, theta[0]))
+        _apply_expected_t_shift(model_name, theta_dict, sim_out_no_noise)
 
     # ADD HISTOGRAMS
     # -------------------------------

tests/conftest.py

@@ -0,0 +1,4 @@
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))

tests/test_variability_collapse.py

@@ -0,0 +1,200 @@
+"""Regression tests for the cartoon-variability collapse helpers in utils/utils.py.
+
+These guard against the misleading behavior where `no_noise=True` cartoon plots in
+ssms-gui rendered one random realization of across-trial variability (e.g. a slope
+of `v + epsilon` for ddm_sdv) instead of the model's deterministic skeleton.
+"""
+
+import numpy as np
+import pytest
+
+from ssms.basic_simulators import Simulator
+
+from utils.utils import (
+    _VARIABILITY_COLLAPSE,
+    _EXPECTED_T_POST_SHIFT,
+    _apply_expected_t_shift,
+    _collapse_theta_for_cartoon,
+    _patch_trajectory_t_with_actual_ndt,
+)
+
+
+# (model_name, theta_inner, sv_or_st_index, collapsed_value)
+COLLAPSE_CASES = [
+    ("ddm_sdv",        [0.5, 1.0, 0.5, 0.3, 1.5], 4, 0.0),
+    ("ddm_st",         [0.5, 1.0, 0.5, 0.3, 0.2], 4, 0.0),
+    ("ddm_truncnormt", [0.5, 1.0, 0.5, 0.4, 0.1], 4, 1e-9),
+    ("ddm_rayleight",  [0.5, 1.0, 0.5, 0.3],      3, 0.0),
+]
+
+
+@pytest.mark.parametrize("model_name,theta_inner,idx,collapsed", COLLAPSE_CASES)
+def test_collapse_replaces_only_variability_slot(model_name, theta_inner, idx, collapsed):
+    out = _collapse_theta_for_cartoon(model_name, [list(theta_inner)])
+    assert len(out) == 1
+    assert len(out[0]) == len(theta_inner)
+    for i, val in enumerate(theta_inner):
+        if i == idx:
+            assert out[0][i] == collapsed
+        else:
+            assert out[0][i] == val
+
+
+def test_collapse_passthrough_for_unmapped_model():
+    theta = [[0.5, 1.0, 0.5, 0.3]]
+    assert _collapse_theta_for_cartoon("ddm", theta) is theta
+
+
+def test_collapse_does_not_mutate_input():
+    theta = [[0.5, 1.0, 0.5, 0.3, 1.5]]
+    original = [list(theta[0])]
+    _collapse_theta_for_cartoon("ddm_sdv", theta)
+    assert theta[0] == original[0]
+
+
+def test_apply_expected_t_shift_rayleigh():
+    st_val = 0.3
+    sim_out = {"metadata": {"t": np.array([0.0], dtype=np.float32)}}
+    _apply_expected_t_shift("ddm_rayleight", {"st": st_val}, sim_out)
+    expected = st_val * np.sqrt(np.pi / 2.0)
+    assert sim_out["metadata"]["t"][0] == pytest.approx(expected, rel=1e-5)
+
+
+def test_apply_expected_t_shift_noop_for_unmapped_model():
+    sim_out = {"metadata": {"t": np.array([0.123], dtype=np.float32)}}
+    _apply_expected_t_shift("ddm_sdv", {"sv": 1.0}, sim_out)
+    assert sim_out["metadata"]["t"][0] == pytest.approx(0.123)
+
+
+# ---------------------------------------------------------------------------
+# End-to-end: simulate with collapsed theta and `no_noise=True`. The trajectory's
+# slope must equal `v` (no spurious tilt from a residual variability draw).
+# Smoke test: with collapsed `sv`/`st`, a `no_noise=True` simulation should be
+# fully deterministic up to integer truncation. We sanity-check the slope of
+# `metadata['trajectory']` against `v`.
+# ---------------------------------------------------------------------------
+
+E2E_CASES = [
+    ("ddm_sdv",        {"v": 0.8, "a": 1.2, "z": 0.5, "t": 0.1, "sv": 1.5}),
+    ("ddm_st",         {"v": 0.8, "a": 1.2, "z": 0.5, "t": 0.1, "st": 0.3}),
+    ("ddm_truncnormt", {"v": 0.8, "a": 1.2, "z": 0.5, "mt": 0.4, "st": 0.1}),
+    ("ddm_rayleight",  {"v": 0.8, "a": 1.2, "z": 0.5, "st": 0.3}),
+]
+
+
+def _theta_list(model_name, theta_dict):
+    from ssms.config import get_model_config
+    params = get_model_config()[model_name]["params"]
+    return [[theta_dict[p] for p in params]]
+
+
+@pytest.mark.parametrize("model_name,theta_dict", E2E_CASES)
+def test_collapsed_cartoon_slope_matches_v(model_name, theta_dict):
+    theta = _theta_list(model_name, theta_dict)
+    theta_cartoon = _collapse_theta_for_cartoon(model_name, theta)
+
+    sim = Simulator(model=model_name)
+    out = sim.simulate(
+        theta=theta_cartoon, n_samples=1,
+        no_noise=True, delta_t=0.001, smooth_unif=False,
+        random_state=42,
+    )
+
+    traj = np.asarray(out["metadata"]["trajectory"]).flatten()
+    valid = traj[traj > -999]
+    # Need at least a handful of points to fit a line; constant-boundary DDMs reach this easily.
+    assert len(valid) >= 10, f"too few trajectory points for {model_name}: {len(valid)}"
+
+    # Drop the last point (boundary crossing — slightly past the bound), fit a line.
+    delta_t = 0.001
+    t_axis = np.arange(len(valid)) * delta_t
+    slope, _ = np.polyfit(t_axis[:-1], valid[:-1], 1)
+    assert slope == pytest.approx(theta_dict["v"], abs=0.05), (
+        f"{model_name}: cartoon slope {slope:.4f} != v {theta_dict['v']}"
+    )
+
+
+def test_rayleight_metadata_t_post_shift_e2e():
+    theta_dict = {"v": 0.8, "a": 1.2, "z": 0.5, "st": 0.3}
+    theta = _theta_list("ddm_rayleight", theta_dict)
+    theta_cartoon = _collapse_theta_for_cartoon("ddm_rayleight", theta)
+
+    sim = Simulator(model="ddm_rayleight")
+    out = sim.simulate(
+        theta=theta_cartoon, n_samples=1,
+        no_noise=True, delta_t=0.001, smooth_unif=False,
+        random_state=42,
+    )
+    _apply_expected_t_shift("ddm_rayleight", theta_dict, out)
+
+    expected = theta_dict["st"] * np.sqrt(np.pi / 2.0)
+    assert out["metadata"]["t"].flat[0] == pytest.approx(expected, rel=1e-4)
+
+
+def test_collapse_table_keys_match_post_shift_keys_subset():
+    # Every model in the post-shift table must also be in the collapse table.
+    assert set(_EXPECTED_T_POST_SHIFT.keys()).issubset(_VARIABILITY_COLLAPSE.keys())
+
+
+# ---------------------------------------------------------------------------
+# Trajectory NDT patch: metadata['t'] is the *input* t, not the per-sample NDT
+# actually drawn. The patch back-derives the actual NDT from rt - decision_time
+# so the trajectory plotter starts each line at its own NDT.
+# ---------------------------------------------------------------------------
+
+
+def test_patch_trajectory_t_recovers_known_ndt():
+    delta_t = 0.001
+    # Trajectory: 50 valid steps then -999. decision_time = 50 * delta_t = 0.05s.
+    traj = np.full(200, -999.0, dtype=np.float32)
+    traj[:51] = np.linspace(0.0, 1.0, 51)  # indices 0..50 valid
+    rt = 0.05 + 0.42  # decision_time + arbitrary NDT
+    sim_out = {
+        "rts": np.array([[[rt]]], dtype=np.float32),
+        "metadata": {
+            "t": np.array([0.0], dtype=np.float32),
+            "trajectory": traj,
+        },
+    }
+    _patch_trajectory_t_with_actual_ndt(sim_out, delta_t=delta_t)
+    assert sim_out["metadata"]["t"][0] == pytest.approx(0.42, abs=1e-5)
+
+
+def test_patch_trajectory_t_e2e_rayleigh_varies_across_calls():
+    """Across n trajectory calls with different seeds, the patched NDTs must vary
+    (Rayleigh-distributed) — not all be identical to the input t=0."""
+    sim = Simulator(model="ddm_rayleight")
+    theta = _theta_list("ddm_rayleight", {"v": 0.8, "a": 1.2, "z": 0.5, "st": 0.3})
+
+    ndts = []
+    for seed in range(20):
+        out = sim.simulate(
+            theta=theta, n_samples=1,
+            no_noise=False, delta_t=0.001,
+            random_state=seed, smooth_unif=False,
+        )
+        _patch_trajectory_t_with_actual_ndt(out, delta_t=0.001)
+        ndts.append(float(out["metadata"]["t"].flat[0]))
+
+    ndts = np.array(ndts)
+    assert (ndts > 0).all(), "all patched NDTs should be strictly positive for Rayleigh"
+    assert ndts.std() > 0.05, f"patched NDTs should vary across seeds, got std={ndts.std():.4f}"
+    # Mean should be in the right ballpark of E[Rayleigh(0.3)] = 0.3 * sqrt(pi/2) ≈ 0.376.
+    # 20 samples is noisy, so a wide tolerance.
+    assert 0.15 < ndts.mean() < 0.65, f"unexpected mean NDT {ndts.mean():.4f}"
+
+
+def test_patch_trajectory_t_noop_for_constant_ndt_model():
+    """For plain ddm (no NDT variability), patched NDT should match input t."""
+    sim = Simulator(model="ddm")
+    input_t = 0.3
+    theta = _theta_list("ddm", {"v": 0.8, "a": 1.2, "z": 0.5, "t": input_t})
+
+    out = sim.simulate(
+        theta=theta, n_samples=1,
+        no_noise=False, delta_t=0.001,
+        random_state=42, smooth_unif=False,
+    )
+    _patch_trajectory_t_with_actual_ndt(out, delta_t=0.001)
+    # Allow a single delta_t of slop from integer truncation of decision time.
+    assert out["metadata"]["t"][0] == pytest.approx(input_t, abs=2e-3)