diff --git a/factors/compute/candidates.py b/factors/compute/candidates.py
index 7770e6d..ea8a135 100644
--- a/factors/compute/candidates.py
+++ b/factors/compute/candidates.py
@@ -98,6 +98,10 @@ def spec(self) -> FactorSpec:
             overnight_boundary=base.overnight_boundary,
             family="reversal",
             min_history_bars=base.min_history_bars,
+            # D4 pre-registration: reversal IS -momentum, same inputs, so the
+            # transitive lookback depth is inherited verbatim from the base spec
+            # (close[t] and close[t-window] -> window + 1 trailing trading days).
+            lookback_depth=base.lookback_depth,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
@@ -162,6 +166,11 @@ def spec(self) -> FactorSpec:
             # pct_change loses row 0 and the rolling std needs ``window`` returns
             # -> the leading ``window`` rows are NaN.
             min_history_bars=self._window,
+            # D4 pre-registration: the value at t is std of returns[t-window+1..t],
+            # and returns[s]=close[s]/close[s-1]-1, so the earliest input close is
+            # close[t-window] -> transitive depth = window + 1 trailing trading days
+            # (the pct_change adds one day beyond the headline window; §六.18).
+            lookback_depth=self._window + 1,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
@@ -233,6 +242,9 @@ def spec(self) -> FactorSpec:
             # rolling mean over ``window`` amounts -> first valid row is
             # ``window-1`` (no pct_change involved, unlike volatility).
             min_history_bars=self._window - 1,
+            # D4 pre-registration: the rolling mean at t reads amount[t-window+1..t]
+            # -> transitive depth = window trailing trading days (no return diff).
+            lookback_depth=self._window,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
@@ -324,6 +336,10 @@ def spec(self) -> FactorSpec:
             # row 0 has no prior close and the rolling sum needs ``window`` full
             # overnight returns -> the leading ``window`` rows are NaN.
             min_history_bars=self._window,
+            # D4 pre-registration: term t reads open[t] and close[t-1]; summed over
+            # t-window+1..t the earliest input is close[t-window] -> transitive
+            # depth = window + 1 trailing trading days (§六.18).
+            lookback_depth=self._window + 1,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
@@ -395,6 +411,9 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="value",
             min_history_bars=0,
+            # D4 pre-registration: a bare same-day column select (1/pe or 1/pb);
+            # the value at d reads only d's published ratio -> depth = 1.
+            lookback_depth=1,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/financial.py b/factors/compute/financial.py
index 97ef172..604d5fe 100644
--- a/factors/compute/financial.py
+++ b/factors/compute/financial.py
@@ -88,6 +88,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family=family,
             min_history_bars=0,
+            # D4 pre-registration: a bare same-day column select of the
+            # ann_date-aligned value; the value at d reads only the single latest
+            # as-of column entry -> depth = 1 (the ann_date lookback lives in the
+            # 400-day fina revision horizon, not in this trailing-window depth).
+            lookback_depth=1,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/amp_marginal_anomaly_vol.py b/factors/compute/minute/amp_marginal_anomaly_vol.py
index bbf3780..4bb68c4 100644
--- a/factors/compute/minute/amp_marginal_anomaly_vol.py
+++ b/factors/compute/minute/amp_marginal_anomaly_vol.py
@@ -326,6 +326,10 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18): a trailing ``lookback_days`` trading-day
+            # pool of (|Δamp|, r) pairs with NO nested baseline (the 5min bars are
+            # derived within-day) -> depth = lookback_days trading days.
+            lookback_depth=self._lookback_days,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/binding.py b/factors/compute/minute/binding.py
new file mode 100644
index 0000000..da5fca7
--- /dev/null
+++ b/factors/compute/minute/binding.py
@@ -0,0 +1,249 @@
+"""Minute-factor raw-compute binding for the D4 materializer (design §3.5).
+
+The materializer is factor-agnostic: it loads cutoff-filtered 1min bars and asks
+"compute this minute factor's raw daily series from these bars". Each minute
+Factor's ``compute(panel)`` only SURFACES a pre-aggregated column (D2), so this
+module binds every minute factor to the free ``compute_*`` function that actually
+does the aggregation, parameterized by the factor INSTANCE (its ``lookback_days``
+and ``name``). All other definition constants stay at their module defaults —
+the SAME single-source constants the eval runners / ``qt.panel_freeze`` recipes
+pass (they pass them explicitly but equal to these defaults), so a materialized
+minute factor matches the frozen baseline's math.
+
+This lives in ``factors/compute/minute`` (the factor layer) rather than duplicated
+in ``qt`` because it is FACTOR knowledge; ``qt.panel_freeze`` keeps its own
+recipe copy (a frozen D1 tool) and D6 may later fold it onto this binding.
+
+Deferred (readable error, D5): ``valley_price_quantile`` also needs the DAILY
+close panel (its reversal neutralization), so it does not fit the pure
+``(factor, bars) -> series`` shape and is NOT bound here — the materializer
+raises for it rather than silently mis-computing.
+
+Layering: factor layer only (never qt / feeds).
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+import pandas as pd
+
+from factors.base import Factor
+from factors.compute.minute.amp_marginal_anomaly_vol import (
+    AmpMarginalAnomalyVolFactor,
+    compute_amp_marginal_anomaly_vol,
+)
+from factors.compute.minute.intraday_amp_cut import (
+    IntradayAmpCutFactor,
+    compute_intraday_amp_cut,
+)
+from factors.compute.minute.jump_amount_corr import (
+    JumpAmountCorrFactor,
+    compute_jump_amount_corr,
+)
+from factors.compute.minute.minute_ideal_amplitude import (
+    MinuteIdealAmplitudeFactor,
+    compute_minute_ideal_amplitude,
+)
+from factors.compute.minute.peak_interval_kurtosis import (
+    PeakIntervalKurtosisFactor,
+    compute_peak_interval_kurtosis,
+)
+from factors.compute.minute.peak_ridge_amount_ratio import (
+    PeakRidgeAmountRatioFactor,
+    compute_peak_ridge_amount_ratio,
+)
+from factors.compute.minute.ridge_minute_return import (
+    RidgeMinuteReturnFactor,
+    compute_ridge_minute_return,
+)
+from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor
+from factors.compute.minute.valley_relative_vwap import (
+    ValleyRelativeVwapFactor,
+    compute_valley_relative_vwap,
+)
+from factors.compute.minute.valley_ridge_vwap_ratio import (
+    ValleyRidgeVwapRatioFactor,
+    compute_valley_ridge_vwap_ratio,
+)
+from factors.compute.minute.primitives import VOLUME_PRV_BASELINE_DAYS
+from factors.compute.minute.volume_peak_count import (
+    VolumePeakCountFactor,
+    compute_volume_peak_count,
+)
+
+#: bars -> daily raw factor Series, parameterized by the factor instance. The
+#: cutoff is already applied by the materializer (bars are pre-filtered to
+#: ``available_time <= decision``), so the compute functions run with their
+#: DEFAULT decision_time (14:50) as a redundant-but-consistent internal cutoff.
+BindingFn = Callable[[Factor, pd.DataFrame], pd.Series]
+
+
+def _bind(compute_fn) -> BindingFn:
+    """A ``(factor, bars)`` binding that calls ``compute_fn`` with the instance's
+    lookback + name and every other definition constant at its module default."""
+
+    def _call(factor: Factor, bars: pd.DataFrame) -> pd.Series:
+        return compute_fn(
+            bars, lookback_days=factor.lookback_days, name=factor.name  # type: ignore[attr-defined]
+        )
+
+    return _call
+
+
+#: factor class -> its bars-based raw compute. Bound: the 10 minute factors whose
+#: raw compute needs ONLY the 1min bars. NOT bound: valley_price_quantile (needs
+#: the daily panel too — deferred, readable error below).
+_MINUTE_BINDINGS: dict[type[Factor], BindingFn] = {
+    JumpAmountCorrFactor: _bind(compute_jump_amount_corr),
+    MinuteIdealAmplitudeFactor: _bind(compute_minute_ideal_amplitude),
+    AmpMarginalAnomalyVolFactor: _bind(compute_amp_marginal_anomaly_vol),
+    VolumePeakCountFactor: _bind(compute_volume_peak_count),
+    IntradayAmpCutFactor: _bind(compute_intraday_amp_cut),
+    PeakIntervalKurtosisFactor: _bind(compute_peak_interval_kurtosis),
+    ValleyRelativeVwapFactor: _bind(compute_valley_relative_vwap),
+    ValleyRidgeVwapRatioFactor: _bind(compute_valley_ridge_vwap_ratio),
+    RidgeMinuteReturnFactor: _bind(compute_ridge_minute_return),
+    PeakRidgeAmountRatioFactor: _bind(compute_peak_ridge_amount_ratio),
+}
+
+#: Minute factors deliberately NOT bound (need extra inputs), with the reason.
+_DEFERRED: dict[type[Factor], str] = {
+    ValleyPriceQuantileFactor: (
+        "valley_price_quantile also consumes the DAILY close panel (its reversal "
+        "neutralization), so it does not fit the pure (factor, bars) binding; the "
+        "materializer defers it to D5 (a readable error, never a silent mis-compute)."
+    ),
+}
+
+
+#: VALID-DAY POOLED factors (design §3.3 review HIGH): their trailing window
+#: counts VALID days (invalid days do NOT occupy a slot — they roll over a
+#: valid-day-indexed series via ``primitives.rolling_valid_days`` or an equivalent
+#: ``.loc[valid_days].rolling(...)`` / valid-day python loop), so the CALENDAR
+#: lookback depth of a trailing ``lookback_days`` valid-day pool is DATA-DEPENDENT
+#: and UNBOUNDED, and loading more history can also re-classify boundary days.
+#: A fixed ``lookback_depth`` trim therefore truncates sparse-valid-day symbols'
+#: pools, making the stored value depend on LOAD GEOMETRY (red line #6: the value
+#: must be f(factor, params, view, data), never the anchor/window shape). The
+#: materializer must load these to SATURATION (design's structural terminal =
+#: real data start) so single-fill and batch-fill agree. Classification is
+#: STRUCTURAL (by the rolling mechanism, code-inspected), NOT by observed
+#: divergence: only ridge_minute_return / valley_ridge_vwap_ratio /
+#: peak_ridge_amount_ratio happened to diverge on the review's data, but
+#: volume_peak_count / peak_interval_kurtosis / intraday_amp_cut /
+#: valley_relative_vwap roll over valid days too and are the "happens-to-be-clean"
+#: representatives (#82 lesson: never fix only what you observed).
+#: valley_price_quantile is DEFERRED (needs the daily panel) but declared HERE so
+#: D5 cannot bind it carrying the fixed-depth defect.
+#: pooled factor class -> its same-slot BASELINE depth in trading days (the
+#: "locking offset", review point 1): a loaded day is CLASSIFICATION-FINAL only
+#: once it has this many strictly-prior trading days (the baseline reaches full
+#: depth and no earlier history can change it). Derived from each factor's own
+#: baseline parameter, NOT hardcoded: the peak family shares
+#: ``VOLUME_PRV_BASELINE_DAYS``; ``intraday_amp_cut`` has NO cross-day baseline
+#: (its classification is within-day), so its locking offset is 0.
+_POOLED_BASELINE_DAYS: dict[type[Factor], int] = {
+    VolumePeakCountFactor: VOLUME_PRV_BASELINE_DAYS,
+    PeakIntervalKurtosisFactor: VOLUME_PRV_BASELINE_DAYS,
+    ValleyRelativeVwapFactor: VOLUME_PRV_BASELINE_DAYS,
+    ValleyRidgeVwapRatioFactor: VOLUME_PRV_BASELINE_DAYS,
+    RidgeMinuteReturnFactor: VOLUME_PRV_BASELINE_DAYS,
+    PeakRidgeAmountRatioFactor: VOLUME_PRV_BASELINE_DAYS,
+    ValleyPriceQuantileFactor: VOLUME_PRV_BASELINE_DAYS,
+    IntradayAmpCutFactor: 0,  # within-day classification, no cross-day baseline
+}
+
+VALID_DAY_POOLED_FACTORS: frozenset[type[Factor]] = frozenset(_POOLED_BASELINE_DAYS)
+
+#: The bounded minute factors (trailing window over ALL trading days, not valid
+#: days) — a fixed lookback_depth trim is load-geometry-free for these. Kept as
+#: an explicit set so the classification is a CLOSED partition of the minute
+#: surface (a new minute factor missing from BOTH sets is a readable error).
+_BOUNDED_MINUTE_FACTORS: frozenset[type[Factor]] = frozenset({
+    JumpAmountCorrFactor,
+    MinuteIdealAmplitudeFactor,
+    AmpMarginalAnomalyVolFactor,
+})
+
+
+def pooled_baseline_days(factor: Factor) -> int:
+    """The factor's same-slot baseline depth = the LOCKING OFFSET (trading days).
+
+    Expanding the load backward can only change the classification of days that
+    sit within ``baseline_days`` trading days of the loaded window's start (the
+    rolling baseline takes the ``baseline_days`` most recent STRICTLY-PRIOR
+    same-slot observations; once it is at full depth, earlier history cannot
+    change it). So the locking is POSITION-MONOTONE: drop the first
+    ``baseline_days`` trading days of the loaded window and every later day's
+    classification is FINAL. Readable error for a non-pooled factor.
+    """
+    cls = type(factor)
+    if cls not in _POOLED_BASELINE_DAYS:
+        raise KeyError(
+            f"{factor.name} ({cls.__name__}) is not a valid-day-pooled factor, so "
+            f"it has no baseline locking offset."
+        )
+    return int(_POOLED_BASELINE_DAYS[cls])
+
+
+def pooled_lookback_days(factor: Factor) -> int:
+    """The factor's trailing pool size in VALID days (its ``lookback_days``)."""
+    if type(factor) not in _POOLED_BASELINE_DAYS:
+        raise KeyError(f"{factor.name} is not a valid-day-pooled factor.")
+    return int(factor.lookback_days)  # type: ignore[attr-defined]
+
+
+def is_valid_day_pooled(factor: Factor) -> bool:
+    """True iff ``factor``'s trailing window counts VALID days (unbounded depth).
+
+    Readable error for a minute factor in NEITHER partition set — a new minute
+    factor must be classified explicitly (never silently treated as bounded,
+    which would reintroduce the load-geometry divergence).
+    """
+    cls = type(factor)
+    if cls in VALID_DAY_POOLED_FACTORS:
+        return True
+    if cls in _BOUNDED_MINUTE_FACTORS:
+        return False
+    raise KeyError(
+        f"{factor.name} ({cls.__name__}) is a minute factor not classified as "
+        f"valid-day-pooled or bounded in factors.compute.minute.binding. Add it to "
+        f"exactly one partition set (VALID_DAY_POOLED_FACTORS if its trailing "
+        f"window counts VALID days, else _BOUNDED_MINUTE_FACTORS) — a missing "
+        f"classification would silently size its saturation load wrong."
+    )
+
+
+def is_minute_bound(factor: Factor) -> bool:
+    """True iff ``factor`` has a bars-only raw-compute binding here."""
+    return type(factor) in _MINUTE_BINDINGS
+
+
+def minute_raw_from_bars(factor: Factor, bars: pd.DataFrame) -> pd.Series:
+    """Compute ``factor``'s raw daily Series from (cutoff-filtered) 1min ``bars``.
+
+    Readable error for a minute factor that is deferred (needs extra inputs) or a
+    non-minute-bound factor — never a silent wrong result.
+    """
+    binding = _MINUTE_BINDINGS.get(type(factor))
+    if binding is not None:
+        return binding(factor, bars)
+    deferred = _DEFERRED.get(type(factor))
+    if deferred is not None:
+        raise NotImplementedError(f"{factor.name}: {deferred}")
+    raise KeyError(
+        f"{factor.name} ({type(factor).__name__}) has no minute-bars binding; it "
+        f"is not a minute-derived factor bound in factors.compute.minute.binding."
+    )
+
+
+__all__ = [
+    "VALID_DAY_POOLED_FACTORS",
+    "BindingFn",
+    "is_minute_bound",
+    "is_valid_day_pooled",
+    "minute_raw_from_bars",
+    "pooled_baseline_days",
+    "pooled_lookback_days",
+]
diff --git a/factors/compute/minute/intraday_amp_cut.py b/factors/compute/minute/intraday_amp_cut.py
index ba10de0..f54bfbe 100644
--- a/factors/compute/minute/intraday_amp_cut.py
+++ b/factors/compute/minute/intraday_amp_cut.py
@@ -448,6 +448,10 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18): a trailing ``lookback_days`` VALID
+            # trading-day window with NO nested baseline (the amplitude-cut stats
+            # are within-day) -> depth = lookback_days trading days.
+            lookback_depth=self._lookback_days,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/jump_amount_corr.py b/factors/compute/minute/jump_amount_corr.py
index 8109d24..5bb2dca 100644
--- a/factors/compute/minute/jump_amount_corr.py
+++ b/factors/compute/minute/jump_amount_corr.py
@@ -260,6 +260,10 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18): the value at d is a trailing-window
+            # correlation over ``lookback_days`` trading days with NO nested
+            # baseline -> transitive depth = lookback_days trailing trading days.
+            lookback_depth=self._lookback_days,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/minute_ideal_amplitude.py b/factors/compute/minute/minute_ideal_amplitude.py
index 280c258..c121a7b 100644
--- a/factors/compute/minute/minute_ideal_amplitude.py
+++ b/factors/compute/minute/minute_ideal_amplitude.py
@@ -273,6 +273,9 @@ def spec(self) -> FactorSpec:
             overnight_boundary="crossed_disclosed",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18): a trailing ``lookback_days`` trading-day
+            # pool with NO nested baseline -> depth = lookback_days trading days.
+            lookback_depth=self._lookback_days,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/peak_interval_kurtosis.py b/factors/compute/minute/peak_interval_kurtosis.py
index d804069..baffa6a 100644
--- a/factors/compute/minute/peak_interval_kurtosis.py
+++ b/factors/compute/minute/peak_interval_kurtosis.py
@@ -400,6 +400,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline ->
+            # transitive depth = lookback_days + baseline_days trailing trading days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/peak_ridge_amount_ratio.py b/factors/compute/minute/peak_ridge_amount_ratio.py
index f606969..6c00387 100644
--- a/factors/compute/minute/peak_ridge_amount_ratio.py
+++ b/factors/compute/minute/peak_ridge_amount_ratio.py
@@ -530,6 +530,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline ->
+            # transitive depth = lookback_days + baseline_days trailing trading days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/ridge_minute_return.py b/factors/compute/minute/ridge_minute_return.py
index 0739569..b43f439 100644
--- a/factors/compute/minute/ridge_minute_return.py
+++ b/factors/compute/minute/ridge_minute_return.py
@@ -494,6 +494,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline ->
+            # transitive depth = lookback_days + baseline_days trailing trading days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/valley_price_quantile.py b/factors/compute/minute/valley_price_quantile.py
index d344bdd..0505a9e 100644
--- a/factors/compute/minute/valley_price_quantile.py
+++ b/factors/compute/minute/valley_price_quantile.py
@@ -696,6 +696,13 @@ def spec(self) -> FactorSpec:
             overnight_boundary="crossed_disclosed",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline
+            # (lookback_days + baseline_days); the daily reversal neutralized
+            # against spans only ``VALLEY_QUANTILE_REVERSAL_DAYS`` (< that sum), so
+            # the minute chain dominates -> depth = lookback_days + baseline_days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/valley_relative_vwap.py b/factors/compute/minute/valley_relative_vwap.py
index ab5a938..542fe60 100644
--- a/factors/compute/minute/valley_relative_vwap.py
+++ b/factors/compute/minute/valley_relative_vwap.py
@@ -407,6 +407,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline ->
+            # transitive depth = lookback_days + baseline_days trailing trading days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/valley_ridge_vwap_ratio.py b/factors/compute/minute/valley_ridge_vwap_ratio.py
index f50681b..f4a8954 100644
--- a/factors/compute/minute/valley_ridge_vwap_ratio.py
+++ b/factors/compute/minute/valley_ridge_vwap_ratio.py
@@ -490,6 +490,11 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): trailing
+            # ``lookback_days`` valid days, each bar classified against a
+            # strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot baseline ->
+            # transitive depth = lookback_days + baseline_days trailing trading days.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/minute/volume_peak_count.py b/factors/compute/minute/volume_peak_count.py
index 07da815..a4c59d0 100644
--- a/factors/compute/minute/volume_peak_count.py
+++ b/factors/compute/minute/volume_peak_count.py
@@ -293,6 +293,12 @@ def spec(self) -> FactorSpec:
             overnight_boundary="none",
             family="microstructure",
             min_history_bars=0,
+            # D4 pre-registration (§六.18, NESTED lookback): a value at d pools the
+            # trailing ``lookback_days`` valid days, and each bar is classified
+            # against its strictly-prior ``VOLUME_PRV_BASELINE_DAYS``-day same-slot
+            # baseline, so the earliest input reaches lookback_days + baseline_days
+            # trailing trading days -> that sum is the transitive depth.
+            lookback_depth=self._lookback_days + VOLUME_PRV_BASELINE_DAYS,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/compute/momentum.py b/factors/compute/momentum.py
index 208f603..2aa8861 100644
--- a/factors/compute/momentum.py
+++ b/factors/compute/momentum.py
@@ -96,6 +96,10 @@ def spec(self) -> FactorSpec:
             # value at t needs bars t-window..t -> the leading ``window`` rows
             # of every symbol are NaN by construction.
             min_history_bars=self._window,
+            # D4 pre-registration: close[t]/close[t-window]-1 reads close over the
+            # inclusive span t-window..t -> transitive depth = window + 1 trailing
+            # trading days (the denominator adds one day beyond the headline; §六.18).
+            lookback_depth=self._window + 1,
         )
 
     def compute(self, panel: pd.DataFrame) -> pd.Series:
diff --git a/factors/materialize.py b/factors/materialize.py
new file mode 100644
index 0000000..e7d8a12
--- /dev/null
+++ b/factors/materialize.py
@@ -0,0 +1,513 @@
+"""The ONE vectorized factor-value engine (factor-refactor D4, design §3.5).
+
+The materializer is the single path that turns injected data-layer inputs into
+RAW factor values for a view over a date range. ``factors.service`` read-through
+routes every store miss here; nothing else computes factor values. It is
+factor-agnostic and view-agnostic-compute-preserving (design §1.4 mechanism 3):
+
+* it reads the factor's ``requires`` to decide DAILY vs MINUTE input;
+* it pulls inputs from INJECTED providers (layering red line #3/#10: ``factors``
+  never touches a feed, a token, or qt — the consumer injects the data access);
+* it applies the (source, view) availability lag from ``factors.view_lag`` (R18:
+  daily prev-day shift with the field-level ``open`` exception; minute
+  ``available_time <= d 14:50`` cutoff; PIT as-of is a no-op here);
+* it trims the input to EXACTLY the factor's ``lookback_depth`` trailing trading
+  days before the emit window, then computes and emits — so a single-date fill
+  and a batch fill produce bit-identical values (design §3.5 P8);
+* ``masked`` factors have their ex-date rows NaN'd via the BOOLEAN ex-date mask
+  (§1.3 note 4; the ``adj_factor`` numeric never enters the frame). The closing
+  14 factors are all non-masked, so this is a no-op for them (fixture-tested).
+
+The daily factor path calls ``factor.compute`` on the lagged panel; the minute
+path calls the ``factors.compute.minute.binding`` for the factor and the
+cutoff-filtered bars. The forward-return boundary is elsewhere: a factor value
+never sees a future return (invariant #1) — the materializer only reads history.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import Protocol
+
+import pandas as pd
+
+from data.availability_policy import STK_MINS_1MIN, OvernightBoundary, View
+from data.clean.intraday_schema import DEFAULT_DECISION_TIME
+from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL
+from factors.base import Factor
+from factors.compute.minute.binding import (
+    is_minute_bound,
+    is_valid_day_pooled,
+    minute_raw_from_bars,
+    pooled_baseline_days,
+    pooled_lookback_days,
+)
+from factors.store.incremental import CacheHorizonConfig
+from factors.view_lag import (
+    daily_decision_lag,
+    ex_date_mask,
+    minute_decision_cutoff,
+)
+
+#: Extra calendar days loaded before the emit window so the exact trailing-trading
+#: -day trim always has enough history (covers weekends + holiday clusters + the
+#: daily shift's one extra day). Generous by design: the CORRECTNESS floor is the
+#: trim to ``lookback_depth`` trading days, not this buffer (design §3.5 P8).
+def _load_buffer_calendar_days(warmup: int) -> int:
+    return int(warmup) * 2 + 25
+
+
+#: One backward-expansion step (calendar days) of the valid-day-pooled saturation
+#: loop. Only a STEP SIZE (how fast the search walks back), never a correctness
+#: input: termination is decided by the structural criterion in
+#: ``_materialize_pooled`` or by the provider's declared earliest boundary, so a
+#: too-small chunk costs iterations, never correctness.
+def _saturation_chunk_calendar_days(warmup: int) -> int:
+    return int(warmup) * 2 + 40
+
+
+#: Hard ceiling on saturation-expansion chunks (defensive; the loop terminates
+#: structurally on value-stability or provider exhaustion long before this).
+_MAX_SATURATION_CHUNKS: int = 500
+
+
+# --------------------------------------------------------------------------- #
+# Injected data providers (the consumer wires these to feed/cache; tests fake)
+# --------------------------------------------------------------------------- #
+class DailyPanelProvider(Protocol):
+    """Loads the CLOSE-view daily factor-input panel (front-adjusted + enriched)."""
+
+    def daily_panel(
+        self, symbols: list[str], start: pd.Timestamp, end: pd.Timestamp
+    ) -> pd.DataFrame:
+        """MultiIndex(date, symbol) panel, values dated at their natural close date."""
+
+
+class MinuteBarProvider(Protocol):
+    """Loads normalized 1min bars (cache-only) for the symbols over the window.
+
+    Providers feeding a VALID-DAY-POOLED factor must also implement
+    :meth:`earliest_available` — the saturation loop needs a RELIABLE data-start
+    signal. Inferring "no more history" from an unchanged row count is WRONG: a
+    long mid-history gap (a suspension) yields an unchanged count while real
+    earlier history still exists (the review's counterexample #2).
+    """
+
+    def minute_bars(
+        self, symbols: list[str], start: pd.Timestamp, end: pd.Timestamp
+    ) -> pd.DataFrame:
+        """MultiIndex(time, symbol) bars (:mod:`data.clean.intraday_schema`)."""
+
+    def earliest_available(self, symbols: list[str]) -> pd.Timestamp:
+        """The earliest date for which ANY of ``symbols`` could have bars.
+
+        The declared lower bound of the provider's data (a cache-coverage floor
+        or a documented constant). Loading from it means "all history that
+        exists is loaded"; a mid-history gap must NEVER be mistaken for it.
+        """
+
+
+class AdjFactorProvider(Protocol):
+    """Loads the RAW adj_factor series (ONLY consumed as an ex-date BOOLEAN)."""
+
+    def adj_factor(
+        self, symbols: list[str], start: pd.Timestamp, end: pd.Timestamp
+    ) -> pd.Series:
+        """MultiIndex(date, symbol) raw adj_factor (never enters the factor frame)."""
+
+
+@dataclass(frozen=True)
+class MaterializeSources:
+    """The injected data access the materializer needs (consumer-wired)."""
+
+    daily: DailyPanelProvider | None = None
+    minute: MinuteBarProvider | None = None
+    adj_factor: AdjFactorProvider | None = None
+
+
+# --------------------------------------------------------------------------- #
+# Live-config wiring (R5 / D3-note: NEVER the CacheHorizonConfig default)
+# --------------------------------------------------------------------------- #
+def build_horizon_config(cache_cfg: object) -> CacheHorizonConfig:
+    """Build the incremental-overlap horizon config from the LIVE cache config.
+
+    Reads ``refresh_recent_days`` and ``fina_tail_days`` off ``cache_cfg`` (duck
+    typed, so ``qt.config`` DataCacheCfg works without a factors->qt import). Both
+    are REQUIRED: the ``CacheHorizonConfig`` 400 default is a test convenience and
+    a config whose fina tail differs from 400 would silently mis-size the fina
+    overlap window (R5) — so a missing attribute is a readable error, never the
+    default.
+    """
+    missing = [a for a in ("refresh_recent_days", "fina_tail_days") if not hasattr(cache_cfg, a)]
+    if missing:
+        raise ValueError(
+            f"build_horizon_config needs the live cache config's {missing} "
+            f"attribute(s); refusing to fall back to the CacheHorizonConfig default "
+            f"(R5: a differing fina tail would silently mis-size the overlap)."
+        )
+    return CacheHorizonConfig(
+        refresh_recent_days=int(cache_cfg.refresh_recent_days),
+        fina_tail_days=int(cache_cfg.fina_tail_days),
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Factor kind
+# --------------------------------------------------------------------------- #
+def is_minute_factor(factor: Factor) -> bool:
+    """True iff any of the factor's declared inputs comes from the 1min endpoint."""
+    return any(r.source == STK_MINS_1MIN for r in (factor.spec.requires or ()))
+
+
+# --------------------------------------------------------------------------- #
+# Trailing-trading-day trim (the P8 correctness floor)
+# --------------------------------------------------------------------------- #
+def _warmup_start(dates: pd.DatetimeIndex, emit_start: pd.Timestamp, warmup: int):
+    """The date ``warmup`` trading days before ``emit_start`` (or the earliest).
+
+    ``dates`` is the sorted unique set of trading dates present in the loaded
+    input. Returns the cutoff date to keep input from; if fewer than ``warmup``
+    trading days precede ``emit_start`` the earliest available date is used
+    (near the data start the emit rows are honestly under-warmed -> NaN, and
+    single/batch stay consistent because both trim to the same earliest date).
+    """
+    order = dates.sort_values()
+    pos = int(order.searchsorted(emit_start, side="left"))  # index of emit_start (or ins.)
+    keep_idx = max(0, pos - int(warmup))
+    return order[keep_idx]
+
+
+# --------------------------------------------------------------------------- #
+# The engine
+# --------------------------------------------------------------------------- #
+def materialize_range(
+    factor: Factor,
+    *,
+    view: object,
+    symbols: list[str],
+    emit_start: pd.Timestamp,
+    emit_end: pd.Timestamp,
+    sources: MaterializeSources,
+    decision_cutoff: str = DEFAULT_DECISION_TIME,
+    warmup: int | None = None,
+) -> pd.Series:
+    """Compute ``factor``'s RAW values for ``view`` over ``[emit_start, emit_end]``.
+
+    Loads a generous window (so the exact trailing trim always has history),
+    applies the view lag, trims to ``warmup`` (= ``spec.lookback_depth``) trailing
+    trading days before ``emit_start``, computes, and returns the emit-window
+    rows. Single-date and batch calls give bit-identical values (design §3.5 P8).
+    """
+    resolved_view = View(view)
+    emit_start = pd.Timestamp(emit_start).normalize()
+    emit_end = pd.Timestamp(emit_end).normalize()
+    w = int(factor.spec.lookback_depth) if warmup is None else int(warmup)
+    if w < 1:
+        raise ValueError(f"{factor.name}: warmup must be >= 1; got {w}.")
+    load_start = emit_start - pd.Timedelta(days=_load_buffer_calendar_days(w))
+
+    if is_minute_factor(factor):
+        raw = _materialize_minute(
+            factor, resolved_view, symbols, load_start, emit_start, emit_end, w,
+            sources, decision_cutoff,
+        )
+    else:
+        raw = _materialize_daily(
+            factor, resolved_view, symbols, load_start, emit_start, emit_end, w, sources,
+        )
+
+    raw = _apply_ex_date_mask(
+        factor, raw, resolved_view, symbols, load_start, emit_end, sources
+    )
+    return _restrict_emit(raw, emit_start, emit_end, factor.name)
+
+
+def _materialize_daily(
+    factor, view, symbols, load_start, emit_start, emit_end, warmup, sources,
+) -> pd.Series:
+    if sources.daily is None:
+        raise ValueError(
+            f"{factor.name} is a daily factor but no DailyPanelProvider was injected."
+        )
+    panel = sources.daily.daily_panel(list(symbols), load_start, emit_end)
+    if panel.empty:
+        return _empty_series(factor.name)
+    if view is View.DECISION:
+        panel = daily_decision_lag(panel)  # prev-day shift, open same-day (R18)
+    trimmed = _trim_daily(panel, emit_start, warmup)
+    return factor.compute(trimmed).rename(factor.name)
+
+
+def _materialize_minute(
+    factor, view, symbols, load_start, emit_start, emit_end, warmup, sources, decision_cutoff,
+) -> pd.Series:
+    if sources.minute is None:
+        raise ValueError(
+            f"{factor.name} is a minute factor but no MinuteBarProvider was injected."
+        )
+    if not is_minute_bound(factor):
+        # Readable error (e.g. valley_price_quantile needs the daily panel too).
+        return minute_raw_from_bars(factor, sources.minute.minute_bars([], load_start, emit_end))
+    # Load one extra calendar day past emit_end so bar cutoffs on emit_end resolve.
+    # POOLED (valid-day trailing window): the pool counts VALID days, so its
+    # CALENDAR depth is data-dependent and UNBOUNDED — a fixed trim would make the
+    # value depend on load geometry (review HIGH / red line #6). Load by
+    # SATURATION-EXPANSION instead (below); bounded factors keep the fixed trim.
+    if is_valid_day_pooled(factor):
+        return _materialize_pooled(
+            factor, view, list(symbols), emit_start, emit_end, warmup, sources, decision_cutoff,
+        )
+    bars = sources.minute.minute_bars(
+        list(symbols), load_start, emit_end + pd.Timedelta(days=1)
+    )
+    if bars.empty:
+        return _empty_series(factor.name)
+    if view is View.DECISION:
+        bars = minute_decision_cutoff(bars, decision_time=decision_cutoff)
+    bars = _trim_minute(bars, emit_start, warmup)
+    return minute_raw_from_bars(factor, bars)
+
+
+def _materialize_pooled(
+    factor, view, symbols, emit_start, emit_end, warmup, sources, decision_cutoff,
+) -> pd.Series:
+    """Saturation-expanding load for a valid-day-pooled factor (design §3.3, review HIGH).
+
+    STRUCTURAL saturation criterion (NOT a value-stability fixed point — that was
+    disproved: one expansion chunk landing entirely inside a long no-bar gap
+    leaves the emit values unchanged and would falsely terminate).
+
+    The locking argument (why the criterion is SUFFICIENT):
+
+    * Expanding the load backward can only change the classification of days
+      within ``baseline_days`` trading days of the loaded window's start — the
+      rolling baseline takes the ``baseline_days`` most recent STRICTLY-PRIOR
+      same-slot observations, and once at full depth no earlier history can move
+      it. Locking is therefore POSITION-MONOTONE: **drop the first
+      ``baseline_days`` trading days of the loaded window and every later day's
+      classification is FINAL.**
+    * The trailing pool takes only the most recent ``lookback_days`` VALID days.
+      So if the LOCKED sub-window already contains >= ``lookback_days`` final
+      valid days at or before the earliest emit date, the pool for every emit
+      date is drawn entirely from final days and can never reach back into the
+      unlocked head — further history cannot change any emit value.
+
+    Termination: the criterion above, OR the load start reaching the provider's
+    DECLARED earliest boundary (then all existing history is loaded and the value
+    is well-defined). A mid-history gap is never mistaken for the data start
+    (that is exactly the row-count inference this replaced).
+
+    Valid days are counted as the factor's own OUTPUT dates: a pooled factor
+    emits one row per valid day, so output dates ARE the valid days — the count
+    is read off the engine's own result rather than a re-implementation of each
+    factor's validity rule, and a gap simply contributes no output dates (the
+    count stalls and the loop keeps expanding, which is the desired behaviour).
+    """
+    chunk = pd.Timedelta(days=_saturation_chunk_calendar_days(warmup))
+    end = emit_end + pd.Timedelta(days=1)
+    floor = _provider_earliest(sources.minute, symbols, factor)
+    baseline_days = pooled_baseline_days(factor)
+    lookback_days = pooled_lookback_days(factor)
+
+    load_start = emit_start - chunk
+    last: pd.Series | None = None
+    for _ in range(_MAX_SATURATION_CHUNKS):
+        at_floor = load_start <= floor
+        if at_floor:
+            load_start = floor
+        bars = sources.minute.minute_bars(symbols, load_start, end)
+        if bars.empty:
+            if at_floor:
+                return _empty_series(factor.name)
+            load_start = load_start - chunk
+            continue
+        work = bars
+        if view is View.DECISION:
+            work = minute_decision_cutoff(work, decision_time=decision_cutoff)
+        full = minute_raw_from_bars(factor, work)
+        last = _restrict_emit(full, emit_start, emit_end, factor.name)
+        if at_floor or _pooled_pool_saturated(
+            full, work, emit_start,
+            baseline_days=baseline_days, lookback_days=lookback_days,
+            symbols=symbols,
+        ):
+            return last
+        load_start = load_start - chunk
+    # Never silently return an UNSATURATED result (red line #9: no silent
+    # degradation). Unreachable in practice — 500 chunks is >160 years of history
+    # — so hitting it means the loop is not converging, which must be loud.
+    raise RuntimeError(
+        f"{factor.name}: pooled saturation did not converge after "
+        f"{_MAX_SATURATION_CHUNKS} expansion chunks (expanded back to "
+        f"{load_start.date()} for {len(symbols)} symbol(s), declared floor "
+        f"{floor.date()}). Refusing to return an unsaturated value, which would "
+        f"depend on load geometry."
+    )
+
+
+def _provider_earliest(provider, symbols, factor) -> pd.Timestamp:
+    """The provider's DECLARED earliest-available date (never inferred)."""
+    getter = getattr(provider, "earliest_available", None)
+    if getter is None:
+        raise ValueError(
+            f"{factor.name} is a valid-day-pooled factor, so its saturation load "
+            f"needs a RELIABLE data-start signal, but the minute provider "
+            f"({type(provider).__name__}) does not implement earliest_available(). "
+            f"Inferring the data start from an unchanged row count is unsound — a "
+            f"long mid-history no-bar gap (a suspension) looks identical to it. "
+            f"Declare the provider's earliest available date."
+        )
+    return pd.Timestamp(getter(list(symbols))).normalize()
+
+
+def _pooled_pool_saturated(
+    full: pd.Series,
+    bars: pd.DataFrame,
+    emit_start: pd.Timestamp,
+    *,
+    baseline_days: int,
+    lookback_days: int,
+    symbols: list[str],
+) -> bool:
+    """The structural criterion (see :func:`_materialize_pooled` for the argument).
+
+    True iff, for EVERY REQUESTED symbol, the LOCKED sub-window (the loaded
+    trading days after dropping the first ``baseline_days``) holds at least
+    ``lookback_days`` valid days at or before ``emit_start`` — valid days being
+    the factor's own output dates.
+
+    ITERATING THE REQUESTED SYMBOLS, NOT THE OUTPUT SYMBOLS, IS LOAD-BEARING
+    (review HIGH): a symbol that produces ZERO rows in the current load window (a
+    long suspension, thin coverage) is absent from the output, so iterating the
+    output would give it NO VETO — yet it is exactly the symbol that a deeper load
+    would unlock. The consequence was worse than a wrong value: a single-date fill
+    dropped such a name from the cross-section entirely while a batch fill kept it
+    (measured: volume_peak_count_20 single ``rows=0`` vs batch ``rows=10``), which
+    is the silent name-dropping the I5a red line forbids (a missing symbol is a
+    sample-coverage bias, and must be loud). Counting an absent symbol as 0 valid
+    days vetoes termination, so the loop expands until it too is saturated or the
+    declared floor is reached.
+
+    COST, stated plainly: any requested symbol that cannot accumulate
+    ``lookback_days`` valid days before ``emit_start`` (a recent listing, a long
+    suspension, a coverage hole) drags the WHOLE universe's load back to the
+    declared floor. That cost already existed for symbols with SOME output but too
+    few valid days; this fix only extends the same conservative behaviour to
+    zero-output symbols. A per-symbol saturation start is the D5 optimization —
+    and NOT a pure one: ``intraday_amp_cut`` z-scores across the loaded
+    cross-section, so changing per-symbol load depth changes its cross-section
+    composition.
+    """
+    loaded_days = pd.DatetimeIndex(
+        pd.unique(pd.DatetimeIndex(bars.index.get_level_values("time")).normalize())
+    ).sort_values()
+    if len(loaded_days) <= baseline_days:
+        return False
+    locked_from = loaded_days[baseline_days]  # first day with a FINAL classification
+
+    out_dates = full.index.get_level_values(DATE_LEVEL)
+    out_symbols = pd.Index(full.index.get_level_values(SYMBOL_LEVEL))
+    for symbol in symbols:  # the REQUESTED universe, not just what produced rows
+        mine = out_dates[out_symbols == str(symbol)]
+        final_valid = mine[(mine >= locked_from) & (mine <= emit_start)]
+        if len(final_valid) < lookback_days:
+            return False
+    return True
+
+
+def _trim_daily(panel: pd.DataFrame, emit_start: pd.Timestamp, warmup: int) -> pd.DataFrame:
+    dates = pd.DatetimeIndex(pd.unique(panel.index.get_level_values(DATE_LEVEL)))
+    keep_from = _warmup_start(dates, emit_start, warmup)
+    return panel[panel.index.get_level_values(DATE_LEVEL) >= keep_from]
+
+
+def _trim_minute(bars: pd.DataFrame, emit_start: pd.Timestamp, warmup: int) -> pd.DataFrame:
+    bar_dates = pd.DatetimeIndex(bars.index.get_level_values("time")).normalize()
+    trading = pd.DatetimeIndex(pd.unique(bar_dates))
+    keep_from = _warmup_start(trading, emit_start, warmup)
+    return bars[bar_dates >= keep_from]
+
+
+def _apply_ex_date_mask(factor, raw, view, symbols, load_start, emit_end, sources):
+    """NaN out ex-date rows for a ``masked`` factor (§1.3 note 4).
+
+    Non-masked factors (all 14 closing) skip this entirely — the adj_factor
+    provider is not even consulted, so no ex-date machinery runs for them.
+    """
+    if factor.spec.overnight_boundary is not OvernightBoundary.MASKED:
+        return raw
+    if view is not View.DECISION or raw.empty:
+        return raw
+    if sources.adj_factor is None:
+        raise ValueError(
+            f"{factor.name} declares overnight_boundary='masked' but no "
+            f"AdjFactorProvider was injected to resolve ex-dates."
+        )
+    af = sources.adj_factor.adj_factor(list(symbols), load_start, emit_end)
+    mask = ex_date_mask(af).reindex(raw.index, fill_value=False)
+    out = raw.copy()
+    out[mask.to_numpy()] = float("nan")
+    return out
+
+
+def _restrict_emit(raw, emit_start, emit_end, name) -> pd.Series:
+    if raw.empty:
+        return _empty_series(name)
+    dates = raw.index.get_level_values(DATE_LEVEL)
+    within = (dates >= emit_start) & (dates <= emit_end)
+    return raw[within].sort_index(kind="mergesort").rename(name)
+
+
+def _empty_series(name: str) -> pd.Series:
+    index = pd.MultiIndex.from_arrays(
+        [pd.DatetimeIndex([]), pd.Index([], dtype=object)],
+        names=[DATE_LEVEL, SYMBOL_LEVEL],
+    )
+    return pd.Series([], index=index, dtype=float, name=name)
+
+
+def make_recompute_fn(
+    factor: Factor,
+    *,
+    view: object,
+    symbols: list[str],
+    sources: MaterializeSources,
+    data_start: pd.Timestamp,
+    decision_cutoff: str = DEFAULT_DECISION_TIME,
+) -> Callable[[pd.Timestamp | None, pd.Timestamp, int], pd.Series]:
+    """A ``RecomputeFn`` for the D3 tail-recompute engine (design §3.3).
+
+    ``recompute(emit_start, end, warmup)`` -> the factor's raw Series over
+    ``[emit_start, end]`` having loaded ``warmup`` trailing trading days of input
+    (``emit_start is None`` -> from ``data_start``). The same trailing-trim floor
+    as ``materialize_range`` keeps the incremental overlap bit-identical to a full
+    column (the D3 batch=incremental invariant).
+    """
+
+    def _recompute(emit_start, end, warmup) -> pd.Series:
+        start = data_start if emit_start is None else emit_start
+        return materialize_range(
+            factor,
+            view=view,
+            symbols=list(symbols),
+            emit_start=start,
+            emit_end=end,
+            sources=sources,
+            decision_cutoff=decision_cutoff,
+            warmup=warmup,
+        )
+
+    return _recompute
+
+
+__all__ = [
+    "AdjFactorProvider",
+    "DailyPanelProvider",
+    "MaterializeSources",
+    "MinuteBarProvider",
+    "build_horizon_config",
+    "is_minute_factor",
+    "make_recompute_fn",
+    "materialize_range",
+]
diff --git a/factors/process/orchestrate.py b/factors/process/orchestrate.py
new file mode 100644
index 0000000..1c78d3c
--- /dev/null
+++ b/factors/process/orchestrate.py
@@ -0,0 +1,75 @@
+"""The single home of the factor-processing ORCHESTRATION (design §3.5 R9, D4).
+
+``drop_missing + winsorize + neutralize + zscore`` is a cross-sectional,
+covariate-dependent transform whose ORCHESTRATION previously lived in
+``qt.pipeline._process_factors`` and was consumed by 10+ modules. The D5 unified
+eval runner will build in the analytics layer, where red line #10 forbids
+``analytics -> qt`` — so the orchestration cannot stay in ``qt``. Left there, the
+eval and backtest paths would each copy it (the P3-6 ``drop_missing``-across-
+columns lesson) or break layering.
+
+It is homed HERE, in ``factors/process`` (where the ``neutralize_by_date``
+primitive already lives), so the eval and backtest paths share ONE processing
+truth. The covariates (industry / market_cap) are INJECTED by the caller —
+``factors`` never touches a feed (red line #3) — and the store stays raw-only
+(red line #7): processing is a consumer-side decision applied to the RAW factor
+panel the service returns, never persisted as a source of truth.
+
+Behavior is preserved verbatim: this constructs the SAME
+:class:`~factors.process.pipeline.ProcessingPipeline` with the SAME arguments the
+retired ``_process_factors`` did and calls ``transform`` — so the phase0 anchor
+and every existing processing number are byte-for-byte unchanged.
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from factors.process.pipeline import ProcessingPipeline
+
+
+def covariates_from_panel(
+    panel: pd.DataFrame,
+) -> tuple[pd.Series | None, pd.Series | None]:
+    """Pull the ``industry`` / ``market_cap`` neutralization covariates off a panel.
+
+    Returns ``(industry, market_cap)``, each ``None`` when the column is absent
+    (the retired ``_process_factors`` did exactly this). The caller placed the
+    covariates on the panel (e.g. ``_maybe_enrich_covariates``); ``factors`` never
+    fetches them itself.
+    """
+    industry = panel["industry"] if "industry" in panel.columns else None
+    market_cap = panel["market_cap"] if "market_cap" in panel.columns else None
+    return industry, market_cap
+
+
+def process_factor_panel(
+    factor_panel: pd.DataFrame,
+    *,
+    drop_missing: bool = True,
+    standardize: bool = True,
+    winsorize: bool = False,
+    neutralize: bool = False,
+    industry: pd.Series | None = None,
+    market_cap: pd.Series | None = None,
+) -> pd.DataFrame:
+    """Run the cross-sectional processing pipeline over a RAW factor panel.
+
+    The ONE orchestration point (design §3.5 R9): eval and backtest both call
+    this, so ``drop_missing`` (cross-column, per date) and neutralization can
+    never drift between the two paths. Covariates are injected; when
+    ``neutralize`` is enabled but they are absent, ``ProcessingPipeline`` raises a
+    readable error rather than silently skipping.
+    """
+    pipeline = ProcessingPipeline(
+        drop_missing=drop_missing,
+        standardize=standardize,
+        winsorize=winsorize,
+        neutralize=neutralize,
+        industry=industry,
+        market_cap=market_cap,
+    )
+    return pipeline.transform(factor_panel)
+
+
+__all__ = ["covariates_from_panel", "process_factor_panel"]
diff --git a/factors/service.py b/factors/service.py
new file mode 100644
index 0000000..8d0dc1b
--- /dev/null
+++ b/factors/service.py
@@ -0,0 +1,217 @@
+"""Factor service: the thin read-through onto the value store (design §3.5, D4).
+
+Two functions and one frozen data class — NOT a behavior-class stack (R25). Both
+functions READ-THROUGH the value store: a hit returns the stored RAW values, a
+miss triggers the ONE materializer engine (``factors.materialize``) to fill the
+gap and append, then re-reads. There is no second implementation path, which is
+what makes ``panel[d] ≡ cross_section(d)`` a trivial read-layer smoke and the
+real guarantee ``single-point fill ≡ batch fill`` (design §3.5 P8).
+
+Contract (all structural, not comments):
+* only data with ``available_time <= decision`` is computed — the materializer
+  applies the (source, view) availability lag; the service never relaxes it;
+* it NEVER touches execution prices — forward returns are born only at the
+  alpha/eval boundary (invariant #1); this returns raw factor VALUES only;
+* it returns RAW values (processed is a consumer-side decision, red line #7);
+* the view x return-basis pairing is enforced at CALL time via the D0/D1
+  ``require_legal_pairing`` (decision<->exec_to_exec / close<->close_to_close);
+  any other pairing is a readable error, not a doc convention.
+
+Layering (red line #10): ``factors.service`` imports the factor layer + the pure
+availability leaf only; the data access (store + providers) is INJECTED by the
+consumer, so ``factors`` never carries a feed, a token, or a qt import.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable, Mapping
+from dataclasses import dataclass
+
+import pandas as pd
+
+from data.availability_policy import ReturnBasis, View, require_legal_pairing
+from data.clean.intraday_schema import DEFAULT_DECISION_TIME
+from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL
+from factors import registry as factor_registry
+from factors.base import Factor
+from factors.materialize import MaterializeSources, materialize_range
+from factors.store.fingerprint import data_fingerprint
+from factors.store.keys import store_key
+from factors.store.values import FactorValueStore
+
+
+@dataclass(frozen=True)
+class DecisionPoint:
+    """A decision instant: the date and the within-day cutoff (default 14:50).
+
+    ``cutoff`` is the same string the FactorSpec intraday block declares as its
+    ``decision_cutoff`` (single source); the materializer filters minute bars to
+    ``available_time <= date + cutoff``.
+    """
+
+    date: pd.Timestamp
+    cutoff: str = DEFAULT_DECISION_TIME
+
+
+def _build_factor(factor_id: str, params_by_id: Mapping[str, Mapping[str, object]] | None) -> Factor:
+    params = (params_by_id or {}).get(factor_id)
+    return factor_registry.build(factor_id, params)
+
+
+def _fingerprint(factor: Factor) -> dict:
+    """The store data fingerprint for ``factor`` (none/returns_invariant here).
+
+    price_level factors need the per-symbol adj_factor event table; the closing
+    14 have zero such members, so ``data_fingerprint`` raises readably for them
+    (never a silent wrong fingerprint) until the service wires an adj-factor
+    source for that case (deferred with the ex-date masking).
+    """
+    return data_fingerprint(adjustment=factor.spec.adjustment)
+
+
+def _ensure_coverage(
+    factor: Factor,
+    key,
+    fingerprint: dict,
+    *,
+    dates: pd.DatetimeIndex,
+    symbols: list[str],
+    store: FactorValueStore,
+    sources: MaterializeSources,
+    view: View,
+    cutoff: str,
+) -> pd.Series:
+    """Read-through: fill any date the store lacks for ``factor``, return the Series.
+
+    A miss over the requested ``dates`` materializes ``[min(missing), max(missing)]``
+    (the single engine) and upserts it; already-present dates in that range are
+    recomputed identically and win the dedup harmlessly. Returns the stored Series
+    (possibly empty). The per-date fill (dates = one) and the batch fill (dates =
+    many) leave BIT-IDENTICAL stores because the materializer's trailing trim is
+    warmup-consistent (design §3.5 P8).
+    """
+    stored = store.read_valid(key, expected_fingerprint=fingerprint)
+    have = (
+        pd.DatetimeIndex(pd.unique(stored.index.get_level_values(DATE_LEVEL)))
+        if stored is not None and not stored.empty
+        else pd.DatetimeIndex([])
+    )
+    missing = dates[~dates.isin(have)]
+    if len(missing):
+        fresh = materialize_range(
+            factor,
+            view=view,
+            symbols=list(symbols),
+            emit_start=missing.min(),
+            emit_end=missing.max(),
+            sources=sources,
+            decision_cutoff=cutoff,
+        )
+        if not fresh.empty:
+            store.upsert(key, fresh, fingerprint=fingerprint)
+        stored = store.read_valid(key, expected_fingerprint=fingerprint)
+    if stored is None:
+        return pd.Series(
+            [],
+            index=pd.MultiIndex.from_arrays(
+                [pd.DatetimeIndex([]), pd.Index([], dtype=object)],
+                names=[DATE_LEVEL, SYMBOL_LEVEL],
+            ),
+            dtype=float,
+            name=factor.name,
+        )
+    return stored
+
+
+def _uniform_cutoff(decisions: list[DecisionPoint]) -> str:
+    cutoffs = {d.cutoff for d in decisions}
+    if len(cutoffs) != 1:
+        raise ValueError(
+            f"panel() requires all DecisionPoints to share one cutoff; got {sorted(cutoffs)}. "
+            f"Split the call per cutoff (the minute cutoff enters the store view semantics)."
+        )
+    return next(iter(cutoffs))
+
+
+def _assemble(
+    factor_ids: list[str],
+    series_by_id: dict[str, pd.Series],
+    dates: pd.DatetimeIndex,
+    symbols: list[str],
+) -> pd.DataFrame:
+    """Slice each factor's stored Series to (dates x universe) and stack to columns."""
+    symbol_set = set(map(str, symbols))
+    columns: dict[str, pd.Series] = {}
+    for fid in factor_ids:
+        s = series_by_id[fid]
+        if s.empty:
+            columns[fid] = s
+            continue
+        d = s.index.get_level_values(DATE_LEVEL)
+        sym = s.index.get_level_values(SYMBOL_LEVEL)
+        keep = d.isin(dates) & pd.Index(sym).isin(symbol_set)
+        columns[fid] = s[keep]
+    frame = pd.concat(columns, axis=1)
+    frame.index = frame.index.set_names([DATE_LEVEL, SYMBOL_LEVEL])
+    return frame.sort_index(kind="mergesort")
+
+
+def panel(
+    factor_ids: Iterable[str],
+    universe: Iterable[str],
+    decisions: list[DecisionPoint],
+    *,
+    store: FactorValueStore,
+    sources: MaterializeSources,
+    view: object = View.DECISION,
+    basis: object = ReturnBasis.EXEC_TO_EXEC,
+    params_by_id: Mapping[str, Mapping[str, object]] | None = None,
+) -> pd.DataFrame:
+    """Raw factor values for ``factor_ids`` over ``decisions`` x ``universe``.
+
+    Read-through: any decision date the store lacks is materialized once (batch)
+    and appended. The view x basis pairing is enforced up front.
+    """
+    resolved_view, _ = require_legal_pairing(view, basis)
+    factor_ids = list(factor_ids)
+    symbols = [str(s) for s in universe]
+    if not decisions:
+        raise ValueError("panel() needs at least one DecisionPoint.")
+    cutoff = _uniform_cutoff(decisions)
+    dates = pd.DatetimeIndex(sorted({pd.Timestamp(d.date).normalize() for d in decisions}))
+
+    series_by_id: dict[str, pd.Series] = {}
+    for fid in factor_ids:
+        factor = _build_factor(fid, params_by_id)
+        fp = _fingerprint(factor)
+        key = store_key(factor, view=resolved_view.value, params=(params_by_id or {}).get(fid))
+        series_by_id[fid] = _ensure_coverage(
+            factor, key, fp, dates=dates, symbols=symbols, store=store,
+            sources=sources, view=resolved_view, cutoff=cutoff,
+        )
+    return _assemble(factor_ids, series_by_id, dates, symbols)
+
+
+def cross_section(
+    factor_ids: Iterable[str],
+    universe: Iterable[str],
+    decision: DecisionPoint,
+    *,
+    store: FactorValueStore,
+    sources: MaterializeSources,
+    view: object = View.DECISION,
+    basis: object = ReturnBasis.EXEC_TO_EXEC,
+    params_by_id: Mapping[str, Mapping[str, object]] | None = None,
+) -> pd.DataFrame:
+    """Raw factor values for ``factor_ids`` at one ``decision`` x ``universe``.
+
+    A convenience over :func:`panel` with a single decision — the SAME read-through
+    and the SAME engine, so ``panel[d] ≡ cross_section(d)`` on the same store.
+    """
+    return panel(
+        factor_ids, universe, [decision], store=store, sources=sources,
+        view=view, basis=basis, params_by_id=params_by_id,
+    )
+
+
+__all__ = ["DecisionPoint", "cross_section", "panel"]
diff --git a/factors/view_lag.py b/factors/view_lag.py
new file mode 100644
index 0000000..9e8c052
--- /dev/null
+++ b/factors/view_lag.py
@@ -0,0 +1,146 @@
+"""Availability-lag transforms for the two views (factor-refactor D4, §1.3/R18).
+
+PURE functions (no data access, no feed, no qt). Given an already-prepared
+close-view daily panel or normalized 1min bars, produce the view-lagged input a
+factor's ``compute`` consumes — so the SAME view-agnostic factor code (design
+§1.4 mechanism 3) serves both views. The materializer composes these.
+
+The decision(d) view is the operational information set at the 14:50 decision:
+
+* daily fields are visible only at ``<= d-1`` EXCEPT ``open`` (fixed at 09:30,
+  known by 14:50 — the ONE field-level same-day exception, R7). Concretely the
+  prev-day columns are shifted forward one trading day PER SYMBOL, so the row at
+  date d carries the value that was final at d-1; ``open`` is left same-day.
+  This is exactly ``<= d-1`` for every prev-day field: a factor reading the
+  shifted column at row d therefore uses only d-1 information for it (and a
+  factor's own trailing lag composes on top consistently). The same-day set is
+  DERIVED from the policy table (:func:`decision_same_day_fields`), never a
+  hand-kept literal.
+* minute bars: only bars with ``available_time <= d 14:50`` survive (the I3
+  cutoff), filtered on per-bar timestamps BEFORE any daily grouping.
+
+The ex-date boundary (§1.3 note 4) is a SEPARATE, type-boundary-clean module:
+:func:`ex_date_mask` turns a raw ``adj_factor`` series into a BOOLEAN mask
+(``af(d) != af(d-1)`` per symbol) and nothing else — the ``adj_factor`` numeric
+never enters the decision-view frame, so "only judge the fact, never use the
+value" is enforced by the return type, not by docstring discipline.
+
+Leaf discipline: numpy / pandas + the pure ``data.availability_policy`` leaf and
+``data.clean.intraday_schema`` constants. Never import qt, feeds, or caches.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from data.availability_policy import (
+    AVAILABILITY_POLICY,
+    DecisionVisibility,
+    RevisionHorizonSource,
+)
+from data.clean.intraday_schema import DEFAULT_DECISION_TIME
+from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL
+
+
+def decision_same_day_fields() -> frozenset[str]:
+    """Factor-input fields legal SAME-DAY in the decision view (policy-derived).
+
+    Reads the §1.3 policy table: a field is decision-view same-day iff its rule
+    is ``DecisionVisibility.SAME_DAY`` AND it is a factor-store input endpoint
+    (``horizon_source != NOT_FACTOR_INPUT`` — stk_limit / suspend_d / namechange
+    are same-day too but never feed a factor, so they are excluded). Today this
+    is exactly ``{"open"}`` (market_daily's field-level exception), but it is
+    derived so a policy-table change flows through automatically.
+    """
+    out: set[str] = set()
+    for (_source, field), rule in AVAILABILITY_POLICY.items():
+        if field is None:
+            continue
+        if (
+            rule.decision_visibility is DecisionVisibility.SAME_DAY
+            and rule.horizon_source is not RevisionHorizonSource.NOT_FACTOR_INPUT
+        ):
+            out.add(field)
+    return frozenset(out)
+
+
+def daily_decision_lag(
+    panel: pd.DataFrame, *, same_day_fields: frozenset[str] | None = None
+) -> pd.DataFrame:
+    """Shift the close-view daily panel into the decision(d) view (R18 daily lag).
+
+    Every column is shifted forward one trading day PER SYMBOL (row d gets row
+    d-1's value = ``<= d-1`` visibility) EXCEPT the same-day fields (``open``),
+    which are left untouched (09:30 fixed, legal at 14:50). The first row of each
+    symbol has no d-1 for the shifted columns, so it becomes NaN (a decision-view
+    value cannot exist on a symbol's very first date). Never mutates ``panel``.
+    """
+    if not isinstance(panel.index, pd.MultiIndex) or list(panel.index.names) != [
+        DATE_LEVEL,
+        SYMBOL_LEVEL,
+    ]:
+        raise ValueError(
+            "daily_decision_lag expects a MultiIndex(date, symbol) panel; got "
+            f"index names {list(panel.index.names)}."
+        )
+    same_day = decision_same_day_fields() if same_day_fields is None else same_day_fields
+    ordered = panel.sort_index(kind="mergesort")
+    shift_cols = [c for c in ordered.columns if c not in same_day]
+    if not shift_cols:
+        return ordered.copy()
+    out = ordered.copy()
+    shifted = ordered[shift_cols].groupby(level=SYMBOL_LEVEL, sort=False).shift(1)
+    out[shift_cols] = shifted
+    return out
+
+
+def minute_decision_cutoff(
+    bars: pd.DataFrame, *, decision_time: str = DEFAULT_DECISION_TIME
+) -> pd.DataFrame:
+    """Keep only bars visible at the decision cutoff (``available_time <= d cutoff``).
+
+    The cutoff for each bar is its own trade date + ``decision_time`` (default
+    14:50). The filter runs on per-bar timestamps (never date-normalized first),
+    so a post-cutoff bar can never leak into a decision. Never mutates ``bars``.
+    """
+    if bars.empty:
+        return bars.copy()
+    bar_end = bars.index.get_level_values("time")
+    trade_date = pd.DatetimeIndex(bar_end).normalize()
+    cutoff = trade_date + pd.Timedelta(decision_time)
+    visible = bars["available_time"].to_numpy() <= cutoff.to_numpy()
+    return bars.loc[visible].copy()
+
+
+def ex_date_mask(adj_factor: pd.Series) -> pd.Series:
+    """BOOLEAN ex-date mask ``af(d) != af(d-1)`` per symbol (§1.3 note 4).
+
+    The ONLY thing this module exposes to the materializer: the fact "day d is an
+    ex-date for this symbol", never the ``adj_factor`` numeric. The first row of
+    each symbol is False (no d-1 to compare). ``af`` is compared with NaN-aware
+    inequality so a missing raw factor is not spuriously flagged.
+    """
+    if not isinstance(adj_factor.index, pd.MultiIndex) or list(
+        adj_factor.index.names
+    ) != [DATE_LEVEL, SYMBOL_LEVEL]:
+        raise ValueError(
+            "ex_date_mask expects a MultiIndex(date, symbol) adj_factor series; got "
+            f"index names {list(adj_factor.index.names)}."
+        )
+    ordered = adj_factor.sort_index(kind="mergesort")
+    prev = ordered.groupby(level=SYMBOL_LEVEL, sort=False).shift(1)
+    cur_v = ordered.to_numpy(dtype=float)
+    prev_v = prev.to_numpy(dtype=float)
+    both_nan = np.isnan(cur_v) & np.isnan(prev_v)
+    # differ, and not the (first-row) case where prev is NaN, and not both-NaN.
+    changed = (~(cur_v == prev_v)) & ~np.isnan(prev_v) & ~both_nan
+    return pd.Series(changed, index=ordered.index, name="ex_date")
+
+
+__all__ = [
+    "daily_decision_lag",
+    "decision_same_day_fields",
+    "ex_date_mask",
+    "minute_decision_cutoff",
+]
diff --git a/qt/factor_hotpath_smoke.py b/qt/factor_hotpath_smoke.py
new file mode 100644
index 0000000..4c7056f
--- /dev/null
+++ b/qt/factor_hotpath_smoke.py
@@ -0,0 +1,213 @@
+"""Hot-path timing smoke for the D4 materializer (design §八 R20).
+
+Measures, on the REAL intraday cache (cache-only, ``stk_mins_live_calls=0``), the
+magnitude gap between:
+
+* NAIVE per-factor loading — each factor materialized separately, so the 1min
+  read + normalize is paid ONCE PER FACTOR (what the D4 service does today: each
+  ``materialize_range`` call loads its own bars); and
+* SHARED loading — the 1min read + normalize + cutoff paid ONCE, then every
+  factor computed from the shared bars (the ``factors.compute.minute.primitives``
+  amortization the design §3.2 calls the load-bearing hot-path mechanism).
+
+The budget is MEASURED, not claimed (§八): the caliber (universe sample, window,
+factor count) and the two wall times are printed as-is, with an explicit caveat
+that the numbers do NOT extrapolate to full-A (~6x the CSI500 read).
+
+Run: ``python -m qt.factor_hotpath_smoke`` (deliberately NOT in qt/cli). It reads
+``artifacts/cache/tushare/v1`` (symlink the main checkout's artifacts into the
+worktree first; remove the symlink after so ``git status`` stays clean).
+"""
+
+from __future__ import annotations
+
+import argparse
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import pandas as pd
+
+from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT
+from data.cache.intraday_cache import READ_COLUMNS
+from data.cache.intraday_parquet_store import IntradayParquetStore
+from data.clean.intraday_schema import (
+    RAW_INTRADAY_FREQ,
+    empty_intraday_bars,
+    normalize_intraday_bars,
+)
+from factors import registry as factor_registry
+from factors.compute.minute.binding import minute_raw_from_bars
+from factors.view_lag import minute_decision_cutoff
+
+DEFAULT_CACHE_ROOT = "artifacts/cache/tushare/v1"
+#: The intraday cache's DECLARED earliest bar date (measured on the real cache:
+#: several CSI500 names carry 1min bars from 2015-01-05). The pooled saturation
+#: loop needs a declared floor; it must never infer one from row counts.
+CACHE_MINUTE_DATA_START = "2015-01-05"
+DEFAULT_FACTORS = (
+    "jump_amount_corr_20",
+    "minute_ideal_amp_10",
+    "amp_marginal_anomaly_vol_20",
+    "volume_peak_count_20",
+    "valley_relative_vwap_20",
+    "peak_ridge_amount_ratio_20",
+)
+DEFAULT_START = "2024-03-01"
+DEFAULT_END = "2024-04-05"
+DEFAULT_N_SYMBOLS = 40
+
+
+class CacheMinuteProvider:
+    """Cache-only MinuteBarProvider: per-symbol read + normalize, zero live calls."""
+
+    def __init__(self, root: str) -> None:
+        self._store = IntradayParquetStore(root)
+        self.calls = 0
+        self.live_calls = 0  # provably 0 — read_range has no fetch closure
+
+    def earliest_available(self, symbols):
+        """The cache's DECLARED minute-data floor (measured: bars from 2015-01-05).
+
+        Declared, never inferred from row counts — a long mid-history no-bar gap
+        (a suspension) is indistinguishable from exhaustion by row count, which
+        is exactly the unsound signal the pooled saturation loop refuses.
+        """
+        return pd.Timestamp(CACHE_MINUTE_DATA_START)
+
+    def minute_bars(self, symbols, start, end):
+        self.calls += 1
+        if not symbols:
+            return empty_intraday_bars()
+        parts = []
+        s = pd.Timestamp(start)
+        e = pd.Timestamp(end)
+        for sym in symbols:
+            part = self._store.read_range(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, s, e)
+            if part.empty:
+                continue
+            parts.append(
+                normalize_intraday_bars(
+                    part.rename(columns={"bar_end": "time"})[READ_COLUMNS], freq=RAW_INTRADAY_FREQ
+                )
+            )
+        if not parts:
+            return empty_intraday_bars()
+        return pd.concat(parts).sort_index(kind="mergesort")
+
+
+def _sample_symbols(root: str, n: int) -> list[str]:
+    """First ``n`` symbols the intraday store has, in a stable directory order."""
+    base = Path(root) / f"{INTRADAY_ENDPOINT}" / f"freq={RAW_INTRADAY_FREQ}"
+    syms: list[str] = []
+    for prefix_dir in sorted(base.glob("symbol_prefix=*")):
+        for sym_dir in sorted(prefix_dir.glob("symbol=*")):
+            syms.append(sym_dir.name.split("=", 1)[1])
+            if len(syms) >= n:
+                return syms
+    return syms
+
+
+@dataclass(frozen=True)
+class SmokeResult:
+    n_symbols: int
+    n_factors: int
+    window: str
+    naive_seconds: float
+    shared_seconds: float
+    ratio: float
+    naive_provider_calls: int
+    shared_provider_calls: int
+    live_calls: int
+    raw_bar_rows: int
+
+
+def run_hotpath_smoke(
+    *,
+    cache_root: str = DEFAULT_CACHE_ROOT,
+    factor_ids=DEFAULT_FACTORS,
+    start: str = DEFAULT_START,
+    end: str = DEFAULT_END,
+    n_symbols: int = DEFAULT_N_SYMBOLS,
+    cutoff: str = "14:50:00",
+) -> SmokeResult:
+    factor_ids = list(factor_ids)
+    symbols = _sample_symbols(cache_root, n_symbols)
+    factors = [factor_registry.build(fid) for fid in factor_ids]
+    emit_start = pd.Timestamp(start).normalize()
+    emit_end = pd.Timestamp(end).normalize()
+
+    # The smoke isolates the READ amortization only, so BOTH paths use the same
+    # BOUNDED load window (never the valid-day-pooled SATURATION expansion, which
+    # is a separate correctness path — its multi-load cost would confound a pure
+    # read-cost comparison). The bounded window comfortably covers each factor's
+    # trailing depth; the shared load uses the max depth so every factor is served.
+    max_depth = max(int(f.spec.lookback_depth) for f in factors)
+    end_bar = emit_end + pd.Timedelta(days=1)
+
+    # -- NAIVE: read + normalize + cutoff ONCE PER FACTOR (its own bounded window).
+    naive_provider = CacheMinuteProvider(cache_root)
+    t0 = time.monotonic()
+    for factor in factors:
+        w = int(factor.spec.lookback_depth)
+        ls = emit_start - pd.Timedelta(days=w * 2 + 25)
+        bars = minute_decision_cutoff(
+            naive_provider.minute_bars(symbols, ls, end_bar), decision_time=cutoff
+        )
+        raw = minute_raw_from_bars(factor, bars)
+        dd = raw.index.get_level_values("date")
+        _ = raw[(dd >= emit_start) & (dd <= emit_end)]
+    naive_seconds = time.monotonic() - t0
+
+    # -- SHARED: read + normalize + cutoff ONCE, then compute every factor.
+    shared_provider = CacheMinuteProvider(cache_root)
+    load_start = emit_start - pd.Timedelta(days=max_depth * 2 + 25)
+    t0 = time.monotonic()
+    bars = shared_provider.minute_bars(symbols, load_start, end_bar)
+    bars = minute_decision_cutoff(bars, decision_time=cutoff)
+    raw_rows = len(bars)
+    for factor in factors:
+        raw = minute_raw_from_bars(factor, bars)
+        dd = raw.index.get_level_values("date")
+        _ = raw[(dd >= emit_start) & (dd <= emit_end)]
+    shared_seconds = time.monotonic() - t0
+
+    return SmokeResult(
+        n_symbols=len(symbols),
+        n_factors=len(factors),
+        window=f"{start}..{end}",
+        naive_seconds=round(naive_seconds, 2),
+        shared_seconds=round(shared_seconds, 2),
+        ratio=round(naive_seconds / shared_seconds, 2) if shared_seconds else float("nan"),
+        naive_provider_calls=naive_provider.calls,
+        shared_provider_calls=shared_provider.calls,
+        live_calls=naive_provider.live_calls + shared_provider.live_calls,
+        raw_bar_rows=raw_rows,
+    )
+
+
+def main(argv=None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--cache-root", default=DEFAULT_CACHE_ROOT)
+    parser.add_argument("--start", default=DEFAULT_START)
+    parser.add_argument("--end", default=DEFAULT_END)
+    parser.add_argument("--n-symbols", type=int, default=DEFAULT_N_SYMBOLS)
+    args = parser.parse_args(argv)
+    r = run_hotpath_smoke(
+        cache_root=args.cache_root, start=args.start, end=args.end, n_symbols=args.n_symbols
+    )
+    print(f"caliber: {r.n_symbols} symbols x {r.n_factors} minute factors, window {r.window}")
+    print(f"raw 1min bar rows (shared load, post-cutoff): {r.raw_bar_rows}")
+    print(f"NAIVE  (per-factor load): {r.naive_seconds}s  ({r.naive_provider_calls} provider loads)")
+    print(f"SHARED (load once):       {r.shared_seconds}s  ({r.shared_provider_calls} provider load)")
+    print(f"naive / shared ratio:     {r.ratio}x")
+    print(f"stk_mins_live_calls:      {r.live_calls}  (cache-only)")
+    print(
+        "CAVEAT: measured on this CSI500 sample/window only; does NOT extrapolate "
+        "to full-A (~6x the read). The budget is measured, not claimed (§八)."
+    )
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover - thin CLI shim
+    raise SystemExit(main())
diff --git a/qt/pipeline.py b/qt/pipeline.py
index 91b341d..98e1c53 100644
--- a/qt/pipeline.py
+++ b/qt/pipeline.py
@@ -57,7 +57,7 @@
 from factors.compute.candidates import ValueFactor
 from factors.compute.financial import SUPPORTED_FIELDS as SUPPORTED_FINANCIAL_FIELDS
 from factors.compute.financial import FinancialFactor
-from factors.process.pipeline import ProcessingPipeline
+from factors.process.orchestrate import covariates_from_panel, process_factor_panel
 from portfolio.construct import TopNEqualWeight
 from qt.config import RootConfig, load_config
 from qt.reports import write_phase0_summary
@@ -967,13 +967,18 @@ def _process_factors(
 ) -> pd.DataFrame:
     """Run drop_missing + (winsorize) + neutralize + z-score from config toggles.
 
-    Neutralization pulls the ``industry`` / ``market_cap`` covariates off the panel
-    (placed there by :func:`_maybe_enrich_covariates`). When neutralize is enabled
-    but they are absent, ``ProcessingPipeline`` raises a readable error.
+    D4 (factor-refactor R9): a THIN delegate to
+    :func:`factors.process.orchestrate.process_factor_panel`, the single home of
+    the processing orchestration (so the eval and backtest paths share ONE truth
+    without ``analytics -> qt``). Behavior is preserved verbatim — same covariate
+    extraction, same pipeline arguments. Neutralization pulls the ``industry`` /
+    ``market_cap`` covariates off the panel (placed there by
+    :func:`_maybe_enrich_covariates`); when neutralize is enabled but they are
+    absent, the orchestrator's ``ProcessingPipeline`` raises a readable error.
     """
-    industry = panel["industry"] if "industry" in panel.columns else None
-    market_cap = panel["market_cap"] if "market_cap" in panel.columns else None
-    pipeline = ProcessingPipeline(
+    industry, market_cap = covariates_from_panel(panel)
+    return process_factor_panel(
+        factor_panel,
         drop_missing=cfg.processing.drop_missing,
         standardize=cfg.processing.standardize.enabled,
         winsorize=cfg.processing.winsorize.enabled,
@@ -981,7 +986,6 @@ def _process_factors(
         industry=industry,
         market_cap=market_cap,
     )
-    return pipeline.transform(factor_panel)
 
 
 def _maybe_enrich_covariates(
diff --git a/tests/test_factor_lookback_depth.py b/tests/test_factor_lookback_depth.py
new file mode 100644
index 0000000..4283591
--- /dev/null
+++ b/tests/test_factor_lookback_depth.py
@@ -0,0 +1,124 @@
+"""D4 Commit 1: the 14 closing factors' pre-registered ``lookback_depth``.
+
+The D3 tail-recompute engine HARD-REQUIRES ``spec.lookback_depth`` (a readable
+error, never a silent 0/headline fallback) to size its incremental overlap
+window. This test pins the pre-registered transitive lookback depth of every
+closing-set factor to a specific value derived from its CODE CONSTANTS (design
+§六.18: headline + nested baseline, NOT the headline window).
+
+Pinning is the point: once declared, the value is a pre-registration — changing
+it is a definition-adjacent change that must be disclosed separately. A wrong
+declaration under-samples the overlap and produces FALSE mismatches every night
+(the D3 engine already has teeth for that); this test is the up-front guard.
+
+The parameterization also proves the declaration is DERIVED from the constructor
+param, not a frozen literal: a non-default window/lookback yields the matching
+depth (so a re-parameterized factor cannot silently keep the default depth).
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from factors import registry as factor_registry
+from factors.compute.candidates import ValueFactor, VolatilityFactor
+from factors.compute.financial import FinancialFactor
+from factors.compute.minute.amp_marginal_anomaly_vol import AmpMarginalAnomalyVolFactor
+from factors.compute.minute.intraday_amp_cut import IntradayAmpCutFactor
+from factors.compute.minute.jump_amount_corr import JumpAmountCorrFactor
+from factors.compute.minute.minute_ideal_amplitude import MinuteIdealAmplitudeFactor
+from factors.compute.minute.peak_interval_kurtosis import PeakIntervalKurtosisFactor
+from factors.compute.minute.peak_ridge_amount_ratio import PeakRidgeAmountRatioFactor
+from factors.compute.minute.primitives import VOLUME_PRV_BASELINE_DAYS
+from factors.compute.minute.ridge_minute_return import RidgeMinuteReturnFactor
+from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor
+from factors.compute.minute.valley_relative_vwap import ValleyRelativeVwapFactor
+from factors.compute.minute.valley_ridge_vwap_ratio import ValleyRidgeVwapRatioFactor
+from factors.compute.minute.volume_peak_count import VolumePeakCountFactor
+from factors.compute.momentum import MomentumFactor
+
+# The 14 closing factors: (factor instance, expected lookback_depth).
+# 3 book factors + 11 minute-derived factors. Baseline (nested-lookback) minute
+# factors = lookback_days (20) + VOLUME_PRV_BASELINE_DAYS (20) = 40.
+_BASELINE = VOLUME_PRV_BASELINE_DAYS  # 20
+_CLOSING_FACTORS = [
+    # -- book factors -----------------------------------------------------
+    (ValueFactor("value_ep"), 1),
+    (ValueFactor("value_bp"), 1),
+    (VolatilityFactor(window=20), 21),  # window + 1 (pct_change adds a day)
+    # -- minute factors, no nested baseline -------------------------------
+    (JumpAmountCorrFactor(), 20),
+    (MinuteIdealAmplitudeFactor(), 10),
+    (AmpMarginalAnomalyVolFactor(), 20),
+    (IntradayAmpCutFactor(), 10),
+    # -- minute factors, nested baseline (lookback 20 + baseline 20) ------
+    (VolumePeakCountFactor(), 20 + _BASELINE),
+    (PeakIntervalKurtosisFactor(), 20 + _BASELINE),
+    (ValleyRelativeVwapFactor(), 20 + _BASELINE),
+    (ValleyRidgeVwapRatioFactor(), 20 + _BASELINE),
+    (RidgeMinuteReturnFactor(), 20 + _BASELINE),
+    (ValleyPriceQuantileFactor(), 20 + _BASELINE),
+    (PeakRidgeAmountRatioFactor(), 20 + _BASELINE),
+]
+
+
+@pytest.mark.parametrize(
+    "factor, expected",
+    _CLOSING_FACTORS,
+    ids=[f.name for f, _ in _CLOSING_FACTORS],
+)
+def test_closing_factor_lookback_depth(factor, expected):
+    """Each closing factor declares its exact pre-registered transitive depth."""
+    depth = factor.spec.lookback_depth
+    assert depth is not None, (
+        f"{factor.name}: lookback_depth is None — the store engine would refuse "
+        f"to size its overlap window (D3 hard-require)."
+    )
+    assert depth == expected, f"{factor.name}: lookback_depth {depth} != {expected}"
+    # It must be a positive int (contract v1.1 validates this at construction).
+    assert isinstance(depth, int) and not isinstance(depth, bool) and depth >= 1
+
+
+def test_closing_factors_count_is_fourteen():
+    """The pinned table is exactly the 14-factor closing set (no drift)."""
+    assert len(_CLOSING_FACTORS) == 14
+
+
+@pytest.mark.parametrize(
+    "factor, expected",
+    [
+        # window-parameterized: depth tracks the constructor param, not a literal.
+        (VolatilityFactor(window=10), 11),
+        (MomentumFactor(window=30), 31),
+        (JumpAmountCorrFactor(lookback_days=30), 30),
+        (MinuteIdealAmplitudeFactor(lookback_days=15), 15),
+        (VolumePeakCountFactor(lookback_days=15), 15 + _BASELINE),
+        (ValleyRidgeVwapRatioFactor(lookback_days=25), 25 + _BASELINE),
+    ],
+)
+def test_lookback_depth_tracks_the_constructor_param(factor, expected):
+    """A non-default window/lookback yields the matching depth (derived, not frozen)."""
+    assert factor.spec.lookback_depth == expected
+
+
+def test_non_closing_daily_factors_also_declare_depth():
+    """The remaining registered factors declare a depth too (no latent store break)."""
+    from factors.compute.candidates import (
+        LiquidityFactor,
+        OvernightMomentumFactor,
+        ReversalFactor,
+    )
+
+    assert MomentumFactor(window=20).spec.lookback_depth == 21
+    assert ReversalFactor(window=20).spec.lookback_depth == 21  # inherits momentum
+    assert LiquidityFactor(window=20).spec.lookback_depth == 20
+    assert OvernightMomentumFactor(window=20).spec.lookback_depth == 21
+    assert FinancialFactor("roe").spec.lookback_depth == 1
+
+
+def test_registry_built_factor_carries_the_same_depth():
+    """Building through the registry (the D4 service path) yields the same depth."""
+    built = factor_registry.build("jump_amount_corr_20")
+    assert built.spec.lookback_depth == 20
+    built_vol = factor_registry.build("volatility_20")
+    assert built_vol.spec.lookback_depth == 21
diff --git a/tests/test_factor_materialize.py b/tests/test_factor_materialize.py
new file mode 100644
index 0000000..eec609b
--- /dev/null
+++ b/tests/test_factor_materialize.py
@@ -0,0 +1,377 @@
+"""D4: the materializer engine (factors.materialize) — R18 lag + leakage + binding.
+
+Network-free. Covers, against the §九 D4 acceptance:
+* R18 full lag: daily prev-day fields at ``<= d-1``, field-level ``open`` same-day,
+  daily_basic/enriched at ``<= d-1``, minute ``available_time <= d 14:50``;
+* decision-after perturbation leaves the decision-view value bit-unchanged, and
+  perturbing ``open[d]`` DOES change it (the reverse teeth);
+* the minute binding is correct (close-view materialize == direct compute_X);
+* ex-date masking (a fixture ``masked`` factor; the 14 closing factors are all
+  non-masked so this path is a no-op for them);
+* the live-config wiring of ``build_horizon_config`` (no default fallback, R5).
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from data.availability_policy import View
+from data.clean.intraday_schema import normalize_intraday_bars
+from factors.base import Factor
+from factors.compute.candidates import OvernightMomentumFactor, ValueFactor, VolatilityFactor
+from factors.compute.minute.jump_amount_corr import (
+    JumpAmountCorrFactor,
+    compute_jump_amount_corr,
+)
+from factors.materialize import (
+    MaterializeSources,
+    build_horizon_config,
+    is_minute_factor,
+    make_recompute_fn,
+    materialize_range,
+)
+from factors.spec import FactorSpec, PanelField
+
+SYMS = ["000001.SZ", "000002.SZ"]
+DATES = pd.bdate_range("2021-01-04", periods=70)
+
+
+# --------------------------------------------------------------------------- #
+# Synthetic data + fake providers
+# --------------------------------------------------------------------------- #
+def _daily(perturb=None):
+    """Enriched close-view daily panel (+ a value_ep column) for the fakes.
+
+    ``perturb`` = (date, symbol, column, new_value) mutates one cell.
+    """
+    rng = np.random.RandomState(0)
+    rows = []
+    for si, s in enumerate(SYMS):
+        px = 100.0 + si * 20 + np.cumsum(rng.normal(0, 1.0, len(DATES)))
+        for d, p in zip(DATES, px):
+            rows.append((d, s, p - 0.3, p + 0.5, p - 0.5, p, 1e5, p * 1e5, 1.0 / (10.0 + p * 0.01)))
+    frame = pd.DataFrame(
+        rows, columns=["date", "symbol", "open", "high", "low", "close", "volume", "amount", "value_ep"]
+    ).set_index(["date", "symbol"]).sort_index()
+    if perturb is not None:
+        d, s, col, val = perturb
+        frame.loc[(d, s), col] = val
+    return frame
+
+
+def _minute(perturb=None):
+    """Two symbols, 40 days, a session spanning across 14:50 so the cutoff bites."""
+    rng = np.random.RandomState(1)
+    rows = []
+    for s in SYMS:
+        for d in DATES[:40]:
+            # bars from 14:40 to 14:59 (across the 14:50 cutoff)
+            base = pd.Timestamp(d) + pd.Timedelta("14:40:00")
+            for i in range(20):
+                t = base + pd.Timedelta(minutes=i)
+                o = 100.0 + rng.normal(0, 1)
+                w = 4.0 if i in (2, 6, 12) else 0.5
+                rows.append((t, s, o, o + w, o - w, o + rng.normal(0, 0.2), 1e4, 1e6 + rng.normal(0, 1e5)))
+    frame = pd.DataFrame(
+        rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"]
+    )
+    bars = normalize_intraday_bars(frame, freq="1min")
+    if perturb is not None:
+        t, s, col, val = perturb
+        bars.loc[(pd.Timestamp(t), s), col] = val
+    return bars
+
+
+class DailyProv:
+    def __init__(self, panel):
+        self._p = panel
+
+    def daily_panel(self, symbols, start, end):
+        m = self._p.index.get_level_values("date")
+        return self._p[(m >= pd.Timestamp(start)) & (m <= pd.Timestamp(end))]
+
+
+class MinuteProv:
+    def __init__(self, bars):
+        self._b = bars
+
+    def minute_bars(self, symbols, start, end):
+        if not symbols:
+            return self._b.iloc[0:0]
+        t = self._b.index.get_level_values("time")
+        return self._b[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))]
+
+
+def _sources(daily=None, minute=None, adj=None):
+    return MaterializeSources(
+        daily=DailyProv(daily) if daily is not None else None,
+        minute=MinuteProv(minute) if minute is not None else None,
+        adj_factor=adj,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Factor kind classification
+# --------------------------------------------------------------------------- #
+def test_is_minute_factor():
+    assert is_minute_factor(JumpAmountCorrFactor())
+    assert not is_minute_factor(VolatilityFactor(window=20))
+    assert not is_minute_factor(ValueFactor("value_ep"))
+
+
+# --------------------------------------------------------------------------- #
+# valid-day POOLED classification (review HIGH) — a CLOSED partition
+# --------------------------------------------------------------------------- #
+def test_valid_day_pooled_classification_is_pinned():
+    """Every minute factor is classified pooled (valid-day trailing window,
+    unbounded calendar depth) or bounded (all-trading-day window). Pinned by CODE
+    INSPECTION of the rolling mechanism, not by observed divergence (only 3 of the
+    8 pooled factors happened to diverge on the review's data — #82 lesson)."""
+    from factors.compute.minute.binding import is_valid_day_pooled
+    from factors.compute.minute.intraday_amp_cut import IntradayAmpCutFactor
+    from factors.compute.minute.minute_ideal_amplitude import MinuteIdealAmplitudeFactor
+    from factors.compute.minute.peak_interval_kurtosis import PeakIntervalKurtosisFactor
+    from factors.compute.minute.peak_ridge_amount_ratio import PeakRidgeAmountRatioFactor
+    from factors.compute.minute.ridge_minute_return import RidgeMinuteReturnFactor
+    from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor
+    from factors.compute.minute.valley_relative_vwap import ValleyRelativeVwapFactor
+    from factors.compute.minute.valley_ridge_vwap_ratio import ValleyRidgeVwapRatioFactor
+    from factors.compute.minute.volume_peak_count import VolumePeakCountFactor
+    from factors.compute.minute.amp_marginal_anomaly_vol import AmpMarginalAnomalyVolFactor
+
+    pooled = [
+        VolumePeakCountFactor(), PeakIntervalKurtosisFactor(), IntradayAmpCutFactor(),
+        ValleyRelativeVwapFactor(), ValleyRidgeVwapRatioFactor(), RidgeMinuteReturnFactor(),
+        PeakRidgeAmountRatioFactor(), ValleyPriceQuantileFactor(),
+    ]
+    bounded = [
+        JumpAmountCorrFactor(), MinuteIdealAmplitudeFactor(), AmpMarginalAnomalyVolFactor(),
+    ]
+    for f in pooled:
+        assert is_valid_day_pooled(f), f"{f.name} must be valid-day pooled"
+    for f in bounded:
+        assert not is_valid_day_pooled(f), f"{f.name} must be bounded"
+    assert len(pooled) == 8 and len(bounded) == 3  # the whole 11-factor minute surface
+
+
+def test_unclassified_minute_factor_raises():
+    """A minute factor in NEITHER partition set is a readable error (never
+    silently treated as bounded, which would reintroduce the divergence)."""
+    from factors.compute.minute.binding import is_valid_day_pooled
+
+    class _UnclassifiedMinute(Factor):
+        name = "unclassified_minute"
+
+        @property
+        def spec(self) -> FactorSpec:
+            return FactorSpec(
+                factor_id="unclassified_minute", version="1.0", description="fixture",
+                expected_ic_sign=1, is_intraday=False, forward_return_horizon=1,
+                return_basis="close_to_close", input_fields=("close",),
+                requires=(PanelField("close", source="stk_mins_1min"),),
+                adjustment="returns_invariant", overnight_boundary="none",
+                family="fixture", min_history_bars=0, lookback_depth=20,
+            )
+
+        def compute(self, panel):  # pragma: no cover - never reached
+            return panel["close"].rename(self.name)
+
+    with pytest.raises(KeyError, match="not classified"):
+        is_valid_day_pooled(_UnclassifiedMinute())
+
+
+# --------------------------------------------------------------------------- #
+# Daily decision-view materialization + R18 daily lag
+# --------------------------------------------------------------------------- #
+def test_daily_decision_materialize_is_finite():
+    src = _sources(daily=_daily())
+    v = materialize_range(VolatilityFactor(window=20), view=View.DECISION, symbols=SYMS,
+                          emit_start=DATES[40], emit_end=DATES[55], sources=src)
+    assert np.isfinite(v.to_numpy()).all() and len(v) == 16 * 2
+
+
+def test_r18_close_is_prev_day_open_is_same_day():
+    """A close-reading factor's decision value at d is UNCHANGED when close[d]
+    is perturbed (decision uses close[d-1]); an open-reading factor's value at d
+    DOES change when open[d] is perturbed (open is field-level same-day, R7)."""
+    d = DATES[55]
+    base_daily = _daily()
+    src0 = _sources(daily=base_daily)
+
+    # (a) close[d] perturbation -> decision-view volatility at d bit-unchanged.
+    vol = VolatilityFactor(window=20)
+    v0 = materialize_range(vol, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src0)
+    src_close = _sources(daily=_daily(perturb=(d, SYMS[0], "close", 999.0)))
+    v1 = materialize_range(vol, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src_close)
+    assert v0.loc[(d, SYMS[0])] == v1.loc[(d, SYMS[0])]  # close[d] not visible at 14:50
+
+    # (b) open[d] perturbation -> overnight (reads open same-day) at d CHANGES.
+    ov = OvernightMomentumFactor(window=20)
+    o0 = materialize_range(ov, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src0)
+    src_open = _sources(daily=_daily(perturb=(d, SYMS[0], "open", 999.0)))
+    o1 = materialize_range(ov, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src_open)
+    assert o0.loc[(d, SYMS[0])] != o1.loc[(d, SYMS[0])]  # open[d] IS visible (reverse teeth)
+
+    # (c) close[d] perturbation -> overnight at d unchanged (it never reads close[d]).
+    o2 = materialize_range(ov, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src_close)
+    assert o0.loc[(d, SYMS[0])] == o2.loc[(d, SYMS[0])]
+
+
+def test_r18_daily_basic_enriched_column_is_prev_day():
+    """value_ep (daily_basic-derived) at d uses the d-1 published ratio (<= d-1)."""
+    d = DATES[55]
+    src0 = _sources(daily=_daily())
+    vf = ValueFactor("value_ep")
+    v0 = materialize_range(vf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src0)
+    # perturbing value_ep[d] must NOT change the decision value at d (uses d-1)...
+    src_d = _sources(daily=_daily(perturb=(d, SYMS[0], "value_ep", 999.0)))
+    v_d = materialize_range(vf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src_d)
+    assert v0.loc[(d, SYMS[0])] == v_d.loc[(d, SYMS[0])]
+    # ...but perturbing value_ep[d-1] MUST change it.
+    src_p = _sources(daily=_daily(perturb=(DATES[54], SYMS[0], "value_ep", 999.0)))
+    v_p = materialize_range(vf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d, sources=src_p)
+    assert v0.loc[(d, SYMS[0])] != v_p.loc[(d, SYMS[0])]
+
+
+# --------------------------------------------------------------------------- #
+# Minute cutoff + leakage + binding correctness
+# --------------------------------------------------------------------------- #
+def test_minute_cutoff_post_1450_bar_does_not_leak():
+    """Perturbing a post-14:50 bar leaves the decision-view value bit-unchanged;
+    perturbing a pre-14:50 bar changes it (the cutoff is load-bearing)."""
+    d = DATES[35]
+    base = _minute()
+    src0 = _sources(minute=base)
+    jf = JumpAmountCorrFactor()
+    j0 = materialize_range(jf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d,
+                           sources=src0, decision_cutoff="14:50:00")
+    # a 14:55 bar (available_time 14:56 > 14:50) is filtered out -> no effect.
+    post_t = pd.Timestamp(d) + pd.Timedelta("14:55:00")
+    src_post = _sources(minute=_minute(perturb=(post_t, SYMS[0], "amount", 9e9)))
+    j_post = materialize_range(jf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d,
+                               sources=src_post, decision_cutoff="14:50:00")
+    assert j0.equals(j_post) or np.array_equal(j0.to_numpy(), j_post.to_numpy(), equal_nan=True)
+    # a 14:42 bar (available_time 14:43 <= 14:50) is inside the cutoff -> changes.
+    pre_t = pd.Timestamp(d) + pd.Timedelta("14:42:00")
+    src_pre = _sources(minute=_minute(perturb=(pre_t, SYMS[0], "amount", 9e9)))
+    j_pre = materialize_range(jf, view=View.DECISION, symbols=SYMS, emit_start=d, emit_end=d,
+                              sources=src_pre, decision_cutoff="14:50:00")
+    assert not np.array_equal(j0.to_numpy(), j_pre.to_numpy(), equal_nan=True)
+
+
+def test_minute_binding_close_view_matches_direct_compute():
+    """close-view minute materialize == the free compute_* on the same bar window
+    (proves the binding calls the right function with the right constants)."""
+    src = _sources(minute=_minute())
+    jf = JumpAmountCorrFactor()
+    got = materialize_range(jf, view=View.CLOSE, symbols=SYMS, emit_start=DATES[30], emit_end=DATES[38], sources=src)
+    win = MinuteProv(_minute()).minute_bars(
+        SYMS, DATES[30] - pd.Timedelta(days=60), DATES[38] + pd.Timedelta(days=1)
+    )
+    direct = compute_jump_amount_corr(win, lookback_days=jf.lookback_days, name=jf.name)
+    dd = direct.index.get_level_values("date")
+    direct_win = direct[(dd >= DATES[30]) & (dd <= DATES[38])].sort_index()
+    got = got.reindex(direct_win.index)
+    # A wrong binding (wrong function or wrong constant) would differ by O(1); the
+    # only residual is the pandas rolling-sum accumulation over a differently-sized
+    # loaded window (~1e-15, an attributable float-reorder, §五), so a tight
+    # tolerance proves the binding while allowing the reorder.
+    assert np.allclose(got.to_numpy(), direct_win.to_numpy(), rtol=1e-9, atol=1e-12, equal_nan=True)
+
+
+# --------------------------------------------------------------------------- #
+# ex-date masking (fixture masked factor; the 14 are non-masked, no-op)
+# --------------------------------------------------------------------------- #
+class _MaskedDailyFactor(Factor):
+    """Fixture-only masked factor: surfaces close; NaN'd on ex-dates by the engine."""
+
+    name = "masked_fixture"
+
+    @property
+    def spec(self) -> FactorSpec:
+        return FactorSpec(
+            factor_id="masked_fixture", version="1.0", description="fixture masked factor",
+            expected_ic_sign=1, is_intraday=False, forward_return_horizon=1,
+            return_basis="close_to_close", input_fields=("close",),
+            requires=(PanelField("close", source="market_daily"),),
+            adjustment="returns_invariant", overnight_boundary="masked",
+            family="fixture", min_history_bars=0, lookback_depth=1,
+        )
+
+    def compute(self, panel):
+        return panel["close"].rename(self.name)
+
+
+class AdjProv:
+    def __init__(self, series):
+        self._s = series
+
+    def adj_factor(self, symbols, start, end):
+        m = self._s.index.get_level_values("date")
+        return self._s[(m >= pd.Timestamp(start)) & (m <= pd.Timestamp(end))]
+
+
+def test_ex_date_masking_nans_the_ex_date_row():
+    """A masked factor's ex-date (d, symbol) is NaN; non-masked factors skip this."""
+    d_ex = DATES[50]
+    # adj_factor jumps on d_ex for SYMS[0] only.
+    af_rows = []
+    for s in SYMS:
+        for di, d in enumerate(DATES):
+            af = 1.0 if (s != SYMS[0] or d < d_ex) else 2.0
+            af_rows.append((d, s, af))
+    af = pd.DataFrame(af_rows, columns=["date", "symbol", "af"]).set_index(["date", "symbol"])["af"]
+    src = _sources(daily=_daily(), adj=AdjProv(af))
+    got = materialize_range(_MaskedDailyFactor(), view=View.DECISION, symbols=SYMS,
+                            emit_start=DATES[48], emit_end=DATES[52], sources=src)
+    assert np.isnan(got.loc[(d_ex, SYMS[0])])            # ex-date masked
+    assert np.isfinite(got.loc[(d_ex, SYMS[1])])         # other symbol untouched
+    assert np.isfinite(got.loc[(DATES[49], SYMS[0])])    # non-ex-date untouched
+
+
+def test_non_masked_factor_never_consults_adj_provider():
+    """The 14 closing factors are non-masked: the ex-date path is a no-op (no
+    AdjFactorProvider needed)."""
+    src = _sources(daily=_daily())  # NO adj provider
+    v = materialize_range(VolatilityFactor(window=20), view=View.DECISION, symbols=SYMS,
+                          emit_start=DATES[45], emit_end=DATES[46], sources=src)
+    assert len(v) == 4  # ran fine without an adj provider
+
+
+# --------------------------------------------------------------------------- #
+# recompute-fn factory (D3 tail-recompute compatibility)
+# --------------------------------------------------------------------------- #
+def test_make_recompute_fn_matches_materialize_range():
+    src = _sources(daily=_daily())
+    vol = VolatilityFactor(window=20)
+    fn = make_recompute_fn(vol, view=View.DECISION, symbols=SYMS, sources=src, data_start=DATES[0])
+    a = fn(DATES[40], DATES[50], vol.spec.lookback_depth)
+    b = materialize_range(vol, view=View.DECISION, symbols=SYMS,
+                          emit_start=DATES[40], emit_end=DATES[50], sources=src)
+    assert np.array_equal(a.sort_index().to_numpy(), b.sort_index().to_numpy(), equal_nan=True)
+
+
+# --------------------------------------------------------------------------- #
+# live-config wiring (R5): no default fallback
+# --------------------------------------------------------------------------- #
+class _LiveCacheCfg:
+    refresh_recent_days = 14
+    fina_tail_days = 400
+
+
+class _PartialCacheCfg:
+    refresh_recent_days = 14  # fina_tail_days missing
+
+
+def test_build_horizon_config_from_live_config():
+    hc = build_horizon_config(_LiveCacheCfg())
+    assert hc.refresh_recent_days == 14
+    assert hc.fina_tail_days == 400
+    assert hc.overrides() == {"fina_indicator": 400}
+
+
+def test_build_horizon_config_refuses_missing_attr():
+    with pytest.raises(ValueError, match="fina_tail_days"):
+        build_horizon_config(_PartialCacheCfg())
diff --git a/tests/test_factor_process_orchestrate.py b/tests/test_factor_process_orchestrate.py
new file mode 100644
index 0000000..9774bfc
--- /dev/null
+++ b/tests/test_factor_process_orchestrate.py
@@ -0,0 +1,71 @@
+"""D4 R9: the processing orchestration home (factors.process.orchestrate).
+
+Proves the orchestrator is behavior-preserving vs a directly-constructed
+ProcessingPipeline (so ``qt.pipeline._process_factors`` delegating to it changes
+no number), and that covariate extraction matches the retired inline logic.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+from factors.process.orchestrate import covariates_from_panel, process_factor_panel
+from factors.process.pipeline import ProcessingPipeline
+
+DATES = pd.bdate_range("2024-01-02", periods=8)
+SYMS = ["A", "B", "C", "D"]
+IDX = pd.MultiIndex.from_product([DATES, SYMS], names=["date", "symbol"])
+
+
+def _factor_panel():
+    rng = np.random.RandomState(3)
+    return pd.DataFrame(
+        {"f1": rng.normal(size=len(IDX)), "f2": rng.normal(size=len(IDX))}, index=IDX
+    )
+
+
+def test_orchestrator_matches_direct_pipeline_no_neutralize():
+    fp = _factor_panel()
+    got = process_factor_panel(fp, drop_missing=True, standardize=True)
+    direct = ProcessingPipeline(drop_missing=True, standardize=True).transform(fp)
+    pd.testing.assert_frame_equal(got, direct)
+
+
+def test_orchestrator_matches_direct_pipeline_with_neutralize():
+    fp = _factor_panel()
+    rng = np.random.RandomState(4)
+    industry = pd.Series(rng.choice(["X", "Y"], size=len(IDX)), index=IDX)
+    market_cap = pd.Series(np.abs(rng.normal(size=len(IDX))) + 1.0, index=IDX)
+    got = process_factor_panel(
+        fp, drop_missing=True, standardize=True, neutralize=True,
+        industry=industry, market_cap=market_cap,
+    )
+    direct = ProcessingPipeline(
+        drop_missing=True, standardize=True, neutralize=True,
+        industry=industry, market_cap=market_cap,
+    ).transform(fp)
+    pd.testing.assert_frame_equal(got, direct)
+
+
+def test_orchestrator_does_not_mutate_input():
+    fp = _factor_panel()
+    before = fp.copy()
+    process_factor_panel(fp, drop_missing=True, standardize=True)
+    pd.testing.assert_frame_equal(fp, before)
+
+
+def test_covariates_from_panel_extracts_present_columns():
+    panel = pd.DataFrame(
+        {"close": 1.0, "industry": "X", "market_cap": 2.0}, index=IDX
+    )
+    industry, market_cap = covariates_from_panel(panel)
+    assert industry is not None and market_cap is not None
+    pd.testing.assert_series_equal(industry, panel["industry"])
+    pd.testing.assert_series_equal(market_cap, panel["market_cap"])
+
+
+def test_covariates_from_panel_returns_none_when_absent():
+    panel = pd.DataFrame({"close": 1.0}, index=IDX)
+    industry, market_cap = covariates_from_panel(panel)
+    assert industry is None and market_cap is None
diff --git a/tests/test_factor_service.py b/tests/test_factor_service.py
new file mode 100644
index 0000000..5c76102
--- /dev/null
+++ b/tests/test_factor_service.py
@@ -0,0 +1,770 @@
+"""D4: the factor service (factors.service) — the §3.5 P8 acceptance.
+
+Network-free. Proves the hardest guarantee: two COLD stores, one filled by
+repeated per-date ``cross_section`` and one by a single batch ``panel``, hold
+BIT-IDENTICAL raw values (design §3.5 P8). Also the read-layer smoke
+(``panel[d] ≡ cross_section(d)``), the view x basis pairing gate, and the
+read-through no-recompute-on-hit property.
+
+The P8 test uses THREE factors that stress the tail-splice / leading-NaN
+boundary differently:
+* ``momentum_20`` — window-local (a two-point ratio, NO accumulation), so single
+  and batch are EXACTLY bit-identical; its transitive depth (21) exceeds its
+  headline window (20), so the warmup trim uses the transitive depth, not the
+  headline (the §六.18 deception guard);
+* ``volatility_20`` — rolling std (pandas accumulation), single==batch up to an
+  attributable float-reorder (<= 1e-12);
+* ``volume_peak_count_20`` — a genuinely NESTED minute factor (depth 40 = 20 +
+  20 baseline), single==batch up to the same float-reorder.
+
+Mutation evidence (recorded): forcing the materializer's warmup BELOW the
+factor's window makes the per-date fill under-warm every date while the batch
+fill only under-warms its first — so the two stores DIVERGE. The test that
+applies this mutation asserts the divergence, proving the warmup trim (and the
+transitive ``lookback_depth``) is load-bearing.
+"""
+
+from __future__ import annotations
+
+import tempfile
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from data.availability_policy import ReturnBasis, View
+from data.clean.intraday_schema import normalize_intraday_bars
+from factors import registry as factor_registry
+from factors import service as service_mod
+from factors.materialize import MaterializeSources, materialize_range
+from factors.service import DecisionPoint, cross_section, panel
+from factors.store.keys import store_key
+from factors.store.values import FactorValueStore
+
+SYMS = ["000001.SZ", "000002.SZ"]
+DATES = pd.bdate_range("2021-01-04", periods=75)
+
+
+# --------------------------------------------------------------------------- #
+# Synthetic data + providers
+# --------------------------------------------------------------------------- #
+def _daily():
+    rng = np.random.RandomState(0)
+    rows = []
+    for si, s in enumerate(SYMS):
+        px = 100.0 + si * 20 + np.cumsum(rng.normal(0, 1.0, len(DATES)))
+        for d, p in zip(DATES, px):
+            rows.append((d, s, p - 0.3, p + 0.5, p - 0.5, p, 1e5, p * 1e5))
+    return (
+        pd.DataFrame(rows, columns=["date", "symbol", "open", "high", "low", "close", "volume", "amount"])
+        .set_index(["date", "symbol"]).sort_index()
+    )
+
+
+def _rich_minute():
+    """55 days x 238-bar sessions with same-slot baseline + eruptions -> the
+    peak/valley taxonomy produces finite nested-factor values."""
+    rng = np.random.RandomState(7)
+    rows = []
+    for si, s in enumerate(SYMS):
+        for d in DATES[:55]:
+            base = pd.Timestamp(d) + pd.Timedelta("09:31:00")
+            price = 100.0 + si * 5 + rng.normal(0, 2)
+            for i in range(238):
+                t = base + pd.Timedelta(minutes=i)
+                price += rng.normal(0, 0.05)
+                slot_base = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0))
+                erupt = 6.0 if (rng.rand() < 0.06) else 1.0
+                vol = slot_base * erupt * (1.0 + 0.1 * rng.rand())
+                w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand())
+                hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand()
+                cl = lo + (hi - lo) * rng.rand()
+                rows.append((t, s, price, hi, lo, cl, vol, cl * vol))
+    frame = pd.DataFrame(rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"])
+    return normalize_intraday_bars(frame, freq="1min")
+
+
+DAILY = _daily()
+MINUTE = _rich_minute()
+
+
+class DailyProv:
+    def daily_panel(self, symbols, start, end):
+        m = DAILY.index.get_level_values("date")
+        return DAILY[(m >= pd.Timestamp(start)) & (m <= pd.Timestamp(end))]
+
+
+class MinuteProv:
+    def __init__(self):
+        self.calls = 0
+
+    def minute_bars(self, symbols, start, end):
+        self.calls += 1
+        if not symbols:
+            return MINUTE.iloc[0:0]
+        t = MINUTE.index.get_level_values("time")
+        return MINUTE[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))]
+
+    def earliest_available(self, symbols):
+        """DECLARED data start of this fixture (never inferred from row counts)."""
+        return DATES[0]
+
+
+def _sparse_minute():
+    """Sparse-valid minute data: every 3rd day is thin (fewer bars) so the
+    valid-day density is < 1 — the window where a fixed lookback_depth trim makes
+    valid-day-POOLED factors (ridge / valley_ridge / peak_ridge) diverge
+    finite<->NaN between a single-date fill and a batch fill (the review HIGH)."""
+    rng = np.random.RandomState(11)
+    rows = []
+    for si, s in enumerate(SYMS):
+        for di, d in enumerate(DATES[:80]):
+            n = 120 if (di % 3 == 0) else 238
+            base = pd.Timestamp(d) + pd.Timedelta("09:31:00")
+            price = 100.0 + si * 5 + rng.normal(0, 2)
+            for i in range(n):
+                t = base + pd.Timedelta(minutes=i)
+                price += rng.normal(0, 0.05)
+                slot = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0))
+                erupt = 6.0 if (rng.rand() < 0.06) else 1.0
+                vol = slot * erupt * (1.0 + 0.1 * rng.rand())
+                w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand())
+                hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand()
+                cl = lo + (hi - lo) * rng.rand()
+                rows.append((t, s, price, hi, lo, cl, vol, cl * vol))
+    frame = pd.DataFrame(rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"])
+    return normalize_intraday_bars(frame, freq="1min")
+
+
+SPARSE_MINUTE = _sparse_minute()
+
+
+class SparseMinuteProv:
+    def minute_bars(self, symbols, start, end):
+        if not symbols:
+            return SPARSE_MINUTE.iloc[0:0]
+        t = SPARSE_MINUTE.index.get_level_values("time")
+        return SPARSE_MINUTE[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))]
+
+    def earliest_available(self, symbols):
+        return DATES[0]
+
+
+#: Gap geometry (the review counterexample, sized to actually exercise it).
+#: One expansion chunk is ``lookback_depth*2 + 40`` CALENDAR days ~= 85 trading
+#: days at depth 40. The gap must be wide enough that a WHOLE expansion step
+#: lands INSIDE it (adding no bars at all -> emit values unchanged -> a
+#: value-stability fixed point terminates falsely) while REAL earlier history
+#: still exists beyond it: 200 trading days spans more than two chunks.
+_GAP_TRADING_DAYS = 200
+_PRE_GAP_DAYS = 60
+_POST_GAP_DAYS = 30
+_GAP_BARS_PER_DAY = 238  # a full session: the pooled factors need enough
+                         # classifiable + ridge/valley bars to yield valid days
+GAP_DATES = pd.bdate_range(
+    "2021-01-04", periods=_PRE_GAP_DAYS + _GAP_TRADING_DAYS + _POST_GAP_DAYS
+)
+
+
+def _gapped_minute():
+    """The REVIEW COUNTEREXAMPLE shape: rich bars, then a LONG no-bar gap
+    (a suspension) wider than one expansion chunk, then rich bars again.
+
+    The emit window sits AFTER the gap and its trailing valid-day pool must reach
+    back ACROSS the gap. A value-stability fixed point terminates falsely here
+    (one chunk lands entirely inside the gap -> emit values unchanged), and an
+    unchanged row count inside the gap looks exactly like the data start — the
+    two failure modes the structural criterion replaces.
+    """
+    rng = np.random.RandomState(23)
+    rows = []
+    pre = GAP_DATES[:_PRE_GAP_DAYS]
+    post = GAP_DATES[_PRE_GAP_DAYS + _GAP_TRADING_DAYS:]
+    for si, s in enumerate(SYMS):
+        for d in list(pre) + list(post):
+            base = pd.Timestamp(d) + pd.Timedelta("09:31:00")
+            price = 100.0 + si * 5 + rng.normal(0, 2)
+            for i in range(_GAP_BARS_PER_DAY):
+                t = base + pd.Timedelta(minutes=i)
+                price += rng.normal(0, 0.05)
+                slot = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0))
+                erupt = 6.0 if (rng.rand() < 0.06) else 1.0
+                vol = slot * erupt * (1.0 + 0.1 * rng.rand())
+                w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand())
+                hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand()
+                cl = lo + (hi - lo) * rng.rand()
+                rows.append((t, s, price, hi, lo, cl, vol, cl * vol))
+    frame = pd.DataFrame(rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"])
+    return normalize_intraday_bars(frame, freq="1min"), list(post)
+
+
+GAPPED_MINUTE, GAPPED_POST_DATES = _gapped_minute()
+
+
+class GappedMinuteProv:
+    def minute_bars(self, symbols, start, end):
+        if not symbols:
+            return GAPPED_MINUTE.iloc[0:0]
+        t = GAPPED_MINUTE.index.get_level_values("time")
+        return GAPPED_MINUTE[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))]
+
+    def earliest_available(self, symbols):
+        """DECLARED start — the mid-history gap must never be mistaken for it."""
+        return GAP_DATES[0]
+
+
+def _sources():
+    return MaterializeSources(daily=DailyProv(), minute=MinuteProv())
+
+
+def _sparse_sources():
+    return MaterializeSources(minute=SparseMinuteProv())
+
+
+def _gapped_sources():
+    return MaterializeSources(minute=GappedMinuteProv())
+
+
+def _store_series(store, factor_id, view=View.DECISION):
+    factor = factor_registry.build(factor_id)
+    key = store_key(factor, view=view.value)
+    return store.read(key)
+
+
+def _fill_single(store, factor_ids, dates, src, universe=None):
+    for d in dates:
+        cross_section(factor_ids, universe or SYMS, DecisionPoint(date=d),
+                      store=store, sources=src)
+
+
+def _fill_batch(store, factor_ids, dates, src, universe=None):
+    panel(factor_ids, universe or SYMS, [DecisionPoint(date=d) for d in dates],
+          store=store, sources=src)
+
+
+# --------------------------------------------------------------------------- #
+# P8: single-fill == batch-fill
+# --------------------------------------------------------------------------- #
+def test_single_fill_equals_batch_fill_momentum_bit_identical():
+    """Window-local momentum: the two cold stores are EXACTLY bit-identical."""
+    dates = list(DATES[40:56])
+    with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+        a, b = FactorValueStore(ta), FactorValueStore(tb)
+        _fill_single(a, ["momentum_20"], dates, _sources())
+        _fill_batch(b, ["momentum_20"], dates, _sources())
+        sa = _store_series(a, "momentum_20").sort_index()
+        sb = _store_series(b, "momentum_20").sort_index()
+        assert sa.index.equals(sb.index)
+        av, bv = sa.to_numpy(), sb.to_numpy()
+        assert np.array_equal(np.isnan(av), np.isnan(bv))
+        assert np.array_equal(av[~np.isnan(av)], bv[~np.isnan(bv)])  # BIT identical
+
+
+def test_single_fill_equals_batch_fill_rolling_and_nested_minute():
+    """volatility (rolling) + volume_peak_count (NESTED minute, depth 40): the two
+    stores match up to the attributable pandas-rolling float-reorder (<= 1e-12)."""
+    dates = list(DATES[42:54])
+    for fid in ("volatility_20", "volume_peak_count_20"):
+        with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+            a, b = FactorValueStore(ta), FactorValueStore(tb)
+            _fill_single(a, [fid], dates, _sources())
+            _fill_batch(b, [fid], dates, _sources())
+            sa = _store_series(a, fid).sort_index()
+            sb = _store_series(b, fid).sort_index()
+            assert sa.index.equals(sb.index), fid
+            av, bv = sa.to_numpy(), sb.to_numpy()
+            assert np.array_equal(np.isnan(av), np.isnan(bv)), fid
+            finite = ~np.isnan(av)
+            # non-vacuous: the nested factor really produced values in this window.
+            assert finite.sum() > 0, fid
+            assert np.allclose(av[finite], bv[finite], rtol=1e-12, atol=1e-12), fid
+
+
+def test_reduced_warmup_breaks_single_equals_batch(monkeypatch):
+    """MUTATION: forcing the warmup below the factor window makes per-date fill
+    under-warm every date while batch under-warms only its first -> the stores
+    DIVERGE. Proves the warmup trim (transitive lookback_depth) is load-bearing."""
+    dates = list(DATES[40:56])
+
+    def _short_warmup(factor, **kw):
+        # 10 << momentum's window of 20: every per-date fill under-warms (all NaN),
+        # but the batch fill (loading 10 before d1) still computes its tail dates
+        # d >= d1 + (window - 10) -> the two NaN patterns diverge.
+        kw["warmup"] = 10
+        return materialize_range(factor, **kw)
+
+    monkeypatch.setattr(service_mod, "materialize_range", _short_warmup)
+    with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+        a, b = FactorValueStore(ta), FactorValueStore(tb)
+        _fill_single(a, ["momentum_20"], dates, _sources())
+        _fill_batch(b, ["momentum_20"], dates, _sources())
+        sa = _store_series(a, "momentum_20").sort_index()
+        sb = _store_series(b, "momentum_20").sort_index()
+        av, bv = sa.to_numpy(), sb.to_numpy()
+        # the NaN pattern (per-date all-NaN vs batch finite near the tail) differs.
+        assert not np.array_equal(np.isnan(av), np.isnan(bv))
+
+
+# --------------------------------------------------------------------------- #
+# P8 EXPANSION: valid-day POOLED factors (review HIGH) on a sparse-valid window
+# --------------------------------------------------------------------------- #
+# WHY these factors and this window (the review's "false confidence"): the P8
+# single==batch test above uses volume_peak_count, which is ALSO valid-day pooled
+# but happened to have dense-enough valid days to stay clean — it is the
+# "happens-to-be-clean nested representative". The review showed that on a sparse-
+# valid window ridge_minute_return / valley_ridge_vwap_ratio / peak_ridge_amount_
+# ratio produce a single-fill=finite / batch-fill=NaN divergence (e.g. the clean
+# cell ridge_minute_return_20 (2021-03-18, 000002.SZ): single=1.6976..., batch=NaN),
+# because the trailing pool counts VALID days and a fixed lookback_depth trim
+# truncates the pool differently for a per-date fill vs a batch fill. The
+# materializer now loads these factors to SATURATION (real data start, no trim),
+# so the value is load-geometry-free and the two stores agree.
+_POOLED_DIVERGENT = ["ridge_minute_return_20", "valley_ridge_vwap_ratio_20", "peak_ridge_amount_ratio_20"]
+
+
+def test_pooled_factor_single_equals_batch_on_sparse_valid_window():
+    """The valid-day-pooled factors that diverge under a fixed trim now agree
+    (saturation load) — single-fill store == batch-fill store, and the pool is
+    fuller than a truncated one so the result is non-vacuous (real finite values)."""
+    dates = list(DATES[55:75])
+    for fid in _POOLED_DIVERGENT:
+        with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+            a, b = FactorValueStore(ta), FactorValueStore(tb)
+            _fill_single(a, [fid], dates, _sparse_sources())
+            _fill_batch(b, [fid], dates, _sparse_sources())
+            sa = _store_series(a, fid).sort_index()
+            sb = _store_series(b, fid).sort_index()
+            assert sa.index.equals(sb.index), fid
+            av, bv = sa.to_numpy(), sb.to_numpy()
+            assert np.array_equal(np.isnan(av), np.isnan(bv)), f"{fid}: NaN mask diverges"
+            finite = ~np.isnan(av)
+            assert finite.sum() > 0, f"{fid}: vacuous (no finite values)"
+            # saturation gives shared-prefix accumulation -> BIT identical.
+            assert np.array_equal(av[finite], bv[finite]), f"{fid}: finite values diverge"
+
+
+def test_pooled_single_equals_batch_across_a_long_no_bar_gap():
+    """THE REVIEW COUNTEREXAMPLE, committed: a no-bar gap WIDER than one expansion
+    chunk (``_GAP_TRADING_DAYS`` = a suspension), emit AFTER the gap, and the trailing
+    valid-day pool must reach back ACROSS it.
+
+    This is what defeated the value-stability fixed point (a chunk landing wholly
+    inside the gap leaves the emit values unchanged -> false 'stable') and the
+    row-count data-start inference (an unchanged count inside the gap looks like
+    exhaustion). The structural criterion counts FINAL valid days in the locked
+    sub-window, so a gap contributes none, the count stalls, and expansion
+    continues to the declared floor.
+
+    MUTATION EVIDENCE (run for this commit):
+    * reverting the criterion to the value-stability fixed point -> this test
+      FAILS (rc=1): both fills terminate inside the gap and emit all-NaN, caught
+      by the non-vacuity assertion below; restored -> passes (rc=0).
+    * dropping the baseline LOCKING OFFSET does NOT fail on THIS fixture (the
+      valid-day count alone already forces enough history here); it IS isolated
+      by ``test_dropping_the_locking_offset_yields_a_wrong_value``, whose sparse
+      recent block puts an unlocked band inside the criterion's margin.
+    """
+    dates = GAPPED_POST_DATES[10:22]  # after the gap, pool still spans it
+    for fid in _POOLED_DIVERGENT:
+        with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+            a, b = FactorValueStore(ta), FactorValueStore(tb)
+            _fill_single(a, [fid], dates, _gapped_sources())
+            _fill_batch(b, [fid], dates, _gapped_sources())
+            sa = _store_series(a, fid).sort_index()
+            sb = _store_series(b, fid).sort_index()
+            assert sa.index.equals(sb.index), fid
+            av, bv = sa.to_numpy(), sb.to_numpy()
+            assert np.array_equal(np.isnan(av), np.isnan(bv)), f"{fid}: NaN mask diverges"
+            finite = ~np.isnan(av)
+            assert finite.sum() > 0, f"{fid}: vacuous (no finite values across the gap)"
+            assert np.allclose(av[finite], bv[finite], rtol=0.0, atol=1e-12), fid
+
+
+#: Zero-output fixture (review HIGH): symbol A has bars throughout; symbol B has
+#: an ancient block and a recent block with a LONG suspension between, so in a
+#: shallow load window B produces ZERO factor rows.
+_ZO_DATES = pd.bdate_range("2021-01-04", periods=130)
+_ZO_A, _ZO_B = "600000.SH", "600519.SH"
+_ZO_SYMS = [_ZO_A, _ZO_B]
+
+
+def _zero_output_minute():
+    rng = np.random.RandomState(99)
+    rows = []
+    plan = ((_ZO_A, list(range(130))), (_ZO_B, list(range(10, 25)) + list(range(120, 130))))
+    for si, (s, days) in enumerate(plan):
+        for di in days:
+            base = pd.Timestamp(_ZO_DATES[di]) + pd.Timedelta("09:31:00")
+            price = 100.0 + si * 5 + rng.normal(0, 2)
+            for i in range(238):
+                t = base + pd.Timedelta(minutes=i)
+                price += rng.normal(0, 0.05)
+                slot = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0))
+                erupt = 6.0 if (rng.rand() < 0.06) else 1.0
+                vol = slot * erupt * (1.0 + 0.1 * rng.rand())
+                w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand())
+                hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand()
+                cl = lo + (hi - lo) * rng.rand()
+                rows.append((t, s, price, hi, lo, cl, vol, cl * vol))
+    frame = pd.DataFrame(rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"])
+    return normalize_intraday_bars(frame, freq="1min")
+
+
+ZERO_OUTPUT_MINUTE = _zero_output_minute()
+
+
+class ZeroOutputMinuteProv:
+    def minute_bars(self, symbols, start, end):
+        if not symbols:
+            return ZERO_OUTPUT_MINUTE.iloc[0:0]
+        t = ZERO_OUTPUT_MINUTE.index.get_level_values("time")
+        sym = ZERO_OUTPUT_MINUTE.index.get_level_values("symbol")
+        keep = (t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end)) & sym.isin(list(symbols))
+        return ZERO_OUTPUT_MINUTE[keep]
+
+    def earliest_available(self, symbols):
+        return _ZO_DATES[0]
+
+
+def test_zero_output_symbol_is_not_silently_dropped():
+    """REVIEW HIGH: a symbol producing ZERO rows in the current load window must
+    still VETO saturation — otherwise a single-date fill drops it from the
+    cross-section entirely while a batch fill keeps it (silent name-dropping =
+    sample-coverage bias, forbidden by the I5a red line).
+
+    Measured before the fix: volume_peak_count_20 single-fill ``600519.SH rows=0``
+    vs batch-fill ``rows=10 finite=6`` — the criterion's per-symbol counts were
+    ``{'600000.SH': 67}`` with 600519 absent entirely. After the fix both fills
+    carry the same symbols and the same rows.
+    """
+    dates = list(_ZO_DATES[90:130])  # the review's emit window (indices 90..129)
+    fid = "volume_peak_count_20"
+    with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+        a, b = FactorValueStore(ta), FactorValueStore(tb)
+        _fill_single(a, [fid], dates, MaterializeSources(minute=ZeroOutputMinuteProv()),
+                     universe=_ZO_SYMS)
+        _fill_batch(b, [fid], dates, MaterializeSources(minute=ZeroOutputMinuteProv()),
+                    universe=_ZO_SYMS)
+        sa = _store_series(a, fid).sort_index()
+        sb = _store_series(b, fid).sort_index()
+        # the thin symbol must be present in BOTH stores (never silently dropped)
+        syms_a = set(map(str, sa.index.get_level_values("symbol")))
+        syms_b = set(map(str, sb.index.get_level_values("symbol")))
+        assert _ZO_B in syms_a, f"{_ZO_B} silently dropped by the single-date fill"
+        assert syms_a == syms_b, f"symbol sets diverge: single={syms_a} batch={syms_b}"
+        assert sa.index.equals(sb.index)
+        av, bv = sa.to_numpy(), sb.to_numpy()
+        assert np.array_equal(np.isnan(av), np.isnan(bv))
+        finite = ~np.isnan(av)
+        assert finite.sum() > 0
+        assert np.allclose(av[finite], bv[finite], rtol=0.0, atol=1e-12)
+
+
+def test_zero_output_symbol_veto_mutation(monkeypatch):
+    """MUTATION for the HIGH: restricting the criterion to OUTPUT symbols (the
+    pre-fix behaviour) reintroduces the silent drop -> this test's assertion of
+    the drop proves the requested-symbol iteration is load-bearing."""
+    import factors.materialize as mat
+
+    real = mat._pooled_pool_saturated
+
+    def output_only(full, bars, emit_start, *, baseline_days, lookback_days, symbols):
+        present = [s for s in symbols if str(s) in set(map(str, full.index.get_level_values("symbol")))]
+        return real(full, bars, emit_start, baseline_days=baseline_days,
+                    lookback_days=lookback_days, symbols=present)
+
+    monkeypatch.setattr(mat, "_pooled_pool_saturated", output_only)
+    dates = list(_ZO_DATES[90:130])  # the review's emit window (indices 90..129)
+    fid = "volume_peak_count_20"
+    with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+        a, b = FactorValueStore(ta), FactorValueStore(tb)
+        _fill_single(a, [fid], dates, MaterializeSources(minute=ZeroOutputMinuteProv()),
+                     universe=_ZO_SYMS)
+        _fill_batch(b, [fid], dates, MaterializeSources(minute=ZeroOutputMinuteProv()),
+                    universe=_ZO_SYMS)
+        sa, sb = _store_series(a, fid), _store_series(b, fid)
+        syms_a = set(map(str, sa.index.get_level_values("symbol"))) if sa is not None else set()
+        syms_b = set(map(str, sb.index.get_level_values("symbol"))) if sb is not None else set()
+        # NON-VACUITY GUARD: the disjunction below is trivially true when BOTH
+        # stores are empty (nothing filled at all), which would make this mutation
+        # test pass without exercising the mutation. Require real content first.
+        assert syms_a, "single-fill store is empty — the mutation test would be vacuous"
+        assert syms_b, "batch-fill store is empty — the mutation test would be vacuous"
+        assert _ZO_B in syms_b, (
+            "the batch fill must still carry the thin symbol; otherwise the "
+            "single-vs-batch contrast below proves nothing"
+        )
+        assert syms_a != syms_b or _ZO_B not in syms_a, (
+            "the output-symbols-only criterion should drop the thin symbol from "
+            "the single-date fill"
+        )
+
+
+class DistantFloorMinuteProv(ZeroOutputMinuteProv):
+    """Same bars, but a DECLARED floor far in the past — so the saturation loop
+    cannot terminate by reaching the floor within a bounded number of chunks."""
+
+    def earliest_available(self, symbols):
+        return pd.Timestamp("1990-01-01")
+
+
+def test_saturation_exhaustion_raises_loudly(monkeypatch):
+    """Exhausting the chunk budget must RAISE, never return an unsaturated value.
+
+    Red line #9 (no silent degradation): the branch is unreachable in practice
+    (500 chunks is >160 years), which is exactly why it needs a test — a raise
+    that never executes is the next AttributeError's host. Recipe: cap the budget
+    at 2 chunks, use a symbol that never accumulates ``lookback_days`` valid days,
+    and declare a floor too distant to reach — so neither termination condition
+    fires and the loop must fall through to the raise.
+    """
+    import factors.materialize as mat
+
+    monkeypatch.setattr(mat, "_MAX_SATURATION_CHUNKS", 2)
+    factor = factor_registry.build("volume_peak_count_20")
+    emit = _ZO_DATES[128]
+    with pytest.raises(RuntimeError, match="pooled saturation did not converge"):
+        mat.materialize_range(
+            factor, view=View.DECISION, symbols=_ZO_SYMS,
+            emit_start=emit, emit_end=emit,
+            sources=MaterializeSources(minute=DistantFloorMinuteProv()),
+        )
+
+
+def test_saturation_exhaustion_message_is_actionable(monkeypatch):
+    """The raise names the factor and the declared floor (an operator must be able
+    to tell WHICH factor stalled and how deep the search went)."""
+    import factors.materialize as mat
+
+    monkeypatch.setattr(mat, "_MAX_SATURATION_CHUNKS", 2)
+    factor = factor_registry.build("volume_peak_count_20")
+    emit = _ZO_DATES[128]
+    with pytest.raises(RuntimeError) as excinfo:
+        mat.materialize_range(
+            factor, view=View.DECISION, symbols=_ZO_SYMS,
+            emit_start=emit, emit_end=emit,
+            sources=MaterializeSources(minute=DistantFloorMinuteProv()),
+        )
+    message = str(excinfo.value)
+    assert factor.name in message           # which factor stalled
+    assert "1990-01-01" in message          # the declared floor
+    assert "expansion chunks" in message    # how far the search went
+
+
+#: Locking-offset isolation fixture (review task 3(b), adopted verbatim in shape).
+_LO_DATES = pd.bdate_range("2020-01-01", periods=210)
+_LO_SYM = "600000.SH"
+_LO_E_IDX = 200
+#: deep history (reachable only by a second expansion) + a SPARSE recent block of
+#: 33 every-other-day bar-days ending exactly at the emit date. In the shallow
+#: window the sparse block is all there is, and its local days 10..19 are
+#: classifiable but NOT yet final (baseline below full depth) — the UNLOCKED band
+#: the offset exists to exclude.
+_LO_BAR_DAYS = list(range(40, 100)) + [_LO_E_IDX - 2 * j for j in range(32, -1, -1)]
+
+
+def _locking_offset_minute():
+    rng = np.random.RandomState(2026)
+    rows = []
+    for di in _LO_BAR_DAYS:
+        base = pd.Timestamp(_LO_DATES[di]) + pd.Timedelta("09:31:00")
+        price = 100.0 + rng.normal(0, 2)
+        for i in range(238):
+            t = base + pd.Timedelta(minutes=i)
+            price += rng.normal(0, 0.05)
+            slot = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0))
+            erupt = 6.0 if (rng.rand() < 0.06) else 1.0
+            vol = slot * erupt * (1.0 + 0.1 * rng.rand())
+            w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand())
+            hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand()
+            cl = lo + (hi - lo) * rng.rand()
+            rows.append((t, _LO_SYM, price, hi, lo, cl, vol, cl * vol))
+    frame = pd.DataFrame(rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"])
+    return normalize_intraday_bars(frame, freq="1min")
+
+
+LOCKING_OFFSET_MINUTE = _locking_offset_minute()
+
+
+class LockingOffsetProv:
+    def minute_bars(self, symbols, start, end):
+        if not symbols:
+            return LOCKING_OFFSET_MINUTE.iloc[0:0]
+        t = LOCKING_OFFSET_MINUTE.index.get_level_values("time")
+        return LOCKING_OFFSET_MINUTE[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))]
+
+    def earliest_available(self, symbols):
+        return _LO_DATES[0]
+
+
+def test_locking_offset_is_load_bearing():
+    """The BASELINE LOCKING OFFSET produces the load-geometry-free value.
+
+    Ground truth = computing on the WHOLE history at once. With the offset the
+    saturation loop expands past the unlocked band and reproduces it exactly;
+    dropping the offset (``loaded_days[baseline_days]`` -> ``loaded_days[0]``)
+    lets the criterion stop one chunk early with the pool still drawing on days
+    whose classification is not final, yielding a silently WRONG value.
+
+    MUTATION rc: with ``baseline_days=0`` forced this test FAILS (rc=1, the
+    mutated value != truth); unmutated it passes (rc=0). This replaces the
+    earlier honest note that the offset was not isolated by the gap fixtures.
+    """
+    import factors.materialize as mat
+    from factors.compute.minute.binding import minute_raw_from_bars
+    from factors.view_lag import minute_decision_cutoff
+
+    factor = factor_registry.build("volume_peak_count_20")
+    emit = _LO_DATES[_LO_E_IDX]
+    truth_full = minute_raw_from_bars(
+        factor, minute_decision_cutoff(LOCKING_OFFSET_MINUTE, decision_time="14:50:00")
+    )
+    truth = float(truth_full.loc[(emit, _LO_SYM)])
+
+    got = mat.materialize_range(
+        factor, view=View.DECISION, symbols=[_LO_SYM], emit_start=emit, emit_end=emit,
+        sources=MaterializeSources(minute=LockingOffsetProv()),
+    )
+    assert float(got.iloc[0]) == truth, (
+        "the saturated value must equal the whole-history (load-geometry-free) value"
+    )
+
+
+def test_dropping_the_locking_offset_yields_a_wrong_value(monkeypatch):
+    """MUTATION evidence for the offset: forcing ``baseline_days=0`` makes the
+    criterion accept the unlocked band and return a value that differs from the
+    whole-history truth (measured on this fixture: 438.0 vs the true 416.0)."""
+    import factors.materialize as mat
+    from factors.compute.minute.binding import minute_raw_from_bars
+    from factors.view_lag import minute_decision_cutoff
+
+    factor = factor_registry.build("volume_peak_count_20")
+    emit = _LO_DATES[_LO_E_IDX]
+    truth = float(
+        minute_raw_from_bars(
+            factor, minute_decision_cutoff(LOCKING_OFFSET_MINUTE, decision_time="14:50:00")
+        ).loc[(emit, _LO_SYM)]
+    )
+    real = mat._pooled_pool_saturated
+
+    def no_offset(full, bars, emit_start, *, baseline_days, lookback_days, symbols):
+        return real(full, bars, emit_start, baseline_days=0,
+                    lookback_days=lookback_days, symbols=symbols)
+
+    monkeypatch.setattr(mat, "_pooled_pool_saturated", no_offset)
+    mutated = mat.materialize_range(
+        factor, view=View.DECISION, symbols=[_LO_SYM], emit_start=emit, emit_end=emit,
+        sources=MaterializeSources(minute=LockingOffsetProv()),
+    )
+    assert float(mutated.iloc[0]) != truth, (
+        "dropping the locking offset must produce a load-geometry-dependent value"
+    )
+
+
+def test_disabling_saturation_reintroduces_pooled_divergence(monkeypatch):
+    """MUTATION: treating a pooled factor as fixed-depth (saturation OFF) brings
+    back the single-fill/batch-fill finite<->NaN divergence -> the saturation
+    load is load-bearing. rc=1 (this asserts the divergence) with the mutation;
+    rc=0 (the test above) without it."""
+    import factors.materialize as mat
+
+    monkeypatch.setattr(mat, "is_valid_day_pooled", lambda factor: False)
+    dates = list(DATES[55:75])
+    any_diverged = False
+    for fid in _POOLED_DIVERGENT:
+        with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb:
+            a, b = FactorValueStore(ta), FactorValueStore(tb)
+            _fill_single(a, [fid], dates, _sparse_sources())
+            _fill_batch(b, [fid], dates, _sparse_sources())
+            sa = _store_series(a, fid).sort_index()
+            sb = _store_series(b, fid).sort_index()
+            av = sa.reindex(sb.index).to_numpy()
+            bv = sb.to_numpy()
+            if not np.array_equal(np.isnan(av), np.isnan(bv)):
+                any_diverged = True
+    assert any_diverged, "saturation-off must reintroduce a pooled divergence"
+
+
+# --------------------------------------------------------------------------- #
+# read-layer smoke + read-through
+# --------------------------------------------------------------------------- #
+def test_panel_equals_cross_section_on_same_store():
+    dates = list(DATES[40:52])
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        pan = panel(["momentum_20", "volatility_20"], SYMS,
+                    [DecisionPoint(date=d) for d in dates], store=store, sources=src)
+        d = dates[6]
+        xs = cross_section(["momentum_20", "volatility_20"], SYMS, DecisionPoint(date=d),
+                           store=store, sources=src)
+        pan_d = pan[pan.index.get_level_values("date") == d].sort_index()
+        xs = xs.sort_index()
+        assert xs.index.equals(pan_d.index)
+        assert np.allclose(xs.to_numpy(), pan_d.to_numpy(), equal_nan=True, rtol=0, atol=0)
+
+
+def test_read_through_hit_does_not_recompute():
+    """A second call over already-covered dates does not re-hit the provider."""
+    dates = list(DATES[40:50])
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        panel(["jump_amount_corr_20"], SYMS, [DecisionPoint(date=d) for d in dates],
+              store=store, sources=src)
+        calls_after_fill = src.minute.calls
+        assert calls_after_fill > 0
+        # second identical call: store hit -> no new minute read
+        panel(["jump_amount_corr_20"], SYMS, [DecisionPoint(date=d) for d in dates],
+              store=store, sources=src)
+        assert src.minute.calls == calls_after_fill
+
+
+def test_service_returns_only_universe_symbols():
+    dates = list(DATES[40:44])
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        pan = panel(["momentum_20"], [SYMS[0]], [DecisionPoint(date=d) for d in dates],
+                    store=store, sources=src)
+        got_syms = set(pan.index.get_level_values("symbol"))
+        assert got_syms <= {SYMS[0]}
+
+
+# --------------------------------------------------------------------------- #
+# pairing gate
+# --------------------------------------------------------------------------- #
+def test_illegal_view_basis_pairing_raises():
+    """close view scored on exec_to_exec (or decision on close_to_close) is a
+    readable construction-time error, not a doc convention (§1.4)."""
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        with pytest.raises(ValueError, match="illegal view/basis pairing"):
+            cross_section(["momentum_20"], SYMS, DecisionPoint(date=DATES[40]),
+                          store=store, sources=src, view=View.CLOSE, basis=ReturnBasis.EXEC_TO_EXEC)
+        with pytest.raises(ValueError, match="illegal view/basis pairing"):
+            cross_section(["momentum_20"], SYMS, DecisionPoint(date=DATES[40]),
+                          store=store, sources=src, view=View.DECISION, basis=ReturnBasis.CLOSE_TO_CLOSE)
+
+
+def test_legal_pairings_are_accepted():
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        # decision <-> exec_to_exec (default) and close <-> close_to_close
+        cross_section(["momentum_20"], SYMS, DecisionPoint(date=DATES[40]),
+                      store=store, sources=src)
+        cross_section(["momentum_20"], SYMS, DecisionPoint(date=DATES[40]),
+                      store=store, sources=src, view=View.CLOSE, basis=ReturnBasis.CLOSE_TO_CLOSE)
+
+
+def test_panel_requires_uniform_cutoff():
+    src = _sources()
+    with tempfile.TemporaryDirectory() as td:
+        store = FactorValueStore(td)
+        with pytest.raises(ValueError, match="share one cutoff"):
+            panel(["momentum_20"], SYMS,
+                  [DecisionPoint(date=DATES[40], cutoff="14:50:00"),
+                   DecisionPoint(date=DATES[41], cutoff="14:45:00")],
+                  store=store, sources=src)
diff --git a/tests/test_factor_view_lag.py b/tests/test_factor_view_lag.py
new file mode 100644
index 0000000..fdfe758
--- /dev/null
+++ b/tests/test_factor_view_lag.py
@@ -0,0 +1,129 @@
+"""D4: availability-lag transforms (factors.view_lag) — R18 per field-class.
+
+These pin the pure transforms the materializer composes: the daily prev-day shift
+with the field-level ``open`` exception, the minute cutoff, and the ex-date
+BOOLEAN. The materializer's end-to-end R18 coverage rides on top (see
+tests/test_factor_materialize.py); here we prove the transforms in isolation.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from data.clean.intraday_schema import normalize_intraday_bars
+from factors.view_lag import (
+    daily_decision_lag,
+    decision_same_day_fields,
+    ex_date_mask,
+    minute_decision_cutoff,
+)
+
+SYMS = ["000001.SZ", "000002.SZ"]
+DATES = pd.bdate_range("2024-01-02", periods=6)
+
+
+def _daily():
+    rows = []
+    for si, s in enumerate(SYMS):
+        for di, d in enumerate(DATES):
+            base = 100.0 + si * 10 + di
+            rows.append((d, s, base + 0.1, base + 0.5, base - 0.5, base, 1e5 + di, base * 1e5))
+    return (
+        pd.DataFrame(rows, columns=["date", "symbol", "open", "high", "low", "close", "volume", "amount"])
+        .set_index(["date", "symbol"]).sort_index()
+    )
+
+
+def test_decision_same_day_fields_is_open_only():
+    """Policy-derived: the ONLY decision-view same-day factor field is open (R7)."""
+    assert decision_same_day_fields() == frozenset({"open"})
+
+
+def test_daily_decision_lag_shifts_prev_day_keeps_open():
+    """close/high/low/volume/amount at d become d-1; open stays same-day."""
+    panel = _daily()
+    lagged = daily_decision_lag(panel)
+    s = SYMS[0]
+    for di in range(1, len(DATES)):
+        d, prev = DATES[di], DATES[di - 1]
+        # open is same-day: unchanged.
+        assert lagged.loc[(d, s), "open"] == panel.loc[(d, s), "open"]
+        # prev-day fields carry the d-1 value.
+        for col in ("close", "high", "low", "volume", "amount"):
+            assert lagged.loc[(d, s), col] == panel.loc[(prev, s), col]
+    # the first date of each symbol has no d-1 -> shifted cols NaN, open present.
+    d0 = DATES[0]
+    assert np.isnan(lagged.loc[(d0, s), "close"])
+    assert lagged.loc[(d0, s), "open"] == panel.loc[(d0, s), "open"]
+
+
+def test_daily_decision_lag_does_not_mutate_input():
+    panel = _daily()
+    before = panel.copy()
+    daily_decision_lag(panel)
+    pd.testing.assert_frame_equal(panel, before)
+
+
+def test_daily_decision_lag_is_per_symbol():
+    """The shift never pulls one symbol's row into another's."""
+    panel = _daily()
+    lagged = daily_decision_lag(panel)
+    # symbol B's first date is NaN independently of symbol A.
+    assert np.isnan(lagged.loc[(DATES[0], SYMS[1]), "close"])
+
+
+def _bars(rows):
+    df = pd.DataFrame(
+        rows, columns=["time", "symbol", "open", "high", "low", "close", "volume", "amount"]
+    )
+    return normalize_intraday_bars(df, freq="1min")
+
+
+def test_minute_decision_cutoff_drops_post_1450_bars():
+    """Bars whose available_time exceeds d 14:50 are removed; earlier kept."""
+    day = "2024-01-02"
+    rows = []
+    # bars at 14:48, 14:49 (kept), 14:50, 14:55 (dropped: available_time = bar_end+1min)
+    for hhmm in ("14:48", "14:49", "14:50", "14:55"):
+        t = pd.Timestamp(f"{day} {hhmm}:00")
+        rows.append((t, SYMS[0], 100.0, 100.5, 99.5, 100.0, 1e4, 1e6))
+    bars = _bars(rows)
+    kept = minute_decision_cutoff(bars, decision_time="14:50:00")
+    kept_times = kept.index.get_level_values("time").strftime("%H:%M").tolist()
+    # available_time = bar_end + 1min: 14:48->14:49(<=14:50 keep), 14:49->14:50(keep),
+    # 14:50->14:51(drop), 14:55->14:56(drop).
+    assert kept_times == ["14:48", "14:49"]
+
+
+def test_minute_decision_cutoff_does_not_mutate_input():
+    day = "2024-01-02"
+    rows = [(pd.Timestamp(f"{day} 14:{m}:00"), SYMS[0], 100.0, 100.5, 99.5, 100.0, 1e4, 1e6)
+            for m in (40, 55)]
+    bars = _bars(rows)
+    before = bars.copy()
+    minute_decision_cutoff(bars)
+    pd.testing.assert_frame_equal(bars, before)
+
+
+def test_ex_date_mask_is_boolean_and_flags_af_changes():
+    """af(d) != af(d-1) per symbol; first row False; NaN-aware."""
+    idx = pd.MultiIndex.from_tuples(
+        [(DATES[0], "A"), (DATES[1], "A"), (DATES[2], "A"),
+         (DATES[0], "B"), (DATES[1], "B")],
+        names=["date", "symbol"],
+    )
+    af = pd.Series([1.0, 1.0, 2.0, 3.0, 3.0], index=idx)
+    mask = ex_date_mask(af)
+    assert mask.dtype == bool
+    assert mask.loc[(DATES[0], "A")] is np.False_ or not mask.loc[(DATES[0], "A")]  # first row
+    assert not mask.loc[(DATES[1], "A")]  # 1.0 -> 1.0 no change
+    assert mask.loc[(DATES[2], "A")]      # 1.0 -> 2.0 change
+    assert not mask.loc[(DATES[0], "B")]  # first row of B
+    assert not mask.loc[(DATES[1], "B")]  # 3.0 -> 3.0
+
+
+def test_ex_date_mask_rejects_non_multiindex():
+    with pytest.raises(ValueError):
+        ex_date_mask(pd.Series([1.0, 2.0]))