diff --git a/src/backtest_engine/wiring.py b/src/backtest_engine/wiring.py index 7353956..52994eb 100644 --- a/src/backtest_engine/wiring.py +++ b/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"}: diff --git a/tests/test_feature_outputs.py b/tests/test_feature_outputs.py index 3864c58..97afddf 100644 --- a/tests/test_feature_outputs.py +++ b/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)