fix: terminate incompatible compiled plans

src/backtest_engine/wiring.py

@@ -75,6 +75,7 @@
 from .basic_runtime import (
     BasicCompiledPlan,
     BasicDecisionStatus,
+    BasicPlanCompatibilityError,
     BasicPlanReplay,
     BasicPlanRuntime,
     PlanEvaluation,
@@ -1598,7 +1599,16 @@ def bind(self, envelope: JobEnvelope, context: JobContext) -> JobBinding:
                 reason_code="REQUIRED_INPUT_UNAVAILABLE",
             )
         manifest = primary[0][1]
-        plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum)
+        try:
+            plan = self._runtime.load(plan_document, compiled_plan_checksum=plan_checksum)
+        except BasicPlanCompatibilityError as exc:
+            # The plan is immutable and addressed by its checksum. A schema,
+            # integrity, or catalog incompatibility therefore cannot become
+            # valid when SQS redelivers the same message. Translate it at the
+            # binding boundary so the existing terminal-result path records
+            # the precise plan-load failure once instead of retrying a
+            # deterministic producer/consumer mismatch to exhaustion.
+            raise JobNotSatisfiable(str(exc), reason_code=exc.failure.value) from exc
         if plan.reference_series[1] == "$DATASET":
             dataset_resolution = str(manifest.get("resolution", ""))
             if dataset_resolution not in {"30m", "1h", "4h", "1d"}:

tests/test_feature_outputs.py

@@ -39,7 +39,7 @@
     JobNotSatisfiable,
     OrchestratorJobHandler,
 )
-from backtest_engine.worker import JobContext
+from backtest_engine.worker import JobContext, JobResult
 from d_reproducibility_testkit import (
     BAR,
     CLOSES,
@@ -737,6 +737,39 @@ def _context() -> JobContext:
     )
 
 
+def test_contract_invalid_compiled_plan_is_one_terminal_binding_failure(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The exact Development failure cannot be repaired by SQS redelivery."""
+    invalid_plan = _plan_document()
+    invalid_plan["executionSnapshot"]["initialCashAmount"] = "100000"
+    envelope = _envelope(pins=[])
+    handler = _handler(Source({}), Reader(b""))
+    handler._plans = StaticCompiledPlanSource(
+        {invalid_plan["planChecksum"]: invalid_plan}
+    )
+
+    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))
+
+    outcome = handler({}, _context())
+
+    assert outcome.result is JobResult.PERMANENT_FAILURE
+    assert outcome.reason_code == "PLAN_CONTRACT_INVALID"
+    assert len(sink.events) == 1
+    assert sink.events[0]["status"] == "FAILED"
+    assert sink.events[0]["failureCode"] == "PLAN_CONTRACT_INVALID"
+    assert sink.events[0]["retryable"] is False
+
+
 def test_job_binding_attaches_only_fully_verified_feature_series() -> None:
     body = _parquet()
     record = _record(body)