Fail incompatible backtest execution windows before replay

src/backtest_engine/wiring.py

@@ -82,6 +82,7 @@
     ReplaySkipReason,
     derive_data_requirements,
 )
+from .calendar import CalendarCoverageError
 from .contracts import build_backtest_result_event
 from .data_availability import AvailabilityAssessment
 from .detail_object_manifest import (
@@ -1377,6 +1378,43 @@ def evaluation_window(manifest: Mapping[str, Any], plan: BasicCompiledPlan) -> t
     return evaluation_from, coverage_end
 
 
+def require_compatible_execution_window(
+    policy: ExecutionPolicy,
+    manifest: Mapping[str, Any],
+    calendar: SessionCalendar,
+) -> None:
+    """Reject immutable execution inputs that cannot describe one run.
+
+    The worker must not silently shrink a policy or manifest period to make an
+    incompatible request executable. This check runs during binding, before a
+    ``RUNNING`` event is published, because redelivery cannot change any input.
+    """
+    problems: list[str] = []
+    manifest_start = _parse_instant(manifest.get("period_start"))
+    manifest_end = _parse_instant(manifest.get("period_end"))
+    if manifest_start != policy.period_start or manifest_end != policy.period_end:
+        problems.append(
+            "dataset manifest period "
+            f"{manifest_start.isoformat()}..{manifest_end.isoformat()} does not match "
+            f"execution policy {policy.version} period "
+            f"{policy.period_start.isoformat()}..{policy.period_end.isoformat()}"
+        )
+
+    zone = ZoneInfo(policy.timezone)
+    first = policy.period_start.astimezone(zone).date()
+    last = (policy.period_end - timedelta(microseconds=1)).astimezone(zone).date()
+    try:
+        calendar.session_schedule(first, last)
+    except CalendarCoverageError as exc:
+        problems.append(f"session calendar does not cover the execution policy period: {exc}")
+
+    if problems:
+        raise JobNotSatisfiable(
+            "execution inputs are mutually incompatible: " + "; ".join(problems),
+            reason_code="REQUIRED_INPUT_UNAVAILABLE",
+        )
+
+
 def _parse_instant(value: object) -> datetime:
     if not isinstance(value, str):  # pragma: no cover - the schema requires a string
         raise JobNotSatisfiable(
@@ -1599,6 +1637,7 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding:
                 reason_code="REQUIRED_INPUT_UNAVAILABLE",
             )
         manifest = primary[0][1]
+        require_compatible_execution_window(policy, manifest, self._calendar)
         try:
             plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum)
         except BasicPlanCompatibilityError as exc:

tests/test_feature_outputs.py

@@ -4,6 +4,7 @@
 import hashlib
 import json
 import uuid
+from dataclasses import replace
 from datetime import UTC, datetime, timedelta
 from decimal import Decimal
 from typing import Any
@@ -770,6 +771,57 @@ def publish(self, event: Any, *, delivery_attempt: int) -> None:
     assert sink.events[0]["retryable"] is False
 
 
+def test_incompatible_development_windows_are_one_terminal_binding_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The exact INT03 policy/manifest/calendar tuple cannot improve on redelivery."""
+    policy = replace(
+        E2E_EXECUTION_POLICY,
+        version="development-official-backtest-2026-q3-v1",
+        release_quarter="2026-Q3",
+        period_start=datetime(2016, 7, 1, 4, tzinfo=UTC),
+        period_end=datetime(2026, 7, 1, 4, tzinfo=UTC),
+    )
+    market_bytes = market_bars_parquet()
+    manifest = dataset_manifest(
+        hashlib.sha256(market_bytes).hexdigest(),
+        row_count=len(CLOSES),
+        coverage_end=EVALUATION_THROUGH,
+    )
+    manifest["period_start"] = "2024-01-01T05:00:00Z"
+    manifest["period_end"] = "2024-02-01T05:00:00Z"
+    handler = _handler(Source({}), Reader(b""))
+    handler._policies = ExecutionPolicyCatalog([policy])
+    handler._manifests = StaticDatasetManifestSource({DATASET_MANIFEST_ID: manifest})
+    envelope = replace(_envelope(pins=[]), execution_policy_version=policy.version)
+
+    class Sink:
+        def __init__(self) -> None:
+            self.events: list[dict[str, Any]] = []
+
+        def publish(self, event: Any, *, delivery_attempt: int) -> None:
+            self.events.append(dict(event) | {"deliveryAttempt": delivery_attempt})
+
+    sink = Sink()
+    handler._sink = sink
+    monkeypatch.setattr(JobEnvelope, "parse", classmethod(lambda _cls, _job: envelope))
+
+    with pytest.raises(JobNotSatisfiable) as failure:
+        handler.bind(envelope, _context())
+    assert "dataset manifest period" in str(failure.value)
+    assert "2016-07-01 is outside the pinned XNYS coverage" in str(failure.value)
+
+    outcome = handler({}, _context())
+
+    assert outcome.result is JobResult.PERMANENT_FAILURE
+    assert outcome.reason_code == "REQUIRED_INPUT_UNAVAILABLE"
+    assert len(sink.events) == 1
+    assert sink.events[0]["status"] == "FAILED"
+    assert sink.events[0]["failureCode"] == "REQUIRED_INPUT_UNAVAILABLE"
+    assert sink.events[0]["retryable"] is False
+    assert sink.events[0]["deliveryAttempt"] == 1
+
+
 def test_job_binding_attaches_only_fully_verified_feature_series() -> None:
     body = _parquet()
     record = _record(body)