refactor(concurrent): typed run plan + recorded estimand/population metadata (#394 pillar 4)

src/language_reading_predictors/statistical_models/concurrent.py

@@ -0,0 +1,280 @@
+# Copyright (c) 2026 Down Syndrome Education International and contributors
+# SPDX-License-Identifier: AGPL-3.0-or-later
+
+"""Typed settings and a resolved run plan for the concurrent-associations family (#394 pillar 4).
+
+Mirrors the ITT / gain-factor / level-factor / DiD run-plan pattern (:mod:`itt`,
+:mod:`gain_factors`, :mod:`level_factors`, :mod:`did`) for the per-wave concurrent
+conditional-associations (``kind="concurrent"``) models. A model module declares its
+settings; the plan is resolved and **validated before any data are loaded or an
+output directory is reset**, then drives data preparation and the ``config.json`` /
+``model_recipe.md`` audit trail. This removes the untyped ``spec.extra`` boundary
+(where a misspelled key silently defaulted) and records the resolved design,
+estimand, causal status, analysis population and missing-data assumption alongside
+every fit.
+
+The concurrent design fits, **at each wave separately**, a between-child
+Beta-Binomial regression of the focal outcome's level on the standardised same-wave
+logits of a predictor skill set (plus age and a group nuisance term), reported side
+by side with matched bivariate refits. Every coefficient is an **adjusted
+association**; the family makes no causal claim, so conditioning on contemporaneous
+(post-treatment) skill levels is intentional and the Table-2 fallacy applies.
+
+Because the model is fit once per wave with a wave-specific usable-predictor subset
+(and the bivariate refits vary ``include_age`` / ``include_group``), the factory is
+called many times by the pipeline; this plan therefore owns the *settings*, the
+single ``load_and_prepare`` call and the recorded metadata, while the per-wave
+factory calls stay in ``fit_concurrent`` using the resolved plan attributes.
+``predictor_slope_sigma`` defaults to ``None`` so the pipeline can fill the factory
+default through ``_default_of`` reflection, keeping the anti-drift single source.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from dataclasses import asdict, dataclass
+from typing import Any
+
+from language_reading_predictors.statistical_models.context import ModelSpec
+
+# The complete, closed set of legacy ``spec.extra`` keys the concurrent family
+# understands. Anything else is a typo and must fail before a fit starts.
+_LEGACY_KEYS = frozenset(
+    {
+        "predictor_symbols",
+        "covariates",
+        "include_age",
+        "include_group",
+        "predictor_slope_sigma",
+    }
+)
+
+_DEFAULT_PREDICTORS = ("L", "B", "TR", "TE", "R", "E")
+
+
+def _tuple_of_strings(value: Any, *, name: str) -> tuple[str, ...]:
+    if isinstance(value, str) or not hasattr(value, "__iter__"):
+        raise TypeError(f"{name} must be a sequence of strings, got {value!r}")
+    out = tuple(value)
+    for item in out:
+        if not isinstance(item, str) or not item:
+            raise TypeError(f"{name} must contain non-empty strings, got {item!r}")
+    return out
+
+
+@dataclass(frozen=True, slots=True)
+class ConcurrentModelSettings:
+    """Immutable settings declared by a single concurrent-associations model module.
+
+    Defaults encode the primary six-skill concurrent read with age and a group
+    nuisance term. ``predictor_slope_sigma`` is ``None`` by default so the pipeline
+    fills the ``build_concurrent_model`` default via ``_default_of`` reflection
+    rather than duplicating the numeric literal here.
+    """
+
+    predictor_symbols: tuple[str, ...] = _DEFAULT_PREDICTORS
+    covariates: tuple[str, ...] = ()
+    include_age: bool = True
+    include_group: bool = True
+    predictor_slope_sigma: float | None = None
+
+    def __post_init__(self) -> None:
+        object.__setattr__(
+            self,
+            "predictor_symbols",
+            _tuple_of_strings(self.predictor_symbols, name="predictor_symbols"),
+        )
+        object.__setattr__(
+            self, "covariates", _tuple_of_strings(self.covariates, name="covariates")
+        )
+        for flag in ("include_age", "include_group"):
+            if not isinstance(getattr(self, flag), bool):
+                raise TypeError(f"{flag} must be bool")
+        sigma = self.predictor_slope_sigma
+        if sigma is not None:
+            # bool is an int subclass but is never a valid slope scale.
+            if isinstance(sigma, bool) or not isinstance(sigma, (int, float)):
+                raise TypeError("predictor_slope_sigma must be a number or None")
+            if sigma <= 0:
+                raise ValueError("predictor_slope_sigma must be positive")
+            object.__setattr__(self, "predictor_slope_sigma", float(sigma))
+
+    @classmethod
+    def from_legacy_extra(
+        cls, extra: Mapping[str, Any], *, model_id: str
+    ) -> ConcurrentModelSettings:
+        """Strictly translate the former ``spec.extra`` dictionary boundary.
+
+        Rejects unknown keys so a misspelling fails before data loading rather than
+        silently taking a default."""
+        unknown = sorted(set(extra) - _LEGACY_KEYS)
+        if unknown:
+            raise ValueError(
+                f"{model_id}: unknown concurrent setting(s): {', '.join(unknown)}. "
+                "Declare ConcurrentModelSettings so misspellings fail fast."
+            )
+        # Pass raw values through so __post_init__ is the single validation/coercion
+        # point. ``predictor_slope_sigma`` absent -> None (pipeline uses the factory
+        # default via _default_of), matching the former .get(key, _default_of(...)).
+        return cls(
+            predictor_symbols=extra.get("predictor_symbols", _DEFAULT_PREDICTORS),
+            covariates=extra.get("covariates", ()),
+            include_age=extra.get("include_age", True),
+            include_group=extra.get("include_group", True),
+            predictor_slope_sigma=extra.get("predictor_slope_sigma"),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ConcurrentRunPlan:
+    """Concrete, validated instructions consumed by preparation and modelling."""
+
+    model_id: str
+    outcome_symbol: str
+    settings_source: str
+    predictor_symbols: tuple[str, ...]
+    covariates: tuple[str, ...]
+    include_age: bool
+    include_group: bool
+    # ``None`` -> the pipeline fills the build_concurrent_model default via _default_of.
+    predictor_slope_sigma: float | None
+    # Recorded audit metadata (#394 pillar 4).
+    design: str
+    estimand: str
+    causal_status: str
+    analysis_population: str
+    missing_data_assumption: str
+
+    @property
+    def measure_outcomes(self) -> tuple[str, ...]:
+        """The outcome plus its predictor skills, de-duplicated and order-preserving."""
+        return tuple(dict.fromkeys((self.outcome_symbol, *self.predictor_symbols)))
+
+    def as_dict(self) -> dict[str, Any]:
+        """Return the JSON-ready run-plan contract for ``config.json``."""
+        return asdict(self)
+
+    def prepare_kwargs(self) -> dict[str, Any]:
+        """Arguments for ``load_and_prepare`` from the resolved plan.
+
+        The concurrent family loads the per-wave ``levels`` panel with the outcome and
+        its predictor skills as bounded-count measures; the trait covariates are
+        t1-measured and enter as baseline covariates broadcast across the waves."""
+        return {
+            "phase_mode": "levels",
+            "outcomes": self.measure_outcomes,
+            "baseline_covariates": self.covariates,
+        }
+
+    def recipe_markdown(self, *, title: str) -> str:
+        """Undergraduate-friendly explanation generated from the resolved plan."""
+        preds = ", ".join(self.predictor_symbols) if self.predictor_symbols else "none"
+        covs = ", ".join(self.covariates) if self.covariates else "none"
+        sigma = (
+            "build default"
+            if self.predictor_slope_sigma is None
+            else f"{self.predictor_slope_sigma:g}"
+        )
+        return (
+            "Note: Generated from the validated concurrent-associations run plan; "
+            "template drafted by an LLM-based AI tool (Claude Code/Opus 4.8).\n\n"
+            f"# Model recipe: {title}\n\n"
+            f"Model ID: `{self.model_id}`.\n\n"
+            f"## Design\n\n{self.design}\n\n"
+            f"## Estimand\n\n{self.estimand}\n\n"
+            f"## Causal status\n\n{self.causal_status}\n\n"
+            f"## Analysis population\n\n{self.analysis_population}\n\n"
+            f"## Missing data\n\n{self.missing_data_assumption}\n\n"
+            "## Terms\n\n"
+            f"Outcome: `{self.outcome_symbol}`. Predictor skills: {preds}. Trait "
+            f"covariates: {covs}. Age term: {self.include_age}. Group nuisance term: "
+            f"{self.include_group}. Predictor-slope prior sigma: {sigma}.\n\n"
+            "## Uncertainty and checks\n\n"
+            "The fit reports a posterior distribution; interpret it only after the "
+            "convergence gate and posterior-predictive checks pass. The saved "
+            "`config.json` contains the same resolved run plan in machine-readable "
+            "form.\n"
+        )
+
+
+def declared_concurrent_settings(
+    spec: ModelSpec,
+) -> tuple[ConcurrentModelSettings, str]:
+    """Return typed settings and their source, rejecting mixed declarations."""
+    settings = spec.model_settings
+    if settings is not None:
+        if spec.extra:
+            raise ValueError(
+                f"{spec.model_id}: concurrent settings cannot be split between "
+                "model_settings and extra"
+            )
+        if not isinstance(settings, ConcurrentModelSettings):
+            raise TypeError(
+                f"{spec.model_id}: kind='concurrent' requires "
+                f"ConcurrentModelSettings, got {type(settings).__name__}"
+            )
+        return settings, "typed"
+    return (
+        ConcurrentModelSettings.from_legacy_extra(spec.extra, model_id=spec.model_id),
+        "legacy_extra",
+    )
+
+
+def resolve_concurrent_run_plan(spec: ModelSpec) -> ConcurrentRunPlan:
+    """Resolve and validate a concurrent-associations spec before any data are loaded."""
+    if spec.kind != "concurrent":
+        raise ValueError(
+            f"{spec.model_id}: expected kind 'concurrent', got {spec.kind!r}"
+        )
+    if not spec.outcome_symbol:
+        raise ValueError(
+            f"{spec.model_id}: outcome_symbol is required for a concurrent model"
+        )
+
+    settings, source = declared_concurrent_settings(spec)
+    own = spec.outcome_symbol
+
+    design = (
+        "Per-wave concurrent conditional associations: at each timepoint a "
+        "between-child Beta-Binomial regression of the outcome level on the "
+        "standardised same-wave logits of the predictor skills, plus optional age and "
+        "a group nuisance term. Four cross-sectional fits reported side by side, each "
+        "with a matched single-predictor (bivariate) refit."
+    )
+    estimand = (
+        "Adjusted per-wave conditional associations (per +1 SD of each predictor's "
+        "same-wave logit), reported alongside the unadjusted bivariate association. "
+        "No causal quantity is estimated: conditioning on contemporaneous "
+        "post-treatment skill levels is intentional and the Table-2 fallacy applies."
+    )
+    causal_status = (
+        "Associational only. The family makes no causal claim; every coefficient is a "
+        "latent-ability-confounded adjusted association at a wave, never a cause."
+    )
+    analysis_population = (
+        "Available-case children observed at each wave (about 53 per wave, "
+        "outcome-complete). The reported associations average over the fitted rows at "
+        "each wave -- a descriptive averaging population."
+    )
+    missing_data_assumption = (
+        "Rows missing the focal outcome are dropped (an outcome cannot be imputed); "
+        "missing predictor values are mean-imputed (a descriptive read -- imputation "
+        "can bias a conditional coefficient when missingness relates to the predictor, "
+        "outcome or other skills)."
+    )
+
+    return ConcurrentRunPlan(
+        model_id=spec.model_id,
+        outcome_symbol=own,
+        settings_source=source,
+        predictor_symbols=settings.predictor_symbols,
+        covariates=settings.covariates,
+        include_age=settings.include_age,
+        include_group=settings.include_group,
+        predictor_slope_sigma=settings.predictor_slope_sigma,
+        design=design,
+        estimand=estimand,
+        causal_status=causal_status,
+        analysis_population=analysis_population,
+        missing_data_assumption=missing_data_assumption,
+    )

src/language_reading_predictors/statistical_models/pipeline.py

@@ -73,6 +73,9 @@
     save_plotcollection,
     save_styled_figure,
 )
+from language_reading_predictors.statistical_models.concurrent import (
+    resolve_concurrent_run_plan,
+)
 from language_reading_predictors.statistical_models.context import (
     ModelSpec,
     StatisticalFitContext,
@@ -7385,35 +7388,39 @@ def fit_concurrent(spec: ModelSpec, config: str = "dev") -> StatisticalFitContex
     detailed probability/items marginals (wave × predictor × {+1 SD, +k items}).
     """
     _require_spec(spec, "concurrent", outcome=True)
-    e = spec.extra
-    outcome = spec.outcome_symbol or "W"
-    predictor_symbols = list(e.get("predictor_symbols", ["L", "B", "TR", "TE", "R", "E"]))
+    # Resolve and validate the family contract before the context resets an output
+    # directory or the loader reads any data (#394 pillar 4). One plan drives
+    # preparation, the teaching recipe and config.json.
+    plan = resolve_concurrent_run_plan(spec)
+    outcome = plan.outcome_symbol
+    predictor_symbols = list(plan.predictor_symbols)
     # Trait covariates (non-verbal ability, hearing, speech, phonological memory),
     # aligned with the gains panel. They are t1-measured, so they enter as
     # baseline covariates broadcast across the waves (there is no per-wave value).
-    covariates = list(e.get("covariates", []))
-    include_age = bool(e.get("include_age", True))
-    include_group = bool(e.get("include_group", True))
-    sigma0 = float(
-        e.get(
-            "predictor_slope_sigma",
-            _default_of(_factories.build_concurrent_model, "predictor_slope_sigma"),
+    covariates = list(plan.covariates)
+    include_age = plan.include_age
+    include_group = plan.include_group
+    # ``predictor_slope_sigma`` is None on the plan when a spec does not set it, so the
+    # build_concurrent_model default is filled via _default_of here — the anti-drift
+    # single source #394 retains until typed family defaults replace it.
+    sigma0 = (
+        float(plan.predictor_slope_sigma)
+        if plan.predictor_slope_sigma is not None
+        else float(
+            _default_of(_factories.build_concurrent_model, "predictor_slope_sigma")
         )
     )
 
     from language_reading_predictors.statistical_models.measures import MEASURES
 
     ctx = make_context(spec, config)
+    ctx.resolved_plan = plan
+    _report.write_model_recipe(ctx)
     hdi = ctx.reporting.ci_prob
     N_focal = MEASURES[outcome].n_trials
 
     section_header("Prepare data")
-    measure_outcomes = tuple(dict.fromkeys([outcome, *predictor_symbols]))
-    prepared_all = load_and_prepare(
-        phase_mode="levels",
-        outcomes=measure_outcomes,
-        baseline_covariates=tuple(covariates),
-    )
+    prepared_all = load_and_prepare(**plan.prepare_kwargs())
 
     # Timepoints present; each wave's row count and its usable predictor set (a
     # predictor whose same-wave logit has positive variance on the wave's rows —

src/language_reading_predictors/statistical_models/reporting.py

@@ -27,6 +27,10 @@
 from language_reading_predictors.statistical_models.context import (
     StatisticalFitContext,
 )
+from language_reading_predictors.statistical_models.concurrent import (
+    ConcurrentRunPlan,
+    resolve_concurrent_run_plan,
+)
 from language_reading_predictors.statistical_models.gain_factors import (
     GainFactorsRunPlan,
     resolve_gain_factors_run_plan,
@@ -2273,12 +2277,22 @@ def _gain_factors_run_plan(context: StatisticalFitContext) -> GainFactorsRunPlan
     return resolve_gain_factors_run_plan(context.spec)
 
 
+def _concurrent_run_plan(context: StatisticalFitContext) -> ConcurrentRunPlan:
+    """Return the concurrent plan resolved before loading, or reconstruct it."""
+    resolved_plan = getattr(context, "resolved_plan", None)
+    if isinstance(resolved_plan, ConcurrentRunPlan):
+        return resolved_plan
+    return resolve_concurrent_run_plan(context.spec)
+
+
 def _resolved_run_plan(context: StatisticalFitContext):
-    """The typed run plan for families that have one (ITT, gain factors), else None."""
+    """The typed run plan for families that have one (ITT, gain, concurrent), else None."""
     if context.spec.kind == "itt":
         return _itt_run_plan(context)
     if context.spec.kind == "gain_factors":
         return _gain_factors_run_plan(context)
+    if context.spec.kind == "concurrent":
+        return _concurrent_run_plan(context)
     return None