fix: require completed XNYS session for SOXL v4 P1

docs/soxl-core-only-p2-v4-free-split-close-research.md

@@ -15,6 +15,8 @@ remain separate.
 
 ## P1 admission rule
 
+Before any source request, P1 proves that the requested XNYS cutoff session has
+closed at `observed_at`; a calendar-valid but in-progress session is parked.
 For each of `SOXL`, `SOXX`, and `BOXX`:
 
 1. Twelve Data provides the candidate canonical split-adjusted close series.

src/us_equity_snapshot_pipelines/lifecycle/soxl_core_only_free_split_close_p1.py

@@ -21,10 +21,12 @@
 import sys
 import tempfile
 from collections.abc import Mapping
-from datetime import date
+from datetime import date, datetime
 from pathlib import Path
 from typing import Protocol
 
+import exchange_calendars as xcals
+import pandas as pd
 from quant_platform_kit.data.multisource_assurance import (
     DATA_ASSURANCE_STATUS_VERIFIED,
     SOURCE_OBSERVATION_READY,
@@ -120,6 +122,34 @@ def _date_cutoff(value: object) -> str:
     return value
 
 
+def validate_soxl_core_only_free_split_close_completed_session(
+    *, date_cutoff: object, observed_at: object
+) -> str:
+    """Require a completed XNYS session before a daily P1 can acquire data.
+
+    A calendar-valid date alone is insufficient: before that session closes,
+    both providers can legitimately expose an in-progress daily bar.  P1 must
+    park before any source request in that case, rather than comparing or
+    materializing provisional prices.
+    """
+    cutoff = _date_cutoff(date_cutoff)
+    if not isinstance(observed_at, str):
+        raise SoxlCoreOnlyFreeSplitCloseP1Error("invalid SOXL free-source observed time")
+    try:
+        observed = datetime.fromisoformat(observed_at)
+    except ValueError as exc:
+        raise SoxlCoreOnlyFreeSplitCloseP1Error("invalid SOXL free-source observed time") from exc
+    if observed.tzinfo is None or observed.utcoffset() is None:
+        raise SoxlCoreOnlyFreeSplitCloseP1Error("invalid SOXL free-source observed time")
+    try:
+        closing = xcals.get_calendar("XNYS").session_close(pd.Timestamp(cutoff))
+    except (AttributeError, TypeError, ValueError) as exc:
+        raise SoxlCoreOnlyFreeSplitCloseP1Error("XNYS calendar is unavailable") from exc
+    if pd.Timestamp(observed) < closing:
+        raise SoxlCoreOnlyFreeSplitCloseP1UnavailableError("SOXL free-source session is not complete")
+    return cutoff
+
+
 def _policy(*, symbol: str, date_cutoff: str) -> MultiSourceDailyBarPolicy:
     return MultiSourceDailyBarPolicy(
         scope_id=f"{_POLICY_SCOPE_PREFIX}:{symbol.lower()}",
@@ -527,6 +557,10 @@ def publish_soxl_core_only_free_split_close_p1_inputs(
 ) -> dict[str, object]:
     """Publish a verified v4 P1 root, or fail closed without a root."""
     destination = _require_new_private_output_root(output_root)
+    validate_soxl_core_only_free_split_close_completed_session(
+        date_cutoff=date_cutoff,
+        observed_at=observed_at,
+    )
     binding = build_soxl_core_only_free_split_close_p1_binding(date_cutoff=date_cutoff)
     expected = expected_soxl_core_only_sessions(date_cutoff)
     canonical_series: dict[str, list[dict[str, object]]] = {}
@@ -674,6 +708,7 @@ def verify_soxl_core_only_free_split_close_p1_input_root(output_root: str | Path
     "canonical_soxl_core_only_free_split_close_series_bytes",
     "publish_soxl_core_only_free_split_close_p1_inputs",
     "soxl_core_only_free_split_close_p1_binding_sha256",
+    "validate_soxl_core_only_free_split_close_completed_session",
     "validate_soxl_core_only_free_split_close_input_manifest",
     "validate_soxl_core_only_free_split_close_assurance_member",
     "validate_soxl_core_only_free_split_close_p1_binding",

tests/test_soxl_core_only_free_split_close_p1.py

@@ -118,6 +118,27 @@ def test_binding_freezes_v4_two_source_split_adjusted_close_identity() -> None:
     assert p1.validate_soxl_core_only_free_split_close_p1_binding(binding) == binding
 
 
+def test_p1_rejects_an_in_progress_xnys_session_before_any_source_observation(tmp_path: Path) -> None:
+    observer = _AssuredObserver()
+    output_root = tmp_path / "in-progress-p1"
+
+    with pytest.raises(p1.SoxlCoreOnlyFreeSplitCloseP1UnavailableError, match="not complete"):
+        p1.publish_soxl_core_only_free_split_close_p1_inputs(
+            observer,
+            output_root=output_root,
+            observed_at="2026-08-18T19:59:59Z",
+            producer=_producer(),
+            date_cutoff=_CUTOFF,
+        )
+
+    assert observer.requests == []
+    assert not output_root.exists()
+    assert p1.validate_soxl_core_only_free_split_close_completed_session(
+        date_cutoff=_CUTOFF,
+        observed_at="2026-08-18T20:00:00Z",
+    ) == _CUTOFF
+
+
 def test_publisher_requires_two_source_close_agreement_then_writes_a_private_root(tmp_path: Path) -> None:
     observer = _AssuredObserver()
     output_root = tmp_path / "free-split-close-p1"