diff --git a/src/language_reading_predictors/statistical_models/level_factors.py b/src/language_reading_predictors/statistical_models/level_factors.py new file mode 100644 index 00000000..83b6b34f --- /dev/null +++ b/src/language_reading_predictors/statistical_models/level_factors.py @@ -0,0 +1,334 @@ +# 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 level-factor family (#389 finding 6). + +Mirrors the ITT / gain-factor run-plan pattern (:mod:`itt`, :mod:`gain_factors`) for +the level-factor (``kind="level_factors"``) 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 a single object drives data preparation, factory +construction 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 level family previously persisted +null ``family`` / ``design`` / ``estimand_type`` / ``causal_status`` metadata while +its report published an unqualified cause-and-effect statement (#389 finding 4). + +The level design is a per-wave levels model: each wave's score is regressed on the +randomised group (entered as a per-timepoint vector when ``group_by_time``), the +ability covariate (optionally wave-varying) and, when ``group_ability``, a +group x ability effect-modification term, with a non-centred child random intercept. +The single randomised quantity is the **t2 group contrast** ``b_grp_time[1]`` (an +items- or risk-difference average marginal effect read at the t2 rows); the other +waves are post-crossover and every ability / interaction term is a +latent-ability-confounded **adjusted association**, never a causal effect. The +precise t2 estimand -- population-standardised average vs conditional-at-a-profile, +and the treatment of the currently time-invariant ``group x ability`` term -- is the +open methodological decision recorded in #389 finding 1; the prose below states the +quantity as it is presently implemented and flags that review. +""" + +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 +from language_reading_predictors.statistical_models.preprocessing import ( + split_confounders_by_timing, + split_covariates_by_wave, +) + +# The complete, closed set of legacy ``spec.extra`` keys the level-factor family +# understands. Anything else is a typo and must fail before a fit starts. +_LEGACY_KEYS = frozenset( + { + "ability_covariate", + "adjust_for", + "group_by_time", + "ability_by_time", + "group_ability", + "likelihood", + } +) + +_LIKELIHOODS = frozenset({"beta_binomial", "bernoulli_offfloor"}) + + +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 LevelFactorsModelSettings: + """Immutable settings declared by a single level-factor model module. + + Defaults encode the primary per-wave levels model: no ability covariate, no + extra adjusters, group and ability both entered as per-timepoint vectors with a + group x ability effect-modification term, and the Beta-Binomial working + likelihood. + """ + + ability_covariate: str | None = None + adjust_for: tuple[str, ...] = () + group_by_time: bool = True + ability_by_time: bool = True + group_ability: bool = True + likelihood: str = "beta_binomial" + + def __post_init__(self) -> None: + object.__setattr__( + self, "adjust_for", _tuple_of_strings(self.adjust_for, name="adjust_for") + ) + if self.ability_covariate is not None and ( + not isinstance(self.ability_covariate, str) or not self.ability_covariate + ): + raise TypeError("ability_covariate must be a non-empty string or None") + for flag in ("group_by_time", "ability_by_time", "group_ability"): + if not isinstance(getattr(self, flag), bool): + raise TypeError(f"{flag} must be bool") + if self.likelihood not in _LIKELIHOODS: + raise ValueError( + f"likelihood must be one of {sorted(_LIKELIHOODS)}, got {self.likelihood!r}" + ) + + @classmethod + def from_legacy_extra( + cls, extra: Mapping[str, Any], *, model_id: str + ) -> LevelFactorsModelSettings: + """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 level-factor setting(s): {', '.join(unknown)}. " + "Declare LevelFactorsModelSettings so misspellings fail fast." + ) + # Pass raw values through so __post_init__ is the single validation/coercion + # point; pre-coercing here (tuple(...)/bool(...)) would silently reshape + # misshaped legacy settings ("hs" -> ('h', 's'), 1 -> True) instead of failing + # fast against the strict checks in __post_init__. The bool flags default True. + return cls( + ability_covariate=extra.get("ability_covariate"), + adjust_for=extra.get("adjust_for", ()), + group_by_time=extra.get("group_by_time", True), + ability_by_time=extra.get("ability_by_time", True), + group_ability=extra.get("group_ability", True), + likelihood=extra.get("likelihood", "beta_binomial"), + ) + + +@dataclass(frozen=True, slots=True) +class LevelFactorsRunPlan: + """Concrete, validated instructions consumed by preparation and modelling.""" + + model_id: str + outcome_symbol: str + settings_source: str + ability_covariate: str | None + adjust_for: tuple[str, ...] + group_by_time: bool + ability_by_time: bool + group_ability: bool + likelihood: str + off_floor: bool + # Covariate loading split by measurement wave (resolved from adjust_for). + baseline_covariates: tuple[str, ...] + pre_covariates: tuple[str, ...] + post_covariates: tuple[str, ...] + # Recorded audit metadata (#389 findings 4 & 6 acceptance criteria). + design: str + estimand: str + causal_status: str + analysis_population: str + missing_data_assumption: str + + @property + def obs_node(self) -> str: + return "y_offfloor" if self.off_floor else "y_post" + + 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 level family loads the per-wave ``levels`` panel with only its own + outcome (no upstream skill baselines); the ability covariate and any + baseline-timed confounders load at t1, interval covariates at the pre row and + contemporaneous confounders (e.g. hearing) at the post row (#247 timing).""" + return { + "phase_mode": "levels", + "outcomes": (self.outcome_symbol,), + "baseline_covariates": self.baseline_covariates, + "covariates": self.pre_covariates, + "post_covariates": self.post_covariates, + } + + def factory_kwargs( + self, *, effective_adjustment: tuple[str, ...] | None = None + ) -> dict[str, Any]: + """Arguments for ``build_level_factors_model`` for this plan.""" + return { + "outcome_symbol": self.outcome_symbol, + "ability_covariate": self.ability_covariate, + "adjust_for": self.adjust_for + if effective_adjustment is None + else effective_adjustment, + "group_by_time": self.group_by_time, + "ability_by_time": self.ability_by_time, + "group_ability": self.group_ability, + "likelihood": self.likelihood, + } + + def recipe_markdown(self, *, title: str) -> str: + """Undergraduate-friendly explanation generated from the resolved plan.""" + adjust = ", ".join(self.adjust_for) if self.adjust_for else "none" + return ( + "Note: Generated from the validated level-factor 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}`. Ability covariate: " + f"{self.ability_covariate or 'none'}. Group entered per timepoint: " + f"{self.group_by_time}. Ability entered per timepoint: " + f"{self.ability_by_time}. Group x ability effect modification: " + f"{self.group_ability}. Requested adjustment terms: {adjust}.\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_level_factors_settings( + spec: ModelSpec, +) -> tuple[LevelFactorsModelSettings, 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}: level-factor settings cannot be split between " + "model_settings and extra" + ) + if not isinstance(settings, LevelFactorsModelSettings): + raise TypeError( + f"{spec.model_id}: kind='level_factors' requires " + f"LevelFactorsModelSettings, got {type(settings).__name__}" + ) + return settings, "typed" + return ( + LevelFactorsModelSettings.from_legacy_extra(spec.extra, model_id=spec.model_id), + "legacy_extra", + ) + + +def resolve_level_factors_run_plan(spec: ModelSpec) -> LevelFactorsRunPlan: + """Resolve and validate a level-factor specification before any data are loaded.""" + if spec.kind != "level_factors": + raise ValueError( + f"{spec.model_id}: expected kind 'level_factors', got {spec.kind!r}" + ) + if not spec.outcome_symbol: + raise ValueError( + f"{spec.model_id}: outcome_symbol is required for a level-factor model" + ) + + settings, source = declared_level_factors_settings(spec) + own = spec.outcome_symbol + off_floor = settings.likelihood == "bernoulli_offfloor" + + # Covariate loading split by measurement wave -- identical to the former inline + # logic in fit_level_factors: the ability covariate and any baseline-timed + # confounders load at t1 (the language-proximal SP/RW confounders are read at the + # pre-randomisation baseline so the t2 causal contrast is not conditioned on a + # treatment-affected descendant), interval covariates at the pre row, hearing + # contemporaneous at the post row (#247 timing; review finding A1). + pre_adj, post_adj = split_covariates_by_wave(settings.adjust_for) + baseline_adj, post_adj = split_confounders_by_timing(post_adj) + baseline_covariates = ( + (settings.ability_covariate,) if settings.ability_covariate else () + ) + baseline_adj + + if off_floor: + design = ( + "Per-wave off-floor levels model: a Bernoulli likelihood for whether the " + "child is above the outcome floor at each wave, with the randomised group " + "entered per timepoint, the ability covariate, an optional group x ability " + "term, and a non-centred child random intercept." + ) + estimand = ( + "The t2 randomised group contrast on the probability of being off the " + "floor (a risk difference read at the t2 rows). The other waves are " + "post-crossover; ability and interaction terms are adjusted associations." + ) + else: + design = ( + "Per-wave levels model: each wave's score is regressed (a Beta-Binomial " + "working likelihood) on the randomised group entered per timepoint, the " + "ability covariate, an optional group x ability term, and a non-centred " + "child random intercept for the repeated observations." + ) + estimand = ( + "The t2 randomised group contrast b_grp_time[1] (an items-scale average " + "marginal effect read at the t2 rows). The other waves are post-crossover; " + "ability and interaction terms are adjusted associations. The precise t2 " + "estimand -- population-standardised average vs conditional-at-a-profile, " + "and the treatment of the currently time-invariant group x ability term -- " + "is under methodological review (#389 finding 1)." + ) + causal_status = ( + "Only the t2 group term is randomised (a contrast on the available-case t2 " + "population); the other timepoints are post-crossover and every ability and " + "group x ability term is a latent-ability-confounded adjusted association, " + "never a causal effect." + ) + analysis_population = ( + "Available-case children observed across the level waves (about 53-54 " + "depending on outcome). The randomised interpretation applies to the t2 " + "contrast on this available-case population, not automatically the complete " + "randomised cohort." + ) + missing_data_assumption = ( + "Available-case analysis under ignorable missingness: missing outcomes and " + "covariates are assumed ignorable given the modelled covariates." + ) + + return LevelFactorsRunPlan( + model_id=spec.model_id, + outcome_symbol=own, + settings_source=source, + ability_covariate=settings.ability_covariate, + adjust_for=settings.adjust_for, + group_by_time=settings.group_by_time, + ability_by_time=settings.ability_by_time, + group_ability=settings.group_ability, + likelihood=settings.likelihood, + off_floor=off_floor, + baseline_covariates=baseline_covariates, + pre_covariates=pre_adj, + post_covariates=post_adj, + design=design, + estimand=estimand, + causal_status=causal_status, + analysis_population=analysis_population, + missing_data_assumption=missing_data_assumption, + ) diff --git a/src/language_reading_predictors/statistical_models/pipeline.py b/src/language_reading_predictors/statistical_models/pipeline.py index aab60fe5..854cdbd8 100644 --- a/src/language_reading_predictors/statistical_models/pipeline.py +++ b/src/language_reading_predictors/statistical_models/pipeline.py @@ -91,6 +91,9 @@ write_itt_analysis_audit, write_itt_ppc_calibration, ) +from language_reading_predictors.statistical_models.level_factors import ( + resolve_level_factors_run_plan, +) from language_reading_predictors.statistical_models.measures import ( ITT_OUTCOMES, MEASURES, @@ -5736,49 +5739,34 @@ def _lf_diag_vars( def fit_level_factors(spec: ModelSpec, config: str = "dev") -> StatisticalFitContext: _require_spec(spec, "level_factors", outcome=True) + # Resolve and validate the family contract before the context resets an output + # directory or the loader reads any data (#389 finding 6). One plan then drives + # preparation, factory arguments, the teaching recipe and config.json. The + # covariate wave-split is resolved into the plan (#247 timing; review finding A1: + # the language-proximal SP/RW confounders load at the pre-randomisation baseline + # so the t2 randomised contrast is not conditioned on a treatment-affected + # descendant; hearing is exogenous and stays contemporaneous). The level model + # takes no measure-skill adjusters (post-treatment mediators). + plan = resolve_level_factors_run_plan(spec) ctx = make_context(spec, config) - extra = spec.extra + ctx.resolved_plan = plan + _report.write_model_recipe(ctx) + + ability_covariate = plan.ability_covariate + off_floor = plan.off_floor + obs_node = plan.obs_node section_header("Prepare data") - ability_covariate = extra.get("ability_covariate") - likelihood = extra.get("likelihood", "beta_binomial") - off_floor = likelihood == "bernoulli_offfloor" - obs_node = "y_offfloor" if off_floor else "y_post" - baseline_covariates = (ability_covariate,) if ability_covariate else () - # Revised-DAG raw-covariate confounders (hearing/speech/phonological memory; #247). - # Timing (review finding A1; team decision 2026-07-13): the language-proximal SP/RW - # confounders (deapp_c/erbto + missing indicators) are read at the pre-randomisation - # BASELINE (t1) — the clean randomised contrast here is the t2 group term, and at t2 - # these language-proximal states may already be treatment-affected, so a - # contemporaneous read would condition the causal contrast on a descendant of the - # exposure. Hearing (hs) is exogenous and stays contemporaneous (post). Re-filter - # after loading so a constant ``_missing`` indicator dropped by the loader is not - # built or gated. The level model takes no measure-skill adjusters (post-treatment - # mediators). - adjust_for = tuple(extra.get("adjust_for", ())) - pre_adj, post_adj = split_covariates_by_wave(adjust_for) - baseline_adj, post_adj = split_confounders_by_timing(post_adj) - prepared = load_and_prepare( - phase_mode="levels", - outcomes=(spec.outcome_symbol,), - baseline_covariates=(*baseline_covariates, *baseline_adj), - covariates=pre_adj, - post_covariates=post_adj, - ) - adjust_for = tuple(c for c in adjust_for if c in prepared.covariates) + prepared = load_and_prepare(**plan.prepare_kwargs()) + # Re-filter after loading — a constant ``_missing`` indicator is dropped by the + # loader and must not be built or reported as adjusted-for. + adjust_for = tuple(c for c in plan.adjust_for if c in prepared.covariates) ctx.prepared = prepared _print_header(ctx) section_header("Build model") built = _factories.build_level_factors_model( - prepared, - outcome_symbol=spec.outcome_symbol, - ability_covariate=ability_covariate, - adjust_for=adjust_for, - group_by_time=bool(extra.get("group_by_time", True)), - ability_by_time=bool(extra.get("ability_by_time", True)), - group_ability=bool(extra.get("group_ability", True)), - likelihood=likelihood, + prepared, **plan.factory_kwargs(effective_adjustment=adjust_for) ) _attach_built(ctx, built) @@ -5796,7 +5784,7 @@ def fit_level_factors(spec: ModelSpec, config: str = "dev") -> StatisticalFitCon _run_ppc(ctx, var_names=[obs_node]) section_header("Extended diagnostics") - _lf_group_by_time = extra.get("group_by_time", True) + _lf_group_by_time = plan.group_by_time # For the shipped group-by-time LF models the flagged-causal term is the t2 # element of the per-timepoint group vector, ``b_grp_time`` (``b_grp_time[1]``, # which reporting.level_t2_marginal_effect reads into the causal ROPE card), so @@ -5813,7 +5801,7 @@ def fit_level_factors(spec: ModelSpec, config: str = "dev") -> StatisticalFitCon section_header("Factor summary") # Only the t2 group contrast (b_grp_time[1]) is the clean randomised effect; # the other timepoints are post-crossover (see the level-model caveat). - causal = ("b_grp_time[1]",) if extra.get("group_by_time", True) else () + causal = ("b_grp_time[1]",) if plan.group_by_time else () fs = _report.factor_summary( ctx.trace, _lf_coef_names(spec, adjust_for), ci_prob=ctx.reporting.ci_prob, causal_terms=causal ) @@ -5856,7 +5844,7 @@ def fit_level_factors(spec: ModelSpec, config: str = "dev") -> StatisticalFitCon delta_items = ROPE_DELTA.get(spec.outcome_symbol) delta_prob = ROPE_DELTA_PROB.get(spec.outcome_symbol) - _gbt = extra.get("group_by_time", True) + _gbt = plan.group_by_time _graded_card = delta_items is not None and not off_floor and _gbt _offfloor_card = off_floor and delta_prob is not None and _gbt if _graded_card or _offfloor_card: diff --git a/src/language_reading_predictors/statistical_models/reporting.py b/src/language_reading_predictors/statistical_models/reporting.py index 8fac06fe..f87ed48a 100644 --- a/src/language_reading_predictors/statistical_models/reporting.py +++ b/src/language_reading_predictors/statistical_models/reporting.py @@ -36,6 +36,10 @@ declared_settings_dict, resolve_itt_run_plan, ) +from language_reading_predictors.statistical_models.level_factors import ( + LevelFactorsRunPlan, + resolve_level_factors_run_plan, +) from language_reading_predictors.statistical_models.provenance import ( run_provenance, write_environment_lock, @@ -2273,12 +2277,22 @@ def _gain_factors_run_plan(context: StatisticalFitContext) -> GainFactorsRunPlan return resolve_gain_factors_run_plan(context.spec) +def _level_factors_run_plan(context: StatisticalFitContext) -> LevelFactorsRunPlan: + """Return the level-factor plan resolved before loading, or reconstruct it.""" + resolved_plan = getattr(context, "resolved_plan", None) + if isinstance(resolved_plan, LevelFactorsRunPlan): + return resolved_plan + return resolve_level_factors_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 / level factors), 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 == "level_factors": + return _level_factors_run_plan(context) return None diff --git a/tests/statistical_models/test_level_factors.py b/tests/statistical_models/test_level_factors.py new file mode 100644 index 00000000..298ea3cf --- /dev/null +++ b/tests/statistical_models/test_level_factors.py @@ -0,0 +1,210 @@ +# Copyright (c) 2026 Down Syndrome Education International and contributors +# SPDX-License-Identifier: AGPL-3.0-or-later + +"""Tests for the typed level-factor settings and resolved run plan (#389 finding 6).""" + +from __future__ import annotations + +import glob +import importlib +import os + +import pytest + +from language_reading_predictors.statistical_models.context import ModelSpec +from language_reading_predictors.statistical_models.level_factors import ( + LevelFactorsModelSettings, + LevelFactorsRunPlan, + resolve_level_factors_run_plan, +) + +_META_FIELDS = ( + "design", + "estimand", + "causal_status", + "analysis_population", + "missing_data_assumption", +) + + +def _level_factor_specs() -> list[ModelSpec]: + """Every registered level-factor model's SPEC.""" + root = os.path.dirname( + importlib.import_module( + "language_reading_predictors.statistical_models.level_factors" + ).__file__ + ) + specs: list[ModelSpec] = [] + for path in sorted(glob.glob(os.path.join(root, "lrp_rli_lf_*.py"))): + mod = importlib.import_module( + "language_reading_predictors.statistical_models." + os.path.basename(path)[:-3] + ) + spec = getattr(mod, "SPEC", None) + if spec is not None and spec.kind == "level_factors": + specs.append(spec) + return specs + + +# --- settings validation ------------------------------------------------------ + + +def test_settings_reject_unknown_likelihood(): + with pytest.raises(ValueError, match="likelihood"): + LevelFactorsModelSettings(likelihood="poisson") + + +def test_settings_reject_non_bool_group_by_time(): + with pytest.raises(TypeError, match="group_by_time"): + LevelFactorsModelSettings(group_by_time=1) # type: ignore[arg-type] + + +def test_settings_reject_non_bool_group_ability(): + with pytest.raises(TypeError, match="group_ability"): + LevelFactorsModelSettings(group_ability=0) # type: ignore[arg-type] + + +def test_settings_reject_string_adjust_for(): + # A bare string is a common mistake for a sequence-of-strings field. + with pytest.raises(TypeError, match="adjust_for"): + LevelFactorsModelSettings(adjust_for="hs") # type: ignore[arg-type] + + +def test_from_legacy_extra_rejects_unknown_key(): + with pytest.raises(ValueError, match="unknown level-factor setting"): + LevelFactorsModelSettings.from_legacy_extra( + {"group_by_time": True, "group_by_tyme": False}, # typo + model_id="lrp-rli-lf-999", + ) + + +def test_from_legacy_extra_round_trips_known_keys(): + settings = LevelFactorsModelSettings.from_legacy_extra( + { + "ability_covariate": "blocks", + "adjust_for": ("hs", "deapp_c"), + "group_by_time": False, + "ability_by_time": False, + "group_ability": False, + "likelihood": "bernoulli_offfloor", + }, + model_id="lrp-rli-lf-999", + ) + assert settings.ability_covariate == "blocks" + assert settings.adjust_for == ("hs", "deapp_c") + assert settings.group_by_time is False + assert settings.ability_by_time is False + assert settings.group_ability is False + assert settings.likelihood == "bernoulli_offfloor" + + +def test_from_legacy_extra_defaults_flags_true(): + # The three structural flags default True (the shipped per-timepoint design). + settings = LevelFactorsModelSettings.from_legacy_extra({}, model_id="lrp-rli-lf-999") + assert settings.group_by_time is True + assert settings.ability_by_time is True + assert settings.group_ability is True + assert settings.likelihood == "beta_binomial" + + +# --- resolve ------------------------------------------------------------------ + + +def _spec(**extra) -> ModelSpec: + return ModelSpec( + model_id="lrp-rli-lf-000", + kind="level_factors", + title="test", + outcome_symbol="W", + extra=extra, + ) + + +def test_resolve_rejects_wrong_kind(): + spec = ModelSpec(model_id="x", kind="gain_factors", title="t", outcome_symbol="W") + with pytest.raises(ValueError, match="expected kind 'level_factors'"): + resolve_level_factors_run_plan(spec) + + +def test_resolve_primary_records_t2_randomised_estimand(): + plan = resolve_level_factors_run_plan(_spec(ability_covariate="blocks")) + assert plan.settings_source == "legacy_extra" + assert not plan.off_floor + assert plan.obs_node == "y_post" + assert "randomised" in plan.causal_status + assert "t2" in plan.estimand + # prepare/factory kwargs are shaped for the loader and the factory. + assert plan.prepare_kwargs()["outcomes"] == ("W",) + assert plan.prepare_kwargs()["phase_mode"] == "levels" + assert plan.factory_kwargs()["group_by_time"] is True + + +def test_resolve_off_floor_sets_bernoulli_node_and_risk_difference(): + plan = resolve_level_factors_run_plan(_spec(likelihood="bernoulli_offfloor")) + assert plan.off_floor + assert plan.obs_node == "y_offfloor" + assert "risk difference" in plan.estimand + + +def test_resolve_splits_adjust_for_by_wave(): + # deapp_c (speech) is a language-proximal confounder → baseline (t1) timing so + # the t2 contrast is not conditioned on a treatment-affected descendant; hs + # (hearing) is exogenous → contemporaneous (post). Mirrors #247 timing. + plan = resolve_level_factors_run_plan( + _spec(ability_covariate="blocks", adjust_for=("hs", "deapp_c")) + ) + assert "blocks" in plan.baseline_covariates + assert "deapp_c" in plan.baseline_covariates + assert "hs" in plan.post_covariates + + +def test_factory_kwargs_apply_effective_adjustment(): + plan = resolve_level_factors_run_plan(_spec(adjust_for=("hs", "deapp_c"))) + kw = plan.factory_kwargs(effective_adjustment=("hs",)) + assert kw["adjust_for"] == ("hs",) + + +def test_typed_settings_are_accepted_and_sourced(): + spec = ModelSpec( + model_id="lrp-rli-lf-000", + kind="level_factors", + title="test", + outcome_symbol="W", + model_settings=LevelFactorsModelSettings(ability_covariate="blocks"), + ) + plan = resolve_level_factors_run_plan(spec) + assert plan.settings_source == "typed" + assert plan.ability_covariate == "blocks" + + +def test_split_settings_between_typed_and_extra_is_rejected(): + spec = ModelSpec( + model_id="lrp-rli-lf-000", + kind="level_factors", + title="test", + outcome_symbol="W", + model_settings=LevelFactorsModelSettings(), + extra={"ability_covariate": "blocks"}, + ) + with pytest.raises(ValueError, match="cannot be split"): + resolve_level_factors_run_plan(spec) + + +# --- registered-specification coverage (acceptance criterion) ----------------- + + +def test_every_registered_level_factor_model_resolves_with_metadata(): + """Every registered level-factor model resolves to a validated plan that records + the design, estimand, causal status, analysis population and missing-data + assumption (#389 findings 4 & 6 acceptance criteria).""" + specs = _level_factor_specs() + assert len(specs) >= 11, f"expected the full level-factor suite, found {len(specs)}" + for spec in specs: + plan = resolve_level_factors_run_plan(spec) + assert isinstance(plan, LevelFactorsRunPlan) + recorded = plan.as_dict() + for field in _META_FIELDS: + assert isinstance(recorded[field], str) and recorded[field], ( + f"{spec.model_id}: {field} not recorded" + ) + # The outcome is always loaded as its own (only) outcome. + assert plan.prepare_kwargs()["outcomes"] == (spec.outcome_symbol,)