Bound INT03 replay calendar to the run window

src/backtest_engine/orchestrator.py

@@ -52,6 +52,7 @@
     FailureKind,
     ResourceMonitor,
 )
+from .calendar import CalendarCoverageError
 from .data_availability import (
     AvailabilityAssessment,
     AvailabilityStatus,
@@ -651,10 +652,12 @@ def __init__(
         self._assessor = assessor or DataAvailabilityAssessor()
         self._publication_lag = publication_lag
 
-    def _schedule(self, policy: ExecutionPolicy) -> OfficialSessionSchedule:
-        zone = ZoneInfo(policy.timezone)
-        first = policy.period_start.astimezone(zone).date()
-        last = (policy.period_end - timedelta(microseconds=1)).astimezone(zone).date()
+    def _schedule(self, job: BacktestJob) -> OfficialSessionSchedule:
+        """Resolve only the immutable run window, not the policy's validity envelope."""
+        zone = ZoneInfo(job.execution_policy.timezone)
+        first = min(item.warmup_from for item in job.requirements).astimezone(zone).date()
+        through = max(item.evaluation_through for item in job.requirements)
+        last = (through - timedelta(microseconds=1)).astimezone(zone).date()
         return self._calendar.session_schedule(first, last)
 
     # -- entry point ------------------------------------------------------
@@ -667,7 +670,18 @@ def run(
         lease: AttemptLease,
         monitor: ResourceMonitor,
     ) -> ReplayOutcome:
-        schedule = self._schedule(job.execution_policy)
+        try:
+            schedule = self._schedule(job)
+        except CalendarCoverageError as exc:
+            return self._abort(
+                job,
+                coordinator,
+                lease,
+                "REQUIRED_INPUT_UNAVAILABLE",
+                retryable=False,
+                status=ReplayStatus.UNAVAILABLE,
+                detail=f"{type(exc).__name__}: {exc}",
+            )
         try:
             events = bar_events_from_batches(
                 self._reader.iter_batches(job.manifest, job.execution_policy),

tests/test_orchestrator.py

@@ -18,6 +18,7 @@
 
 import hashlib
 from collections.abc import Mapping
+from dataclasses import replace
 from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from fractions import Fraction
@@ -541,6 +542,53 @@ def materialized_read_is_forbidden(*_args: object, **_kwargs: object) -> object:
     assert outcome.status is ReplayStatus.COMPLETED
 
 
+def test_schedule_uses_the_run_requirement_window_not_the_policy_applicability_window(
+    tmp_path: Path,
+) -> None:
+    """INT03's decade policy must not make a January 2024 run request 2016 sessions."""
+    path = tmp_path / "bars.parquet"
+    write_bars(path)
+    policy = replace(
+        D17_EXECUTION_POLICY_FIXTURE,
+        period_start=datetime(2016, 7, 1, 4, tzinfo=timezone.utc),
+        period_end=datetime(2026, 7, 1, 4, tzinfo=timezone.utc),
+    )
+    job = replace(_job(manifest_for(path)), execution_policy=policy)
+    harness = Harness(tmp_path, StubRuntime(), RecordingEngine(), RecordingPublisher())
+
+    schedule = harness.orchestrator()._schedule(job)
+
+    assert schedule.sessions[0].trading_date_et == date(2024, 1, 2)
+    assert schedule.sessions[-1].trading_date_et == date(2024, 1, 2)
+
+
+def test_calendar_coverage_mismatch_is_one_permanent_input_failure(tmp_path: Path) -> None:
+    path = tmp_path / "bars.parquet"
+    write_bars(path)
+    outside = DataRequirement(
+        requirement_id="outside-calendar",
+        instrument_id=AAPL,
+        data_kind=DATA_KIND,
+        resolution=RESOLUTION,
+        warmup_from=datetime(2023, 12, 29, 14, 30, tzinfo=timezone.utc),
+        evaluation_from=datetime(2023, 12, 29, 14, 30, tzinfo=timezone.utc),
+        evaluation_through=datetime(2023, 12, 29, 15, 0, tzinfo=timezone.utc),
+    )
+    job = _job(manifest_for(path), requirements=(outside,))
+
+    outcome, harness, coordinator = _run(tmp_path, job=job)
+
+    assert outcome.status is ReplayStatus.UNAVAILABLE
+    assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE"
+    assert outcome.retryable is False
+    assert outcome.failure_detail is not None
+    assert outcome.failure_detail.startswith("CalendarCoverageError:")
+    assert harness.publisher.requests == []
+    assert coordinator.state is RunState.FAILED
+    assert len(coordinator.attempts) == 1
+    assert coordinator.attempts[0].state is AttemptState.PERMANENT_FAILED
+
+
 # --------------------------------------------------------------------------
 # The replay loop is genuinely driven by the event clock.
 # --------------------------------------------------------------------------